diff --git a/nebula/config/config.py b/nebula/config/config.py index 1e0a5eaa7..e2d2b9cb2 100755 --- a/nebula/config/config.py +++ b/nebula/config/config.py @@ -87,7 +87,7 @@ def __default_config(self): def __set_default_logging(self, mode="w"): experiment_name = self.participant["scenario_args"]["name"] - self.log_dir = os.path.join(self.participant["tracking_args"]["log_dir"], experiment_name) + self.log_dir =self.participant["tracking_args"]["log_dir"] if not os.path.exists(self.log_dir): os.makedirs(self.log_dir) self.log_filename = f"{self.log_dir}/participant_{self.participant['device_args']['idx']}" diff --git a/nebula/controller/federation/controllers/docker_federation_controller.py b/nebula/controller/federation/controllers/docker_federation_controller.py index 441c420d8..4508489e4 100644 --- a/nebula/controller/federation/controllers/docker_federation_controller.py +++ b/nebula/controller/federation/controllers/docker_federation_controller.py @@ -30,6 +30,8 @@ def __init__(self): self.federation_round: int = 0 self.federation_deployment_lock = Locker("federation_deployment_lock", async_lock=True) self.participants_alive_lock = Locker("participants_alive_lock", async_lock=True) + self.config_dir = "" + self.log_dir = "" async def get_additionals_to_be_deployed(self, config) -> list: async with self.federation_deployment_lock: @@ -89,7 +91,7 @@ async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str) federation = await self._add_nebula_federation_to_pool(federation_id, user) scenario_info = {} if federation: - scenario_builder = ScenarioBuilder(federation_id) + scenario_builder = ScenarioBuilder(federation_id, user=user) await self._initialize_scenario(scenario_builder, scenario_data, federation) generate_ca_certificate(dir_path=self.cert_dir) await self._load_configuration_and_start_nodes(scenario_builder, federation) @@ -301,12 +303,14 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat # Initialize Scenario builder using scenario_data from user self.logger.info("๐Ÿ”ง Initializing Scenario Builder using scenario data") sb.set_scenario_data(scenario_data) - scenario_name = sb.get_scenario_name() + scenario_name = sb.get_scenario_name(user_to=True) 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.environ.get("NEBULA_LOGS_DIR") + # 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) + 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") @@ -317,17 +321,17 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat self.url = f"{os.environ.get('NEBULA_CONTROLLER_HOST')}:{os.environ.get('NEBULA_FEDERATION_CONTROLLER_PORT')}" # Create Scenario management dirs - os.makedirs(self.config_dir, exist_ok=True) - os.makedirs(os.path.join(self.log_dir, scenario_name), exist_ok=True) + os.makedirs(federation.config_dir, exist_ok=True) + os.makedirs(federation.log_dir, exist_ok=True) os.makedirs(self.cert_dir, exist_ok=True) # Give permissions to the directories - os.chmod(self.config_dir, 0o777) - os.chmod(os.path.join(self.log_dir, scenario_name), 0o777) + os.chmod(federation.config_dir, 0o777) + os.chmod(federation.log_dir, 0o777) os.chmod(self.cert_dir, 0o777) # Save the scenario configuration - scenario_file = os.path.join(self.config_dir, "scenario.json") + scenario_file = os.path.join(federation.config_dir, "scenario.json") with open(scenario_file, "w") as f: json.dump(scenario_data, f, sort_keys=False, indent=2) @@ -337,13 +341,13 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat settings = { "scenario_name": scenario_name, "root_path": self.root_path, - "config_dir": self.config_dir, - "log_dir": self.log_dir, + "config_dir": federation.config_dir, + "log_dir": federation.log_dir, "cert_dir": self.cert_dir, "env": None, } - settings_file = os.path.join(self.config_dir, "settings.json") + settings_file = os.path.join(federation.config_dir, "settings.json") with open(settings_file, "w") as f: json.dump(settings, f, sort_keys=False, indent=2) @@ -359,7 +363,7 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat self.logger.info(f"Creating .json file for participant: {index}, Configuration: {node}") node_config = node try: - participant_file = os.path.join(self.config_dir, f"participant_{node_config['id']}.json") + participant_file = os.path.join(federation.config_dir, f"participant_{node_config['id']}.json") self.logger.info(f"Filename: {participant_file}") os.makedirs(os.path.dirname(participant_file), exist_ok=True) except Exception as e: @@ -383,7 +387,7 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federation: NebulaFederationDocker): self.logger.info("๐Ÿ”ง Loading Scenario configuration...") # Get participants configurations - participant_files = glob.glob(f"{self.config_dir}/participant_*.json") + participant_files = glob.glob(f"{federation.config_dir}/participant_*.json") participant_files.sort() if len(participant_files) == 0: raise ValueError("No participant files found in config folder") @@ -408,19 +412,19 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat self.logger.info("๐Ÿ”ง Building preload configuration for initial nodes...") for i in range(n_nodes): try: - with open(f"{self.config_dir}/participant_" + str(i) + ".json") as f: + with open(f"{federation.config_dir}/participant_" + str(i) + ".json") as f: participant_config = json.load(f) except Exception as e: self.logger.info(f"ERROR: open/load participant .json") self.logger.info(f"Building preload conf for participant {i}") try: - sb.build_preload_initial_node_configuration(i, participant_config, self.log_dir, self.config_dir, self.cert_dir, self.advanced_analytics) + sb.build_preload_initial_node_configuration(i, participant_config, federation.log_dir, federation.config_dir, self.cert_dir, self.advanced_analytics) except Exception as e: self.logger.info(f"ERROR: cannot build preload configuration") try: - with open(f"{self.config_dir}/participant_" + str(i) + ".json", "w") as f: + with open(f"{federation.config_dir}/participant_" + str(i) + ".json", "w") as f: json.dump(participant_config, f, sort_keys=False, indent=2) except Exception as e: self.logger.info(f"ERROR: cannot dump preload configuration into participant .json file") @@ -441,7 +445,7 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat federation.config.set_participants_config(participant_files) # Add role to the topology (visualization purposes) - sb.visualize_topology(config_participants, path=f"{self.config_dir}/topology.png", plot=False) + sb.visualize_topology(config_participants, path=f"{federation.config_dir}/topology.png", plot=False) # Additional participants self.logger.info("๐Ÿ”ง Building preload configuration for additional nodes...") @@ -451,7 +455,7 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat last_participant_index = len(participant_files) for i, _ in enumerate(additional_participants): - additional_participant_file = f"{self.config_dir}/participant_{last_participant_index + i}.json" + additional_participant_file = f"{federation.config_dir}/participant_{last_participant_index + i}.json" shutil.copy(last_participant_file, additional_participant_file) with open(additional_participant_file) as f: @@ -475,7 +479,7 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat self.logger.info("โœ… Loading Scenario configuration done") # Build dataset - dataset = sb.configure_dataset(self.config_dir) + dataset = sb.configure_dataset(federation.config_dir) self.logger.info(f"๐Ÿ”ง Splitting {sb.get_dataset_name()} dataset...") dataset.initialize_dataset() self.logger.info(f"โœ… Splitting {sb.get_dataset_name()} dataset... Done") @@ -502,7 +506,7 @@ def _get_participant_container_name(self, scenario_name, idx: int) -> str: def _start_initial_nodes(self, sb: ScenarioBuilder, federation: NebulaFederationDocker): self.logger.info("Starting nodes using Docker Compose...") - federation.network_name = self._get_network_name(f"{sb.get_scenario_name()}-net-scenario") + federation.network_name = self._get_network_name(f"{sb.get_scenario_name(user_to=True)}-net-scenario") federation.base_network_name = self._get_network_name("net-base") # Create the Docker network @@ -520,7 +524,7 @@ def _start_initial_nodes(self, sb: ScenarioBuilder, federation: NebulaFederation # deploy initial nodes self.logger.info(f"Deployment starting for participant {idx}") federation.round_per_participant[idx] = 0 - deployed_successfully = self._start_node(sb.get_scenario_name(), node, federation.network_name, federation.base_network_name, federation.base, federation.last_index_deployed, federation) + deployed_successfully = self._start_node(sb.get_scenario_name(user_to=True), node, federation.network_name, federation.base_network_name, federation.base, federation.last_index_deployed, federation) if deployed_successfully: federation.last_index_deployed += 1 federation.participants_alive += 1 @@ -568,8 +572,8 @@ def _start_node(self, scenario_name, node, network_name, base_network_name, base ), base_network_name: client.api.create_endpoint_config(), }) - node["tracking_args"]["log_dir"] = "/nebula/app/logs" - node["tracking_args"]["config_dir"] = f"/nebula/app/config/{scenario_name}" + node["tracking_args"]["log_dir"] = federation.log_dir + node["tracking_args"]["config_dir"] = federation.config_dir node["scenario_args"]["controller"] = self.url node["scenario_args"]["deployment"] = "docker" node["security_args"]["certfile"] = f"/nebula/app/certs/participant_{node['device_args']['idx']}_cert.pem" @@ -583,7 +587,7 @@ def _start_node(self, scenario_name, node, network_name, base_network_name, base except docker.errors.NotFound: pass # No conflict, safe to proceed # Write the config file in config directory - with open(f"{self.config_dir}/participant_{node['device_args']['idx']}.json", "w") as f: + with open(f"{federation.config_dir}/participant_{node['device_args']['idx']}.json", "w") as f: json.dump(node, f, indent=4) try: container_id = client.api.create_container( @@ -610,14 +614,14 @@ def _start_node(self, scenario_name, node, network_name, base_network_name, base # Write scenario-level metadata for cleanup scenario_metadata = {"containers": container_names, "network": network_name} - with open(os.path.join(self.config_dir, "scenario.metadata"), "a") as f: + with open(os.path.join(federation.config_dir, "scenario.metadata"), "a") as f: if i == 2: json.dump(scenario_metadata, f, indent=2) else: - with open(os.path.join(self.config_dir, "scenario.metadata"), "r") as f: + with open(os.path.join(federation.config_dir, "scenario.metadata"), "r") as f: metadata = json.load(f) metadata["containers"].extend(container_names) - with open(os.path.join(self.config_dir, "scenario.metadata"), "w") as f: + with open(os.path.join(federation.config_dir, "scenario.metadata"), "w") as f: json.dump(metadata, f, indent=2) return success diff --git a/nebula/controller/federation/controllers/processes_federation_controller.py b/nebula/controller/federation/controllers/processes_federation_controller.py index 90869f9b6..efb40b23f 100644 --- a/nebula/controller/federation/controllers/processes_federation_controller.py +++ b/nebula/controller/federation/controllers/processes_federation_controller.py @@ -30,6 +30,8 @@ def __init__(self): self.federation_round: int = 0 self.federation_deployment_lock = Locker("federation_deployment_lock", async_lock=True) self.participants_alive_lock = Locker("participants_alive_lock", async_lock=True) + self.config_dir = "" + self.log_dir = "" async def get_additionals_to_be_deployed(self, config) -> list: async with self.federation_deployment_lock: @@ -89,7 +91,7 @@ async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str) federation = await self._add_nebula_federation_to_pool(federation_id, user) scenario_info = {} if federation: - scenario_builder = ScenarioBuilder(federation_id) + scenario_builder = ScenarioBuilder(federation_id, user=user) await self._initialize_scenario(scenario_builder, scenario_data, federation) generate_ca_certificate(dir_path=self.cert_dir) await self._load_configuration_and_start_nodes(scenario_builder, federation) @@ -250,12 +252,14 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat # Initialize Scenario builder using scenario_data from user self.logger.info("๐Ÿ”ง Initializing Scenario Builder using scenario data") sb.set_scenario_data(scenario_data) - scenario_name = sb.get_scenario_name() + scenario_name = sb.get_scenario_name(user_to=True) 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.environ.get("NEBULA_LOGS_DIR") + # 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) + 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") @@ -266,17 +270,17 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat self.url = f"127.0.0.1:{os.environ.get('NEBULA_FEDERATION_CONTROLLER_PORT')}" # Create Scenario management dirs - os.makedirs(self.config_dir, exist_ok=True) - os.makedirs(os.path.join(self.log_dir, scenario_name), exist_ok=True) + os.makedirs(federation.config_dir, exist_ok=True) + os.makedirs(federation.log_dir, exist_ok=True) os.makedirs(self.cert_dir, exist_ok=True) # Give permissions to the directories - os.chmod(self.config_dir, 0o777) - os.chmod(os.path.join(self.log_dir, scenario_name), 0o777) + os.chmod(federation.config_dir, 0o777) + os.chmod(federation.log_dir, 0o777) os.chmod(self.cert_dir, 0o777) # Save the scenario configuration - scenario_file = os.path.join(self.config_dir, "scenario.json") + scenario_file = os.path.join(federation.config_dir, "scenario.json") with open(scenario_file, "w") as f: json.dump(scenario_data, f, sort_keys=False, indent=2) @@ -286,13 +290,13 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat settings = { "scenario_name": scenario_name, "root_path": self.root_path, - "config_dir": self.config_dir, - "log_dir": self.log_dir, + "config_dir": federation.config_dir, + "log_dir": federation.log_dir, "cert_dir": self.cert_dir, "env": None, } - settings_file = os.path.join(self.config_dir, "settings.json") + settings_file = os.path.join(federation.config_dir, "settings.json") with open(settings_file, "w") as f: json.dump(settings, f, sort_keys=False, indent=2) @@ -308,7 +312,7 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat self.logger.info(f"Creating .json file for participant: {index}, Configuration: {node}") node_config = node try: - participant_file = os.path.join(self.config_dir, f"participant_{node_config['id']}.json") + participant_file = os.path.join(federation.config_dir, f"participant_{node_config['id']}.json") self.logger.info(f"Filename: {participant_file}") os.makedirs(os.path.dirname(participant_file), exist_ok=True) except Exception as e: @@ -332,7 +336,7 @@ async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federat async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federation: NebulaFederationProcesses): self.logger.info("๐Ÿ”ง Loading Scenario configuration...") # Get participants configurations - participant_files = glob.glob(f"{self.config_dir}/participant_*.json") + participant_files = glob.glob(f"{federation.config_dir}/participant_*.json") participant_files.sort() if len(participant_files) == 0: raise ValueError("No participant files found in config folder") @@ -356,19 +360,19 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat self.logger.info("๐Ÿ”ง Building preload configuration for initial nodes...") for i in range(n_nodes): try: - with open(f"{self.config_dir}/participant_" + str(i) + ".json") as f: + with open(f"{federation.config_dir}/participant_" + str(i) + ".json") as f: participant_config = json.load(f) except Exception as e: self.logger.info(f"ERROR: open/load participant .json") self.logger.info(f"Building preload conf for participant {i}") try: - sb.build_preload_initial_node_configuration(i, participant_config, self.log_dir, self.config_dir, self.cert_dir, self.advanced_analytics) + sb.build_preload_initial_node_configuration(i, participant_config, federation.log_dir, federation.config_dir, self.cert_dir, self.advanced_analytics) except Exception as e: self.logger.info(f"ERROR: cannot build preload configuration") try: - with open(f"{self.config_dir}/participant_" + str(i) + ".json", "w") as f: + with open(f"{federation.config_dir}/participant_" + str(i) + ".json", "w") as f: json.dump(participant_config, f, sort_keys=False, indent=2) except Exception as e: self.logger.info(f"ERROR: cannot dump preload configuration into participant .json file") @@ -389,7 +393,7 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat federation.config.set_participants_config(participant_files) # Add role to the topology (visualization purposes) - sb.visualize_topology(config_participants, path=f"{self.config_dir}/topology.png", plot=False) + sb.visualize_topology(config_participants, path=f"{federation.config_dir}/topology.png", plot=False) # Additional participants self.logger.info("๐Ÿ”ง Building preload configuration for additional nodes...") @@ -399,7 +403,7 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat last_participant_index = len(participant_files) for i, _ in enumerate(additional_participants): - additional_participant_file = f"{self.config_dir}/participant_{last_participant_index + i}.json" + additional_participant_file = f"{federation.config_dir}/participant_{last_participant_index + i}.json" shutil.copy(last_participant_file, additional_participant_file) with open(additional_participant_file) as f: @@ -423,7 +427,7 @@ async def _load_configuration_and_start_nodes(self, sb: ScenarioBuilder, federat self.logger.info("โœ… Loading Scenario configuration done") # Build dataset - dataset = sb.configure_dataset(self.config_dir) + dataset = sb.configure_dataset(federation.config_dir) self.logger.info(f"๐Ÿ”ง Splitting {sb.get_dataset_name()} dataset...") dataset.initialize_dataset() self.logger.info(f"โœ… Splitting {sb.get_dataset_name()} dataset... Done") @@ -456,7 +460,7 @@ def _start_initial_nodes(self, sb: ScenarioBuilder, federation: NebulaFederation federation.participants_alive += 1 if federation.config.participants and commands: - self._write_commands_on_file(commands) + self._write_commands_on_file(commands, federation) else: self.logger.info("ERROR: No commands on a proccesses deployment..") @@ -464,8 +468,8 @@ def _start_node(self, sb: ScenarioBuilder, node, network_name, base_network_name self.processes_root_path = os.path.join(os.path.dirname(__file__), "..", "..") node_idx = node['device_args']['idx'] # Include additional config to the participants - node["tracking_args"]["log_dir"] = os.path.join(self.root_path, "app", "logs") - node["tracking_args"]["config_dir"] = os.path.join(self.root_path, "app", "config", sb.get_scenario_name()) + node["tracking_args"]["log_dir"] = os.path.join(self.root_path, "app", "logs", sb.get_scenario_name(user_to=True)) + node["tracking_args"]["config_dir"] = os.path.join(self.root_path, "app", "config", sb.get_scenario_name(user_to=True)) node["scenario_args"]["controller"] = self.url node["scenario_args"]["deployment"] = sb.get_deployment() node["security_args"]["certfile"] = os.path.join( @@ -476,7 +480,7 @@ def _start_node(self, sb: ScenarioBuilder, node, network_name, base_network_name ) node["security_args"]["cafile"] = os.path.join(self.root_path, "app", "certs", "ca_cert.pem") # Write the config file in config directory - with open(f"{self.config_dir}/participant_{node['device_args']['idx']}.json", "w") as f: + with open(f"{federation.config_dir}/participant_{node['device_args']['idx']}.json", "w") as f: json.dump(node, f, indent=4) self.logger.info(f"Configuration file created successfully: {node_idx}") @@ -488,10 +492,10 @@ def _start_node(self, sb: ScenarioBuilder, node, network_name, base_network_name else: commands += "Start-Sleep -Seconds 2\n" commands += f'Write-Host "Running node {node["device_args"]["idx"]}..."\n' - commands += f'$OUT_FILE = "{self.root_path}\\app\\logs\\{sb.get_scenario_name()}\\participant_{node["device_args"]["idx"]}.out"\n' - commands += f'$ERROR_FILE = "{self.root_path}\\app\\logs\\{sb.get_scenario_name()}\\participant_{node["device_args"]["idx"]}.err"\n' + commands += f'$OUT_FILE = "{self.root_path}\\app\\logs\\{sb.get_scenario_name(user_to=True)}\\participant_{node["device_args"]["idx"]}.out"\n' + commands += f'$ERROR_FILE = "{self.root_path}\\app\\logs\\{sb.get_scenario_name(user_to=True)}\\participant_{node["device_args"]["idx"]}.err"\n' # Use Start-Process for executing Python in background and capture PID - commands += f"""$process = Start-Process -FilePath "python" -ArgumentList "{self.root_path}\\nebula\\core\\node.py {self.root_path}\\app\\config\\{sb.get_scenario_name()}\\participant_{node["device_args"]["idx"]}.json" -PassThru -NoNewWindow -RedirectStandardOutput $OUT_FILE -RedirectStandardError $ERROR_FILE + commands += f"""$process = Start-Process -FilePath "python" -ArgumentList "{self.root_path}\\nebula\\core\\node.py {self.root_path}\\app\\config\\{sb.get_scenario_name(user_to=True)}\\participant_{node["device_args"]["idx"]}.json" -PassThru -NoNewWindow -RedirectStandardOutput $OUT_FILE -RedirectStandardError $ERROR_FILE Add-Content -Path $PID_FILE -Value $process.Id """ else: @@ -500,8 +504,8 @@ def _start_node(self, sb: ScenarioBuilder, node, network_name, base_network_name else: commands += "sleep 2\n" commands += f'echo "Running node {node["device_args"]["idx"]}..."\n' - commands += f"OUT_FILE={self.root_path}/app/logs/{sb.get_scenario_name()}/participant_{node['device_args']['idx']}.out\n" - commands += f"python {self.root_path}/nebula/core/node.py {self.root_path}/app/config/{sb.get_scenario_name()}/participant_{node['device_args']['idx']}.json &\n" + commands += f"OUT_FILE={self.root_path}/app/logs/{sb.get_scenario_name(user_to=True)}/participant_{node['device_args']['idx']}.out\n" + commands += f"python {self.root_path}/nebula/core/node.py {self.root_path}/app/config/{sb.get_scenario_name(user_to=True)}/participant_{node['device_args']['idx']}.json &\n" commands += "echo $! >> $PID_FILE\n\n" except Exception as e: raise Exception(f"Error starting nodes as processes: {e}") @@ -524,19 +528,19 @@ def _build_initial_commands(self): raise Exception(f"Error starting nodes as processes: {e}") return commands - def _write_commands_on_file(self, commands: str): + def _write_commands_on_file(self, commands: str, federation: NebulaFederationProcesses): try: if self.host_platform == "windows": commands += 'Write-Host "All nodes started. PIDs stored in $PID_FILE"\n' - with open(f"{self.config_dir}/current_scenario_commands.ps1", "w") as f: + with open(f"{federation.config_dir}/current_scenario_commands.ps1", "w") as f: #self.logger.info(f"Process commands: {commands}") f.write(commands) - os.chmod(f"{self.config_dir}/current_scenario_commands.ps1", 0o755) + os.chmod(f"{federation.config_dir}/current_scenario_commands.ps1", 0o755) else: commands += 'echo "All nodes started. PIDs stored in $PID_FILE"\n' - with open(f"{self.config_dir}/current_scenario_commands.sh", "w") as f: + with open(f"{federation.config_dir}/current_scenario_commands.sh", "w") as f: #self.logger.info(f"Process commands: {commands}") f.write(commands) - os.chmod(f"{self.config_dir}/current_scenario_commands.sh", 0o755) + os.chmod(f"{federation.config_dir}/current_scenario_commands.sh", 0o755) except Exception as e: raise Exception(f"Error starting nodes as processes: {e}") diff --git a/nebula/controller/federation/scenario_builder.py b/nebula/controller/federation/scenario_builder.py index 4fa219cb0..ab4f2ffd9 100644 --- a/nebula/controller/federation/scenario_builder.py +++ b/nebula/controller/federation/scenario_builder.py @@ -9,13 +9,14 @@ from nebula.core.datasets.nebuladataset import NebulaDataset, factory_nebuladataset, factory_dataset_setup class ScenarioBuilder(): - def __init__(self, federation_id): + def __init__(self, federation_id, user): self._scenario_data = None self._config_setup = None self.logger = logging.getLogger("Federation-Controller") self._topology_manager: TopologyManager = None self._scenario_name = "" self._federation_id = federation_id + self._user = user @property def sd(self): @@ -27,8 +28,9 @@ def tm(self): """Topology Manager""" return self._topology_manager - def get_scenario_name(self): - return self._scenario_name + def get_scenario_name(self, user_to=False): + scenario_path = self._user+"_"+self._scenario_name if user_to else self._scenario_name + return scenario_path def set_scenario_data(self, scenario_data: dict): self._scenario_data = scenario_data @@ -51,7 +53,7 @@ def get_deployment(self) -> str: return self.sd["deployment"] def get_scenario_info(self) -> dict: - return {"federation_id": self._federation_id, "start_time": datetime.now()} + return {"federation_id": self._federation_id, "start_time": datetime.now().strftime('%d/%m/%Y %H:%M:%S'), "alias": self.sd["scenario_title"] , "scenario_name": self._scenario_name} """ ############################### # SCENARIO CONFIG NODE # diff --git a/nebula/controller/hub.py b/nebula/controller/hub.py index f2a2ffda2..7602c1d27 100755 --- a/nebula/controller/hub.py +++ b/nebula/controller/hub.py @@ -1,6 +1,6 @@ import argparse import asyncio -import datetime +from datetime import datetime import importlib import ipaddress import json @@ -302,13 +302,19 @@ async def run_scenario(run_scenario_request: controller_requests.RunScenarioRequ Returns: str: The name of the scenario that was started. """ + import hashlib + def generate_id(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + response = None + try: fed_controller_port = os.environ.get("NEBULA_FEDERATION_CONTROLLER_PORT") fed_controller_host = os.environ.get("NEBULA_CONTROLLER_HOST") - url_init_fed_controller = f"http://{fed_controller_host}:{fed_controller_port}" + federation_requests.factory_requests_path("init") url_run_scenario = f"http://{fed_controller_host}:{fed_controller_port}" + federation_requests.factory_requests_path("run") #init_fed_req = InitFederationRequest(experiment_type="docker") - run_scenario_req = federation_requests.RunScenarioRequest(scenario_data=run_scenario_request.scenario_data, federation_id=f"id_nebula_{1}", user=run_scenario_request.user) #TODO ID per experiment + federation_id = generate_id(f"nebula_{run_scenario_request.user}_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}") + run_scenario_req = federation_requests.RunScenarioRequest(scenario_data=run_scenario_request.scenario_data, federation_id=federation_id, user=run_scenario_request.user) #TODO ID per experiment #await APIUtils.post(url_init_fed_controller, init_fed_req.model_dump()) response = await APIUtils.post(url_run_scenario, run_scenario_req.model_dump()) except Exception as e: @@ -325,24 +331,36 @@ async def run_scenario(run_scenario_request: controller_requests.RunScenarioRequ if response: try: - await update_scenario( - scenario_name=response["federation_id"], + payload = controller_requests.ScenarioUpdateRequest( + alias=response["alias"], + scenario_name=response["scenario_name"], start_time=response["start_time"], end_time="", scenario=run_scenario_request.scenario_data, status="running", username=run_scenario_request.user, + ).model_dump() + path = controller_requests.factory_requests_path( + "update", federation_id=federation_id ) - + await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) return response["federation_id"] except Exception as e: logging.info(e) else: raise HTTPException(500, detail={"failed running scenario"}) -@app.post(controller_requests.Routes.STOP) #TODO redo method +@app.post(controller_requests.Routes.STOP) # TODO redo method async def stop_scenario( - scenario_name: str = Body(..., embed=True), + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Federation identifier", + ), + ], all: bool = Body(False, embed=True), ): """ @@ -355,8 +373,7 @@ async def stop_scenario( - Optionally finalizes all active scenarios if the 'all' flag is set. Args: - scenario_name (str): Name of the scenario to stop. - username (str): User who initiated the stop operation. + federation_id (str): Identifier of the scenario to stop. all (bool): Whether to stop all running scenarios instead of just one (default: False). Raises: @@ -367,55 +384,56 @@ async def stop_scenario( """ fed_controller_port = os.environ.get("NEBULA_FEDERATION_CONTROLLER_PORT") fed_controller_host = os.environ.get("NEBULA_CONTROLLER_HOST") - url_stop_scenario = f"http://{fed_controller_host}:{fed_controller_port}" + federation_requests.factory_requests_path("stop") - stop_scenario_req = federation_requests.StopScenarioRequest(federation_id="id_nebula") + url_stop_scenario = ( + f"http://{fed_controller_host}:{fed_controller_port}" + + federation_requests.factory_requests_path("stop") + ) + stop_scenario_req = federation_requests.StopScenarioRequest( + experiment_type="docker", federation_id=federation_id + ) try: - path = federation_requests.factory_requests_path("stop") - payload = federation_requests.StopScenarioRequest(scenario_name=scenario_name, all=all).model_dump() + path = controller_requests.factory_requests_path( + "stop", federation_id=federation_id + ) + payload = controller_requests.ScenarioStopRequest(all=all).model_dump() await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) await APIUtils.post(url_stop_scenario, stop_scenario_req.model_dump()) except Exception as e: logging.info(f"ERROR: sending stop scenario to federation Controller: {e}") - # from nebula.controller.scenarios import ScenarioManagement - - # ScenarioManagement.cleanup_scenario_containers() - # try: - # if all: - # await scenario_set_all_status_to_finished() - # else: - # await scenario_set_status_to_finished(scenario_name) - # except Exception as e: - # logging.exception(f"Error setting scenario {scenario_name} to finished: {e}") - # raise HTTPException(status_code=500, detail="Internal server error") - @app.post(controller_requests.Routes.REMOVE) async def remove_scenario( - scenario_name: str = Body(..., embed=True), + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Federation identifier", + ), + ], + request: controller_requests.ScenarioRemoveRequest, ): """ - Removes a scenario from the database by its name. - - Args: - scenario_name (str): Name of the scenario to remove. - - Returns: - dict: A message indicating successful removal. + Removes a scenario from the database by its federation identifier. """ from nebula.controller.scenarios import ScenarioManagement try: - path = controller_requests.factory_requests_path("remove") - payload = controller_requests.ScenarioRemoveRequest(scenario_name=scenario_name).model_dump() - await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) - ScenarioManagement.remove_files_by_scenario(scenario_name) + path = controller_requests.factory_requests_path( + "remove", federation_id=federation_id + ) + await APIUtils.post(f"{DATABASE_API_URL}{path}") + ScenarioManagement.remove_files_by_scenario(request.scenario_name) except Exception as e: - logging.exception(f"Error removing scenario {scenario_name}: {e}") + logging.exception( + f"Error removing scenario {request.scenario_name} ({federation_id}): {e}" + ) raise HTTPException(status_code=500, detail="Internal server error") - return {"message": f"Scenario {scenario_name} removed successfully"} + return {"message": f"Scenario {request.scenario_name} removed successfully"} @app.get(controller_requests.Routes.GET_SCENARIOS_BY_USER) @@ -443,63 +461,73 @@ async def get_scenarios( @app.post(controller_requests.Routes.UPDATE) async def update_scenario( - scenario_name: str = Body(..., embed=True), - start_time: str = Body(..., embed=True), - end_time: str = Body(..., embed=True), - scenario: dict = Body(..., embed=True), - status: str = Body(..., embed=True), - username: str = Body(..., embed=True), + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Federation identifier", + ), + ], + request: controller_requests.ScenarioUpdateRequest, ): """ - Updates the status and metadata of a scenario. + Updates the status and metadata of a scenario identified by its federation ID. Args: - scenario_name (str): Name of the scenario. - start_time (str): Start time of the scenario. - end_time (str): End time of the scenario. - scenario (dict): Scenario configuration. - status (str): New status of the scenario (e.g., "running", "finished"). - username (str): User performing the update. + federation_id (str): Identifier of the scenario to update. + request (ScenarioUpdateRequest): Payload containing alias, scenario name, timing, configuration, status and username. Returns: dict: A message confirming the update. """ try: - payload = controller_requests.ScenarioUpdateRequest( - scenario_name=scenario_name, - start_time=start_time, - end_time=end_time, - scenario=scenario, - status=status, - username=username, - ).model_dump() - path = controller_requests.factory_requests_path("update") + payload = request.model_dump() + path = controller_requests.factory_requests_path( + "update", federation_id=federation_id + ) return await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) except Exception as e: - logging.exception(f"Error updating scenario {scenario_name}: {e}") + logging.exception( + f"Error updating scenario {request.scenario_name} ({federation_id}): {e}" + ) raise HTTPException(status_code=500, detail="Internal server error") @app.post(controller_requests.Routes.FINISH) async def set_scenario_status_to_finished( - scenario_name: str = Body(..., embed=True), all: bool = Body(False, embed=True) + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Federation identifier", + ), + ], + all: bool = Body(False, embed=True), ): """ Sets the status of a scenario (or all scenarios) to 'finished'. Args: - scenario_name (str): Name of the scenario to mark as finished. + federation_id (str): Identifier of the scenario to mark as finished. all (bool): If True, sets all scenarios to finished. Returns: dict: A message confirming the operation. """ try: - payload = controller_requests.ScenarioFinishRequest(scenario_name=scenario_name, all=all).model_dump() - path = controller_requests.factory_requests_path("finish") + payload = controller_requests.ScenarioFinishRequest(all=all).model_dump() + path = controller_requests.factory_requests_path( + "finish", federation_id=federation_id + ) return await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) except Exception as e: - logging.exception(f"Error setting scenario {scenario_name} to finished: {e}") + logging.exception( + f"Error setting scenario {federation_id} to finished: {e}" + ) raise HTTPException(status_code=500, detail="Internal server error") @@ -526,8 +554,8 @@ async def get_running_scenario_endpoint(get_all: bool = False): async def check_scenario( user: Annotated[str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid username")], role: Annotated[str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid role")], - scenario_name: Annotated[ - str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid scenario name") + federation_id: Annotated[ + str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=64, description="Valid federation identifier") ], ): """ @@ -541,64 +569,90 @@ async def check_scenario( dict: Whether the scenario is allowed for the role. """ try: - path = controller_requests.factory_requests_path("check_scenario", user=user, role=role, scenario_name=scenario_name) + path = controller_requests.factory_requests_path( + "check_scenario", + user=user, + role=role, + federation_id=federation_id, + ) return await APIUtils.get(f"{DATABASE_API_URL}{path}") except Exception as e: logging.exception(f"Error checking scenario with role: {e}") raise HTTPException(status_code=500, detail="Internal server error") -@app.get(controller_requests.Routes.GET_SCENARIOS_BY_SCENARIO_NAME) -async def get_scenario_by_name_endpoint( - scenario_name: Annotated[ - str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid scenario name") +@app.get(controller_requests.Routes.GET_SCENARIO_BY_FEDERATION_ID) +async def get_scenario_by_federation_id( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), ], ): """ - Fetches a scenario by its name. + Fetches a scenario by its federation identifier. Args: - scenario_name (str): The name of the scenario. + federation_id (str): The identifier of the scenario. Returns: - dict: The scenario data. + dict: The scenario data returned by the database API. """ try: - path = controller_requests.factory_requests_path("get_scenarios_by_scenario_name", scenario_name=scenario_name) + path = controller_requests.factory_requests_path( + "get_scenarios_by_scenario_name", federation_id=federation_id + ) return await APIUtils.get(f"{DATABASE_API_URL}{path}") except Exception as e: - logging.exception(f"Error obtaining scenario {scenario_name}: {e}") + logging.exception(f"Error obtaining scenario {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") -@app.get(controller_requests.Routes.NODES_BY_SCENARIO_NAME) -async def list_nodes_by_scenario_name_endpoint( - scenario_name: Annotated[ - str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid scenario name") +@app.get(controller_requests.Routes.NODES_BY_FEDERATION_ID) +async def list_nodes_by_federation_id_endpoint( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), ], ): """ - Lists all nodes associated with a specific scenario. + Lists all nodes associated with a specific federation identifier. Args: - scenario_name (str): Name of the scenario. + federation_id (str): Identifier of the scenario whose nodes should be listed. Returns: list: List of nodes. """ try: - path = controller_requests.factory_requests_path("get_nodes_by_scenario_name", scenario_name=scenario_name) + path = controller_requests.factory_requests_path( + "get_nodes_by_scenario_name", federation_id=federation_id + ) return await APIUtils.get(f"{DATABASE_API_URL}{path}") except Exception as e: - logging.exception(f"Error obtaining nodes: {e}") + logging.exception(f"Error obtaining nodes for {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") -@app.post(controller_requests.Routes.NODES_UPDATE_BY_SCENARIO) +@app.post(controller_requests.Routes.NODES_UPDATE_BY_FEDERATION) async def update_nodes( - scenario_name: Annotated[ + federation_id: Annotated[ str, - Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid scenario name"), + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), ], request: Request, ): @@ -606,7 +660,7 @@ async def update_nodes( Updates the configuration of a node in the database and notifies the frontend. Args: - scenario_name (str): The scenario to which the node belongs. + federation_id (str): Identifier of the scenario to update. request (Request): The HTTP request containing the node data. Returns: @@ -614,7 +668,7 @@ async def update_nodes( """ try: config:dict = await request.json() - config["timestamp"] = str(datetime.datetime.now()) + config["timestamp"] = str(datetime.now()) mobility_args = config.get("mobility_args", None) if not mobility_args: @@ -626,6 +680,8 @@ async def update_nodes( # Build payload and include extras with mobility data payload = validated.model_dump() payload["extras"] = payload.get("mobility_args", {}) + payload.setdefault("scenario_args", {}) + payload["scenario_args"]["federation"] = federation_id # Update the node in database with validated data and extras path = controller_requests.factory_requests_path("update_nodes") @@ -634,6 +690,7 @@ async def update_nodes( logging.exception(f"Error updating nodes: {e}") raise HTTPException(status_code=500, detail="Internal server error") + scenario_name = validated.scenario_args.name url = ( f"http://{os.environ['NEBULA_ENV_TAG']}_{os.environ['NEBULA_PREFIX_TAG']}_{os.environ['NEBULA_USER_TAG']}_nebula-frontend/platform/dashboard/{scenario_name}/node/update" ) @@ -669,82 +726,118 @@ async def node_done( @app.post(controller_requests.Routes.NODES_REMOVE) -async def remove_nodes_by_scenario_name_endpoint(scenario_name: str = Body(..., embed=True)): +async def remove_nodes_by_federation_id_endpoint( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), + ] +): """ - Endpoint to remove all nodes associated with a scenario. - - Body Parameters: - - scenario_name: Name of the scenario whose nodes should be removed. + Endpoint to remove all nodes associated with a scenario identified by federation ID. Returns a success message or an error if something goes wrong. """ try: - path = controller_requests.factory_requests_path("remove_nodes") - payload = controller_requests.NodesRemoveRequest(scenario_name=scenario_name).model_dump() - await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) + path = controller_requests.factory_requests_path( + "remove_nodes", federation_id=federation_id + ) + await APIUtils.post(f"{DATABASE_API_URL}{path}") except Exception as e: logging.exception(f"Error removing nodes: {e}") raise HTTPException(status_code=500, detail="Internal server error") - return {"message": f"Nodes for scenario {scenario_name} removed successfully"} + return {"message": f"Nodes for federation {federation_id} removed successfully"} -@app.get(controller_requests.Routes.NOTES_BY_SCENARIO_NAME) -async def get_notes_by_scenario_name( - scenario_name: Annotated[ - str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid scenario name") +@app.get(controller_requests.Routes.NOTES_BY_FEDERATION_ID) +async def get_notes_by_federation_id( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), ], ): """ Endpoint to retrieve notes associated with a scenario. """ try: - path = controller_requests.factory_requests_path("get_notes_by_scenario_name", scenario_name=scenario_name) + path = controller_requests.factory_requests_path( + "get_notes_by_scenario_name", federation_id=federation_id + ) return await APIUtils.get(f"{DATABASE_API_URL}{path}") except Exception as e: - logging.exception(f"Error obtaining notes for scenario {scenario_name}: {e}") + logging.exception(f"Error obtaining notes for federation {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.post(controller_requests.Routes.NOTES_UPDATE) -async def update_notes_by_scenario_name(scenario_name: str = Body(..., embed=True), notes: str = Body(..., embed=True)): +async def update_notes_by_federation_id( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), + ], + notes: str = Body(..., embed=True), +): """ Endpoint to update notes for a given scenario. Body Parameters: - - scenario_name: Name of the scenario. - notes: Text content to store as notes. Returns a success message or an error if something goes wrong. """ try: - payload = controller_requests.NotesUpdateRequest(scenario_name=scenario_name, notes=notes).model_dump() - path = controller_requests.factory_requests_path("update_notes") + payload = controller_requests.NotesUpdateRequest(notes=notes).model_dump() + path = controller_requests.factory_requests_path( + "update_notes", federation_id=federation_id + ) return await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) except Exception as e: - logging.exception(f"Error updating notes: {e}") + logging.exception(f"Error updating notes for federation {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.post(controller_requests.Routes.NOTES_REMOVE) -async def remove_notes_by_scenario_name_endpoint(scenario_name: str = Body(..., embed=True)): +async def remove_notes_by_federation_id_endpoint( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), + ] +): """ - Endpoint to remove notes associated with a scenario. - - Body Parameters: - - scenario_name: Name of the scenario. + Endpoint to remove notes associated with a scenario identified by federation ID. Returns a success message or an error if something goes wrong. """ try: - path = controller_requests.factory_requests_path("remove_notes") - payload = controller_requests.NotesRemoveRequest(scenario_name=scenario_name).model_dump() - await APIUtils.post(f"{DATABASE_API_URL}{path}", data=payload) + path = controller_requests.factory_requests_path( + "remove_notes", federation_id=federation_id + ) + await APIUtils.post(f"{DATABASE_API_URL}{path}") except Exception as e: - logging.exception(f"Error removing notes: {e}") + logging.exception(f"Error removing notes for federation {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") - return {"message": f"Notes for scenario {scenario_name} removed successfully"} + return {"message": f"Notes for federation {federation_id} removed successfully"} @app.get(controller_requests.Routes.USER_LIST) @@ -765,25 +858,33 @@ async def list_users_controller(all_info: bool = False): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error retrieving users: {e}") -@app.get(controller_requests.Routes.USER_BY_SCENARIO_NAME) -async def get_user_by_scenario_name_endpoint( - scenario_name: Annotated[ - str, Path(regex="^[a-zA-Z0-9_-]+$", min_length=1, max_length=50, description="Valid scenario name") +@app.get(controller_requests.Routes.USER_BY_FEDERATION_ID) +async def get_user_by_federation_id_endpoint( + federation_id: Annotated[ + str, + Path( + regex="^[a-zA-Z0-9_-]+$", + min_length=1, + max_length=64, + description="Valid federation identifier", + ), ], ): """ - Endpoint to retrieve the user assigned to a scenario. + Endpoint to retrieve the user assigned to a scenario identified by federation ID. Path Parameters: - - scenario_name: Name of the scenario. + - federation_id: Identifier of the scenario. Returns user info or raises an HTTPException on error. """ try: - path = controller_requests.factory_requests_path("get_user_by_scenario_name", scenario_name=scenario_name) + path = controller_requests.factory_requests_path( + "get_user_by_scenario_name", federation_id=federation_id + ) return await APIUtils.get(f"{DATABASE_API_URL}{path}") except Exception as e: - logging.exception(f"Error obtaining user for scenario {scenario_name}: {e}") + logging.exception(f"Error obtaining user for federation {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @@ -920,13 +1021,13 @@ async def get_physical_node_state(ip: str): # Physical ยท aggregate state for an entire scenario # โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ @app.get(controller_requests.Routes.PHYSICAL_SCENARIO_STATE, tags=["physical"]) -async def get_physical_scenario_state(scenario_name: str): +async def get_physical_scenario_state(federation_id: str): """ Check the training state of *every* physical node assigned to a scenario. Parameters ---------- - scenario_name : str + federation_id : str Scenario identifier. Returns @@ -940,11 +1041,11 @@ async def get_physical_scenario_state(scenario_name: str): } """ # 1) Retrieve scenario metadata and node list from the DB - scenario = await get_scenario_by_name_endpoint(scenario_name) + scenario = await get_scenario_by_federation_id(federation_id) if not scenario: raise HTTPException(status_code=404, detail="Scenario not found") - nodes = await list_nodes_by_scenario_name_endpoint(scenario_name) + nodes = await list_nodes_by_federation_id_endpoint(federation_id) if not nodes: raise HTTPException(status_code=404, detail="No nodes found for scenario") diff --git a/nebula/controller/utils_requests.py b/nebula/controller/utils_requests.py index 9a66a4c6e..240168886 100644 --- a/nebula/controller/utils_requests.py +++ b/nebula/controller/utils_requests.py @@ -13,30 +13,30 @@ class Routes: # Scenarios (Controller + DB API routing) RUN = "/scenarios/run" - UPDATE = "/scenarios/update" - STOP = "/scenarios/stop" - REMOVE = "/scenarios/remove" - FINISH = "/scenarios/set_status_to_finished" + UPDATE = "/scenarios/{federation_id}/update" + STOP = "/scenarios/{federation_id}/stop" + REMOVE = "/scenarios/{federation_id}/remove" + FINISH = "/scenarios/{federation_id}/set_status_to_finished" RUNNING = "/scenarios/running" - CHECK_SCENARIO = "/scenarios/check/{user}/{role}/{scenario_name}" + CHECK_SCENARIO = "/scenarios/check/{user}/{role}/{federation_id}" GET_SCENARIOS_BY_USER = "/scenarios/{user}/{role}" - GET_SCENARIOS_BY_SCENARIO_NAME = "/scenarios/{scenario_name}" + GET_SCENARIO_BY_FEDERATION_ID = "/scenarios/{federation_id}" # Nodes - NODES_BY_SCENARIO_NAME = "/nodes/{scenario_name}" + NODES_BY_FEDERATION_ID = "/nodes/{federation_id}" NODES_UPDATE = "/nodes/update" - NODES_UPDATE_BY_SCENARIO = "/nodes/{scenario_name}/update" + NODES_UPDATE_BY_FEDERATION = "/nodes/{federation_id}/update" NODES_DONE_BY_SCENARIO = "/nodes/{scenario_name}/done" - NODES_REMOVE = "/nodes/remove" + NODES_REMOVE = "/nodes/{federation_id}/remove" # Notes - NOTES_BY_SCENARIO_NAME = "/notes/{scenario_name}" - NOTES_UPDATE = "/notes/update" - NOTES_REMOVE = "/notes/remove" + NOTES_BY_FEDERATION_ID = "/notes/{federation_id}" + NOTES_UPDATE = "/notes/{federation_id}/update" + NOTES_REMOVE = "/notes/{federation_id}/remove" # Users USER_LIST = "/user/list" - USER_BY_SCENARIO_NAME = "/user/{scenario_name}" + USER_BY_FEDERATION_ID = "/user/{federation_id}" USER_ADD = "/user/add" USER_DELETE = "/user/delete" USER_UPDATE = "/user/update" @@ -48,7 +48,7 @@ class Routes: PHYSICAL_STOP = "/physical/stop" PHYSICAL_SETUP = "/physical/setup" PHYSICAL_STATE = "/physical/state" - PHYSICAL_SCENARIO_STATE = "/physical/{scenario_name}/state" + PHYSICAL_SCENARIO_STATE = "/physical/{federation_id}/state" class RunScenarioRequest(BaseModel): @@ -62,6 +62,7 @@ class RunScenarioRequest(BaseModel): class ScenarioUpdateRequest(BaseModel): + alias: str scenario_name: str start_time: str end_time: str @@ -71,7 +72,6 @@ class ScenarioUpdateRequest(BaseModel): class ScenarioStopRequest(BaseModel): - scenario_name: str all: bool = False @@ -80,23 +80,13 @@ class ScenarioRemoveRequest(BaseModel): class ScenarioFinishRequest(BaseModel): - scenario_name: str all: bool = False class NotesUpdateRequest(BaseModel): - scenario_name: str notes: str -class NotesRemoveRequest(BaseModel): - scenario_name: str - - -class NodesRemoveRequest(BaseModel): - scenario_name: str - - class UserAddRequest(BaseModel): user: str password: str @@ -160,7 +150,12 @@ class NodesUpdateRequest(BaseModel): timestamp: str -def factory_requests_path(resource: str, user: str = "", role: str = "", scenario_name: str = "") -> str: +def factory_requests_path( + resource: str, + user: str = "", + role: str = "", + federation_id: str = "", +) -> str: """Build paths for requests to the Database API from the Controller. This factory only maps DB API resources; controller endpoints do not require mapping here. @@ -168,40 +163,40 @@ def factory_requests_path(resource: str, user: str = "", role: str = "", scenari if resource == "init": return Routes.INIT elif resource == "update": - return Routes.UPDATE + return Routes.UPDATE.format(federation_id=federation_id) elif resource == "stop": - return Routes.STOP + return Routes.STOP.format(federation_id=federation_id) elif resource == "remove": - return Routes.REMOVE + return Routes.REMOVE.format(federation_id=federation_id) elif resource == "finish": - return Routes.FINISH + return Routes.FINISH.format(federation_id=federation_id) elif resource == "running": return Routes.RUNNING elif resource == "check_scenario": - return Routes.CHECK_SCENARIO.format(user=user, role=role, scenario_name=scenario_name) + return Routes.CHECK_SCENARIO.format(user=user, role=role, federation_id=federation_id) elif resource == "get_scenarios_by_user": return Routes.GET_SCENARIOS_BY_USER.format(user=user, role=role) elif resource == "get_scenarios_by_scenario_name": - return Routes.GET_SCENARIOS_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.GET_SCENARIO_BY_FEDERATION_ID.format(federation_id=federation_id) # Nodes elif resource == "get_nodes_by_scenario_name": - return Routes.NODES_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.NODES_BY_FEDERATION_ID.format(federation_id=federation_id) elif resource == "update_nodes": return Routes.NODES_UPDATE elif resource == "remove_nodes": - return Routes.NODES_REMOVE + return Routes.NODES_REMOVE.format(federation_id=federation_id) # Notes elif resource == "get_notes_by_scenario_name": - return Routes.NOTES_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.NOTES_BY_FEDERATION_ID.format(federation_id=federation_id) elif resource == "update_notes": - return Routes.NOTES_UPDATE + return Routes.NOTES_UPDATE.format(federation_id=federation_id) elif resource == "remove_notes": - return Routes.NOTES_REMOVE + return Routes.NOTES_REMOVE.format(federation_id=federation_id) # Users elif resource == "list_users": return Routes.USER_LIST elif resource == "get_user_by_scenario_name": - return Routes.USER_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.USER_BY_FEDERATION_ID.format(federation_id=federation_id) elif resource == "add_user": return Routes.USER_ADD elif resource == "delete_user": diff --git a/nebula/core/training/lightning.py b/nebula/core/training/lightning.py index edaa7f1c5..5c68c91c3 100755 --- a/nebula/core/training/lightning.py +++ b/nebula/core/training/lightning.py @@ -133,7 +133,7 @@ def __init__(self, model, datamodule, config=None): self.round = 0 self.experiment_name = self.config.participant["scenario_args"]["name"] self.idx = self.config.participant["device_args"]["idx"] - self.log_dir = os.path.join(self.config.participant["tracking_args"]["log_dir"], self.experiment_name) + self.log_dir = self.config.participant["tracking_args"]["log_dir"] self._logger = None self.create_logger() enable_deterministic(seed=self.config.participant["scenario_args"]["random_seed"]) diff --git a/nebula/database/adapters/postgress/docker/init-configs.sql b/nebula/database/adapters/postgress/docker/init-configs.sql index 9370a31d7..7119931ee 100644 --- a/nebula/database/adapters/postgress/docker/init-configs.sql +++ b/nebula/database/adapters/postgress/docker/init-configs.sql @@ -35,10 +35,16 @@ ALTER TABLE IF EXISTS nodes ADD COLUMN IF NOT EXISTS extras JSONB; -- Drop legacy columns for latitude/longitude if present -ALTER TABLE IF EXISTS nodes - DROP COLUMN IF EXISTS latitude; -ALTER TABLE IF EXISTS nodes - DROP COLUMN IF EXISTS longitude; +-- ALTER TABLE IF EXISTS nodes +-- DROP COLUMN IF EXISTS latitude; +-- ALTER TABLE IF EXISTS nodes +-- DROP COLUMN IF EXISTS longitude; +-- AlTER TABLE IF EXISTS scenarios +-- ADD COLUMN IF NOT EXISTS federation_id TEXT; +-- ALTER TABLE IF EXISTS scenarios +-- DROP CONSTRAINT scenarios_pkey; +-- ALTER TABLE IF EXISTS scenarios +-- ADD CONSTRAINT scenarios_pkey PRIMARY KEY (federation_id); -- 3) Configs as JSONB DROP INDEX IF EXISTS idx_configs_config_gin; @@ -52,7 +58,9 @@ CREATE INDEX idx_configs_config_gin ON configs USING GIN (config); -- 4) Scenarios table as JSONB CREATE TABLE IF NOT EXISTS scenarios ( - name TEXT PRIMARY KEY, + federation_id TEXT PRIMARY KEY, + alias TEXT NOT NULL, + name TEXT NOT NULL, username TEXT NOT NULL, status TEXT, start_time TEXT, @@ -67,6 +75,6 @@ CREATE INDEX IF NOT EXISTS idx_scenarios_config_gin -- 5) Notes table CREATE TABLE IF NOT EXISTS notes ( - scenario TEXT PRIMARY KEY, + federation_id TEXT PRIMARY KEY, scenario_notes TEXT ); diff --git a/nebula/database/adapters/postgress/postgress.py b/nebula/database/adapters/postgress/postgress.py index 7d2c1a37e..ec6210516 100755 --- a/nebula/database/adapters/postgress/postgress.py +++ b/nebula/database/adapters/postgress/postgress.py @@ -186,7 +186,7 @@ async def _update_user(self, user:str, password:str, role:str): # --- Node Management Functions --- - async def _list_nodes(self, scenario_name:str=None, sort_by:str="idx"): + async def _list_nodes(self, federation_id:str=None, sort_by:str="idx"): """ Retrieves a list of nodes from the nodes database, optionally filtered by scenario and sorted. """ @@ -197,10 +197,10 @@ async def _list_nodes(self, scenario_name:str=None, sort_by:str="idx"): try: async with self.pool.acquire() as conn: - if scenario_name: + if federation_id: # Using f-string for column names is generally safe if validated as above - command = f"SELECT * FROM nodes WHERE scenario = $1 ORDER BY {sort_by};" - result = await conn.fetch(command, scenario_name) + command = f"SELECT * FROM nodes WHERE federation = $1 ORDER BY {sort_by};" + result = await conn.fetch(command, federation_id) else: command = f"SELECT * FROM nodes ORDER BY {sort_by};" result = await conn.fetch(command) @@ -227,14 +227,14 @@ async def _list_nodes(self, scenario_name:str=None, sort_by:str="idx"): return None - async def _list_nodes_by_scenario_name(self, scenario_name:str): + async def _list_nodes_by_federation_id(self, federation_id:str): """ Fetches all nodes associated with a specific scenario, ordered by their index as integers. """ try: async with self.pool.acquire() as conn: - command = "SELECT * FROM nodes WHERE scenario = $1 ORDER BY CAST(idx AS INTEGER) ASC;" - result = await conn.fetch(command, scenario_name) + command = "SELECT * FROM nodes WHERE federation = $1 ORDER BY CAST(idx AS INTEGER) ASC;" + result = await conn.fetch(command, federation_id) rows = [] for record in result: row = dict(record) @@ -342,12 +342,12 @@ async def _remove_all_nodes(self): await conn.execute("TRUNCATE nodes CASCADE;") # Use CASCADE if there are foreign key dependencies - async def _remove_nodes_by_scenario_name(self, scenario_name:str): + async def _remove_nodes_by_federation_id(self, federation_id:str): """ Deletes all nodes associated with a specific scenario from the database. """ async with self.pool.acquire() as conn: - await conn.execute("DELETE FROM nodes WHERE scenario = $1;", scenario_name) + await conn.execute("DELETE FROM nodes WHERE federation = $1;", federation_id) # --- Scenario Management Functions --- @@ -379,6 +379,7 @@ async def _get_all_scenarios(self, username:str, role:str, sort_by:str="start_ti # Select direct columns and relevant fields from config JSONB command = """ SELECT + federation_id, name, username, status, @@ -430,6 +431,7 @@ async def _get_all_scenarios_and_check_completed(self, user:str, role:str, sort_ # Base query that extracts fields from the JSONB using the ->> operator command = f""" SELECT + federation_id, name, username, status, @@ -456,8 +458,8 @@ async def _get_all_scenarios_and_check_completed(self, user:str, role:str, sort_ re_fetch_required = False for scenario in scenarios_to_return: if scenario["status"] == "running": - if await self._check_scenario_federation_completed(scenario["name"]): - await self._scenario_set_status_to_completed(scenario["name"]) + if await self._check_scenario_federation_completed(scenario["federation_id"]): + await self._scenario_set_status_to_completed(scenario["federation_id"]) re_fetch_required = True break @@ -468,7 +470,7 @@ async def _get_all_scenarios_and_check_completed(self, user:str, role:str, sort_ return scenarios_to_return - async def _scenario_update_record(self, scenario_name:str, start_time:datetime, end_time:datetime, scenario:dict, status:str, username:str): + async def _scenario_update_record(self, federation_id:str, alias:str, scenario_name:str, start_time:datetime, end_time:datetime, scenario:dict, status:str, username:str): """ Inserts or updates a scenario record using the PostgreSQL "UPSERT" pattern. All configuration is saved in the 'config' column of type JSONB. @@ -483,9 +485,11 @@ async def _scenario_update_record(self, scenario_name:str, start_time:datetime, return command = """ - INSERT INTO scenarios (name, start_time, end_time, username, status, config) - VALUES ($1, $2, $3, $4, $5, $6::jsonb) - ON CONFLICT (name) DO UPDATE SET + INSERT INTO scenarios (federation_id, alias, name, start_time, end_time, username, status, config) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) + ON CONFLICT (federation_id) DO UPDATE SET + alias = EXCLUDED.alias, + name = EXCLUDED.name, start_time = EXCLUDED.start_time, end_time = EXCLUDED.end_time, username = EXCLUDED.username, @@ -493,7 +497,7 @@ async def _scenario_update_record(self, scenario_name:str, start_time:datetime, config = scenarios.config || EXCLUDED.config; -- Merge JSONB """ async with self.pool.acquire() as conn: - await conn.execute(command, scenario_name, start_time, end_time, username, status, json.dumps(scenario)) + await conn.execute(command, federation_id, alias, scenario_name, start_time, end_time, username, status, json.dumps(scenario)) async def _scenario_set_all_status_to_finished(self): @@ -515,7 +519,7 @@ async def _scenario_set_all_status_to_finished(self): await conn.execute(command, current_time, json.dumps(current_time)) - async def _scenario_set_status_to_finished(self, scenario_name:str): + async def _scenario_set_status_to_finished(self, federation_id:str): """ Sets the status of a specific scenario to 'finished' and updates its 'end_time'. Updates both the direct columns and the JSONB 'config'. @@ -530,13 +534,13 @@ async def _scenario_set_status_to_finished(self, scenario_name:str): jsonb_set(config, '{status}', '"finished"'), '{end_time}', $2::jsonb ) - WHERE name = $3; + WHERE federation_id = $3; """ async with self.pool.acquire() as conn: - await conn.execute(command, current_time, json.dumps(current_time), scenario_name) + await conn.execute(command, current_time, json.dumps(current_time), federation_id) - async def _scenario_set_status_to_completed(self, scenario_name:str): + async def _scenario_set_status_to_completed(self, federation_id:str): """ Sets the status of a specific scenario to 'completed'. Updates both the direct column and the JSONB 'config'. @@ -546,20 +550,20 @@ async def _scenario_set_status_to_completed(self, scenario_name:str): SET status = 'completed', config = jsonb_set(config, '{status}', '"completed"') - WHERE name = $1; + WHERE federation_id = $1; """ async with self.pool.acquire() as conn: - await conn.execute(command, scenario_name) + await conn.execute(command, federation_id) - async def _finish_scenario(self, scenario_name: str, all: bool = False): + async def _finish_scenario(self, federation_id: str, all: bool = False): """ Consolidated method to set scenarios to finished. """ if all: await self._scenario_set_all_status_to_finished() else: - await self._scenario_set_status_to_finished(scenario_name) + await self._scenario_set_status_to_finished(federation_id) async def _get_running_scenario(self, username:str=None, get_all:bool=False): @@ -570,7 +574,7 @@ async def _get_running_scenario(self, username:str=None, get_all:bool=False): async with self.pool.acquire() as conn: params = ["running"] # Select all columns to get both direct and config data - command = "SELECT name, username, status, start_time, end_time, config FROM scenarios WHERE status = $1" + command = "SELECT federation_id, name, username, status, start_time, end_time, config FROM scenarios WHERE status = $1" if username: command += " AND username = $2" @@ -603,12 +607,12 @@ async def _get_scenarios(self, user: str, role: str): return {"scenarios": scenarios, "scenario_running": scenario_running} - async def _get_scenario_by_name(self, scenario_name:str): + async def _get_scenario_by_federation_id(self, federation_id:str): """ Retrieves the complete record of a scenario by its name. """ async with self.pool.acquire() as conn: - result_row = await conn.fetchrow("SELECT name, start_time, end_time, username, status, config FROM scenarios WHERE name = $1;", scenario_name) + result_row = await conn.fetchrow("SELECT name, start_time, end_time, username, status, config FROM scenarios WHERE federation_id = $1;", federation_id) result = dict(result_row) if result_row else None @@ -629,69 +633,69 @@ async def _get_scenario_by_name(self, scenario_name:str): return result - async def _get_user_by_scenario_name(self, scenario_name:str): + async def _get_user_by_federation_id(self, federation_id:str): """ Retrieves the username associated with a scenario (from the direct 'username' column). """ async with self.pool.acquire() as conn: - return await conn.fetchval("SELECT username FROM scenarios WHERE name = $1;", scenario_name) + return await conn.fetchval("SELECT username FROM scenarios WHERE federation_id = $1;", federation_id) - async def _remove_scenario_by_name(self, scenario_name:str): + async def _remove_scenario_by_federation_id(self, federation_id:str): """ Delete a scenario from the database by its unique name. """ try: async with self.pool.acquire() as conn: - await conn.execute("DELETE FROM scenarios WHERE name = $1;", scenario_name) - logging.info(f"Scenario '{scenario_name}' successfully removed.") + await conn.execute("DELETE FROM scenarios WHERE federation_id = $1;", federation_id) + logging.info(f"Scenario '{federation_id}' successfully removed.") except asyncpg.PostgresError as e: - logging.error(f"Error occurred while deleting scenario '{scenario_name}': {e}") + logging.error(f"Error occurred while deleting scenario '{federation_id}': {e}") - async def _check_scenario_federation_completed(self, scenario_name:str): + async def _check_scenario_federation_completed(self, federation_id:str): """ Check if all nodes in a given scenario have completed the required federation rounds. """ try: async with self.pool.acquire() as conn: # Retrieve the total rounds for the scenario from the 'config' JSONB column - scenario_rounds_str = await conn.fetchval("SELECT config->>'rounds' AS rounds FROM scenarios WHERE name = $1;", scenario_name) + scenario_rounds_str = await conn.fetchval("SELECT config->>'rounds' AS rounds FROM scenarios WHERE federation_id = $1;", federation_id) if not scenario_rounds_str: - logging.warning(f"Scenario '{scenario_name}' not found or 'rounds' not defined.") + logging.warning(f"Scenario '{federation_id}' not found or 'rounds' not defined.") return False # Ensure total_rounds is an integer for comparison try: total_rounds = int(scenario_rounds_str) except (ValueError, TypeError): - logging.error(f"Invalid 'rounds' value for scenario '{scenario_name}': {scenario_rounds_str}") + logging.error(f"Invalid 'rounds' value for scenario '{federation_id}': {scenario_rounds_str}") return False # Fetch the current round progress of all nodes in that scenario - nodes = await conn.fetch("SELECT round FROM nodes WHERE scenario = $1;", scenario_name) + nodes = await conn.fetch("SELECT round FROM nodes WHERE federation = $1;", federation_id) if not nodes: - logging.info(f"No nodes found for scenario '{scenario_name}'. Federation not considered completed.") + logging.info(f"No nodes found for federation '{federation_id}'. Federation not considered completed.") return False # Check if all nodes have completed the total rounds return all(int(node["round"]) >= total_rounds for node in nodes) except asyncpg.PostgresError as e: - logging.error(f"PostgreSQL error during check_scenario_federation_completed for '{scenario_name}': {e}") + logging.error(f"PostgreSQL error during check_scenario_federation_completed for '{federation_id}': {e}") return False except ValueError as e: - logging.error(f"Data error during check_scenario_federation_completed for '{scenario_name}': {e}") + logging.error(f"Data error during check_scenario_federation_completed for '{federation_id}': {e}") return False - async def _check_scenario_with_role(self, role:str, scenario_name:str, user:str=None): + async def _check_scenario_with_role(self, role:str, federation_id:str, user:str=None): """ Verify if a scenario exists that the user with the given role and username can access. """ - scenario_info = await self._get_scenario_by_name(scenario_name) + scenario_info = await self._get_scenario_by_federation_id(federation_id) if not scenario_info: return False # Scenario does not exist @@ -709,7 +713,7 @@ async def _check_scenario_with_role(self, role:str, scenario_name:str, user:str= # --- Notes Management Functions --- - async def _save_notes(self, scenario: str, notes: str): + async def _save_notes(self, federation_id: str, notes: str): """ Save or update notes associated with a specific scenario. """ @@ -717,30 +721,30 @@ async def _save_notes(self, scenario: str, notes: str): async with self.pool.acquire() as conn: await conn.execute( """ - INSERT INTO notes (scenario, scenario_notes) VALUES ($1, $2) - ON CONFLICT(scenario) DO UPDATE SET scenario_notes = EXCLUDED.scenario_notes; + INSERT INTO notes (federation_id, scenario_notes) VALUES ($1, $2) + ON CONFLICT(federation_id) DO UPDATE SET scenario_notes = EXCLUDED.scenario_notes; """, - scenario, notes, + federation_id, notes, ) except asyncpg.PostgresError as e: logging.error(f"PostgreSQL error during save_notes: {e}") - async def _get_notes(self, scenario: str): + async def _get_notes(self, federation_id: str): """ Retrieve notes associated with a specific scenario. """ async with self.pool.acquire() as conn: - row = await conn.fetchrow("SELECT * FROM notes WHERE scenario = $1;", scenario) + row = await conn.fetchrow("SELECT * FROM notes WHERE federation_id = $1;", federation_id) if row is None: # No notes stored for this scenario yet return None return dict(row) - async def _remove_note(self, scenario: str): + async def _remove_note(self, federation_id: str): """ Delete the note associated with a specific scenario. """ async with self.pool.acquire() as conn: - await conn.execute("DELETE FROM notes WHERE scenario = $1;", scenario) + await conn.execute("DELETE FROM notes WHERE federation_id = $1;", federation_id) diff --git a/nebula/database/database_adapter_interface.py b/nebula/database/database_adapter_interface.py index 250464fb1..49ff82d7b 100644 --- a/nebula/database/database_adapter_interface.py +++ b/nebula/database/database_adapter_interface.py @@ -62,13 +62,13 @@ async def _update_user(self, user, password, role): # --- Node Management Functions --- @abstractmethod - async def _list_nodes(self, scenario_name=None, sort_by="idx"): + async def _list_nodes(self, federation_id=None, sort_by="idx"): """Retrieves a list of nodes.""" raise NotImplementedError @abstractmethod - async def _list_nodes_by_scenario_name(self, scenario_name): - """Fetches all nodes for a specific scenario.""" + async def _list_nodes_by_federation_id(self, federation_id): + """Fetches all nodes for a specific federation.""" raise NotImplementedError @abstractmethod @@ -97,8 +97,8 @@ async def _remove_all_nodes(self): raise NotImplementedError @abstractmethod - async def _remove_nodes_by_scenario_name(self, scenario_name): - """Deletes all nodes for a specific scenario.""" + async def _remove_nodes_by_federation_id(self, federation_id): + """Deletes all nodes for a specific federation.""" raise NotImplementedError # --- Scenario Management Functions --- @@ -114,7 +114,7 @@ async def _get_all_scenarios_and_check_completed(self, username, role, sort_by=" raise NotImplementedError @abstractmethod - async def _scenario_update_record(self, name, start_time, end_time, scenario_config, status, username): + async def _scenario_update_record(self, federation_id, name, start_time, end_time, scenario_config, status, username): """Inserts or updates a scenario record.""" raise NotImplementedError @@ -124,13 +124,13 @@ async def _scenario_set_all_status_to_finished(self): raise NotImplementedError @abstractmethod - async def _scenario_set_status_to_finished(self, scenario_name): - """Sets the status of a specific scenario to 'finished'.""" + async def _scenario_set_status_to_finished(self, federation_id): + """Sets the status of a specific scenario (by federation_id) to 'finished'.""" raise NotImplementedError @abstractmethod - async def _scenario_set_status_to_completed(self, scenario_name): - """Sets the status of a specific scenario to 'completed'.""" + async def _scenario_set_status_to_completed(self, federation_id): + """Sets the status of a specific scenario (by federation_id) to 'completed'.""" raise NotImplementedError @abstractmethod @@ -144,28 +144,28 @@ async def _get_completed_scenario(self): raise NotImplementedError @abstractmethod - async def _get_scenario_by_name(self, scenario_name): - """Retrieves a scenario by its name.""" + async def _get_scenario_by_federation_id(self, federation_id): + """Retrieves a scenario by its federation_id.""" raise NotImplementedError @abstractmethod - async def _get_user_by_scenario_name(self, scenario_name): - """Retrieves the user associated with a scenario.""" + async def _get_user_by_federation_id(self, federation_id): + """Retrieves the user associated with a scenario by federation_id.""" raise NotImplementedError @abstractmethod - async def _remove_scenario_by_name(self, scenario_name): - """Deletes a scenario by its name.""" + async def _remove_scenario_by_federation_id(self, federation_id): + """Deletes a scenario by its federation_id.""" raise NotImplementedError @abstractmethod - async def _check_scenario_federation_completed(self, scenario_name): + async def _check_scenario_federation_completed(self, federation_id): """Checks if a scenario's federation is complete.""" raise NotImplementedError @abstractmethod - async def _check_scenario_with_role(self, role, scenario_name, current_username=None): - """Verifies if a user can access a scenario.""" + async def _check_scenario_with_role(self, role, federation_id, user=None): + """Verifies if a user can access a scenario by federation_id.""" raise NotImplementedError # --- Notes Management Functions --- @@ -188,8 +188,8 @@ async def _remove_note(self, scenario): # --- Scenario Finish (no API logic) --- @abstractmethod - async def _finish_scenario(self, scenario_name, all: bool = False): - """Sets status to finished for one scenario or all running scenarios.""" + async def _finish_scenario(self, federation_id, all: bool = False): + """Sets status to finished for one scenario (by federation_id) or all running scenarios.""" raise NotImplementedError @abstractmethod diff --git a/nebula/database/database_api.py b/nebula/database/database_api.py index 579a11a05..bd74a7df7 100644 --- a/nebula/database/database_api.py +++ b/nebula/database/database_api.py @@ -13,11 +13,8 @@ Routes, ScenarioUpdateRequest, ScenarioStopRequest, - ScenarioRemoveRequest, ScenarioFinishRequest, NotesUpdateRequest, - NotesRemoveRequest, - NodesRemoveRequest, UserAddRequest, UserDeleteRequest, UserUpdateRequest, @@ -26,11 +23,7 @@ GetScenariosRequest, GetRunningScenarioRequest, CheckScenarioRequest, - GetScenarioByNameRequest, - ListNodesByScenarioNameRequest, - GetNotesByScenarioNameRequest, ListUsersRequest, - GetUserByScenarioNameRequest, ) # Get a database instance @@ -96,10 +89,12 @@ async def read_root(): # Scenarios @app.post(Routes.UPDATE) async def update_scenario( + federation_id: str, request: ScenarioUpdateRequest, ): try: await db._scenario_update_record( + federation_id = federation_id, **request.model_dump() ) return {"message": f"Scenario {request.scenario_name} updated successfully"} @@ -112,28 +107,29 @@ async def update_scenario( @app.post(Routes.STOP) async def stop_scenario( + federation_id: str, request: ScenarioStopRequest, ): try: - await db._finish_scenario(request.scenario_name, request.all) + await db._finish_scenario(federation_id, request.all) return {"message": "Finished status set successfully"} except Exception as e: logging.exception( - f"Error stopping scenario {request.scenario_name}: {e}" + f"Error stopping scenario {federation_id}: {e}" ) raise HTTPException(status_code=500, detail="Internal server error") @app.post(Routes.REMOVE) async def remove_scenario( - request: ScenarioRemoveRequest, + federation_id: str ): try: - await db._remove_scenario_by_name(request.scenario_name) - return {"message": f"Scenario {request.scenario_name} removed successfully"} + await db._remove_scenario_by_federation_id(federation_id) + return {"message": f"Scenario {federation_id} removed successfully"} except Exception as e: logging.exception( - f"Error removing scenario {request.scenario_name}: {e}" + f"Error removing scenario {federation_id}: {e}" ) raise HTTPException(status_code=500, detail="Internal server error") @@ -151,16 +147,17 @@ async def get_scenarios( @app.post(Routes.FINISH) async def set_scenario_status_to_finished( + federation_id: str, request: ScenarioFinishRequest, ): try: await db._finish_scenario( - request.scenario_name, request.all + federation_id, request.all ) return {"message": "Finished status set successfully"} except Exception as e: logging.exception( - f"Error setting scenario {request.scenario_name} to finished: {e}" + f"Error setting scenario {federation_id} to finished: {e}" ) raise HTTPException(status_code=500, detail="Internal server error") @@ -188,23 +185,23 @@ async def check_scenario( @app.get(Routes.GET_SCENARIOS_BY_SCENARIO_NAME) async def get_scenario_by_name_endpoint( - request: GetScenarioByNameRequest = Depends(), + federation_id: str ): try: - scenario = await db._get_scenario_by_name(request.scenario_name) + scenario = await db._get_scenario_by_federation_id(federation_id) return scenario except Exception as e: - logging.exception(f"Error obtaining scenario {request.scenario_name}: {e}") + logging.exception(f"Error obtaining scenario {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") # Nodes -@app.get(Routes.NODES_BY_SCENARIO_NAME) -async def list_nodes_by_scenario_name_endpoint( - request: ListNodesByScenarioNameRequest = Depends() +@app.get(Routes.NODES_BY_FEDERATION_ID) +async def list_nodes_by_federation_id_endpoint( + federation_id: str ): try: - nodes = await db._list_nodes_by_scenario_name(request.scenario_name) + nodes = await db._list_nodes_by_federation_id(federation_id) return nodes except Exception as e: logging.exception(f"Error obtaining nodes: {e}") @@ -241,43 +238,43 @@ async def update_node_record(request: NodesUpdateRequest): @app.post(Routes.NODES_REMOVE) -async def remove_nodes_by_scenario_name_endpoint(request: NodesRemoveRequest): +async def remove_nodes_by_federation_id_endpoint(federation_id: str): try: - await db._remove_nodes_by_scenario_name(request.scenario_name) - return {"message": f"Nodes for scenario {request.scenario_name} removed successfully"} + await db._remove_nodes_by_federation_id(federation_id) + return {"message": f"Nodes for federation {federation_id} removed successfully"} except Exception as e: logging.exception(f"Error removing nodes: {e}") raise HTTPException(status_code=500, detail="Internal server error") # Notes -@app.get(Routes.NOTES_BY_SCENARIO_NAME) -async def get_notes_by_scenario_name( - request: GetNotesByScenarioNameRequest = Depends() +@app.get(Routes.NOTES_BY_FEDERATION_ID) +async def get_notes_by_federation_id( + federation_id: str ): try: - notes_record = await db._get_notes(request.scenario_name) + notes_record = await db._get_notes(federation_id) return notes_record except Exception as e: - logging.exception(f"Error obtaining notes for scenario {request.scenario_name}: {e}") + logging.exception(f"Error obtaining notes for federation {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.post(Routes.NOTES_UPDATE) -async def update_notes_by_scenario_name(request: NotesUpdateRequest): +async def update_notes_by_scenario_name(federation_id: str, request: NotesUpdateRequest): try: - await db._save_notes(**request.model_dump()) - return {"message": f"Notes for scenario {request.scenario_name} updated successfully"} + await db._save_notes(federation_id ,**request.model_dump()) + return {"message": f"Notes for federation {federation_id} updated successfully"} except Exception as e: logging.exception(f"Error updating notes: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.post(Routes.NOTES_REMOVE) -async def remove_notes_by_scenario_name_endpoint(request: NotesRemoveRequest): +async def remove_notes_by_federation_id_endpoint(federation_id: str): try: - await db._remove_note(request.scenario_name) - return {"message": f"Notes for scenario {request.scenario_name} removed successfully"} + await db._remove_note(federation_id) + return {"message": f"Notes for federation {federation_id} removed successfully"} except Exception as e: logging.exception(f"Error removing notes: {e}") raise HTTPException(status_code=500, detail="Internal server error") @@ -293,15 +290,15 @@ async def list_users_controller(request: ListUsersRequest = Depends()): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error retrieving users: {e}") -@app.get(Routes.USER_BY_SCENARIO_NAME) -async def get_user_by_scenario_name_endpoint( - request: GetUserByScenarioNameRequest = Depends() +@app.get(Routes.USER_BY_FEDERATION_ID) +async def get_user_by_federation_id_endpoint( + federation_id: str ): try: - user = await db._get_user_by_scenario_name(request.scenario_name) + user = await db._get_user_by_federation_id(federation_id) return user except Exception as e: - logging.exception(f"Error obtaining user for scenario {request.scenario_name}: {e}") + logging.exception(f"Error obtaining user for federation {federation_id}: {e}") raise HTTPException(status_code=500, detail="Internal server error") diff --git a/nebula/database/utils_requests.py b/nebula/database/utils_requests.py index 8d9f88da2..59dde8244 100644 --- a/nebula/database/utils_requests.py +++ b/nebula/database/utils_requests.py @@ -6,28 +6,28 @@ class Routes: # Scenarios INIT = "/" - UPDATE = "/scenarios/update" - STOP = "/scenarios/stop" - REMOVE = "/scenarios/remove" - FINISH = "/scenarios/set_status_to_finished" + UPDATE = "/scenarios/{federation_id}/update" + STOP = "/scenarios/{federation_id}/stop" + REMOVE = "/scenarios/{federation_id}/remove" + FINISH = "/scenarios/{federation_id}/set_status_to_finished" RUNNING = "/scenarios/running" - CHECK_SCENARIO = "/scenarios/check/{user}/{role}/{scenario_name}" + CHECK_SCENARIO = "/scenarios/check/{user}/{role}/{federation_id}" GET_SCENARIOS_BY_USER = "/scenarios/{user}/{role}" - GET_SCENARIOS_BY_SCENARIO_NAME = "/scenarios/{scenario_name}" + GET_SCENARIOS_BY_SCENARIO_NAME = "/scenarios/{federation_id}" # Nodes - NODES_BY_SCENARIO_NAME = "/nodes/{scenario_name}" + NODES_BY_FEDERATION_ID = "/nodes/{federation_id}" NODES_UPDATE = "/nodes/update" - NODES_REMOVE = "/nodes/remove" + NODES_REMOVE = "/nodes/{federation_id}/remove" # Notes - NOTES_BY_SCENARIO_NAME = "/notes/{scenario_name}" - NOTES_UPDATE = "/notes/update" - NOTES_REMOVE = "/notes/remove" + NOTES_BY_FEDERATION_ID = "/notes/{federation_id}" + NOTES_UPDATE = "/notes/{federation_id}/update" + NOTES_REMOVE = "/notes/{federation_id}/remove" # Users USER_LIST = "/user/list" - USER_BY_SCENARIO_NAME = "/user/{scenario_name}" + USER_BY_FEDERATION_ID = "/user/{federation_id}" USER_ADD = "/user/add" USER_DELETE = "/user/delete" USER_UPDATE = "/user/update" @@ -35,6 +35,7 @@ class Routes: class ScenarioUpdateRequest(BaseModel): + alias: str scenario_name: str start_time: str end_time: str @@ -44,31 +45,19 @@ class ScenarioUpdateRequest(BaseModel): class ScenarioStopRequest(BaseModel): - scenario_name: str all: bool = False -class ScenarioRemoveRequest(BaseModel): - scenario_name: str - - class ScenarioFinishRequest(BaseModel): - scenario_name: str all: bool = False class NotesUpdateRequest(BaseModel): - scenario_name: str notes: str -class NotesRemoveRequest(BaseModel): - scenario_name: str -class NodesRemoveRequest(BaseModel): - scenario_name: str - class UserAddRequest(BaseModel): user: str @@ -144,30 +133,16 @@ class GetRunningScenarioRequest(BaseModel): class CheckScenarioRequest(BaseModel): user: str role: str - scenario_name: str - - -class GetScenarioByNameRequest(BaseModel): - scenario_name: str - - -class ListNodesByScenarioNameRequest(BaseModel): - scenario_name: str - - -class GetNotesByScenarioNameRequest(BaseModel): - scenario_name: str + federation_id: str class ListUsersRequest(BaseModel): all_info: bool = False -class GetUserByScenarioNameRequest(BaseModel): - scenario_name: str -def factory_requests_path(resource: str, user: str = "", role: str = "", scenario_name: str = "") -> str: +def factory_requests_path(resource: str, user: str = "", role: str = "", federation_id: str = "") -> str: if resource == "init": return Routes.INIT elif resource == "update": @@ -181,30 +156,30 @@ def factory_requests_path(resource: str, user: str = "", role: str = "", scenari elif resource == "running": return Routes.RUNNING elif resource == "check_scenario": - return Routes.CHECK_SCENARIO.format(user=user, role=role, scenario_name=scenario_name) + return Routes.CHECK_SCENARIO.format(user=user, role=role, federation_id=federation_id) elif resource == "get_scenarios_by_user": return Routes.GET_SCENARIOS_BY_USER.format(user=user, role=role) elif resource == "get_scenarios_by_scenario_name": - return Routes.GET_SCENARIOS_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.GET_SCENARIOS_BY_SCENARIO_NAME.format(federation_id=federation_id) # Nodes elif resource == "get_nodes_by_scenario_name": - return Routes.NODES_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.NODES_BY_FEDERATION_ID.format(federation_id=federation_id) elif resource == "update_nodes": return Routes.NODES_UPDATE elif resource == "remove_nodes": - return Routes.NODES_REMOVE + return Routes.NODES_REMOVE.format(federation_id=federation_id) # Notes elif resource == "get_notes_by_scenario_name": - return Routes.NOTES_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.NOTES_BY_FEDERATION_ID.format(federation_id=federation_id) elif resource == "update_notes": return Routes.NOTES_UPDATE elif resource == "remove_notes": - return Routes.NOTES_REMOVE + return Routes.NOTES_REMOVE.format(federation_id=federation_id) # Users elif resource == "list_users": return Routes.USER_LIST elif resource == "get_user_by_scenario_name": - return Routes.USER_BY_SCENARIO_NAME.format(scenario_name=scenario_name) + return Routes.USER_BY_FEDERATION_ID.format(federation_id=federation_id) elif resource == "add_user": return Routes.USER_ADD elif resource == "delete_user":