diff --git a/src/fastcache_api/config.py b/src/fastcache_api/config.py index 174adc8..4523e42 100644 --- a/src/fastcache_api/config.py +++ b/src/fastcache_api/config.py @@ -45,6 +45,9 @@ class Settings(BaseSettings): OIDC_JWKS_URI: str | None = None OIDC_AUDIENCES: Annotated[list[str], BeforeValidator(parse_comma_list)] = [] + CACHE_PORT_START: int = 30000 + CACHE_PORT_END: int = 30100 + # Cache process management # Path to the fastcache executable (overridable; defaults to PATH lookup). FASTCACHE_BINARY: Path = Path("lclstream-fastcache") diff --git a/src/fastcache_api/models.py b/src/fastcache_api/models.py index 5206533..ee77180 100644 --- a/src/fastcache_api/models.py +++ b/src/fastcache_api/models.py @@ -62,10 +62,9 @@ def from_cache_config(cls, config: CacheConfig) -> FastcacheConfig: class CacheRequest(BaseModel): transfer_id: str - hostname: str - # later relax to none - pull_port: int - push_port: int + # Human who initiated the transfer upstream (bearer token is a shared + # service identity, so attribution must travel in the request body). + requested_by: str class CachePublic(BaseModel): diff --git a/src/fastcache_api/process.py b/src/fastcache_api/process.py index 63778ba..82c1624 100644 --- a/src/fastcache_api/process.py +++ b/src/fastcache_api/process.py @@ -2,7 +2,7 @@ import logging import socket import subprocess -from functools import lru_cache +from collections.abc import Iterable from uuid import UUID import psutil @@ -13,11 +13,26 @@ logger = logging.getLogger(__name__) -@lru_cache(maxsize=1) -def local_hostnames() -> set[str]: - names = {socket.gethostname(), socket.getfqdn()} - names |= {name.split(".", 1)[0] for name in names} - return {name.lower() for name in names if name} +def canonical_hostname() -> str: + """This host's preferred public name for cache ZMQ URIs.""" + return (socket.getfqdn() or socket.gethostname()).lower() + + +def ports_in_use(configs: Iterable[CacheConfig]) -> set[int]: + used: set[int] = set() + for config in configs: + for uri in (config.pull_uri, config.push_uri): + if uri.port is not None: + used.add(uri.port) + return used + + +def allocate_port_pair(in_use: set[int], start: int, end: int) -> tuple[int, int]: + for pull in range(start, end + 1, 2): + push = pull + 1 + if push <= end and pull not in in_use and push not in in_use: + return pull, push + raise RuntimeError(f"no free cache port pair in range [{start}, {end}]") def start_cache(cache_id: UUID, config: CacheConfig) -> CacheProcess: diff --git a/src/fastcache_api/routes/cache.py b/src/fastcache_api/routes/cache.py index 3663463..f9f1843 100644 --- a/src/fastcache_api/routes/cache.py +++ b/src/fastcache_api/routes/cache.py @@ -16,7 +16,13 @@ CachesPublic, CacheStatus, ) -from ..process import local_hostnames, start_cache, stop_cache +from ..process import ( + allocate_port_pair, + canonical_hostname, + ports_in_use, + start_cache, + stop_cache, +) from ..tables import Cache router = APIRouter( @@ -54,19 +60,33 @@ async def create_cache( user: Annotated[TokenPayload, Depends(require_user)], session: Annotated[AsyncSession, Depends(get_session)], ): - if req.hostname.lower() not in local_hostnames(): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=( - f"Cache must run on this api server's host; request specified " - f"'{req.hostname}'" - ), + # The cache always runs as a local subprocess of this api server, so the + # ZMQ URIs are published under this host's canonical (FQDN) name. + hostname = canonical_hostname() + + # Derive the ports already bound by live caches and allocate the next free pair. + active = await session.execute( + select(Cache).where( + Cache.state.in_([s.value for s in CacheState if not s.is_final()]) ) + ) + in_use = ports_in_use( + CacheConfig.model_validate(cache.config) for cache in active.scalars().all() + ) + try: + pull_port, push_port = allocate_port_pair( + in_use, settings.CACHE_PORT_START, settings.CACHE_PORT_END + ) + except RuntimeError as exc: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=str(exc), + ) from exc config = CacheConfig( - hostname=req.hostname, - pull_uri=f"tcp://{req.hostname}:{req.pull_port}", - push_uri=f"tcp://{req.hostname}:{req.push_port}", + hostname=hostname, + pull_uri=f"tcp://{hostname}:{pull_port}", + push_uri=f"tcp://{hostname}:{push_port}", ) cache_id = uuid4() @@ -83,7 +103,7 @@ async def create_cache( transfer_id=req.transfer_id, pid=proc.pid, create_time=proc.create_time, - user=user.email, + user=req.requested_by, status=CacheStatus.active, log_path=str(proc.log_path), config=config.model_dump(mode="json"),