From f812a044708eec029d9ee651bcde687626185ea9 Mon Sep 17 00:00:00 2001 From: "R. Garcia-Dias" Date: Mon, 20 Jul 2026 15:55:04 +0100 Subject: [PATCH] feat(fl): tolerate an unreachable fl-server behind PER_JOB_FL_SERVER Phase 0 of the per-job fl-server scale-to-zero effort (FLIP#735). Today fl-api-net-1 raises fatally at boot and check_server_status errors whenever fl-server is unreachable, and keep_fl_api_session_alive pings every net regardless of whether a job is running. Once flip-api starts scaling fl-server to zero between jobs, an unreachable server becomes the normal idle state instead of an outage, so none of that is safe to ship until the services can tell the two apart. Adds a PER_JOB_FL_SERVER flag (default false, mirrored into both flip-api and fl-api-net-1's env from one Terraform variable) so this is a no-op until explicitly enabled: with the flag off, an unreachable fl-server still crash-loops fl-api-net-1 exactly as today, preserving that as the alerting signal for a real outage. Signed-off-by: R. Garcia-Dias --- .env.development.example | 7 ++ deploy/compose.development.nvflare.yml | 2 + deploy/compose.development.yml | 4 + deploy/compose.production.nvflare.yml | 3 + deploy/compose.production.yml | 4 + deploy/providers/AWS/Makefile | 6 ++ deploy/providers/AWS/locals.tf | 5 ++ deploy/providers/AWS/variables.tf | 12 +++ .../nvflare/fl-api-base/fl_api/config.py | 6 ++ .../fl-api-base/fl_api/routers/system.py | 15 +++- .../fl_api/startup/session_manager.py | 24 +++++- .../fl-api-base/fl_api/utils/flip_session.py | 29 ++++++- .../fl-api-base/tests/routers/test_system.py | 30 +++++++- .../tests/startup/test_session_manager.py | 64 +++++++++++++++ .../tests/utils/test_flip_session.py | 77 ++++++++++++++++++- flip-api/src/flip_api/config.py | 18 +++++ .../services/fl_scheduler_service.py | 25 ++++++ .../fl_services/services/fl_service.py | 8 ++ .../services/test_fl_scheduler_service.py | 24 +++++- .../fl_services/services/test_fl_service.py | 41 ++++++++++ 20 files changed, 392 insertions(+), 12 deletions(-) diff --git a/.env.development.example b/.env.development.example index 5fef162f0..5942b9f04 100644 --- a/.env.development.example +++ b/.env.development.example @@ -220,6 +220,13 @@ FL_BACKEND=nvflare JOB_RESOURCE_SPEC_NUM_GPUS=0 JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB=0 +# Scale fl-server-net-1 to zero between training jobs instead of running it always-on +# (FLIP#735, NVFLARE only). Default false everywhere. Threaded into both flip-api and +# fl-api-net-1 (compose.development.nvflare.yml) — both must agree on whether an unreachable +# fl-server is an unplanned outage (false) or the normal idle-by-design state (true). Uncomment +# to exercise the Phase 0 lazy-session/offline-status behavior locally. +# PER_JOB_FL_SERVER=false + # ── MFA Enforcement ─────────────────────────────────────────────────────── # Controls whether the API requires TOTP enrollment for all authenticated # routes. Default is "true" (enforced) in production. In development, the diff --git a/deploy/compose.development.nvflare.yml b/deploy/compose.development.nvflare.yml index ee641f863..adefa2c11 100644 --- a/deploy/compose.development.nvflare.yml +++ b/deploy/compose.development.nvflare.yml @@ -29,6 +29,8 @@ services: # mem <= the client's advertised mem_per_gpu_in_GiB) so jobs train on the GPU. - JOB_RESOURCE_SPEC_NUM_GPUS=${JOB_RESOURCE_SPEC_NUM_GPUS:-0} - JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB=${JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB:-0} + # FLIP#735: mirrors flip-api's PER_JOB_FL_SERVER above — same source var. + - PER_JOB_FL_SERVER=${PER_JOB_FL_SERVER:-false} ports: - "${FL_API_PORT}:8000" - "5679:5679" diff --git a/deploy/compose.development.yml b/deploy/compose.development.yml index 63a1c8694..b3eebd06c 100644 --- a/deploy/compose.development.yml +++ b/deploy/compose.development.yml @@ -124,6 +124,10 @@ services: # Settings.ENFORCE_MFA defaults to True elsewhere. Flip to `true` # here (and `make restart`) to test the MFA flow end-to-end locally. - ENFORCE_MFA=false + # FLIP#735: scale fl-server-net-1 to zero between jobs. Default false (unchanged + # always-on behavior); set PER_JOB_FL_SERVER=true in .env.development (and mirror on + # fl-api-net-1 below) to exercise the Phase 0 lazy-session/offline-status behavior locally. + - PER_JOB_FL_SERVER=${PER_JOB_FL_SERVER:-false} # Settings.LOG_LEVEL defaults to INFO so unknown environments emit # at INFO. Local dev defaults to DEBUG when LOG_LEVEL is unset # (override via LOG_LEVEL in .env.development) so debug-level diff --git a/deploy/compose.production.nvflare.yml b/deploy/compose.production.nvflare.yml index 6830cf812..a5e1d7b40 100644 --- a/deploy/compose.production.nvflare.yml +++ b/deploy/compose.production.nvflare.yml @@ -28,6 +28,9 @@ services: # mem <= the client's advertised mem_per_gpu_in_GiB) so jobs train on the GPU. - JOB_RESOURCE_SPEC_NUM_GPUS=${JOB_RESOURCE_SPEC_NUM_GPUS:-0} - JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB=${JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB:-0} + # FLIP#735: mirrors flip-api's PER_JOB_FL_SERVER (compose.production.yml) — same source + # var, both services must agree on whether fl-server going unreachable is an outage or idle. + - PER_JOB_FL_SERVER=${PER_JOB_FL_SERVER:-false} ports: - "${FL_API_PORT}:8000" security_opt: diff --git a/deploy/compose.production.yml b/deploy/compose.production.yml index b13c1df4c..0943e5b61 100644 --- a/deploy/compose.production.yml +++ b/deploy/compose.production.yml @@ -28,6 +28,10 @@ services: environment: - ENV=production - ENFORCE_MFA=${ENFORCE_MFA:-true} + # FLIP#735: scale fl-server-net-1 to zero between jobs. Default false everywhere; + # mirrored into fl-api-net-1's own PER_JOB_FL_SERVER (compose.production.nvflare.yml) — + # both services must agree on whether an unreachable fl-server is an outage or idle-by-design. + - PER_JOB_FL_SERVER=${PER_JOB_FL_SERVER:-false} - LOG_LEVEL=${LOG_LEVEL:-INFO} # - AWS_PROFILE=${AWS_PROFILE} # in production, AWS credentials are retrieved from the instance role - AWS_REGION=${AWS_REGION} diff --git a/deploy/providers/AWS/Makefile b/deploy/providers/AWS/Makefile index 384c444cd..bd6aad600 100644 --- a/deploy/providers/AWS/Makefile +++ b/deploy/providers/AWS/Makefile @@ -118,6 +118,12 @@ export TF_VAR_MIN_CLIENTS=${MIN_CLIENTS} ifneq ($(JOB_RESOURCE_SPEC_NUM_GPUS),) export TF_VAR_JOB_RESOURCE_SPEC_NUM_GPUS=${JOB_RESOURCE_SPEC_NUM_GPUS} endif +# FLIP#735: scale fl-server-net-1 to zero between jobs. Default false in variables.tf; set +# PER_JOB_FL_SERVER=true in the env file (stag first) once Phase 2 has landed. Conditional export +# for the same reason as the GPU resource-spec vars above (empty string ≠ bool for Terraform). +ifneq ($(PER_JOB_FL_SERVER),) +export TF_VAR_PER_JOB_FL_SERVER=${PER_JOB_FL_SERVER} +endif ifneq ($(JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB),) export TF_VAR_JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB=${JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB} endif diff --git a/deploy/providers/AWS/locals.tf b/deploy/providers/AWS/locals.tf index e4cb7b70f..dc15eacb5 100644 --- a/deploy/providers/AWS/locals.tf +++ b/deploy/providers/AWS/locals.tf @@ -126,6 +126,7 @@ locals { # complete within 30 minutes or they time out at the edge. MAX_MODEL_FILE_BYTES = "5000000000" FL_KIT_SLOT_NAMES = var.FL_KIT_SLOT_NAMES + PER_JOB_FL_SERVER = tostring(var.PER_JOB_FL_SERVER) }) fl_server = { LOCAL_DEV = "false" @@ -160,6 +161,10 @@ locals { # CPU-only. Default 0; set via TF_VAR_JOB_RESOURCE_SPEC_* for GPU jobs. JOB_RESOURCE_SPEC_NUM_GPUS = tostring(var.JOB_RESOURCE_SPEC_NUM_GPUS) JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB = tostring(var.JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB) + # Mirrors flip_api's PER_JOB_FL_SERVER above — same Terraform variable, single source of + # truth (FLIP#735). Both services must agree on whether an unreachable fl-server is an + # unplanned outage or the normal idle state between jobs. + PER_JOB_FL_SERVER = tostring(var.PER_JOB_FL_SERVER) } } } diff --git a/deploy/providers/AWS/variables.tf b/deploy/providers/AWS/variables.tf index 0e5c3025d..ff25ef325 100644 --- a/deploy/providers/AWS/variables.tf +++ b/deploy/providers/AWS/variables.tf @@ -196,6 +196,18 @@ variable "JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB" { default = 0 } +# FLIP#735: scale fl-server-net-1 to zero between training jobs instead of running it 24/7. +# Threaded as an identical env var into both flip-api and fl-api-net-1 (see ecs_task_env in +# locals.tf) — both services must agree on whether an unreachable fl-server is an unplanned +# outage (false, default: fl-api-net-1 crash-loops as today) or the normal idle-by-design state +# between jobs (true: flip-api drives ecs:UpdateService, fl-api-net-1 tolerates the server being +# down). Default false everywhere; set true in `.env.stag` only once Phase 2 has landed. +variable "PER_JOB_FL_SERVER" { + description = "Scale fl-server-net-1 to zero between jobs instead of running it always-on" + type = bool + default = false +} + # Gates the whole FL-on-ECS stack, not just the file system: both FL task # definitions (fl-api-net-1, fl-server-net-1 in ecs_tasks.tf) carry # `count = var.enable_efs ? 1 : 0`, so their EFS volumes (including the shared diff --git a/fl-services/nvflare/fl-api-base/fl_api/config.py b/fl-services/nvflare/fl-api-base/fl_api/config.py index 586759336..7f900743a 100644 --- a/fl-services/nvflare/fl-api-base/fl_api/config.py +++ b/fl-services/nvflare/fl-api-base/fl_api/config.py @@ -25,6 +25,12 @@ class Settings(BaseSettings): TIMEOUT_SESSION_CONNECT: float = 20.0 + # Mirrors flip-api's PER_JOB_FL_SERVER (FLIP#735) — threaded in as an env var from the same + # shared Terraform variable so both services agree on whether fl-server going unreachable is + # an unplanned outage (flag off: fail fast at boot, as today) or the normal idle-by-design + # state between jobs (flag on: tolerate it and connect lazily on first use). + PER_JOB_FL_SERVER: bool = False + # GPU resources that the submitted NVFLARE jobs need in order to schedule correctly. # TODO Currently this is set globally for all jobs, but we should allow per-job overrides in the future. # See https://github.com/londonaicentre/flip/issues/41 diff --git a/fl-services/nvflare/fl-api-base/fl_api/routers/system.py b/fl-services/nvflare/fl-api-base/fl_api/routers/system.py index 46f3c548e..d11748977 100644 --- a/fl-services/nvflare/fl-api-base/fl_api/routers/system.py +++ b/fl-services/nvflare/fl-api-base/fl_api/routers/system.py @@ -13,7 +13,10 @@ # System and service status functions from fastapi import APIRouter, Depends, Query, status +from nvflare.apis.fl_exception import FLCommunicationError +from nvflare.fuel.flare_api.api_spec import NoConnection +from fl_api.config import get_settings from fl_api.core.dependencies import get_session from fl_api.utils.flip_session import FLIP_Session from fl_api.utils.schemas import ClientInfoModel, ServerInfoModel, SystemInfoModel, TargetType @@ -26,13 +29,23 @@ def check_server_status(session: FLIP_Session = Depends(get_session)) -> ServerI """ Checks the status of the server. + With PER_JOB_FL_SERVER on, an unreachable fl-server is the normal idle-between-jobs state + (FLIP#735), not an error — this returns status="STOPPED" so flip-api can poll this endpoint as + a readiness probe. With the flag off, an unreachable server is still an unplanned outage and + this raises, unchanged. + Args: session (FLIP_Session): the FLIP session instance. Returns: ServerInfoModel: status information about the server. """ - return session.check_server_status() + try: + return session.check_server_status() + except (NoConnection, FLCommunicationError): + if not get_settings().PER_JOB_FL_SERVER: + raise + return ServerInfoModel(status="STOPPED", start_time=0.0) @router.get("/check_client_status", response_model=list[ClientInfoModel]) diff --git a/fl-services/nvflare/fl-api-base/fl_api/startup/session_manager.py b/fl-services/nvflare/fl-api-base/fl_api/startup/session_manager.py index 1aa0dfd17..e26d8bc36 100644 --- a/fl-services/nvflare/fl-api-base/fl_api/startup/session_manager.py +++ b/fl-services/nvflare/fl-api-base/fl_api/startup/session_manager.py @@ -11,6 +11,9 @@ # +from nvflare.apis.fl_exception import FLCommunicationError +from nvflare.fuel.flare_api.api_spec import NoConnection + from fl_api.config import get_settings from fl_api.utils.flip_session import FLIP_Session from fl_api.utils.logger import logger @@ -20,6 +23,13 @@ def create_fl_session() -> FLIP_Session: """ Initialize NVIDIA FLARE admin workspace and return a session object for interacting with it. + With PER_JOB_FL_SERVER off, an unreachable fl-server at boot is fatal (unchanged) — this is the + normal alerting signal for an unplanned outage. With it on, fl-server is expected to be down + most of the time (scaled to zero between jobs, FLIP#735), so a connection failure here is + logged and tolerated: the session connects lazily on first use (see FLIP_Session._do_command). + A non-connectivity failure (bad config, auth) still raises either way — only "can't reach the + server" is tolerated. + Returns: FLIP_Session: An initialized FLIP_Session object. """ @@ -43,10 +53,18 @@ def create_fl_session() -> FLIP_Session: debug=debug, ) - # Try connecting the session here, so that we can catch any connection issues at startup - session.try_connect(get_settings().TIMEOUT_SESSION_CONNECT) - logger.info(f"Upload directory set to: {session.upload_dir}") logger.info(f"Download directory set to: {session.download_dir}") + # Try connecting the session here, so that we can catch any connection issues at startup + try: + session.try_connect(get_settings().TIMEOUT_SESSION_CONNECT) + except (NoConnection, FLCommunicationError): + if not get_settings().PER_JOB_FL_SERVER: + raise + logger.warning( + "fl-server unreachable at boot; PER_JOB_FL_SERVER is on, treating this as the normal " + "idle-between-jobs state. Will connect lazily on first use." + ) + return session diff --git a/fl-services/nvflare/fl-api-base/fl_api/utils/flip_session.py b/fl-services/nvflare/fl-api-base/fl_api/utils/flip_session.py index 5858356a3..afda713c2 100644 --- a/fl-services/nvflare/fl-api-base/fl_api/utils/flip_session.py +++ b/fl-services/nvflare/fl-api-base/fl_api/utils/flip_session.py @@ -31,9 +31,20 @@ def __init__( self._debug = debug super(FLIP_Session, self).__init__(username, startup_path, secure_mode, debug) self._error_buffer = None + # Tracked ourselves rather than inspecting nvflare's private `api.closed` — a freshly + # constructed AdminAPI is neither "closed" nor actually connected, so `api.closed` can't + # distinguish "never connected" (PER_JOB_FL_SERVER's lazy-boot case) from a live session. + self._connected = False + + def try_connect(self, timeout: float) -> None: + """Connect the underlying admin API, tracking success so `_do_command` knows whether a + lazy first-use connect is needed (see PER_JOB_FL_SERVER in session_manager.py).""" + super().try_connect(timeout) + self._connected = True def _reconnect(self) -> None: """Re-initialise the underlying admin API and log in again after the session was closed.""" + self._connected = False Session.__init__(self, self.username, self._startup_path, self._secure_mode, self._debug) self.try_connect(timeout=5.0) @@ -41,14 +52,19 @@ def _do_command(self, cmd: str): """ Override the _do_command method to add error handling for session inactivity or closure. - On ``InternalError`` with "session_inactive", reconnects via ``try_connect`` and retries once. - On ``SessionClosed`` (e.g. idle timeout, fl-server restart, network blip), fully re-initialises - the admin session via ``_reconnect`` and retries once. Any exception on the retry is logged and - re-raised immediately — there is no further retry loop. + If never successfully connected (PER_JOB_FL_SERVER let boot proceed with an unreachable + fl-server), connects lazily here on first use. On ``InternalError`` with "session_inactive", + reconnects via ``try_connect`` and retries once. On ``SessionClosed`` (e.g. idle timeout, + fl-server restart, network blip), fully re-initialises the admin session via ``_reconnect`` + and retries once. Any exception on the retry is logged and re-raised immediately — there is + no further retry loop. Args: cmd (str): The command to be executed. """ + if not self._connected: + logger.info("Session not yet connected; connecting now before running command: %s", cmd) + self.try_connect(timeout=5.0) try: return super()._do_command(cmd) except InternalError as e: @@ -66,6 +82,11 @@ def _do_command(self, cmd: str): logger.error("Retry after reconnect failed for command: %s", cmd) raise + @property + def is_connected(self) -> bool: + """Whether the admin session has ever successfully connected (see PER_JOB_FL_SERVER).""" + return self._connected + def check_server_status(self) -> ServerInfoModel: """ Checks the status of the server. diff --git a/fl-services/nvflare/fl-api-base/tests/routers/test_system.py b/fl-services/nvflare/fl-api-base/tests/routers/test_system.py index d1109184f..398640f22 100644 --- a/fl-services/nvflare/fl-api-base/tests/routers/test_system.py +++ b/fl-services/nvflare/fl-api-base/tests/routers/test_system.py @@ -10,10 +10,11 @@ # limitations under the License. # -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest from fastapi import status +from nvflare.fuel.flare_api.api_spec import NoConnection from fl_api.app import app from fl_api.core.dependencies import get_session @@ -64,6 +65,33 @@ def test_check_server_status(client, mock_session): assert resp_json["start_time"] == 123.0 +def test_check_server_status_returns_stopped_when_unreachable_and_flag_on(client, mock_session): + """✅ PER_JOB_FL_SERVER on: an unreachable server reports STOPPED, not a 500 (FLIP#735).""" + app.dependency_overrides[system.get_session] = lambda: mock_session + mock_session.check_server_status.side_effect = NoConnection("cannot connect to server") + + with patch("fl_api.routers.system.get_settings") as mock_get_settings: + mock_get_settings.return_value.PER_JOB_FL_SERVER = True + resp = client.get("/check_server_status") + + assert resp.status_code == status.HTTP_200_OK + resp_json = resp.json() + assert resp_json["status"] == "STOPPED" + assert resp_json["start_time"] == 0.0 + + +def test_check_server_status_raises_when_unreachable_and_flag_off(client, mock_session): + """❌ PER_JOB_FL_SERVER off (default): an unreachable server still errors, unchanged.""" + app.dependency_overrides[system.get_session] = lambda: mock_session + mock_session.check_server_status.side_effect = NoConnection("cannot connect to server") + + with patch("fl_api.routers.system.get_settings") as mock_get_settings: + mock_get_settings.return_value.PER_JOB_FL_SERVER = False + resp = client.get("/check_server_status") + + assert resp.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + + # --------------------------------------------------------------------- # check_client_status # --------------------------------------------------------------------- diff --git a/fl-services/nvflare/fl-api-base/tests/startup/test_session_manager.py b/fl-services/nvflare/fl-api-base/tests/startup/test_session_manager.py index 36f42634c..f207d6298 100644 --- a/fl-services/nvflare/fl-api-base/tests/startup/test_session_manager.py +++ b/fl-services/nvflare/fl-api-base/tests/startup/test_session_manager.py @@ -14,6 +14,8 @@ from unittest.mock import MagicMock, patch import pytest +from nvflare.apis.fl_exception import FLCommunicationError +from nvflare.fuel.flare_api.api_spec import AuthenticationError, NoConnection from fl_api.startup.session_manager import create_fl_session @@ -30,6 +32,7 @@ class Settings: JOB_RESOURCE_SPEC_NUM_GPUS = 2 JOB_RESOURCE_SPEC_MEM_PER_GPU_IN_GIB = 8 TIMEOUT_SESSION_CONNECT = 5.0 + PER_JOB_FL_SERVER = False os.makedirs(Settings.FL_ADMIN_DIRECTORY, exist_ok=True) return Settings() @@ -50,3 +53,64 @@ def test_create_fl_session_success(fake_settings): assert session == mock_session assert session.upload_dir == "/tmp/upload" assert session.download_dir == "/tmp/download" + + +def test_create_fl_session_raises_when_unreachable_and_flag_off(fake_settings): + """❌ PER_JOB_FL_SERVER off (default): an unreachable server at boot still raises, unchanged.""" + fake_settings.PER_JOB_FL_SERVER = False + mock_session = MagicMock() + mock_session.try_connect.side_effect = NoConnection("cannot connect to server") + + with ( + patch("fl_api.startup.session_manager.get_settings", return_value=fake_settings), + patch("fl_api.startup.session_manager.FLIP_Session", return_value=mock_session), + ): + with pytest.raises(NoConnection): + create_fl_session() + + +def test_create_fl_session_tolerates_unreachable_server_when_flag_on(fake_settings): + """✅ PER_JOB_FL_SERVER on: an unreachable server at boot is tolerated (idle-by-design, FLIP#735).""" + fake_settings.PER_JOB_FL_SERVER = True + mock_session = MagicMock() + mock_session.upload_dir = "/tmp/upload" + mock_session.download_dir = "/tmp/download" + mock_session.try_connect.side_effect = NoConnection("cannot connect to server") + + with ( + patch("fl_api.startup.session_manager.get_settings", return_value=fake_settings), + patch("fl_api.startup.session_manager.FLIP_Session", return_value=mock_session), + ): + session = create_fl_session() + + assert session == mock_session + + +def test_create_fl_session_tolerates_fl_communication_error_when_flag_on(fake_settings): + """✅ PER_JOB_FL_SERVER on: a raw FLCommunicationError (not remapped to NoConnection) is tolerated too.""" + fake_settings.PER_JOB_FL_SERVER = True + mock_session = MagicMock() + mock_session.try_connect.side_effect = FLCommunicationError("comm error") + + with ( + patch("fl_api.startup.session_manager.get_settings", return_value=fake_settings), + patch("fl_api.startup.session_manager.FLIP_Session", return_value=mock_session), + ): + session = create_fl_session() + + assert session == mock_session + + +def test_create_fl_session_still_raises_on_non_connectivity_error_when_flag_on(fake_settings): + """❌ PER_JOB_FL_SERVER on: a non-connectivity failure (e.g. auth) still raises — only + unreachability is tolerated, not arbitrary startup failures.""" + fake_settings.PER_JOB_FL_SERVER = True + mock_session = MagicMock() + mock_session.try_connect.side_effect = AuthenticationError("bad cert") + + with ( + patch("fl_api.startup.session_manager.get_settings", return_value=fake_settings), + patch("fl_api.startup.session_manager.FLIP_Session", return_value=mock_session), + ): + with pytest.raises(AuthenticationError): + create_fl_session() diff --git a/fl-services/nvflare/fl-api-base/tests/utils/test_flip_session.py b/fl-services/nvflare/fl-api-base/tests/utils/test_flip_session.py index 88604a8a1..0052abd1c 100644 --- a/fl-services/nvflare/fl-api-base/tests/utils/test_flip_session.py +++ b/fl-services/nvflare/fl-api-base/tests/utils/test_flip_session.py @@ -15,7 +15,7 @@ from unittest.mock import call, patch import pytest -from nvflare.fuel.flare_api.api_spec import InternalError, SessionClosed +from nvflare.fuel.flare_api.api_spec import InternalError, NoConnection, SessionClosed from fl_api.utils.flip_session import FLIP_Session from fl_api.utils.schemas import ServerInfoModel, SystemInfoModel @@ -28,9 +28,15 @@ def session(mock_session_init): """ Create a FLIP_Session without invoking real NVFlare Session initialization. + + Starts marked as already connected: session_inactive/SessionClosed retry paths can only be hit + after a real connection existed, so that's the realistic starting point for testing them. Tests + of the lazy first-connect path (FLIP#735) explicitly reset ``_connected`` to False. """ mock_session_init.return_value = None - return FLIP_Session(username="u", startup_path="p", secure_mode=False, debug=False) + session = FLIP_Session(username="u", startup_path="p", secure_mode=False, debug=False) + session._connected = True + return session def test_do_command_retries_once_on_session_inactive(session): @@ -84,6 +90,73 @@ def test_do_command_propagates_exception_if_retry_fails_after_reconnect(session) assert parent_do_command.call_count == 2 +def test_do_command_connects_lazily_when_never_connected(session): + """✅ First _do_command call connects if the session was never successfully connected + (PER_JOB_FL_SERVER let boot proceed with an unreachable fl-server, FLIP#735).""" + session._connected = False + with ( + patch("nvflare.fuel.flare_api.flare_api.Session._do_command", return_value={"ok": True}) as parent_do_command, + patch.object(session, "try_connect") as try_connect, + ): + result = session._do_command("CMD") + + assert result == {"ok": True} + try_connect.assert_called_once_with(timeout=5.0) + parent_do_command.assert_called_once_with("CMD") + + +def test_do_command_skips_connect_when_already_connected(session): + """✅ No extra connect attempt once the session has already connected successfully.""" + session._connected = True + with ( + patch("nvflare.fuel.flare_api.flare_api.Session._do_command", return_value={"ok": True}), + patch.object(session, "try_connect") as try_connect, + ): + result = session._do_command("CMD") + + assert result == {"ok": True} + try_connect.assert_not_called() + + +def test_try_connect_marks_session_connected_on_success(session): + """✅ try_connect flips _connected to True only after the parent call succeeds.""" + session._connected = False + with patch("nvflare.fuel.flare_api.flare_api.Session.try_connect") as parent_try_connect: + session.try_connect(5.0) + + parent_try_connect.assert_called_once_with(5.0) + assert session._connected is True + assert session.is_connected is True + + +def test_try_connect_does_not_mark_connected_on_failure(session): + """✅ try_connect leaves _connected False when the parent call raises.""" + session._connected = False + with patch("nvflare.fuel.flare_api.flare_api.Session.try_connect", side_effect=NoConnection("down")): + with pytest.raises(NoConnection): + session.try_connect(5.0) + + assert session._connected is False + assert session.is_connected is False + + +def test_reconnect_resets_connected_flag_before_reinitializing(session): + """✅ _reconnect marks _connected False up front, before re-init and reconnect, so a failed + reconnect never leaves a stale True from the previous (now-closed) session.""" + session.username = "u" # real Session.__init__ sets this; the fixture mocks it out + with ( + patch("nvflare.fuel.flare_api.flare_api.Session.__init__", return_value=None) as mock_init, + patch.object(session, "try_connect") as mock_try_connect, + ): + session._reconnect() + + mock_init.assert_called_once() + mock_try_connect.assert_called_once_with(timeout=5.0) + # try_connect is mocked out here, so nothing re-sets _connected to True in this test — + # confirms _reconnect itself is what clears the flag, not a side effect of try_connect. + assert session._connected is False + + def test_check_server_status_returns_server_info(session): sys_info = SystemInfoModel( server_info=ServerInfoModel(status="running", start_time=123.0), diff --git a/flip-api/src/flip_api/config.py b/flip-api/src/flip_api/config.py index c97f91c85..b0aaaa07c 100644 --- a/flip-api/src/flip_api/config.py +++ b/flip-api/src/flip_api/config.py @@ -86,6 +86,24 @@ class Settings(BaseSettings): 30 # How often to check for projects with unimported studies (in minutes) ) + # Per-job fl-server gate (FLIP#735). Defaults to False so the fl-server ECS service stays + # always-on everywhere until explicitly enabled (stag first). The same value is threaded as an + # env var into fl-api-net-1's task definition (see fl-services/nvflare/fl-api-base's own + # PER_JOB_FL_SERVER setting) from one shared Terraform variable — both services must agree on + # whether an unreachable fl-server is an unplanned outage (flag off, fl-api-net-1 crash-loops + # as today) or the normal idle state (flag on). + PER_JOB_FL_SERVER: bool = False + + @field_validator("PER_JOB_FL_SERVER", mode="before") + @classmethod + def coerce_empty_per_job_fl_server(cls, v: str | bool | None) -> bool: + """Treat empty-string or None PER_JOB_FL_SERVER as the default False.""" + if v is None or v == "": + return False + if isinstance(v, bool): + return v + return v.lower() in ("true", "1") # type: ignore[union-attr] + # Database settings DB_PORT: int DB_HOST: str = "localhost" diff --git a/flip-api/src/flip_api/fl_services/services/fl_scheduler_service.py b/flip-api/src/flip_api/fl_services/services/fl_scheduler_service.py index fab060c5e..3560d54fa 100644 --- a/flip-api/src/flip_api/fl_services/services/fl_scheduler_service.py +++ b/flip-api/src/flip_api/fl_services/services/fl_scheduler_service.py @@ -291,6 +291,31 @@ def get_nets(session: Session) -> list[INetDetails]: raise DatabaseError("Error getting nets") from e +def is_net_busy(net_name: str, session: Session) -> bool: + """Check whether the net's scheduler is BUSY (a job is actually running). + + Used to gate PER_JOB_FL_SERVER-only behavior (FLIP#735) that should skip an idle-by-design net + — e.g. keep_fl_api_session_alive has no session worth keeping alive while nothing is training. + + Args: + net_name (str): Name of the net (``FLNets.name``). + session (Session): SQLModel session. + + Returns: + bool: True if a ``FLScheduler`` row for this net exists with status BUSY. + + Raises: + DatabaseError: If the query fails at the DB layer. + """ + try: + statement = select(FLScheduler).join(FLNets).where(FLNets.name == net_name) + scheduler = session.exec(statement).first() + return scheduler is not None and scheduler.status == NetStatus.BUSY + except SQLAlchemyError as e: + logger.error(f"Error checking scheduler status for net {net_name}: {e}") + raise DatabaseError(f"Error checking scheduler status for net {net_name}") from e + + def resolve_backend(session: Session, net: INetDetails | None = None) -> FLBackend: """Resolve the active FL backend at runtime from the nets (never from a static env var). diff --git a/flip-api/src/flip_api/fl_services/services/fl_service.py b/flip-api/src/flip_api/fl_services/services/fl_service.py index ab87a1e63..ce3f8179f 100644 --- a/flip-api/src/flip_api/fl_services/services/fl_service.py +++ b/flip-api/src/flip_api/fl_services/services/fl_service.py @@ -1004,6 +1004,10 @@ def keep_fl_api_session_alive() -> None: A periodic function to keep the FL API session alive by making a simple request. This is useful to prevent the session from going idle or being shut down by the server. + With PER_JOB_FL_SERVER on, a net whose scheduler isn't BUSY is idle by design (FLIP#735) — its + fl-server is expected to be unreachable, so there's no session worth keeping alive and pinging + it would just be continuous log noise. Skip those nets entirely. + TODO This was developed for the NVFLARE backend and might need to be revisited for the Flower backend. See https://github.com/NVIDIA/NVFlare/discussions/3526#discussioncomment-13574644 """ @@ -1017,10 +1021,14 @@ def keep_fl_api_session_alive() -> None: # NOTE In the old implementation, we had 3 'nets' in the database, each with its own FLAdminAPI. So each net had a # separate FLAdminAPI endpoint. Here, there should just be 1 net for now. If we add more nets in the future, they # might all have the same FLARE_API endpoint, if the FLARE_API controls all controllers/clients. + per_job_fl_server = get_settings().PER_JOB_FL_SERVER with Session(get_engine()) as db: nets = fl_scheduler_service.get_nets(db) for net in nets: + if per_job_fl_server and not fl_scheduler_service.is_net_busy(net.name, db): + logger.debug(f"Net {net.name} is idle by design (PER_JOB_FL_SERVER); skipping keep-alive ping.") + continue try: fetch_server_status(net.endpoint) except Exception as e: diff --git a/flip-api/tests/unit/fl_services/services/test_fl_scheduler_service.py b/flip-api/tests/unit/fl_services/services/test_fl_scheduler_service.py index 17e185fa5..c2cdcbf9a 100644 --- a/flip-api/tests/unit/fl_services/services/test_fl_scheduler_service.py +++ b/flip-api/tests/unit/fl_services/services/test_fl_scheduler_service.py @@ -14,6 +14,7 @@ from uuid import uuid4 import pytest +from sqlalchemy.exc import SQLAlchemyError from flip_api.domain.interfaces.fl import ( IJobResponse, @@ -24,7 +25,7 @@ from flip_api.domain.schemas.status import JobStatus, ModelStatus, NetStatus from flip_api.domain.schemas.types import FLBackend from flip_api.fl_services.services import fl_scheduler_service -from flip_api.utils.exceptions import NotFoundError +from flip_api.utils.exceptions import DatabaseError, NotFoundError @pytest.fixture @@ -234,6 +235,27 @@ def test_get_nets_no_results(fake_session): fl_scheduler_service.get_nets(fake_session) +def test_is_net_busy_true_when_busy(fake_session): + fake_session.exec.return_value.first.return_value = MagicMock(status=NetStatus.BUSY) + assert fl_scheduler_service.is_net_busy("net-1", fake_session) is True + + +def test_is_net_busy_false_when_available(fake_session): + fake_session.exec.return_value.first.return_value = MagicMock(status=NetStatus.AVAILABLE) + assert fl_scheduler_service.is_net_busy("net-1", fake_session) is False + + +def test_is_net_busy_false_when_no_scheduler_row(fake_session): + fake_session.exec.return_value.first.return_value = None + assert fl_scheduler_service.is_net_busy("net-missing", fake_session) is False + + +def test_is_net_busy_raises_database_error_on_sqlalchemy_error(fake_session): + fake_session.exec.side_effect = SQLAlchemyError("boom") + with pytest.raises(DatabaseError): + fl_scheduler_service.is_net_busy("net-1", fake_session) + + def test_check_for_available_net(fake_session): scheduler = MagicMock() scheduler.id = uuid4() diff --git a/flip-api/tests/unit/fl_services/services/test_fl_service.py b/flip-api/tests/unit/fl_services/services/test_fl_service.py index 57ab7a3d1..1bdce5686 100644 --- a/flip-api/tests/unit/fl_services/services/test_fl_service.py +++ b/flip-api/tests/unit/fl_services/services/test_fl_service.py @@ -1127,6 +1127,47 @@ def test_keep_fl_api_session_alive_swallows_errors(mock_session, mock_get_nets, mock_fetch.assert_called_once_with("http://net1:8000") +@patch("flip_api.fl_services.services.fl_scheduler_service.is_net_busy", return_value=False) +@patch("flip_api.fl_services.get_status.fetch_server_status") +@patch("flip_api.fl_services.services.fl_scheduler_service.get_nets") +@patch("flip_api.fl_services.services.fl_service.Session") +@patch("flip_api.fl_services.services.fl_service.get_settings") +def test_keep_fl_api_session_alive_skips_idle_net_when_flag_on( + mock_get_settings, mock_session, mock_get_nets, mock_fetch, mock_is_net_busy +): + # PER_JOB_FL_SERVER on + scheduler AVAILABLE (idle by design) -> no ping, no session to keep alive. + mock_get_settings.return_value.PER_JOB_FL_SERVER = True + mock_session.return_value.__enter__.return_value = MagicMock() + net = MagicMock(endpoint="http://net1:8000") + net.name = "net-1" + mock_get_nets.return_value = [net] + + fl_service.keep_fl_api_session_alive() + + mock_is_net_busy.assert_called_once_with("net-1", mock_session.return_value.__enter__.return_value) + mock_fetch.assert_not_called() + + +@patch("flip_api.fl_services.services.fl_scheduler_service.is_net_busy", return_value=True) +@patch("flip_api.fl_services.get_status.fetch_server_status") +@patch("flip_api.fl_services.services.fl_scheduler_service.get_nets") +@patch("flip_api.fl_services.services.fl_service.Session") +@patch("flip_api.fl_services.services.fl_service.get_settings") +def test_keep_fl_api_session_alive_pings_busy_net_when_flag_on( + mock_get_settings, mock_session, mock_get_nets, mock_fetch, mock_is_net_busy +): + # PER_JOB_FL_SERVER on but scheduler BUSY -> still ping, a job is actually training. + mock_get_settings.return_value.PER_JOB_FL_SERVER = True + mock_session.return_value.__enter__.return_value = MagicMock() + net = MagicMock(endpoint="http://net1:8000") + net.name = "net-1" + mock_get_nets.return_value = [net] + + fl_service.keep_fl_api_session_alive() + + mock_fetch.assert_called_once_with("http://net1:8000") + + # --- submit_job success path ---------------------------------------------------------------------