From 8dbbd38049499ea9647c09044d7337b8d1e5b388 Mon Sep 17 00:00:00 2001 From: "Alejandro.A.S" Date: Wed, 24 Sep 2025 14:26:04 +0200 Subject: [PATCH 1/2] feature remove files endpoint & refactor request factory --- .../docker_federation_controller.py | 56 ++++++++++++++++--- .../processes_federation_controller.py | 52 ++++++++++++++--- .../controller/federation/federation_api.py | 15 ++++- .../federation/federation_controller.py | 6 +- .../controller/federation/utils_requests.py | 32 ++++++----- 5 files changed, 126 insertions(+), 35 deletions(-) diff --git a/nebula/controller/federation/controllers/docker_federation_controller.py b/nebula/controller/federation/controllers/docker_federation_controller.py index 890aa1262..cadda93d1 100644 --- a/nebula/controller/federation/controllers/docker_federation_controller.py +++ b/nebula/controller/federation/controllers/docker_federation_controller.py @@ -7,8 +7,8 @@ import docker from nebula.controller.federation.federation_controller import FederationController from nebula.controller.federation.scenario_builder import ScenarioBuilder -from nebula.controller.federation.utils_requests import factory_requests_path -from nebula.controller.federation.utils_requests import NodeUpdateRequest, NodeDoneRequest +from nebula.controller.federation.utils_requests import factory_requests +from nebula.controller.federation.utils_requests import RemoveScenarioRequest, NodeUpdateRequest, NodeDoneRequest from typing import Dict from nebula.config.config import Config from nebula.core.utils.certificate import generate_ca_certificate @@ -104,6 +104,8 @@ async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str) return None else: self.logger.info(f"ERROR: federation ID: ({federation_id}) already exists..") + asyncio.create_task(self.stop_scenario(federation_id)) + asyncio.create_task(self.remove_scenario(federation_id, RemoveScenarioRequest(experiment_type="docker", user=user, scenario_name=nebula_federation.scenario_name))) return scenario_info async def stop_scenario(self, federation_id: str): @@ -112,6 +114,7 @@ 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. """ + await asyncio.sleep(20) federation = await self._remove_nebula_federation_from_pool(federation_id) if not federation: return False @@ -209,6 +212,37 @@ async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest asyncio.create_task(self._send_to_hub("done", payload, federation_id=federation_id)) return {"message": "Nodes done received successfully"} + async def remove_scenario(self, federation_id: str, remove_scenario_request: RemoveScenarioRequest): + await asyncio.sleep(40) + if(await self._check_active_federation(federation_id)): + self.logger.info(f"WARNING: Cannot remove files from active federation: ({federation_id})") + return False + + folder_name = remove_scenario_request.user+"_"+remove_scenario_request.scenario_name + scenario_config_path = os.path.join(self.config_dir, folder_name) + scenario_log_path = os.path.join(self.log_dir, folder_name) + + if not os.path.exists(scenario_config_path): + self.logger.info(f"ERROR {scenario_config_path} - no config folder found") + if not os.path.exists(scenario_log_path): + self.logger.info(f"ERROR {scenario_log_path} - no log folder found") + + try: + shutil.rmtree(scenario_config_path) + self.logger.info(f"Removed config folder {scenario_config_path}") + except Exception as e: + self.logger.info(f"Could not remove config folder {scenario_config_path}: {e}") + return False + + try: + shutil.rmtree(scenario_log_path) + self.logger.info(f"Removed log folder {scenario_log_path}") + except Exception as e: + self.logger.info(f"Could not remove log folder {scenario_log_path}: {e}") + return False + + return True + """ ############################### # FUNCTIONALITIES # ############################### @@ -235,6 +269,13 @@ async def _remove_nebula_federation_from_pool(self, federation_id: str) -> Nebul self.logger.info(f"ERROR: trying to remove ({federation_id}) from federations pool..") return None + async def _check_active_federation(self, federation_id: str) -> bool: + async with self._federations_dict_lock: + if federation_id in self.nfp: + return True + else: + return False + async def _update_federation_on_pool(self, federation_id: str, user: str, nf: NebulaFederationDocker): updated = False async with self._federations_dict_lock: @@ -246,11 +287,9 @@ async def _update_federation_on_pool(self, federation_id: str, user: str, nf: Ne self.logger.info(f"ERROR: trying to update ({federation_id}) on federations pool..") return updated - async def _send_to_hub(self, path, payload, scenario_name="", federation_id="" ): + async def _send_to_hub(self, operation, payload, **kwargs): try: - url_request = self._hub_url + factory_requests_path(path, scenario_name, federation_id) - # self.logger.info(f"Sending to hub, url: {url_request}") - # self.logger.info(f"payload sent to hub, data: {payload}") + url_request = self._hub_url + factory_requests(operation, **kwargs) await APIUtils.post(url_request, payload) except Exception as e: self.logger.info(f"Failed to send update to Hub: {e}") @@ -263,13 +302,12 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat self.root_path = os.environ.get("NEBULA_ROOT_HOST") self.host_platform = os.environ.get("NEBULA_HOST_PLATFORM") - # self.config_dir = os.path.join(os.environ.get("NEBULA_CONFIG_DIR"), scenario_name) - # self.log_dir = os.path.join(os.environ.get("NEBULA_LOGS_DIR"), scenario_name) + self.config_dir = os.environ.get("NEBULA_CONFIG_DIR") + self.log_dir = os.environ.get("NEBULA_LOGS_DIR") federation.config_dir = os.path.join(os.environ.get("NEBULA_CONFIG_DIR"), scenario_name) federation.log_dir = os.path.join(os.environ.get("NEBULA_LOGS_DIR"), scenario_name) self.cert_dir = os.environ.get("NEBULA_CERTS_DIR") self.advanced_analytics = os.environ.get("NEBULA_ADVANCED_ANALYTICS", "False") == "True" - #self.config = Config(entity="FederationController") self.env_tag = os.environ.get("NEBULA_ENV_TAG", "dev") self.prefix_tag = os.environ.get("NEBULA_PREFIX_TAG", "dev") self.user_tag = os.environ.get("NEBULA_USER_TAG", os.environ.get("USER", "unknown")) diff --git a/nebula/controller/federation/controllers/processes_federation_controller.py b/nebula/controller/federation/controllers/processes_federation_controller.py index f7f208677..20eef4666 100644 --- a/nebula/controller/federation/controllers/processes_federation_controller.py +++ b/nebula/controller/federation/controllers/processes_federation_controller.py @@ -7,8 +7,8 @@ import docker from nebula.controller.federation.federation_controller import FederationController from nebula.controller.federation.scenario_builder import ScenarioBuilder -from nebula.controller.federation.utils_requests import factory_requests_path -from nebula.controller.federation.utils_requests import NodeUpdateRequest, NodeDoneRequest +from nebula.controller.federation.utils_requests import factory_requests +from nebula.controller.federation.utils_requests import RemoveScenarioRequest, NodeUpdateRequest, NodeDoneRequest from typing import Dict from fastapi import Request from nebula.config.config import Config @@ -195,6 +195,36 @@ async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest asyncio.create_task(self._send_to_hub("done", payload, federation_id=federation_id)) return {"message": "Nodes done received successfully"} + async def remove_scenario(self, federation_id: str, remove_scenario_request: RemoveScenarioRequest): + if(await self._check_active_federation(federation_id)): + self.logger.info(f"WARNING: Cannot remove files from active federation: ({federation_id})") + return False + + folder_name = remove_scenario_request.user+"_"+remove_scenario_request.scenario_name + scenario_config_path = os.path.join(self.config_dir, folder_name) + scenario_log_path = os.path.join(self.log_dir, folder_name) + + if not os.path.exists(scenario_config_path): + self.logger.info(f"ERROR {scenario_config_path} - no config folder found") + if not os.path.exists(scenario_log_path): + self.logger.info(f"ERROR {scenario_log_path} - no log folder found") + + try: + shutil.rmtree(scenario_config_path) + self.logger.info(f"Removed config folder {scenario_config_path}") + except Exception as e: + self.logger.info(f"Could not remove config folder {scenario_config_path}: {e}") + return False + + try: + shutil.rmtree(scenario_log_path) + self.logger.info(f"Removed log folder {scenario_log_path}") + except Exception as e: + self.logger.info(f"Could not remove log folder {scenario_log_path}: {e}") + return False + + return True + """ ############################### # FUNCTIONALITIES # ############################### @@ -220,12 +250,17 @@ async def _remove_nebula_federation_from_pool(self, federation_id: str) -> Nebul else: self.logger.info(f"ERROR: trying to remove ({federation_id}) from federations pool..") return None + + async def _check_active_federation(self, federation_id: str) -> bool: + async with self._federations_dict_lock: + if federation_id in self.nfp: + return True + else: + return False - async def _send_to_hub(self, path, payload, scenario_name="", federation_id="" ): + async def _send_to_hub(self, operation, payload, **kwargs): try: - url_request = self._hub_url + factory_requests_path(path, scenario_name, federation_id) - # self.logger.info(f"Seding to hub, url: {url_request}") - # self.logger.info(f"payload sent to hub, data: {payload}") + url_request = self._hub_url + factory_requests(operation, **kwargs) await APIUtils.post(url_request, payload) except Exception as e: self.logger.info(f"Failed to send update to Hub: {e}") @@ -238,13 +273,12 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat self.root_path = os.environ.get("NEBULA_ROOT_HOST") self.host_platform = os.environ.get("NEBULA_HOST_PLATFORM") - # self.config_dir = os.path.join(os.environ.get("NEBULA_CONFIG_DIR"), scenario_name) - # self.log_dir = os.path.join(os.environ.get("NEBULA_LOGS_DIR"), scenario_name) + self.config_dir = os.environ.get("NEBULA_CONFIG_DIR") + self.log_dir = os.environ.get("NEBULA_LOGS_DIR") federation.config_dir = os.path.join(os.environ.get("NEBULA_CONFIG_DIR"), scenario_name) federation.log_dir = os.path.join(os.environ.get("NEBULA_LOGS_DIR"), scenario_name) self.cert_dir = os.environ.get("NEBULA_CERTS_DIR") self.advanced_analytics = os.environ.get("NEBULA_ADVANCED_ANALYTICS", "False") == "True" - # self.config = Config(entity="scenarioManagement") self.env_tag = os.environ.get("NEBULA_ENV_TAG", "dev") self.prefix_tag = os.environ.get("NEBULA_PREFIX_TAG", "dev") self.user_tag = os.environ.get("NEBULA_USER_TAG", os.environ.get("USER", "unknown")) diff --git a/nebula/controller/federation/federation_api.py b/nebula/controller/federation/federation_api.py index b6edd00b5..f7513c828 100644 --- a/nebula/controller/federation/federation_api.py +++ b/nebula/controller/federation/federation_api.py @@ -10,7 +10,7 @@ from nebula.utils import LoggerUtils from nebula.controller.federation.federation_controller import FederationController from nebula.controller.federation.factory_federation_controller import federation_controller_factory -from nebula.controller.federation.utils_requests import RunScenarioRequest, StopScenarioRequest, NodeUpdateRequest, NodeDoneRequest, Routes +from nebula.controller.federation.utils_requests import RemoveScenarioRequest, RunScenarioRequest, StopScenarioRequest, NodeUpdateRequest, NodeDoneRequest, Routes fed_controllers: Dict[str, FederationController] = {} @@ -103,6 +103,19 @@ async def node_done( return await controller.node_done(federation_id, node_done_request) else: return {"message": "Experiment type not allowed on responde for Node done message.."} + +@app.post(Routes.REMOVE) +async def scenario_remove( + federation_id: str, + remove_scenario_request: RemoveScenarioRequest, +): + global fed_controllers + experiment_type = remove_scenario_request.experiment_type + controller = fed_controllers.get(experiment_type, None) + if controller: + return await controller.remove_scenario(federation_id, remove_scenario_request) + else: + return {"message": "Experiment type not allowed on responde for scenario remove message.."} if __name__ == "__main__": # Parse args from command line diff --git a/nebula/controller/federation/federation_controller.py b/nebula/controller/federation/federation_controller.py index 99bdda0bc..290aae62d 100644 --- a/nebula/controller/federation/federation_controller.py +++ b/nebula/controller/federation/federation_controller.py @@ -2,7 +2,7 @@ from fastapi import Request from typing import Dict from nebula.controller.federation.scenario_builder import ScenarioBuilder -from nebula.controller.federation.utils_requests import NodeUpdateRequest, NodeDoneRequest +from nebula.controller.federation.utils_requests import NodeUpdateRequest, NodeDoneRequest, RemoveScenarioRequest import logging class NebulaFederation(ABC): @@ -32,4 +32,8 @@ async def update_nodes(self, federation_id: str, node_update_request: NodeUpdate abstractmethod async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest): + pass + + abstractmethod + async def remove_scenario(self, federation_id: str, remove_scenario_request: RemoveScenarioRequest): pass \ No newline at end of file diff --git a/nebula/controller/federation/utils_requests.py b/nebula/controller/federation/utils_requests.py index e6fe76c2b..e51d7f22a 100644 --- a/nebula/controller/federation/utils_requests.py +++ b/nebula/controller/federation/utils_requests.py @@ -19,6 +19,11 @@ class NodeDoneRequest(BaseModel): name: str federation_id: str +class RemoveScenarioRequest(BaseModel): + experiment_type: str + user: str + scenario_name: str + class Routes: INIT = "/init" RUN = "/scenarios/run" @@ -26,21 +31,18 @@ class Routes: UPDATE = "/nodes/{federation_id}/update" DONE = "/nodes/{federation_id}/done" FINISH = "/scenarios/{federation_id}/finish" + REMOVE = "scenario/{federation_id}/remove" + + @classmethod + def format(cls, route: str, **kwargs) -> str: + return getattr(cls, route).format(**kwargs) -def factory_requests_path(resource: str, scenario_name: str = "", federation_id: str = "") -> str: - if resource == "init": - return "/init" - elif resource == "run": - return Routes.RUN - elif resource == "stop": - return Routes.STOP.format(federation_id=federation_id) - elif resource == "update": - return Routes.UPDATE.format(federation_id=federation_id) - elif resource == "done": - return Routes.DONE.format(federation_id=federation_id) - elif resource == "finish": - return Routes.FINISH.format(federation_id=federation_id) - else: - raise Exception(f"resource not found: {resource}") +def factory_requests(resource: str, **kwargs) -> str: + try: + return Routes.format(resource.upper(), **kwargs) + except AttributeError: + raise ValueError(f"Resource not found: {resource}") + except KeyError as e: + raise ValueError(f"Missing parameter for route '{resource}': {e}") \ No newline at end of file From af19e8e2395bb53078cc17496919fffbdc6ba99a Mon Sep 17 00:00:00 2001 From: "Alejandro.A.S" Date: Wed, 24 Sep 2025 14:35:41 +0200 Subject: [PATCH 2/2] fix remove unnecesary proofs --- .../federation/controllers/docker_federation_controller.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/nebula/controller/federation/controllers/docker_federation_controller.py b/nebula/controller/federation/controllers/docker_federation_controller.py index cadda93d1..5fa2f3b66 100644 --- a/nebula/controller/federation/controllers/docker_federation_controller.py +++ b/nebula/controller/federation/controllers/docker_federation_controller.py @@ -104,8 +104,6 @@ async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str) return None else: self.logger.info(f"ERROR: federation ID: ({federation_id}) already exists..") - asyncio.create_task(self.stop_scenario(federation_id)) - asyncio.create_task(self.remove_scenario(federation_id, RemoveScenarioRequest(experiment_type="docker", user=user, scenario_name=nebula_federation.scenario_name))) return scenario_info async def stop_scenario(self, federation_id: str): @@ -114,7 +112,6 @@ 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. """ - await asyncio.sleep(20) federation = await self._remove_nebula_federation_from_pool(federation_id) if not federation: return False @@ -213,7 +210,6 @@ async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest return {"message": "Nodes done received successfully"} async def remove_scenario(self, federation_id: str, remove_scenario_request: RemoveScenarioRequest): - await asyncio.sleep(40) if(await self._check_active_federation(federation_id)): self.logger.info(f"WARNING: Cannot remove files from active federation: ({federation_id})") return False