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 ---------------------------------------------------------------------