diff --git a/nebula/controller/federation/controllers/docker_federation_controller.py b/nebula/controller/federation/controllers/docker_federation_controller.py index 4508489e4..890aa1262 100644 --- a/nebula/controller/federation/controllers/docker_federation_controller.py +++ b/nebula/controller/federation/controllers/docker_federation_controller.py @@ -10,7 +10,6 @@ from nebula.controller.federation.utils_requests import factory_requests_path from nebula.controller.federation.utils_requests import NodeUpdateRequest, NodeDoneRequest from typing import Dict -from fastapi import Request from nebula.config.config import Config from nebula.core.utils.certificate import generate_ca_certificate from nebula.core.utils.locker import Locker @@ -113,93 +112,50 @@ async def stop_scenario(self, federation_id: str): Reads ALL scenario.metadata and removes all listed containers and the network, then deletes the metadata file. Also forcibly stops and removes any containers still attached to the network before removing it. """ - federation_scenario_name = await self._remove_nebula_federation_from_pool(federation_id) - if not federation_scenario_name: + federation = await self._remove_nebula_federation_from_pool(federation_id) + if not federation: return False - # Try multiple possible config directory locations. This depends on where the user called the function from. - possible_config_dirs = [ - os.environ.get("NEBULA_CONFIG_DIR"), - "/nebula/app/config", - "./app/config", - os.path.join(os.getcwd(), "app", "config"), - os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "app", "config"), - ] - - config_dir = None - for dir_path in possible_config_dirs: - if dir_path and os.path.exists(dir_path): - config_dir = dir_path - break - - if not config_dir: - self.logger.info("No valid config directory found, skipping cleanup") - return - - scenario_dirs = [] - self.logger.info(f"Config directory: {config_dir}") - if os.path.exists(config_dir): - for item in os.listdir(config_dir): - scenario_path = os.path.join(config_dir, item) - if os.path.isdir(scenario_path): - metadata_file = os.path.join(scenario_path, "scenario.metadata") - if os.path.exists(metadata_file): - scenario_dirs.append(scenario_path) - - self.logger.info(f"Removing scenario containers for {scenario_dirs}") - if not scenario_dirs: - self.logger.info("No active scenarios found to clean up") - return - client = docker.from_env() - - for scenario_dir in scenario_dirs: - if scenario_dir != federation_scenario_name: - continue - - metadata_path = os.path.join(scenario_dir, "scenario.metadata") - if not os.path.exists(metadata_path): - self.logger.info(f"Skipping {scenario_dir} - no scenario.metadata found") - continue - - with open(metadata_path) as f: - meta = json.load(f) - - # Remove containers listed in metadata - for name in meta.get("containers", []): - try: - container = client.containers.get(name) - container.remove(force=True) - self.logger.info(f"Removed scenario container {name}") - except Exception as e: - self.logger.info(f"Could not remove scenario container {name}: {e}") - - # Remove network, but first forcibly remove any containers still attached - network_name = meta.get("network") - if network_name: - try: - network = client.networks.get(network_name) - attached_containers = network.attrs.get("Containers") or {} - for container_id in attached_containers: - try: - c = client.containers.get(container_id) - c.remove(force=True) - self.logger.info(f"Force-removed container {c.name} attached to {network_name}") - except Exception as e: - self.logger.info(f"Could not force-remove container {container_id}: {e}") - network.remove() - self.logger.info(f"Removed scenario network {network_name}") - except Exception as e: - self.logger.info(f"Could not remove scenario network {network_name}: {e}") - - # Remove metadata file + metadata_path = os.path.join(federation.config_dir, "scenario.metadata") + + if not os.path.exists(metadata_path): + self.logger.info(f"ERROR {metadata_path} - no 'scenario.metadata' found") + return False + + with open(metadata_path) as f: + meta = json.load(f) + # Remove containers listed in metadata + for name in meta.get("containers", []): try: - os.remove(metadata_path) + container = client.containers.get(name) + container.remove(force=True) + self.logger.info(f"Removed scenario container {name}") except Exception as e: - self.logger.info(f"Could not remove scenario.metadata: {e}") - - if scenario_dir == federation_scenario_name: - break + self.logger.info(f"Could not remove scenario container {name}: {e}") + # Remove network, but first forcibly remove any containers still attached + network_name = meta.get("network") + if network_name: + try: + network = client.networks.get(network_name) + attached_containers = network.attrs.get("Containers") or {} + for container_id in attached_containers: + try: + c = client.containers.get(container_id) + c.remove(force=True) + self.logger.info(f"Force-removed container {c.name} attached to {network_name}") + except Exception as e: + self.logger.info(f"Could not force-remove container {container_id}: {e}") + network.remove() + self.logger.info(f"Removed scenario network {network_name}") + except Exception as e: + self.logger.info(f"Could not remove scenario network {network_name}: {e}") + # Remove metadata file + try: + os.remove(metadata_path) + except Exception as e: + self.logger.info(f"Could not remove scenario.metadata: {e}") + return False return True #TODO care about cases @@ -269,15 +225,15 @@ async def _add_nebula_federation_to_pool(self, federation_id: str, user: str): self.logger.info(f"ERROR: trying to add ({federation_id}) to federations pool..") return fed - async def _remove_nebula_federation_from_pool(self, federation_id: str): + async def _remove_nebula_federation_from_pool(self, federation_id: str) -> NebulaFederationDocker | None: async with self._federations_dict_lock: if federation_id in self.nfp: federation = self.nfp.pop(federation_id) self.logger.info(f"SUCCESS: Federation ID: ({federation_id}) removed from pool") - return federation.scenario_name + return federation else: self.logger.info(f"ERROR: trying to remove ({federation_id}) from federations pool..") - return "" + return None async def _update_federation_on_pool(self, federation_id: str, user: str, nf: NebulaFederationDocker): updated = False diff --git a/nebula/controller/federation/controllers/processes_federation_controller.py b/nebula/controller/federation/controllers/processes_federation_controller.py index efb40b23f..f7f208677 100644 --- a/nebula/controller/federation/controllers/processes_federation_controller.py +++ b/nebula/controller/federation/controllers/processes_federation_controller.py @@ -125,44 +125,26 @@ async def stop_scenario(self, federation_id: str = ""): - Supports both Linux/macOS ('.sh') and Windows ('.ps1') script files. - Any errors during file removal are logged with the traceback. """ - federation_name = await self._remove_nebula_federation_from_pool(federation_id) - if not federation_name: + federation = await self._remove_nebula_federation_from_pool(federation_id) + if not federation: return False - # When stopping the nodes, we need to remove the current_scenario_commands.sh file -> it will cause the nodes to stop using PIDs try: - nebula_config_dir = os.environ.get("NEBULA_CONFIG_DIR") - if not nebula_config_dir: - current_dir = os.path.dirname(__file__) - nebula_base_dir = os.path.abspath(os.path.join(current_dir, "..", "..")) - nebula_config_dir = os.path.join(nebula_base_dir, "app", "config") - self.logger.info(f"NEBULA_CONFIG_DIR not found. Using default path: {nebula_config_dir}") - if federation_id: - if os.environ.get("NEBULA_HOST_PLATFORM") == "windows": - scenario_commands_file = os.path.join( - nebula_config_dir, federation_name, "current_scenario_commands.ps1" - ) - else: - scenario_commands_file = os.path.join( - nebula_config_dir, federation_name, "current_scenario_commands.sh" - ) - if os.path.exists(scenario_commands_file): - os.remove(scenario_commands_file) + if os.environ.get("NEBULA_HOST_PLATFORM") == "windows": + scenario_commands_file = os.path.join( + federation.config_dir, "current_scenario_commands.ps1" + ) else: - if os.environ.get("NEBULA_HOST_PLATFORM") == "windows": - files = glob.glob( - os.path.join(nebula_config_dir, "**/current_scenario_commands.ps1"), recursive=True - ) - else: - files = glob.glob( - os.path.join(nebula_config_dir, "**/current_scenario_commands.sh"), recursive=True - ) - for file in files: - os.remove(file) - return True + scenario_commands_file = os.path.join( + federation.config_dir, "current_scenario_commands.sh" + ) + if os.path.exists(scenario_commands_file): + os.remove(scenario_commands_file) + self.logger.info(f"Scenario commands file removed: {scenario_commands_file}") + else: + self.logger.info(f"Scenario commands file not found: {scenario_commands_file}") except Exception as e: - self.logger.exception(f"Error while removing current_scenario_commands.sh file: {e}") - return False + self.logger.exception(f"Error while removing current_scenario_commands file: {e}") async def update_nodes(self, federation_id: str, node_update_request: NodeUpdateRequest): config = node_update_request.config @@ -229,15 +211,15 @@ async def _add_nebula_federation_to_pool(self, federation_id: str, user: str): self.logger.info(f"ERROR: trying to add ({federation_id}) to federations pool..") return fed - async def _remove_nebula_federation_from_pool(self, federation_id: str): + async def _remove_nebula_federation_from_pool(self, federation_id: str) -> NebulaFederationProcesses | None: async with self._federations_dict_lock: if federation_id in self.nfp: federation = self.nfp.pop(federation_id) self.logger.info(f"SUCCESS: Federation ID: ({federation_id}) removed from pool") - return federation.scenario_name + return federation else: self.logger.info(f"ERROR: trying to remove ({federation_id}) from federations pool..") - return "" + return None async def _send_to_hub(self, path, payload, scenario_name="", federation_id="" ): try: diff --git a/nebula/controller/federation/federation_api.py b/nebula/controller/federation/federation_api.py index 746696c4c..b6edd00b5 100644 --- a/nebula/controller/federation/federation_api.py +++ b/nebula/controller/federation/federation_api.py @@ -64,14 +64,17 @@ async def run_scenario(run_scenario_request: RunScenarioRequest): return {"message": "Experiment type not allowed"} @app.post(Routes.STOP) -async def stop_scenario(stop_scenario_request: StopScenarioRequest): +async def stop_scenario( + federation_id: str, + stop_scenario_request: StopScenarioRequest +): global fed_controllers experiment_type = stop_scenario_request.experiment_type controller = fed_controllers.get(experiment_type, None) logger = logging.getLogger("Federation-Controller") logger.info(f"[API]: stop experiment request for federation ID: {stop_scenario_request.federation_id}") if controller: - return await controller.stop_scenario(stop_scenario_request.federation_id) + return await controller.stop_scenario(federation_id) else: return {"message": "Experiment type not allowed"} diff --git a/nebula/controller/federation/utils_requests.py b/nebula/controller/federation/utils_requests.py index e501b3377..e6fe76c2b 100644 --- a/nebula/controller/federation/utils_requests.py +++ b/nebula/controller/federation/utils_requests.py @@ -22,7 +22,7 @@ class NodeDoneRequest(BaseModel): class Routes: INIT = "/init" RUN = "/scenarios/run" - STOP = "/scenarios/stop" + STOP = "/scenarios/{federation_id}/stop" UPDATE = "/nodes/{federation_id}/update" DONE = "/nodes/{federation_id}/done" FINISH = "/scenarios/{federation_id}/finish" @@ -33,7 +33,7 @@ def factory_requests_path(resource: str, scenario_name: str = "", federation_id: elif resource == "run": return Routes.RUN elif resource == "stop": - return Routes.STOP + return Routes.STOP.format(federation_id=federation_id) elif resource == "update": return Routes.UPDATE.format(federation_id=federation_id) elif resource == "done":