diff --git a/docs/user_guide/data_scientist_guide/recipe_api.rst b/docs/user_guide/data_scientist_guide/recipe_api.rst index 652c9b4bb8..69c1f2fc2f 100644 --- a/docs/user_guide/data_scientist_guide/recipe_api.rst +++ b/docs/user_guide/data_scientist_guide/recipe_api.rst @@ -98,6 +98,18 @@ File packaging helpers: ``recipe.add_server_file(file_path)`` Bundle a file or directory into the generated server app package. +Helpers that accept ``clients`` target specific generated client apps. This +requires per-site client apps: construct the recipe with the +``per_site_config`` constructor argument on recipes that support it, and each +name in ``clients`` must match an existing per-site client app. With the +default all-clients topology, targeted calls raise an error rather than +silently dropping the change from the generated job, and unknown site names +raise an error rather than deploying a bare app to that site. Calling +``set_per_site_config`` after construction records the configuration for +``configured_sites()`` but does not yet rebuild an existing all-clients app +into per-site apps; recipes will interpret helper-provided per-site config as +follow-up work. + Filter helpers: ``recipe.add_client_input_filter(filter, tasks=None, clients=None)`` @@ -123,9 +135,11 @@ Shared helpers: Utility helpers: -``add_experiment_tracking(recipe, tracking_type, tracking_config=None, client_side=False, server_side=True)`` +``add_experiment_tracking(recipe, tracking_type, tracking_config=None, client_side=False, server_side=True, clients=None)`` Add supported experiment tracking receivers such as TensorBoard, MLflow, or - Weights & Biases. + Weights & Biases. With ``client_side=True``, ``clients`` limits which sites + receive the client-side receiver; call once per site with different + ``tracking_config`` values for per-site tracking destinations. ``add_cross_site_evaluation(recipe, submit_model_timeout=600, validation_timeout=6000, participating_clients=None)`` Add cross-site evaluation to a training recipe when the recipe/framework diff --git a/nvflare/recipe/spec.py b/nvflare/recipe/spec.py index abd62e7313..63e9006e88 100644 --- a/nvflare/recipe/spec.py +++ b/nvflare/recipe/spec.py @@ -314,26 +314,63 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs obj: Object to add to clients. clients: Optional list of specific client names. If None, applies to all clients. **kwargs: Extra options forwarded to `job.to()`/`job.to_clients()`. + + Raises: + TypeError: If clients is not a list. + ValueError: If clients is empty or contains a non-client name; if specific + clients are targeted while the recipe's client app applies to all clients + (per-site placement cannot be expressed in the generated job in that + topology); or if clients names a site with no existing client app while + per-site client apps exist (that would deploy a bare, executor-less app + to that site). """ + from nvflare.apis.job_def import ALL_SITES, SERVER_SITE_NAME + from nvflare.job_config.defs import JobTargetType + + # FedJob has no public API to list per-site deploy targets, so we inspect + # private deploy map to preserve existing per-site client topology. + deploy_map = getattr(self.job, "_deploy_map", {}) + existing_client_sites = [ + target + for target in deploy_map.keys() + if target not in [ALL_SITES, SERVER_SITE_NAME] + and JobTargetType.get_target_type(target) == JobTargetType.CLIENT + ] if clients is None: - from nvflare.apis.job_def import ALL_SITES, SERVER_SITE_NAME - from nvflare.job_config.defs import JobTargetType - - # FedJob has no public API to list per-site deploy targets, so we inspect - # private deploy map to preserve existing per-site client topology. - deploy_map = getattr(self.job, "_deploy_map", {}) - existing_client_sites = [ - target - for target in deploy_map.keys() - if target not in [ALL_SITES, SERVER_SITE_NAME] - and JobTargetType.get_target_type(target) == JobTargetType.CLIENT - ] if existing_client_sites: for site in existing_client_sites: self.job.to(obj, site, **kwargs) else: self.job.to_clients(obj, **kwargs) else: + # A bare string would iterate per character; an empty list would be a silent no-op. + if not isinstance(clients, list): + raise TypeError(f"clients must be a list of client names, got {type(clients).__name__}") + if not clients: + raise ValueError("clients must not be empty; omit it to apply to all clients") + for client in clients: + if not isinstance(client, str) or client in (ALL_SITES, SERVER_SITE_NAME): + raise ValueError(f"invalid client name {client!r}: client names must name specific client sites") + if ALL_SITES in deploy_map: + # The generated job has one client app deployed to all clients. Exporting + # both an all-clients app and per-site apps is not expressible (the + # all-clients app wins and per-site apps are dropped), so fail loudly + # instead of silently losing the placement. + raise ValueError( + "cannot target specific clients: this recipe's client app applies to all clients. " + "Construct the recipe with per-site client apps (e.g. the per_site_config constructor " + "argument on recipes that support it) or omit clients to apply to all clients." + ) + if existing_client_sites: + # Targeting a site with no app would create a bare, executor-less app for + # that site in the exported job — the same class of quiet misconfiguration + # as the ALL_SITES case above, so fail loudly instead. + unknown = [c for c in clients if c not in existing_client_sites] + if unknown: + raise ValueError( + f"unknown client site(s) {unknown}: this recipe has per-site client apps " + f"only for {sorted(existing_client_sites)}" + ) for client in clients: self.job.to(obj, client, **kwargs) diff --git a/nvflare/recipe/utils.py b/nvflare/recipe/utils.py index 8eaeff97d8..763133b919 100644 --- a/nvflare/recipe/utils.py +++ b/nvflare/recipe/utils.py @@ -214,6 +214,7 @@ def add_experiment_tracking( tracking_config: Optional[dict] = None, client_side: bool = False, server_side: bool = True, + clients: Optional[List[str]] = None, ): """Add experiment tracking to a recipe. @@ -223,8 +224,16 @@ def add_experiment_tracking( recipe: Recipe instance to augment with experiment tracking. tracking_type: Type of tracking to enable ("mlflow", "tensorboard", or "wandb"). tracking_config: Optional configuration dict for the tracking receiver. - client_side: If True, add tracking to all clients (each client tracks locally). + client_side: If True, add tracking to clients (each client tracks locally). server_side: If True, add tracking to server (aggregates metrics from all clients). Default: True. + clients: Optional list of client names for client-side tracking. If None, the + client-side receiver is added to all clients. Only valid with client_side=True. + To give sites different receiver configs (e.g. per-site tracking_uri), call this + function once per site with that site's tracking_config and clients=[site]. + Targeting specific clients requires the recipe's client apps to be per-site + (e.g. recipes constructed with the per_site_config constructor argument), and + each name must match an existing per-site client app; with the default + all-clients topology or unknown site names, targeted placement raises ValueError. Examples: # Server-side tracking (default - federated metrics) @@ -235,6 +244,16 @@ def add_experiment_tracking( # Both server and client tracking add_experiment_tracking(recipe, "mlflow", {...}, client_side=True, server_side=True) + + # Per-site client tracking configs (one call per site) + add_experiment_tracking( + recipe, "mlflow", {"tracking_uri": "file:///tmp/site-1/mlruns"}, + client_side=True, server_side=False, clients=["site-1"], + ) + add_experiment_tracking( + recipe, "mlflow", {"tracking_uri": "file:///tmp/site-2/mlruns"}, + client_side=True, server_side=False, clients=["site-2"], + ) """ tracking_config = tracking_config or {} if tracking_type not in TRACKING_REGISTRY: @@ -243,6 +262,14 @@ def add_experiment_tracking( if not server_side and not client_side: raise ValueError("At least one of server_side or client_side must be True") + if clients is not None: + if not client_side: + raise ValueError("clients is only used for client-side tracking; set client_side=True") + if not isinstance(clients, list) or not all(isinstance(c, str) for c in clients): + raise TypeError(f"clients must be a list of str, got {clients!r}") + if not clients: + raise ValueError("clients must not be empty; omit it to add tracking to all clients") + _, flag = optional_import(TRACKING_REGISTRY[tracking_type]["package"]) if not flag: raise ValueError( @@ -267,7 +294,9 @@ def add_experiment_tracking( client_config["events"] = [ANALYTIC_EVENT_TYPE] client_receiver = receiver_class(**client_config) - recipe.job.to_clients(client_receiver, id="client_receiver") + # Route through the recipe placement layer so existing per-site client apps + # are preserved (to_clients would target ALL_SITES even when per-site apps exist). + recipe._add_to_client_apps(client_receiver, clients=clients, id="client_receiver") def add_cross_site_evaluation( diff --git a/tests/unit_test/recipe/spec_test.py b/tests/unit_test/recipe/spec_test.py index d14f1ed52d..d8c1730858 100644 --- a/tests/unit_test/recipe/spec_test.py +++ b/tests/unit_test/recipe/spec_test.py @@ -756,3 +756,69 @@ def __init__(self): with pytest.raises(TypeError, match=match): set_per_site_config(BasicRecipe(), config) + + +class TestClientPlacementHardening: + """Test _add_to_client_apps validation through the public clients=-taking helpers. + + These guards fix pre-existing silent failures: targeting specific clients while + the job has an all-clients app used to silently drop the placement at export, + and malformed clients values were silently ignored or iterated per character. + """ + + def _make_recipe(self, name="test_placement_hardening"): + from nvflare.recipe.spec import Recipe + + return Recipe(FedJob(name=name, min_clients=1)) + + def test_targeted_config_rejects_all_clients_topology(self): + recipe = self._make_recipe() + # Default recipe topology: one client app for all clients. + recipe.job.to_clients({"executor_standin": True}) + + with pytest.raises(ValueError, match="applies to all clients"): + recipe.add_client_config({"timeout": 600}, clients=["site-1"]) + + def test_targeted_file_rejects_all_clients_topology(self, tmp_path): + src_file = tmp_path / "wrapper.sh" + src_file.write_text("#!/bin/sh\n") + recipe = self._make_recipe() + recipe.job.to_clients({"executor_standin": True}) + + with pytest.raises(ValueError, match="applies to all clients"): + recipe.add_client_file(str(src_file), clients=["site-1"]) + + def test_targeted_config_works_with_per_site_topology(self): + recipe = self._make_recipe() + recipe.job.to({"site_arg": 1}, "site-1") + recipe.job.to({"site_arg": 2}, "site-2") + + recipe.add_client_config({"timeout": 600}, clients=["site-1"]) + + assert recipe.job._deploy_map["site-1"].app_config.additional_params["timeout"] == 600 + assert "timeout" not in recipe.job._deploy_map["site-2"].app_config.additional_params + + def test_targeted_config_rejects_unknown_site_with_per_site_topology(self): + recipe = self._make_recipe() + recipe.job.to({"site_arg": 1}, "site-1") + recipe.job.to({"site_arg": 2}, "site-2") + + with pytest.raises(ValueError, match=r"unknown client site.*site-3"): + recipe.add_client_config({"timeout": 600}, clients=["site-3"]) + + def test_clients_must_be_a_list(self): + recipe = self._make_recipe() + # A bare string would otherwise iterate per character and create per-char apps. + with pytest.raises(TypeError, match="must be a list"): + recipe.add_client_config({"timeout": 600}, clients="site-1") + + def test_clients_must_not_be_empty(self): + recipe = self._make_recipe() + with pytest.raises(ValueError, match="must not be empty"): + recipe.add_client_config({"timeout": 600}, clients=[]) + + @pytest.mark.parametrize("bad_site", ["server", "@ALL"]) + def test_clients_must_name_client_sites(self, bad_site): + recipe = self._make_recipe() + with pytest.raises(ValueError, match="invalid client name"): + recipe.add_client_config({"timeout": 600}, clients=[bad_site]) diff --git a/tests/unit_test/recipe/utils_test.py b/tests/unit_test/recipe/utils_test.py index 883a29b89a..73ec1f8069 100644 --- a/tests/unit_test/recipe/utils_test.py +++ b/tests/unit_test/recipe/utils_test.py @@ -523,3 +523,167 @@ def __init__(self, *args, **kwargs): add_cross_site_evaluation(recipe, participating_clients=participating_clients) assert captured_kwargs["participating_clients"] == participating_clients + + +class TestAddExperimentTrackingClients: + """Test client targeting and per-site configs in add_experiment_tracking.""" + + @pytest.fixture + def dummy_tracking(self, monkeypatch): + """Register a dependency-free tracking type backed by types.SimpleNamespace.""" + import nvflare.recipe.utils as utils_mod + + monkeypatch.setitem( + utils_mod.TRACKING_REGISTRY, + "dummy", + # json is always importable; argparse.Namespace(**config) acts as the receiver + # (unlike types.SimpleNamespace it exposes __module__, so it also survives export). + {"package": "json", "receiver_module": "argparse", "receiver_class": "Namespace"}, + ) + return "dummy" + + def _make_recipe(self, name="test_tracking_clients"): + return Recipe(FedJob(name=name, min_clients=1)) + + def test_client_side_tracking_specific_clients(self, dummy_tracking): + from nvflare.apis.analytix import ANALYTIC_EVENT_TYPE + from nvflare.apis.job_def import ALL_SITES + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + add_experiment_tracking( + recipe, + dummy_tracking, + {"tracking_uri": "file:///tmp/site-1/mlruns"}, + client_side=True, + server_side=False, + clients=["site-1"], + ) + + receiver = recipe.job._deploy_map["site-1"].app_config.components["client_receiver"] + assert receiver.tracking_uri == "file:///tmp/site-1/mlruns" + # Local (non-federated) analytics events are configured by default. + assert receiver.events == [ANALYTIC_EVENT_TYPE] + assert ALL_SITES not in recipe.job._deploy_map + + def test_client_side_tracking_per_site_configs(self, dummy_tracking): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + # Real recipes have per-site client apps (e.g. from per_site_config) before + # tracking is added; targeting only existing sites is enforced. + for site in ("site-1", "site-2"): + recipe.job.to({"site_arg": site}, site) + for site in ("site-1", "site-2"): + add_experiment_tracking( + recipe, + dummy_tracking, + {"tracking_uri": f"file:///tmp/{site}/mlruns"}, + client_side=True, + server_side=False, + clients=[site], + ) + + for site in ("site-1", "site-2"): + receiver = recipe.job._deploy_map[site].app_config.components["client_receiver"] + assert receiver.tracking_uri == f"file:///tmp/{site}/mlruns" + + def test_client_side_tracking_all_clients_by_default(self, dummy_tracking): + from nvflare.apis.job_def import ALL_SITES + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + add_experiment_tracking(recipe, dummy_tracking, {"tracking_uri": "u"}, client_side=True, server_side=False) + + receiver = recipe.job._deploy_map[ALL_SITES].app_config.components["client_receiver"] + assert receiver.tracking_uri == "u" + + def test_server_side_tracking_unaffected_by_clients_feature(self, dummy_tracking): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + add_experiment_tracking(recipe, dummy_tracking, {"tracking_uri": "u"}) + + components = recipe.job._deploy_map["server"].app_config.components + assert components["receiver"].tracking_uri == "u" + # Server receiver keeps federated (default) events. + assert not hasattr(components["receiver"], "events") + + def test_clients_without_client_side_raises(self, dummy_tracking): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + with pytest.raises(ValueError, match="client_side=True"): + add_experiment_tracking(recipe, dummy_tracking, {"tracking_uri": "u"}, clients=["site-1"]) + + @pytest.mark.parametrize("bad_clients", ["site-1", [1, 2], [None]]) + def test_clients_must_be_list_of_str(self, dummy_tracking, bad_clients): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + with pytest.raises(TypeError, match="clients must be a list of str"): + add_experiment_tracking( + recipe, dummy_tracking, {"tracking_uri": "u"}, client_side=True, clients=bad_clients + ) + + def test_clients_empty_list_raises(self, dummy_tracking): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + with pytest.raises(ValueError, match="must not be empty"): + add_experiment_tracking(recipe, dummy_tracking, {"tracking_uri": "u"}, client_side=True, clients=[]) + + def test_clients_targeting_rejects_unknown_site(self, dummy_tracking): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + recipe.job.to({"site_arg": 1}, "site-1") + recipe.job.to({"site_arg": 2}, "site-2") + + with pytest.raises(ValueError, match="unknown client site"): + add_experiment_tracking( + recipe, + dummy_tracking, + {"tracking_uri": "u"}, + client_side=True, + server_side=False, + clients=["site-3"], + ) + + def test_clients_targeting_rejects_all_sites_topology(self, dummy_tracking): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe() + # Default recipe topology: one client app for all clients. + recipe.job.to_clients({"executor_standin": True}) + + with pytest.raises(ValueError, match="applies to all clients"): + add_experiment_tracking( + recipe, + dummy_tracking, + {"tracking_uri": "u"}, + client_side=True, + server_side=False, + clients=["site-1"], + ) + + def test_per_site_client_receiver_survives_export(self, dummy_tracking, tmp_path): + from nvflare.recipe.utils import add_experiment_tracking + + recipe = self._make_recipe("test_tracking_export") + recipe.job.to_server({"server_arg": True}) + add_experiment_tracking( + recipe, + dummy_tracking, + {"tracking_uri": "file:///tmp/site-1/mlruns"}, + client_side=True, + server_side=False, + clients=["site-1"], + ) + + recipe.job.export_job(str(tmp_path)) + with open(tmp_path / "test_tracking_export" / "app_site-1" / "config" / "config_fed_client.json") as f: + client_cfg = json.load(f) + + entry = next(c for c in client_cfg["components"] if c["id"] == "client_receiver") + assert entry["path"].endswith("Namespace")