Source code for nvflare.tool.deploy.k8s_deploy

# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Kubernetes-specific validation and preparation for ``nvflare deploy``."""

from __future__ import annotations

import hashlib
import re
import shutil
from pathlib import Path
from typing import Any

import yaml

from nvflare.tool.deploy.deploy_common import (
    COMM_CONFIG_JSON,
    FED_CLIENT_JSON,
    FED_SERVER_JSON,
    K8S_CLIENT_LAUNCHER,
    K8S_SERVER_LAUNCHER,
    RESOURCES_JSON_DEFAULT,
    ROLE_SERVER,
    RUNTIME_K8S,
    STUDY_DATA_YAML,
    WORKSPACE_MOUNT_PATH,
    KitInfo,
    _ensure_study_runtime_template,
    _fail,
    _internal_resources,
    _load_or_default_comm_config,
    _mapping,
    _optional_mapping,
    _optional_str,
    _patch_resources,
    _relocate_server_storage_to_workspace,
    _remove_start_scripts,
    _required_str,
    _validate_allowed_keys,
    _write_json,
)

HELM_CHART_DIR = "helm_chart"
WORKSPACE_VOLUME_NAME = "workspace"
K8S_PARENT_PYTHON_PATH = "/usr/local/bin/python3"
HELM_TEMPLATES_DIR = Path(__file__).resolve().parent / "templates" / "helm"
K8S_LAUNCHER_PATHS = {K8S_CLIENT_LAUNCHER, K8S_SERVER_LAUNCHER}
K8S_NAME_PATTERN = re.compile(r"^[a-z]([-a-z0-9]*[a-z0-9])?$")
K8S_NAMESPACE_PATTERN = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$")
K8S_SECRET_NAME_PATTERN = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$")
K8S_NAMESPACE_MAX_LENGTH = 63
K8S_SERVICE_NAME_MAX_LENGTH = 63
K8S_SECRET_NAME_MAX_LENGTH = 253
HELM_RELEASE_NAME_MAX_LENGTH = 53
DEFAULT_K8S_SERVER_SERVICE_NAME = "nvflare-server"


