diff --git a/framework/py/flwr/supercore/cli/flower_superexec.py b/framework/py/flwr/supercore/cli/flower_superexec.py index 2f25afd11229..fd480585a0a0 100644 --- a/framework/py/flwr/supercore/cli/flower_superexec.py +++ b/framework/py/flwr/supercore/cli/flower_superexec.py @@ -33,6 +33,7 @@ from flwr.supercore.constant import EXEC_PLUGIN_SECTION, ExecutorType from flwr.supercore.exit import ExitCode, flwr_exit from flwr.supercore.grpc_health import add_args_health +from flwr.supercore.runtime import RuntimeHttpStub as CoreRuntimeHttpStub from flwr.supercore.superexec.executor.config import ( ExecutorConfig, ExecutorConfigError, @@ -49,6 +50,9 @@ from flwr.supercore.update_check import warn_if_flwr_update_available from flwr.supercore.utils import disable_process_dumping from flwr.supercore.version import package_version +from flwr.superlink.runtime import RuntimeHttpStub as SuperLinkRuntimeHttpStub + +RuntimeStubClass = type[RuntimeStub] | type[CoreRuntimeHttpStub] def flower_superexec() -> None: @@ -102,7 +106,9 @@ def flower_superexec() -> None: ExecPluginType.SERVER_APP, ) - plugin_class, stub_class = _get_plugin_and_stub_class(args.plugin_type) + plugin_class, stub_class = _get_plugin_and_stub_class( + args.plugin_type, args.enable_http_api + ) superexec_auth_secret = None if args.superexec_auth_secret_file is not None: try: @@ -126,7 +132,7 @@ def flower_superexec() -> None: run_superexec( plugin_class=plugin_class, - stub_class=stub_class, # type: ignore + stub_class=stub_class, runtime_api_address=args.runtime_api_address, insecure=args.insecure, root_certificates_path=args.root_certificates, @@ -137,6 +143,7 @@ def flower_superexec() -> None: runtime_dependency_install=args.runtime_dependency_install, executor_type=args.executor, executor_config=executor_config, + enable_http_api=args.enable_http_api, ) @@ -158,6 +165,12 @@ def _parse_args() -> argparse.ArgumentParser: required=True, help="Address of the Runtime API", ) + parser.add_argument( + "--enable-http-api", + action="store_true", + default=False, + help="EXPERIMENTAL: Connect to the Runtime API over HTTP instead of gRPC.", + ) parser.add_argument( "--plugin-type", type=str, @@ -221,14 +234,17 @@ def _load_executor_config( def _get_plugin_and_stub_class( plugin_type: str, -) -> tuple[type[ExecPlugin], type[object]]: + enable_http_api: bool = False, +) -> tuple[type[ExecPlugin], RuntimeStubClass]: """Get the plugin class and stub class based on the plugin type.""" - mapping: dict[str, tuple[type[ExecPlugin], type[object]]] = { - ExecPluginType.CLIENT_APP: (ClientAppExecPlugin, RuntimeStub), - ExecPluginType.SERVER_APP: (ServerAppExecPlugin, RuntimeStub), + clientapp_stub = CoreRuntimeHttpStub if enable_http_api else RuntimeStub + serverapp_stub = SuperLinkRuntimeHttpStub if enable_http_api else RuntimeStub + mapping: dict[str, tuple[type[ExecPlugin], RuntimeStubClass]] = { + ExecPluginType.CLIENT_APP: (ClientAppExecPlugin, clientapp_stub), + ExecPluginType.SERVER_APP: (ServerAppExecPlugin, serverapp_stub), ExecPluginType.SERVER_APP_EPHEMERAL: ( ServerAppEphemeralExecPlugin, - RuntimeStub, + serverapp_stub, ), } if plugin_type in mapping: diff --git a/framework/py/flwr/supercore/cli/flower_superexec_test.py b/framework/py/flwr/supercore/cli/flower_superexec_test.py index f78fd442bb79..29bc54752610 100644 --- a/framework/py/flwr/supercore/cli/flower_superexec_test.py +++ b/framework/py/flwr/supercore/cli/flower_superexec_test.py @@ -24,7 +24,9 @@ from flwr.common.constant import ExecPluginType from flwr.proto.runtime_pb2_grpc import RuntimeStub from flwr.supercore.constant import ExecutorType +from flwr.supercore.runtime import RuntimeHttpStub as CoreRuntimeHttpStub from flwr.supercore.version import package_version +from flwr.superlink.runtime import RuntimeHttpStub as SuperLinkRuntimeHttpStub from .flower_superexec import _parse_args @@ -63,6 +65,21 @@ def test_parse_superexec_accepts_kubernetes_executor_config() -> None: assert args.executor_config == "executor.yaml" +def test_parse_superexec_accepts_http_api_flag() -> None: + """SuperExec should accept the temporary Runtime HTTP API flag.""" + args = _parse_args().parse_args( + [ + "--appio-api-address", + "127.0.0.1:8000", + "--plugin-type", + ExecPluginType.CLIENT_APP, + "--enable-http-api", + ] + ) + + assert args.enable_http_api is True + + def test_flower_superexec_checks_for_update( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -117,6 +134,7 @@ def test_flower_superexec_clientapp_allows_missing_secret( health_server_address=None, runtime_dependency_install=False, executor=ExecutorType.SUBPROCESS, + enable_http_api=False, ) captured: dict[str, object] = {} @@ -137,7 +155,7 @@ def _run_superexec(**kwargs: object) -> None: monkeypatch.setattr( flower_superexec_module, "_get_plugin_and_stub_class", - lambda _plugin_type: (object, RuntimeStub), + lambda _plugin_type, _enable_http_api: (object, RuntimeStub), ) monkeypatch.setattr(flower_superexec_module, "run_superexec", _run_superexec) @@ -161,6 +179,7 @@ def test_flower_superexec_serverapp_allows_missing_secret( health_server_address=None, runtime_dependency_install=False, executor=ExecutorType.SUBPROCESS, + enable_http_api=False, ) class _Parser: @@ -182,7 +201,7 @@ def _run_superexec(**kwargs: object) -> None: monkeypatch.setattr( flower_superexec_module, "_get_plugin_and_stub_class", - lambda _plugin_type: (object, RuntimeStub), + lambda _plugin_type, _enable_http_api: (object, RuntimeStub), ) monkeypatch.setattr(flower_superexec_module, "run_superexec", _run_superexec) @@ -206,6 +225,7 @@ def test_flower_superexec_passes_executor_to_run_superexec( health_server_address=None, runtime_dependency_install=False, executor=ExecutorType.SUBPROCESS, + enable_http_api=False, ) parser = Mock() parser.parse_args.return_value = args @@ -222,7 +242,7 @@ def test_flower_superexec_passes_executor_to_run_superexec( monkeypatch.setattr( flower_superexec_module, "_get_plugin_and_stub_class", - lambda _plugin_type: (object, RuntimeStub), + lambda _plugin_type, _enable_http_api: (object, RuntimeStub), ) monkeypatch.setattr(flower_superexec_module, "run_superexec", run_superexec_mock) @@ -233,3 +253,22 @@ def test_flower_superexec_passes_executor_to_run_superexec( run_superexec_mock.call_args.kwargs["executor_type"] == ExecutorType.SUBPROCESS ) assert run_superexec_mock.call_args.kwargs["executor_config"] is None + + +@pytest.mark.parametrize( + ("plugin_type", "expected_stub"), + [ + (ExecPluginType.CLIENT_APP, CoreRuntimeHttpStub), + (ExecPluginType.SERVER_APP, SuperLinkRuntimeHttpStub), + (ExecPluginType.SERVER_APP_EPHEMERAL, SuperLinkRuntimeHttpStub), + ], +) +def test_get_plugin_and_stub_class_selects_http_stub( + plugin_type: str, expected_stub: type[object] +) -> None: + """HTTP mode should select the Runtime stub for each plugin surface.""" + _, stub_class = flower_superexec_module._get_plugin_and_stub_class( + plugin_type, enable_http_api=True + ) + + assert stub_class is expected_stub diff --git a/framework/py/flwr/supercore/heartbeat.py b/framework/py/flwr/supercore/heartbeat.py index 82f64e1193e1..067fa30a9ec0 100644 --- a/framework/py/flwr/supercore/heartbeat.py +++ b/framework/py/flwr/supercore/heartbeat.py @@ -21,6 +21,7 @@ from collections.abc import Callable import grpc +import httpx from flwr.common.constant import ( HEARTBEAT_BASE_MULTIPLIER, @@ -33,6 +34,7 @@ from flwr.proto.runtime_pb2 import SendTaskHeartbeatRequest from flwr.proto.runtime_pb2_grpc import RuntimeStub from flwr.supercore.retry import RetryInvoker, exponential +from flwr.supercore.runtime import RuntimeHttpStub # pylint: enable=E0611 @@ -154,3 +156,22 @@ def fn() -> bool: return True return fn + + +def make_task_heartbeat_fn_http( + stub: RuntimeHttpStub, +) -> Callable[[], bool]: + """Get the function to send a heartbeat to an HTTP Runtime endpoint.""" + req = SendTaskHeartbeatRequest() + + def fn() -> bool: + try: + res = stub.SendTaskHeartbeat(req) + except httpx.TransportError: + return False + + if not res.success: + signal.raise_signal(signal.SIGINT) + return True + + return fn diff --git a/framework/py/flwr/supercore/heartbeat_test.py b/framework/py/flwr/supercore/heartbeat_test.py index 0cfec7f4cb4f..8b1ac1a91d29 100644 --- a/framework/py/flwr/supercore/heartbeat_test.py +++ b/framework/py/flwr/supercore/heartbeat_test.py @@ -19,7 +19,11 @@ import unittest from unittest.mock import Mock -from .heartbeat import HeartbeatSender +import httpx + +from flwr.proto.runtime_pb2 import SendTaskHeartbeatResponse # pylint: disable=E0611 + +from .heartbeat import HeartbeatSender, make_task_heartbeat_fn_http # pylint: disable=protected-access @@ -90,3 +94,19 @@ def test_heartbeat_fail_and_retry(self) -> None: def test_thread_is_daemon(self) -> None: """Test that the thread is a daemon thread.""" self.assertTrue(self.heartbeat_sender._thread.daemon) + + +def test_http_heartbeat_returns_true_on_success() -> None: + """HTTP heartbeat should report a successful Runtime response.""" + stub = Mock() + stub.SendTaskHeartbeat.return_value = SendTaskHeartbeatResponse(success=True) + + assert make_task_heartbeat_fn_http(stub)() is True + + +def test_http_heartbeat_retries_transport_errors() -> None: + """HTTP heartbeat should make transport failures retryable.""" + stub = Mock() + stub.SendTaskHeartbeat.side_effect = httpx.ConnectError("connection failed") + + assert make_task_heartbeat_fn_http(stub)() is False diff --git a/framework/py/flwr/supercore/superexec/executor/kubernetes_executor.py b/framework/py/flwr/supercore/superexec/executor/kubernetes_executor.py index f77b47e4fead..014d2cac6cb1 100644 --- a/framework/py/flwr/supercore/superexec/executor/kubernetes_executor.py +++ b/framework/py/flwr/supercore/superexec/executor/kubernetes_executor.py @@ -517,6 +517,8 @@ def _taskexecutor_args( "--token-file", APPIO_TOKEN_FILE_PATH, ] + if spec.enable_http_api: + args.append("--enable-http-api") if spec.insecure: args.append("--insecure") diff --git a/framework/py/flwr/supercore/superexec/executor/kubernetes_executor_test.py b/framework/py/flwr/supercore/superexec/executor/kubernetes_executor_test.py index 0f2599a3ee60..186f2d12e036 100644 --- a/framework/py/flwr/supercore/superexec/executor/kubernetes_executor_test.py +++ b/framework/py/flwr/supercore/superexec/executor/kubernetes_executor_test.py @@ -407,6 +407,20 @@ def test_build_taskexecutor_pod_supports_clientapp_insecure_args() -> None: ] +def test_build_taskexecutor_pod_forwards_http_api_flag() -> None: + """Test Pod construction forwards Runtime HTTP mode.""" + pod = _as_dict( + _build_taskexecutor_pod( + _execution_spec(enable_http_api=True), + _executor_config(runtime_root_certificates=None), + None, + _LAUNCH_ATTEMPT_ID, + ) + ) + + assert "--enable-http-api" in pod["spec"]["containers"][0]["args"] + + def test_build_taskexecutor_pod_supports_secure_default_trust_store() -> None: """Test secure Pod args can rely on container default trust store.""" spec = _execution_spec() diff --git a/framework/py/flwr/supercore/superexec/executor/subprocess_executor.py b/framework/py/flwr/supercore/superexec/executor/subprocess_executor.py index 6e9e6f1d132b..50c51fd0b507 100644 --- a/framework/py/flwr/supercore/superexec/executor/subprocess_executor.py +++ b/framework/py/flwr/supercore/superexec/executor/subprocess_executor.py @@ -40,6 +40,8 @@ def launch(self, spec: ExecutionSpec) -> LaunchResult: "--token", spec.token, ] + if spec.enable_http_api: + args.append("--enable-http-api") if spec.insecure: args.append("--insecure") diff --git a/framework/py/flwr/supercore/superexec/executor/subprocess_executor_test.py b/framework/py/flwr/supercore/superexec/executor/subprocess_executor_test.py index 83377689664a..82bdf4668c93 100644 --- a/framework/py/flwr/supercore/superexec/executor/subprocess_executor_test.py +++ b/framework/py/flwr/supercore/superexec/executor/subprocess_executor_test.py @@ -93,6 +93,14 @@ def test_launch_renders_runtime_dependency_install_flag() -> None: assert "--allow-runtime-dependency-installation" in popen_mock.call_args.args[0] +def test_launch_renders_http_api_flag() -> None: + """Test subprocess executor forwards Runtime HTTP mode.""" + with patch.object(subprocess, "Popen") as popen_mock: + SubprocessExecutor().launch(_execution_spec(enable_http_api=True)) + + assert "--enable-http-api" in popen_mock.call_args.args[0] + + def test_launch_renders_parent_pid_flag() -> None: """Test subprocess executor renders subprocess parent PID flag.""" with patch.object(subprocess, "Popen") as popen_mock: diff --git a/framework/py/flwr/supercore/superexec/executor/types.py b/framework/py/flwr/supercore/superexec/executor/types.py index ea10daa3d917..a23d0d9fc47b 100644 --- a/framework/py/flwr/supercore/superexec/executor/types.py +++ b/framework/py/flwr/supercore/superexec/executor/types.py @@ -36,6 +36,7 @@ class ExecutionSpec: # pylint: disable=too-many-instance-attributes parent_pid: int | None suppress_output: bool task_id: int + enable_http_api: bool = False def __post_init__(self) -> None: """Validate fields required by all executors.""" diff --git a/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin.py b/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin.py index 454ef803e656..97afe13c94b6 100644 --- a/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin.py +++ b/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin.py @@ -57,6 +57,8 @@ def launch_task(self, token: str, task: Task) -> None: # type: ignore[override] elif self.root_certificates_path: cmds += ["--root-certificates", self.root_certificates_path] cmds += [self.runtime_api_address_arg, self.runtime_api_address] + if self.enable_http_api: + cmds += ["--enable-http-api"] cmds += ["--token", token] cmds += ["--parent-pid", str(os.getpid())] if self.runtime_dependency_install: diff --git a/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin_test.py b/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin_test.py index 942c3ee117ba..fc4b8126d10c 100644 --- a/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin_test.py +++ b/framework/py/flwr/supercore/superexec/plugin/base_ephemeral_exec_plugin_test.py @@ -42,12 +42,13 @@ class _EphemeralExecPlugin(BaseEphemeralExecPlugin): runtime_api_address_arg = "--serverappio-api-address" -def _get_ephemeral_plugin() -> _EphemeralExecPlugin: +def _get_ephemeral_plugin(*, enable_http_api: bool = False) -> _EphemeralExecPlugin: return _EphemeralExecPlugin( runtime_api_address="127.0.0.1:9091", get_run=_get_run, insecure=True, root_certificates_path=None, + enable_http_api=enable_http_api, ) @@ -119,3 +120,18 @@ def test_launch_task_calls_cleanup_before_launch() -> None: # Assert assert call_log == ["cleanup", "subprocess"] + + +def test_launch_task_forwards_http_api_flag() -> None: + """Launch should forward Runtime HTTP mode to the app process.""" + plugin = _get_ephemeral_plugin(enable_http_api=True) + + with ( + patch( + "flwr.supercore.superexec.plugin.base_ephemeral_exec_plugin.subprocess.run" + ) as run, + patch("flwr.supercore.superexec.plugin.base_ephemeral_exec_plugin.flwr_exit"), + ): + plugin.launch_task(token="token-123", task=_get_task(task_id=5)) + + assert "--enable-http-api" in run.call_args.args[0] diff --git a/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin.py b/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin.py index d9a1ccb67be6..0e503b07950e 100644 --- a/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin.py +++ b/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin.py @@ -49,6 +49,7 @@ def __init__( # pylint: disable=R0913, R0917 runtime_dependency_install: bool = RUNTIME_DEPENDENCY_INSTALL, *, executor: Executor, + enable_http_api: bool = False, ) -> None: super().__init__( runtime_api_address=runtime_api_address, @@ -57,6 +58,7 @@ def __init__( # pylint: disable=R0913, R0917 get_run=get_run, runtime_dependency_install=runtime_dependency_install, executor=executor, + enable_http_api=enable_http_api, ) self.executor: Executor = executor @@ -103,6 +105,7 @@ def _build_execution_spec( self.suppress_output and task_type not in self.visible_output_task_types ), task_id=task_id, + enable_http_api=self.enable_http_api, ) def _get_supported_task_type(self, task: Task) -> TaskType | None: diff --git a/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin_test.py b/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin_test.py index 525c8765f1ce..3b04dcd71682 100644 --- a/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin_test.py +++ b/framework/py/flwr/supercore/superexec/plugin/base_exec_plugin_test.py @@ -159,6 +159,23 @@ def test_launch_task_forwards_runtime_dependency_install_flag() -> None: assert spec.task_id == 7 +def test_launch_task_forwards_http_api_flag() -> None: + """Ensure execution spec forwards Runtime HTTP mode.""" + executor = Mock() + plugin = DummyExecPlugin( + runtime_api_address="127.0.0.1:8000", + insecure=True, + root_certificates_path=None, + get_run=Mock(), + executor=executor, + enable_http_api=True, + ) + + plugin.launch_task(token="token-123", task=_get_task(task_id=7)) + + assert _execution_spec_from_executor(executor).enable_http_api is True + + def test_launch_task_skips_optional_runtime_flags_by_default() -> None: """Ensure execution spec omits optional runtime install flags by default.""" executor = Mock() diff --git a/framework/py/flwr/supercore/superexec/plugin/exec_plugin.py b/framework/py/flwr/supercore/superexec/plugin/exec_plugin.py index 1681065dc2a0..023a2a6d159d 100644 --- a/framework/py/flwr/supercore/superexec/plugin/exec_plugin.py +++ b/framework/py/flwr/supercore/superexec/plugin/exec_plugin.py @@ -36,6 +36,7 @@ def __init__( # pylint: disable=R0913, R0917 get_run: Callable[[int], Run], runtime_dependency_install: bool = RUNTIME_DEPENDENCY_INSTALL, executor: Executor | None = None, + enable_http_api: bool = False, ) -> None: self.runtime_api_address = runtime_api_address self.insecure = insecure @@ -44,6 +45,7 @@ def __init__( # pylint: disable=R0913, R0917 self.runtime_dependency_install = runtime_dependency_install # Non-ephemeral plugins use the executor to start task processes. self.executor = executor + self.enable_http_api = enable_http_api @abstractmethod def select_run_id(self, candidate_run_ids: Sequence[int]) -> int | None: diff --git a/framework/py/flwr/supercore/superexec/run_superexec.py b/framework/py/flwr/supercore/superexec/run_superexec.py index 2c0c1ee975c8..56a1df469707 100644 --- a/framework/py/flwr/supercore/superexec/run_superexec.py +++ b/framework/py/flwr/supercore/superexec/run_superexec.py @@ -16,8 +16,9 @@ import time +from collections.abc import Callable from logging import ERROR, WARNING -from typing import Any +from typing import Any, cast import grpc @@ -38,13 +39,17 @@ from flwr.supercore.grpc_health import run_health_server_grpc_no_tls from flwr.supercore.interceptors import ( RuntimeVersionClientInterceptor, + RuntimeVersionHttpInterceptor, SuperExecAuthClientInterceptor, + SuperExecAuthHttpInterceptor, ) from flwr.supercore.interceptors.superexec_auth_interceptor import ( RUNTIME_SUPEREXEC_METHODS, ) +from flwr.supercore.protobuf.client import ProtobufClient, ProtobufClientInterceptor from flwr.supercore.retry import make_simple_grpc_retry_invoker, wrap_stub from flwr.supercore.run import Run +from flwr.supercore.runtime import RuntimeHttpStub from flwr.supercore.telemetry import EventType from flwr.supercore.tls import validate_and_resolve_root_certificates @@ -103,7 +108,7 @@ def _handle_launch_result(result: LaunchResult | None, task: Task) -> None: def run_superexec( # pylint: disable=R0912,R0913,R0914,R0915,R0917 plugin_class: type[ExecPlugin], - stub_class: type[RuntimeStub], + stub_class: type[RuntimeStub] | type[RuntimeHttpStub], runtime_api_address: str, insecure: bool, root_certificates_path: str | None = None, @@ -114,6 +119,7 @@ def run_superexec( # pylint: disable=R0912,R0913,R0914,R0915,R0917 runtime_dependency_install: bool = RUNTIME_DEPENDENCY_INSTALL, executor_type: ExecutorType = ExecutorType.SUBPROCESS, executor_config: ExecutorConfig | None = None, + enable_http_api: bool = False, ) -> None: """Run Flower SuperExec. @@ -147,22 +153,62 @@ def run_superexec( # pylint: disable=R0912,R0913,R0914,R0915,R0917 The executor to use for non-ephemeral app processes. executor_config : Optional[ExecutorConfig] (default: None) Parsed executor configuration. + enable_http_api : bool (default: False) + Whether to connect to the Runtime API over HTTP instead of gRPC. """ try: executor = get_executor(executor_type, executor_config=executor_config) except ValueError as err: flwr_exit(ExitCode.SUPEREXEC_INVALID_EXECUTOR_CONFIG, str(err)) - interceptors: list[grpc.UnaryUnaryClientInterceptor] = [ - RuntimeVersionClientInterceptor(component_name="SuperExec") - ] - auth_interceptor: SuperExecAuthClientInterceptor | None = None - if superexec_auth_secret: - auth_interceptor = SuperExecAuthClientInterceptor( - master_secret=superexec_auth_secret, - protected_methods=RUNTIME_SUPEREXEC_METHODS, + auth_interceptor: ( + SuperExecAuthClientInterceptor | SuperExecAuthHttpInterceptor | None + ) = None + close_runtime_connection: Callable[[], None] + stub: RuntimeStub | RuntimeHttpStub + if enable_http_api: + http_interceptors: list[ProtobufClientInterceptor] = [ + RuntimeVersionHttpInterceptor(component_name="SuperExec") + ] + if superexec_auth_secret: + auth_interceptor = SuperExecAuthHttpInterceptor( + master_secret=superexec_auth_secret, + protected_methods=RUNTIME_SUPEREXEC_METHODS, + ) + http_interceptors.append(auth_interceptor) + validate_and_resolve_root_certificates(root_certificates_path, insecure) + http_stub_class = cast(type[RuntimeHttpStub], stub_class) + scheme = "http" if insecure else "https" + stub = http_stub_class( + f"{scheme}://{runtime_api_address}", + interceptors=http_interceptors, + verify=False if insecure else root_certificates_path or True, ) - interceptors.append(auth_interceptor) + close_runtime_connection = cast(ProtobufClient, stub).close + else: + grpc_interceptors: list[grpc.UnaryUnaryClientInterceptor] = [ + RuntimeVersionClientInterceptor(component_name="SuperExec") + ] + if superexec_auth_secret: + auth_interceptor = SuperExecAuthClientInterceptor( + master_secret=superexec_auth_secret, + protected_methods=RUNTIME_SUPEREXEC_METHODS, + ) + grpc_interceptors.append(auth_interceptor) + + channel = create_channel( + server_address=runtime_api_address, + insecure=insecure, + root_certificates=validate_and_resolve_root_certificates( + root_certificates_path, insecure + ), + interceptors=grpc_interceptors, + ) + channel.subscribe(on_channel_state_change) + grpc_stub_class = cast(type[RuntimeStub], stub_class) + stub = grpc_stub_class(channel) + wrap_stub(stub, make_simple_grpc_retry_invoker()) + close_runtime_connection = channel.close # Start monitoring the parent process if a PID is provided if parent_pid is not None: @@ -174,29 +220,14 @@ def run_superexec( # pylint: disable=R0912,R0913,R0914,R0915,R0917 health_server = run_health_server_grpc_no_tls(health_server_address) grpc_servers.append(health_server) - # Create the channel to the Runtime API - channel = create_channel( - server_address=runtime_api_address, - insecure=insecure, - root_certificates=validate_and_resolve_root_certificates( - root_certificates_path, insecure - ), - interceptors=interceptors, - ) - channel.subscribe(on_channel_state_change) - - # Register exit handlers to close the channel on exit + # Register exit handlers to close the Runtime API connection on exit register_signal_handlers( event_type=EventType.RUN_SUPEREXEC_LEAVE, exit_message="SuperExec terminated gracefully.", grpc_servers=grpc_servers, - exit_handlers=[lambda: channel.close()], # pylint: disable=W0108 + exit_handlers=[close_runtime_connection], ) - # Create the gRPC stub for the Runtime API - stub = stub_class(channel) - wrap_stub(stub, make_simple_grpc_retry_invoker()) - def get_run(run_id: int) -> Run: _req = GetRunRequest(run_id=run_id) _res = stub.GetRun(_req) @@ -210,6 +241,7 @@ def get_run(run_id: int) -> Run: get_run=get_run, runtime_dependency_install=runtime_dependency_install, executor=executor, + enable_http_api=enable_http_api, ) # Load plugin configuration from file if provided @@ -263,4 +295,4 @@ def cleanup_auth_secret() -> None: # Sleep for a while before checking again time.sleep(1) finally: - channel.close() + close_runtime_connection() diff --git a/framework/py/flwr/supercore/superexec/run_superexec_test.py b/framework/py/flwr/supercore/superexec/run_superexec_test.py index 183fd7586d66..dc80b266082d 100644 --- a/framework/py/flwr/supercore/superexec/run_superexec_test.py +++ b/framework/py/flwr/supercore/superexec/run_superexec_test.py @@ -24,7 +24,9 @@ from flwr.supercore.constant import ExecutorType from flwr.supercore.interceptors import ( RuntimeVersionClientInterceptor, + RuntimeVersionHttpInterceptor, SuperExecAuthClientInterceptor, + SuperExecAuthHttpInterceptor, ) from flwr.supercore.superexec.executor import LaunchResult, LaunchResultStatus @@ -112,6 +114,53 @@ def _create_channel(**kwargs: Any) -> Mock: ) +@pytest.mark.parametrize( + ("superexec_auth_secret", "expected_interceptor_types"), + [ + (None, (RuntimeVersionHttpInterceptor,)), + ( + b"superexec-secret", + (RuntimeVersionHttpInterceptor, SuperExecAuthHttpInterceptor), + ), + ], +) +def test_run_superexec_configures_http_runtime_stub( + monkeypatch: pytest.MonkeyPatch, + superexec_auth_secret: bytes | None, + expected_interceptor_types: tuple[type[object], ...], +) -> None: + """SuperExec should create an HTTP stub with version and auth interceptors.""" + stub = Mock() + stub.PullPendingTasks.side_effect = KeyboardInterrupt() + stub_class = Mock(return_value=stub) + create_channel = Mock() + + monkeypatch.setattr(run_superexec_module, "create_channel", create_channel) + monkeypatch.setattr(run_superexec_module, "register_signal_handlers", Mock()) + + with pytest.raises(KeyboardInterrupt): + run_superexec_module.run_superexec( + plugin_class=Mock(), + stub_class=stub_class, + runtime_api_address="127.0.0.1:8000", + insecure=True, + superexec_auth_secret=superexec_auth_secret, + enable_http_api=True, + ) + + create_channel.assert_not_called() + assert stub_class.call_args.args == ("http://127.0.0.1:8000",) + assert stub_class.call_args.kwargs["verify"] is False + assert ( + tuple( + type(interceptor) + for interceptor in stub_class.call_args.kwargs["interceptors"] + ) + == expected_interceptor_types + ) + stub.close.assert_called_once_with() + + def test_run_superexec_passes_executor_config_to_factory( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/framework/py/flwr/supernode/cli/flower_supernode.py b/framework/py/flwr/supernode/cli/flower_supernode.py index bfeb1f291568..942c3a17db74 100644 --- a/framework/py/flwr/supernode/cli/flower_supernode.py +++ b/framework/py/flwr/supernode/cli/flower_supernode.py @@ -204,6 +204,9 @@ def flower_supernode() -> None: trusted_entities=config.trusted_entities, superexec_auth_secret=config.superexec_auth_secret, runtime_dependency_install=config.runtime_dependency_install, + runtime_http_api_address=( + f"{config.host}:{config.port}" if config.enable_http_api else None + ), ) finally: if http_server is not None and http_thread is not None: diff --git a/framework/py/flwr/supernode/cli/flower_supernode_test.py b/framework/py/flwr/supernode/cli/flower_supernode_test.py index fe49d0d2cc3a..422e88eec4dd 100644 --- a/framework/py/flwr/supernode/cli/flower_supernode_test.py +++ b/framework/py/flwr/supernode/cli/flower_supernode_test.py @@ -224,6 +224,8 @@ def test_flower_supernode_injects_state_factory_and_manages_http_api( """SuperNode should share state and stop its background HTTP API.""" config = Mock() config.enable_http_api = True + config.host = "127.0.0.1" + config.port = 8000 objectstore_factory = Mock() state_factory = Mock() start_client_internal = Mock() @@ -258,6 +260,10 @@ def test_flower_supernode_injects_state_factory_and_manages_http_api( node_state_factory.assert_called_once_with(objectstore_factory=objectstore_factory) assert start_client_internal.call_args.kwargs["state_factory"] is state_factory + assert ( + start_client_internal.call_args.kwargs["runtime_http_api_address"] + == "127.0.0.1:8000" + ) start_http_api.assert_called_once_with(config, state_factory) assert http_server.should_exit is True http_thread.join.assert_called_once_with() diff --git a/framework/py/flwr/supernode/cli/flwr_clientapp.py b/framework/py/flwr/supernode/cli/flwr_clientapp.py index 3136cb4a95c3..aa0dfac5860c 100644 --- a/framework/py/flwr/supernode/cli/flwr_clientapp.py +++ b/framework/py/flwr/supernode/cli/flwr_clientapp.py @@ -48,6 +48,8 @@ def flwr_clientapp() -> None: ), parent_pid=args.parent_pid, runtime_dependency_install=args.runtime_dependency_install, + enable_http_api=args.enable_http_api, + root_certificates_path=args.root_certificates, ) @@ -64,5 +66,11 @@ def _parse_args_run_flwr_clientapp() -> argparse.ArgumentParser: help="Address of SuperNode's Runtime API (IPv4, IPv6, or a domain name)." f"By default, it is set to {SUPERNODE_RUNTIME_API_DEFAULT_CLIENT_ADDRESS}.", ) + parser.add_argument( + "--enable-http-api", + action="store_true", + default=False, + help="EXPERIMENTAL: Connect to the Runtime API over HTTP instead of gRPC.", + ) add_args_flwr_app_common(parser=parser) return parser diff --git a/framework/py/flwr/supernode/cli/flwr_clientapp_test.py b/framework/py/flwr/supernode/cli/flwr_clientapp_test.py index 7688b09e51a7..c9a509e35144 100644 --- a/framework/py/flwr/supernode/cli/flwr_clientapp_test.py +++ b/framework/py/flwr/supernode/cli/flwr_clientapp_test.py @@ -52,6 +52,7 @@ def test_parse_flwr_clientapp_parses_tokenized_invocation() -> None: "--parent-pid", "1234", "--allow-runtime-dependency-installation", + "--enable-http-api", ] ) @@ -60,6 +61,7 @@ def test_parse_flwr_clientapp_parses_tokenized_invocation() -> None: assert args.insecure is True assert args.parent_pid == 1234 assert args.runtime_dependency_install is True + assert args.enable_http_api is True def test_flwr_clientapp_forwards_cli_args() -> None: @@ -71,6 +73,7 @@ def test_flwr_clientapp_forwards_cli_args() -> None: root_certificates=None, parent_pid=321, runtime_dependency_install=True, + enable_http_api=True, ) class _Parser: @@ -93,3 +96,5 @@ def parse_args(self) -> SimpleNamespace: assert kwargs["certificates"] is None assert kwargs["parent_pid"] == 321 assert kwargs["runtime_dependency_install"] is True + assert kwargs["enable_http_api"] is True + assert kwargs["root_certificates_path"] is None diff --git a/framework/py/flwr/supernode/runtime/run_clientapp.py b/framework/py/flwr/supernode/runtime/run_clientapp.py index fe0f0ec3086c..66fc395701ba 100644 --- a/framework/py/flwr/supernode/runtime/run_clientapp.py +++ b/framework/py/flwr/supernode/runtime/run_clientapp.py @@ -18,6 +18,7 @@ from logging import DEBUG, ERROR import grpc +import httpx from flwr.app import Context, Message from flwr.app.error import Error @@ -49,7 +50,11 @@ from flwr.supercore.exit import ExitCode, flwr_exit, register_signal_handlers from flwr.supercore.fab import Fab from flwr.supercore.grpc import create_channel, on_channel_state_change -from flwr.supercore.heartbeat import HeartbeatSender, make_task_heartbeat_fn_grpc +from flwr.supercore.heartbeat import ( + HeartbeatSender, + make_task_heartbeat_fn_grpc, + make_task_heartbeat_fn_http, +) from flwr.supercore.inflatable.inflatable_object import ( get_all_nested_objects, get_object_tree, @@ -66,10 +71,14 @@ ) from flwr.supercore.interceptors import ( RuntimeTokenClientInterceptor, + RuntimeTokenHttpInterceptor, RuntimeVersionClientInterceptor, + RuntimeVersionHttpInterceptor, ) +from flwr.supercore.protobuf.client import ProtobufClientInterceptor from flwr.supercore.retry import make_simple_grpc_retry_invoker, wrap_stub from flwr.supercore.run import Run +from flwr.supercore.runtime import RuntimeHttpStub from flwr.supercore.superexec.dependency_installer import ( RuntimeDependencyInstallationError, cleanup_app_runtime_environment, @@ -78,13 +87,15 @@ from flwr.supercore.telemetry import EventType, event -def run_clientapp( # pylint: disable=R0913, R0914, R0915, R0917 +def run_clientapp( # pylint: disable=R0912, R0913, R0914, R0915, R0917 runtime_api_address: str, token: str, insecure: bool, certificates: bytes | None = None, parent_pid: int | None = None, runtime_dependency_install: bool = RUNTIME_DEPENDENCY_INSTALL, + enable_http_api: bool = False, + root_certificates_path: str | None = None, ) -> None: """Run Flower ClientApp process.""" # Monitor the main process in case of SIGKILL @@ -93,19 +104,37 @@ def run_clientapp( # pylint: disable=R0913, R0914, R0915, R0917 event(EventType.FLWR_CLIENTAPP_RUN_ENTER) - channel = create_channel( - server_address=runtime_api_address, - insecure=insecure, - root_certificates=certificates, - interceptors=[ - RuntimeVersionClientInterceptor(component_name="flwr-clientapp"), - RuntimeTokenClientInterceptor(token), - ], - ) - channel.subscribe(on_channel_state_change) - stub = RuntimeStub(channel) - retry_invoker = make_simple_grpc_retry_invoker() - wrap_stub(stub, retry_invoker) + retry_invoker = None + stub: RuntimeStub | RuntimeHttpStub + if enable_http_api: + http_interceptors: list[ProtobufClientInterceptor] = [ + RuntimeVersionHttpInterceptor(component_name="flwr-clientapp"), + RuntimeTokenHttpInterceptor(token), + ] + scheme = "http" if insecure else "https" + stub = RuntimeHttpStub( + f"{scheme}://{runtime_api_address}", + interceptors=http_interceptors, + verify=False if insecure else root_certificates_path or True, + ) + close_runtime_connection = stub.close + heartbeat_fn = make_task_heartbeat_fn_http(stub) + else: + channel = create_channel( + server_address=runtime_api_address, + insecure=insecure, + root_certificates=certificates, + interceptors=[ + RuntimeVersionClientInterceptor(component_name="flwr-clientapp"), + RuntimeTokenClientInterceptor(token), + ], + ) + channel.subscribe(on_channel_state_change) + stub = RuntimeStub(channel) + retry_invoker = make_simple_grpc_retry_invoker() + wrap_stub(stub, retry_invoker) + close_runtime_connection = channel.close + heartbeat_fn = make_task_heartbeat_fn_grpc(stub) # Initialize variables for exit handler heartbeat_sender = None @@ -119,7 +148,8 @@ def run_clientapp( # pylint: disable=R0913, R0914, R0915, R0917 def on_exit() -> None: # Set Grpc max retries to 1 to avoid blocking on exit - retry_invoker.max_tries = 1 + if retry_invoker is not None: + retry_invoker.max_tries = 1 # Push final status and context (if available) push_task_output( @@ -132,7 +162,7 @@ def on_exit() -> None: # Stop heartbeat sender if heartbeat_sender is not None and heartbeat_sender.is_running: heartbeat_sender.stop() - channel.close() + close_runtime_connection() cleanup_app_runtime_environment(runtime_env_dir) @@ -144,7 +174,7 @@ def on_exit() -> None: try: # Start task heartbeat - heartbeat_sender = HeartbeatSender(make_task_heartbeat_fn_grpc(stub)) + heartbeat_sender = HeartbeatSender(heartbeat_fn) heartbeat_sender.start() # Pull Message, Context, Run and FAB from SuperNode @@ -234,7 +264,9 @@ def on_exit() -> None: ) -def pull_task_input(stub: RuntimeStub) -> tuple[Message, Context, Run, Fab]: +def pull_task_input( + stub: RuntimeStub | RuntimeHttpStub, +) -> tuple[Message, Context, Run, Fab]: """Pull TaskInput from SuperNode.""" # Pull Context, Run and FAB res: PullTaskInputResponse = stub.PullTaskInput(PullTaskInputRequest()) @@ -264,7 +296,9 @@ def pull_task_input(stub: RuntimeStub) -> tuple[Message, Context, Run, Fab]: return message, context, run, fab -def push_message(stub: RuntimeStub, message: Message, context: Context) -> None: +def push_message( + stub: RuntimeStub | RuntimeHttpStub, message: Message, context: Context +) -> None: """Push reply message to SuperNode.""" # Set message ID message.metadata.__dict__["_message_id"] = message.object_id @@ -302,7 +336,7 @@ def push_message(stub: RuntimeStub, message: Message, context: Context) -> None: def push_task_output( # pylint: disable=R0913, R0917 - stub: RuntimeStub, + stub: RuntimeStub | RuntimeHttpStub, context: Context | None, sub_status: str, details: str, @@ -317,5 +351,5 @@ def push_task_output( # pylint: disable=R0913, R0917 details=details, ) ) - except grpc.RpcError as err: + except (grpc.RpcError, httpx.HTTPError) as err: log(ERROR, "Failed to push task output: %s", str(err)) diff --git a/framework/py/flwr/supernode/runtime/run_clientapp_test.py b/framework/py/flwr/supernode/runtime/run_clientapp_test.py index 2e8f1219d00e..e98090f3b784 100644 --- a/framework/py/flwr/supernode/runtime/run_clientapp_test.py +++ b/framework/py/flwr/supernode/runtime/run_clientapp_test.py @@ -31,7 +31,9 @@ from flwr.supercore.fab import Fab from flwr.supercore.interceptors import ( RuntimeTokenClientInterceptor, + RuntimeTokenHttpInterceptor, RuntimeVersionClientInterceptor, + RuntimeVersionHttpInterceptor, ) from .run_clientapp import pull_task_input, run_clientapp @@ -74,6 +76,26 @@ def test_run_clientapp_adds_client_interceptors(self) -> None: # pylint: disable-next=protected-access self.assertEqual(interceptors[0]._metadata.component_name, "flwr-clientapp") + def test_run_clientapp_adds_http_interceptors(self) -> None: + """`run_clientapp` should configure its HTTP Runtime stub.""" + with patch( + "flwr.supernode.runtime.run_clientapp.RuntimeHttpStub", + side_effect=RuntimeError, + ) as runtime_http_stub: + with self.assertRaises(RuntimeError): + run_clientapp( + "127.0.0.1:8000", + insecure=True, + token="test-token", + enable_http_api=True, + ) + + self.assertEqual(runtime_http_stub.call_args.args, ("http://127.0.0.1:8000",)) + self.assertIs(runtime_http_stub.call_args.kwargs["verify"], False) + interceptors = runtime_http_stub.call_args.kwargs["interceptors"] + self.assertIsInstance(interceptors[0], RuntimeVersionHttpInterceptor) + self.assertIsInstance(interceptors[1], RuntimeTokenHttpInterceptor) + def test_run_clientapp_exits_nonzero_on_grpc_error(self) -> None: """`run_clientapp` should not report success after Runtime API failures.""" with ( diff --git a/framework/py/flwr/supernode/start_client_internal.py b/framework/py/flwr/supernode/start_client_internal.py index 1bdd365bfd76..a52387591699 100644 --- a/framework/py/flwr/supernode/start_client_internal.py +++ b/framework/py/flwr/supernode/start_client_internal.py @@ -107,6 +107,7 @@ def start_client_internal( trusted_entities: dict[str, str] | None = None, superexec_auth_secret: bytes | None = None, runtime_dependency_install: bool = RUNTIME_DEPENDENCY_INSTALL, + runtime_http_api_address: str | None = None, ) -> None: """Start a Flower client node which connects to a Flower server. @@ -171,6 +172,8 @@ def start_client_internal( Secret used by Runtime API SuperExec metadata auth. runtime_dependency_install : bool (default: False) Whether runtime dependency installation is allowed. + runtime_http_api_address : Optional[str] (default: None) + Runtime HTTP API address. When provided, SuperExec uses HTTP instead of gRPC. """ if insecure is None: insecure = root_certificates is None @@ -230,8 +233,14 @@ def start_client_internal( if isolation == ISOLATION_MODE_SUBPROCESS: # `bound_address` contains the actual address when the port is set to :0 # which means let the OS choose a free port. - runtime_address = resolve_bind_address(runtime_server.bound_address) + runtime_address = ( + runtime_http_api_address + if runtime_http_api_address is not None + else resolve_bind_address(runtime_server.bound_address) + ) command = ["flower-superexec"] + if runtime_http_api_address is not None: + command += ["--enable-http-api"] command += get_client_tls_args( insecure=runtime_certificates is None, root_certificates_path=runtime_root_certificates_path, diff --git a/framework/py/flwr/supernode/start_client_internal_test.py b/framework/py/flwr/supernode/start_client_internal_test.py index 088c2c22168b..10ce7267845a 100644 --- a/framework/py/flwr/supernode/start_client_internal_test.py +++ b/framework/py/flwr/supernode/start_client_internal_test.py @@ -416,6 +416,7 @@ def _run_until_connection_start( runtime_root_certificates_path: str | None = None, runtime_api_address: str = "127.0.0.1:9094", bound_address: str = "127.0.0.1:9094", + runtime_http_api_address: str | None = None, ) -> tuple[Mock, Mock]: """Run startup only far enough to inspect Runtime API and SuperExec wiring.""" objectstore_factory = Mock() @@ -445,6 +446,7 @@ def _run_until_connection_start( runtime_api_address=runtime_api_address, runtime_certificates=runtime_certificates, runtime_root_certificates_path=runtime_root_certificates_path, + runtime_http_api_address=runtime_http_api_address, ) assert run_runtime.call_args.kwargs["state_factory"] is state_factory @@ -497,3 +499,12 @@ def test_start_client_internal_launches_superexec_with_bound_runtime_address() - command = popen.call_args.args[0] assert command[command.index("--appio-api-address") + 1] == "localhost:54321" + + +def test_start_client_internal_launches_http_superexec() -> None: + """Subprocess SuperExec should use the Runtime HTTP endpoint when enabled.""" + _, popen = _run_until_connection_start(runtime_http_api_address="127.0.0.1:8000") + + command = popen.call_args.args[0] + assert "--enable-http-api" in command + assert command[command.index("--appio-api-address") + 1] == "127.0.0.1:8000"