[docs] def validate_config(config: dict[str, Any]) -> None: _validate_allowed_keys( config, {"runtime", "namespace", "server_service_name", "parent", "job_launcher"}, "k8s config", ) parent = _mapping(config.get("parent"), "parent") job_launcher = _mapping(config.get("job_launcher", {}), "job_launcher") _validate_allowed_keys( parent, { "docker_image", "parent_port", "workspace_pvc", "workspace_mount_path", "python_path", "resources", "pod_security_context", "image_pull_secrets", }, "parent", ) _validate_allowed_keys( job_launcher, { "config_file_path", "pending_timeout", "default_python_path", "job_pod_security_context", "image_pull_secrets", }, "job_launcher", ) _required_str(parent, "docker_image", "parent") if "namespace" in config: _validate_k8s_namespace(config, "namespace", "k8s config") if "server_service_name" in config: _validate_k8s_service_name(config, "server_service_name", "k8s config") _optional_int(parent, "parent_port", "parent") _optional_str(parent, "workspace_pvc", "parent") _optional_str(parent, "workspace_mount_path", "parent") _optional_str(parent, "python_path", "parent") _optional_mapping(parent, "resources", "parent") _optional_mapping(parent, "pod_security_context", "parent") _optional_k8s_secret_name_list(parent, "image_pull_secrets", "parent image pull references") _optional_str(job_launcher, "config_file_path", "job_launcher") _optional_str(job_launcher, "default_python_path", "job_launcher") _optional_non_negative_int(job_launcher, "pending_timeout", "job_launcher") _optional_mapping(job_launcher, "job_pod_security_context", "job_launcher") _optional_k8s_secret_name_list(job_launcher, "image_pull_secrets", "job launcher image pull references")
[docs] def prepare(kit_info: KitInfo, final_output: Path, config: dict[str, Any]) -> dict[str, Any]: namespace = config.get("namespace", "default") parent = config.get("parent") or {} job_launcher = config.get("job_launcher") or {} parent_port = parent.get("parent_port", 8102) workspace_mount_path = parent.get("workspace_mount_path", WORKSPACE_MOUNT_PATH) server_service_name = config.get("server_service_name", DEFAULT_K8S_SERVER_SERVICE_NAME) launcher_path = K8S_SERVER_LAUNCHER if kit_info.role == ROLE_SERVER else K8S_CLIENT_LAUNCHER launcher_args = { "config_file_path": job_launcher.get("config_file_path"), "namespace": namespace, "default_python_path": job_launcher.get("default_python_path", K8S_PARENT_PYTHON_PATH), "workspace_mount_path": workspace_mount_path, } if (kit_info.kit_dir / "local" / STUDY_DATA_YAML).exists(): # legacy v1 kit: keep the study data mounts working; new kits use study_runtime.yaml launcher_args["study_data_pvc_file_path"] = f"{workspace_mount_path}/local/{STUDY_DATA_YAML}" if "pending_timeout" in job_launcher: launcher_args["pending_timeout"] = job_launcher["pending_timeout"] if job_launcher.get("job_pod_security_context"): launcher_args["security_context"] = job_launcher["job_pod_security_context"] if job_launcher.get("image_pull_secrets") is not None: launcher_args["image_pull_secrets"] = job_launcher["image_pull_secrets"] _patch_resources(kit_info.kit_dir, "k8s_launcher", launcher_path, launcher_args) _patch_comm_config_for_k8s(kit_info.kit_dir, kit_info.role, kit_info.name, parent_port, server_service_name) _ensure_study_runtime_template(kit_info.kit_dir) if kit_info.role == ROLE_SERVER: _relocate_server_storage_to_workspace(kit_info.kit_dir, workspace_mount_path) _remove_start_scripts(kit_info.kit_dir, keep=set()) _write_helm_chart(kit_info, config) release_name = _k8s_release_name(kit_info.name) final_chart_dir = final_output / HELM_CHART_DIR return { "runtime": RUNTIME_K8S, "role": kit_info.role, "name": kit_info.name, "namespace": namespace, "output": str(final_output), "helm_chart": str(final_chart_dir), "helm_command": f"helm upgrade --install {release_name} {final_chart_dir} --namespace {namespace}", "resources": str(final_output / "local" / RESOURCES_JSON_DEFAULT), }
def _patch_comm_config_for_k8s( kit_dir: Path, role: str, site_name: str, parent_port: int, server_service_name: str = DEFAULT_K8S_SERVER_SERVICE_NAME, ) -> None: comm_config_path = kit_dir / "local" / COMM_CONFIG_JSON comm_config = _load_or_default_comm_config(comm_config_path) resources = _internal_resources(comm_config) resources.update( { "host": _k8s_parent_service_name(role, site_name, server_service_name), "port": parent_port, "connection_security": "clear", } ) internal = comm_config.setdefault("internal", {}) internal["scheme"] = "tcp" _write_json(comm_config_path, comm_config) def _k8s_parent_service_name( role: str, site_name: str, server_service_name: str = DEFAULT_K8S_SERVER_SERVICE_NAME ) -> str: if role == ROLE_SERVER: return server_service_name return _k8s_service_name(site_name) def _k8s_service_name(site_name: str) -> str: return _stable_k8s_name(site_name, K8S_SERVICE_NAME_MAX_LENGTH) def _stable_k8s_name(raw_name: str, max_length: int) -> str: if len(raw_name) <= max_length and K8S_NAME_PATTERN.match(raw_name): return raw_name normalized = re.sub(r"[^a-z0-9-]", "-", raw_name.lower()) normalized = re.sub(r"-+", "-", normalized).strip("-") if not normalized: normalized = "site" if not normalized[0].isalpha(): normalized = f"site-{normalized}" digest = hashlib.sha256(raw_name.encode("utf-8")).hexdigest()[:8] head_max = max_length - len(digest) - 1 head = normalized[:head_max].rstrip("-") or "site" return f"{head}-{digest}" def _write_helm_chart(kit_info: KitInfo, config: dict[str, Any]) -> Path: parent = config.get("parent") or {} docker_image = parent["docker_image"] parent_port = parent.get("parent_port", 8102) workspace_pvc = parent.get("workspace_pvc", "nvflws") workspace_mount_path = parent.get("workspace_mount_path", WORKSPACE_MOUNT_PATH) parent_python_path = parent.get("python_path") or K8S_PARENT_PYTHON_PATH server_service_name = config.get("server_service_name", DEFAULT_K8S_SERVER_SERVICE_NAME) chart_dir = kit_info.kit_dir / HELM_CHART_DIR if chart_dir.exists(): shutil.rmtree(chart_dir) templates_dir = chart_dir / "templates" templates_dir.mkdir(parents=True) if kit_info.role == ROLE_SERVER: _write_server_helm_chart( kit_info, chart_dir, docker_image, parent_port, workspace_pvc, workspace_mount_path, parent_python_path, parent, server_service_name, ) _copy_helm_templates("server", templates_dir) else: _write_client_helm_chart( kit_info, chart_dir, docker_image, parent_port, workspace_pvc, workspace_mount_path, parent_python_path, parent, ) _copy_helm_templates("client", templates_dir) return chart_dir def _write_server_helm_chart( kit_info: KitInfo, chart_dir: Path, docker_image: str, parent_port: int, workspace_pvc: str, workspace_mount_path: str, parent_python_path: str, parent: dict[str, Any], server_service_name: str, ) -> None: repo, tag = _split_image(docker_image) _write_yaml( chart_dir / "Chart.yaml", { "apiVersion": "v2", "name": "nvflare-server", "description": f"NVFlare federated learning server for {kit_info.name}", "type": "application", "version": "0.1.0", "appVersion": tag or "latest", "keywords": ["nvflare", "federated-learning"], "maintainers": [], }, ) fed_learn_port = kit_info.fed_learn_port or 8002 admin_port = kit_info.admin_port if kit_info.admin_port != fed_learn_port else None values = { "name": kit_info.name, "serviceName": server_service_name, "image": {"repository": repo, "tag": tag, "pullPolicy": "IfNotPresent"}, "imagePullSecrets": _image_pull_secret_refs(parent), "serviceAccount": {"create": True, "annotations": {}, "automountServiceAccountToken": True}, "podAnnotations": {}, "rbac": {"create": True}, "persistence": { "workspace": { "claimName": workspace_pvc, "volumeName": WORKSPACE_VOLUME_NAME, "mountPath": workspace_mount_path, } }, "workspaceConfig": { "namespace": None, "local": {"configMapName": None, "items": []}, "startup": {"secretName": None, "items": []}, }, "fedLearnPort": fed_learn_port, "adminPort": admin_port, "parentPort": parent_port, "resources": parent.get("resources") or {"requests": {"cpu": "2", "memory": "8Gi"}}, "securityContext": parent.get("pod_security_context") or {}, "hostPortEnabled": False, "tcpConfigMapEnabled": False, "service": {"type": "ClusterIP", "loadBalancerIP": None, "annotations": {}}, "command": [parent_python_path], "args": [ "-u", "-m", "nvflare.private.fed.app.server.server_train", "-m", workspace_mount_path, "-s", FED_SERVER_JSON, "--set", "secure_train=true", "config_folder=config", f"org={kit_info.org}", ], } _write_yaml(chart_dir / "values.yaml", values) def _write_client_helm_chart( kit_info: KitInfo, chart_dir: Path, docker_image: str, parent_port: int, workspace_pvc: str, workspace_mount_path: str, parent_python_path: str, parent: dict[str, Any], ) -> None: repo, tag = _split_image(docker_image) _write_yaml( chart_dir / "Chart.yaml", { "apiVersion": "v2", "name": "nvflare-client", "description": f"NVFlare federated learning client deployment and service for {kit_info.name}", "type": "application", "version": "0.1.0", "appVersion": tag or "latest", "keywords": ["nvflare", "federated-learning"], "maintainers": [], }, ) values = { "name": kit_info.name, "siteName": kit_info.name, "serviceName": _k8s_parent_service_name(kit_info.role, kit_info.name), "image": {"repository": repo, "tag": tag, "pullPolicy": "Always"}, "imagePullSecrets": _image_pull_secret_refs(parent), "serviceAccount": {"create": True, "annotations": {}, "automountServiceAccountToken": True}, "podAnnotations": {}, "rbac": {"create": True}, "persistence": { "workspace": { "claimName": workspace_pvc, "volumeName": WORKSPACE_VOLUME_NAME, "mountPath": workspace_mount_path, } }, "workspaceConfig": { "namespace": None, "local": {"configMapName": None, "items": []}, "startup": {"secretName": None, "items": []}, }, "port": parent_port, "service": {"annotations": {}}, "securityContext": parent.get("pod_security_context") or {}, "resources": parent.get("resources") or {"requests": {"cpu": "2", "memory": "8Gi"}}, "command": [parent_python_path], "args": [ "-u", "-m", "nvflare.private.fed.app.client.client_train", "-m", workspace_mount_path, "-s", FED_CLIENT_JSON, "--set", "secure_train=true", "config_folder=config", f"org={kit_info.org}", ], } _write_yaml(chart_dir / "values.yaml", values) def _copy_helm_templates(role: str, templates_dir: Path) -> None: files = { "server": [ ("_helpers.tpl", "_helpers.tpl"), ("deployment.yaml", "server-deployment.yaml"), ("service.yaml", "server-service.yaml"), ("tcp-services.yaml", "server-tcp-services.yaml"), ("serviceaccount.yaml", "serviceaccount.yaml"), ("role.yaml", "role.yaml"), ], "client": [ ("_helpers.tpl", "_helpers.tpl"), ("deployment.yaml", "client-deployment.yaml"), ("service.yaml", "service.yaml"), ("serviceaccount.yaml", "serviceaccount.yaml"), ("role.yaml", "role.yaml"), ], } for src_name, dst_name in files[role]: shutil.copy(_helm_src(role, src_name), templates_dir / dst_name) def _split_image(docker_image: str) -> tuple[str, str]: if ":" in docker_image: return docker_image.rsplit(":", 1) return docker_image, "" def _image_pull_secret_refs(parent: dict[str, Any]) -> list[dict[str, str]]: return [{"name": name} for name in parent.get("image_pull_secrets") or []] def _helm_src(role: str, filename: str) -> Path: return HELM_TEMPLATES_DIR / role / filename def _k8s_release_name(name: str) -> str: return _stable_k8s_name(name, HELM_RELEASE_NAME_MAX_LENGTH) def _validate_k8s_namespace( data: dict[str, Any], key: str, where: str, error_code: str = "INVALID_CONFIG", hint: str = "Fix the runtime config.", ) -> None: namespace = data.get(key) if not isinstance(namespace, str) or not namespace: _fail(error_code, f"{where}.{key} must be a non-empty string.", hint) return if len(namespace) > K8S_NAMESPACE_MAX_LENGTH or not K8S_NAMESPACE_PATTERN.fullmatch(namespace): _fail( error_code, f"{where}.{key} must be a valid Kubernetes namespace (DNS-1123 label): {namespace!r}.", "Use lower case alphanumeric characters or '-', start and end with an alphanumeric character, " f"and keep length <= {K8S_NAMESPACE_MAX_LENGTH}.", ) def _validate_k8s_service_name(data: dict[str, Any], key: str, where: str) -> None: service_name = data.get(key) if not isinstance(service_name, str) or not service_name: _fail("INVALID_CONFIG", f"{where}.{key} must be a non-empty string.", "Fix the runtime config.") if len(service_name) > K8S_SERVICE_NAME_MAX_LENGTH or not K8S_NAME_PATTERN.fullmatch(service_name): _fail( "INVALID_CONFIG", f"{where}.{key} must be a valid Kubernetes Service name (DNS-1035 label): {service_name!r}.", "Use lower case alphanumeric characters or '-', start with a letter, end with an alphanumeric character, " f"and keep length <= {K8S_SERVICE_NAME_MAX_LENGTH}.", ) def _validate_k8s_secret_name( name: str, label: str, error_code: str = "INVALID_CONFIG", hint: str = "Fix the runtime config." ) -> None: if ( not isinstance(name, str) or not name or len(name) > K8S_SECRET_NAME_MAX_LENGTH or not K8S_SECRET_NAME_PATTERN.fullmatch(name) ): _fail( error_code, f"{label} must contain valid Kubernetes object names.", "Use lower case alphanumeric characters, '-', or '.', start and end with an alphanumeric character, " f"and keep length <= 253. {hint}", ) def _optional_k8s_secret_name_list(data: dict[str, Any], key: str, label: str) -> list[str] | None: if key not in data or data[key] is None: return None names = data[key] if not isinstance(names, list): _fail("INVALID_CONFIG", f"{label} must be a list of Kubernetes object names.", "Fix the runtime config.") for name in names: _validate_k8s_secret_name(name, label) return names def _optional_int(data: dict[str, Any], key: str, where: str) -> int | None: if key not in data or data[key] is None: return None if not isinstance(data[key], int): _fail("INVALID_CONFIG", f"{where}.{key} must be an integer.", "Fix the runtime config.") return data[key] def _optional_non_negative_int(data: dict[str, Any], key: str, where: str) -> int | None: value = _optional_int(data, key, where) if isinstance(value, bool): _fail("INVALID_CONFIG", f"{where}.{key} must be an integer.", "Fix the runtime config.") if value is not None and value < 0: _fail("INVALID_CONFIG", f"{where}.{key} must be a non-negative integer.", "Fix the runtime config.") return value def _write_yaml(path: Path, data: dict[str, Any]) -> None: path.write_text(yaml.safe_dump(data, default_flow_style=False, sort_keys=False), encoding="utf-8")