diff --git a/app/deployer.py b/app/deployer.py index 95bffda7e..7b624358e 100644 --- a/app/deployer.py +++ b/app/deployer.py @@ -17,7 +17,7 @@ from watchdog.observers import Observer from nebula.addons.env import check_environment -from nebula.controller.controller import TermEscapeCodeFormatter +from nebula.controller.web_app_controller import TermEscapeCodeFormatter from nebula.controller.scenarios import ScenarioManagement from nebula.utils import DockerUtils, FileUtils, SocketUtils @@ -637,6 +637,16 @@ def __init__(self, args): logging.exception(warning_msg) sys.exit(1) + self.controller_port = int(args.controllerport) if hasattr(args, "controllerport") else 5050 + self.federation_controller_port = int(args.federationcontrollerport) if hasattr(args, "federationcontrollerport") else 5051 + self.waf_port = int(args.wafport) if hasattr(args, "wafport") else 6000 + self.frontend_port = int(args.webport) if hasattr(args, "webport") else 6060 + self.grafana_port = int(args.grafanaport) if hasattr(args, "grafanaport") else 6040 + self.loki_port = int(args.lokiport) if hasattr(args, "lokiport") else 6010 + self.statistics_port = int(args.statsport) if hasattr(args, "statsport") else 8080 + self.production = args.production if hasattr(args, "production") else False + self.dev = args.developement if hasattr(args, "developement") else False + self.advanced_analytics = args.advanced_analytics if hasattr(args, "advanced_analytics") else False self.databases_dir = args.databases if hasattr(args, "databases") else "/nebula/app/databases" self.config_dir = args.config self.log_dir = args.logs @@ -843,6 +853,9 @@ def start(self): # Check ports available if not SocketUtils.is_port_open(self.controller_port): self.controller_port = SocketUtils.find_free_port(start_port=self.controller_port) + + if not SocketUtils.is_port_open(self.federation_controller_port): + self.federation_controller_port = SocketUtils.find_free_port(start_port=self.federation_controller_port) if not SocketUtils.is_port_open(self.frontend_port): self.frontend_port = SocketUtils.find_free_port(start_port=self.frontend_port) @@ -1110,11 +1123,13 @@ def run_controller(self): "NEBULA_ROOT_HOST": self.root_path, "NEBULA_DATABASES_DIR": "/nebula/app/databases", "NEBULA_CONTROLLER_LOG": "/nebula/app/logs/controller.log", + "NEBULA_FEDERATION_CONTROLLER_LOG": "/nebula/app/logs/federation.log", "NEBULA_CONFIG_DIR": "/nebula/app/config/", "NEBULA_LOGS_DIR": "/nebula/app/logs/", "NEBULA_CERTS_DIR": "/nebula/app/certs/", "NEBULA_HOST_PLATFORM": self.host_platform, "NEBULA_CONTROLLER_PORT": self.controller_port, + "NEBULA_FEDERATION_CONTROLLER_PORT" : self.federation_controller_port, "NEBULA_CONTROLLER_HOST": self.controller_host, "NEBULA_FRONTEND_PORT": self.frontend_port, "DB_HOST": self.get_container_name("nebula-database"), @@ -1126,7 +1141,7 @@ def run_controller(self): volumes = ["/nebula", "/var/run/docker.sock"] - ports = [self.controller_port] + ports = [self.controller_port, self.federation_controller_port] host_config = client.api.create_host_config( binds=[ @@ -1135,16 +1150,15 @@ def run_controller(self): f"{self.databases_dir}:/nebula/app/databases", ], extra_hosts={"host.docker.internal": "host-gateway"}, - port_bindings={self.controller_port: self.controller_port}, - device_requests=[ - { - "Driver": "nvidia", - "Count": -1, - "Capabilities": [["gpu"]], - } - ] - if self.gpu_available - else None, + port_bindings={ + self.controller_port: self.controller_port, + self.federation_controller_port: self.federation_controller_port + }, + device_requests=[{ + "Driver": "nvidia", + "Count": -1, + "Capabilities": [["gpu"]], + }] if self.gpu_available else None, ) networking_config = client.api.create_networking_config({ diff --git a/app/main.py b/app/main.py index e13d1641e..e822bfb4d 100755 --- a/app/main.py +++ b/app/main.py @@ -17,6 +17,14 @@ help="Controller port (default: 5050)", ) +argparser.add_argument( + "-fcp", + "--federationcontrollerport", + dest="federationcontrollerport", + default=5051, + help="federation controller port port (default: 5051)", +) + argparser.add_argument( "--grafanaport", dest="grafanaport", diff --git a/nebula/addons/attacks/attacks.py b/nebula/addons/attacks/attacks.py index f5587f365..3d7bd71a7 100755 --- a/nebula/addons/attacks/attacks.py +++ b/nebula/addons/attacks/attacks.py @@ -129,7 +129,7 @@ def create_attack(engine) -> Attack: } # Get attack name and parameters from the engine configuration - attack_params = engine.config.participant["adversarial_args"].get("attack_params", {}) + attack_params = engine.config.participant["addons"]["adversarial_args"].get("attack_params", {}) attack_name = attack_params.get("attacks", None) if attack_name is None: raise AttackException("No attack specified") diff --git a/nebula/addons/gps/nebulagps.py b/nebula/addons/gps/nebulagps.py index 571a5ab53..b1025cf22 100644 --- a/nebula/addons/gps/nebulagps.py +++ b/nebula/addons/gps/nebulagps.py @@ -74,8 +74,8 @@ async def is_running(self): return self._running.is_set() async def get_geoloc(self): - latitude = self._config.participant["mobility_args"]["latitude"] - longitude = self._config.participant["mobility_args"]["longitude"] + latitude = self._config.participant["addons"]["mobility"]["latitude"] + longitude = self._config.participant["addons"]["mobility"]["longitude"] return (latitude, longitude) async def calculate_distance(self, self_lat, self_long, other_lat, other_long): diff --git a/nebula/addons/mobility.py b/nebula/addons/mobility.py index b46f7fe88..fdf4265a4 100755 --- a/nebula/addons/mobility.py +++ b/nebula/addons/mobility.py @@ -57,18 +57,18 @@ def __init__(self, config, verbose=False): self._mobility_task = None # Track the background task # Mobility configuration - self.mobility = self.config.participant["mobility_args"]["mobility"] - self.mobility_type = self.config.participant["mobility_args"]["mobility_type"] - self.grace_time = self.config.participant["mobility_args"]["grace_time_mobility"] - self.period = self.config.participant["mobility_args"]["change_geo_interval"] + self.mobility = self.config.participant["addons"]["mobility"]["enabled"] + self.mobility_type = self.config.participant["addons"]["mobility"]["mobility_type"] + self.grace_time = self.config.participant["addons"]["mobility"]["grace_time_mobility"] + self.period = self.config.participant["addons"]["mobility"]["change_geo_interval"] # INFO: These values may change according to the needs of the federation self.max_distance_with_direct_connections = 150 # meters self.max_movement_random_strategy = 50 # meters self.max_movement_nearest_strategy = 50 # meters self.max_initiate_approximation = self.max_distance_with_direct_connections * 1.2 - self.radius_federation = float(config.participant["mobility_args"]["radius_federation"]) - self.scheme_mobility = config.participant["mobility_args"]["scheme_mobility"] - self.round_frequency = int(config.participant["mobility_args"]["round_frequency"]) + self.radius_federation = float(config.participant["addons"]["mobility"]["radius_federation"]) + self.scheme_mobility = config.participant["addons"]["mobility"]["scheme_mobility"] + self.round_frequency = int(config.participant["addons"]["mobility"]["round_frequency"]) # Logging box with mobility information mobility_msg = f"Mobility: {self.mobility}\nMobility type: {self.mobility_type}\nRadius federation: {self.radius_federation}\nScheme mobility: {self.scheme_mobility}\nEach {self.round_frequency} rounds" print_msg_box(msg=mobility_msg, indent=2, title="Mobility information") @@ -267,11 +267,11 @@ async def set_geo_location(self, latitude, longitude): if latitude < -90 or latitude > 90 or longitude < -180 or longitude > 180: # If the new location is out of bounds, we keep the old location - latitude = self.config.participant["mobility_args"]["latitude"] - longitude = self.config.participant["mobility_args"]["longitude"] + latitude = self.config.participant["addons"]["mobility"]["latitude"] + longitude = self.config.participant["addons"]["mobility"]["longitude"] - self.config.participant["mobility_args"]["latitude"] = latitude - self.config.participant["mobility_args"]["longitude"] = longitude + self.config.participant["addons"]["mobility"]["latitude"] = latitude + self.config.participant["addons"]["mobility"]["longitude"] = longitude if self._verbose: logging.info(f"πŸ“ New geo location: {latitude}, {longitude}") cle = ChangeLocationEvent(latitude, longitude) @@ -301,8 +301,8 @@ async def change_geo_location(self): """ if self.mobility and (self.mobility_type == "topology" or self.mobility_type == "both"): random.seed(time.time() + self.config.participant["device_args"]["idx"]) - latitude = float(self.config.participant["mobility_args"]["latitude"]) - longitude = float(self.config.participant["mobility_args"]["longitude"]) + latitude = float(self.config.participant["addons"]["mobility"]["latitude"]) + longitude = float(self.config.participant["addons"]["mobility"]["longitude"]) if True: # Get neighbor closer to me async with self._nodes_distances_lock: diff --git a/nebula/addons/networksimulation/nebulanetworksimulator.py b/nebula/addons/networksimulation/nebulanetworksimulator.py index 9af1f768e..85efbb2a2 100644 --- a/nebula/addons/networksimulation/nebulanetworksimulator.py +++ b/nebula/addons/networksimulation/nebulanetworksimulator.py @@ -35,7 +35,7 @@ def cm(self): async def start(self): logging.info("🌐 Nebula Network Simulator starting...") self._running.set() - grace_time = self.cm.config.participant["mobility_args"]["grace_time_mobility"] + grace_time = self.cm.config.participant["addons"]["mobility"]["grace_time_mobility"] # if self._verbose: logging.info(f"Waiting {grace_time}s to start applying network conditions based on distances between devices") # await asyncio.sleep(grace_time) await EventManager.get_instance().subscribe_addonevent( diff --git a/nebula/addons/reporter.py b/nebula/addons/reporter.py index 376f6a208..5b3952519 100755 --- a/nebula/addons/reporter.py +++ b/nebula/addons/reporter.py @@ -4,6 +4,7 @@ import logging import os import sys +from nebula.controller.federation.utils_requests import NodeUpdateRequest, NodeDoneRequest from typing import TYPE_CHECKING import aiohttp @@ -54,7 +55,7 @@ def __init__(self, config, trainer): self.frequency = self.config.participant["reporter_args"]["report_frequency"] self.grace_time = self.config.participant["reporter_args"]["grace_time_reporter"] self.data_queue = asyncio.Queue() - self.url = f"http://{self.config.participant['scenario_args']['controller']}/nodes/{self.config.participant['scenario_args']['name']}/update" + self.url = f"http://{self.config.participant['scenario_args']['controller']}/nodes/{self.config.participant['scenario_args']['federation_id']}/update" self.counter = 0 self.first_net_metrics = True @@ -170,8 +171,18 @@ async def report_scenario_finished(self): might be temporarily overloaded. - Logs exceptions if the connection attempt to the controller fails. """ - url = f"http://{self.config.participant['scenario_args']['controller']}/nodes/{self.config.participant['scenario_args']['name']}/done" - data = json.dumps({"idx": self.config.participant["device_args"]["idx"]}) + url = f"http://{self.config.participant['scenario_args']['controller']}/nodes/{self.config.participant['scenario_args']['federation_id']}/done" + node_done_req = NodeDoneRequest(idx=self.config.participant["device_args"]["idx"], + deployment=self.config.participant["scenario_args"]["deployment"], + name=self.config.participant["scenario_args"]["name"], + federation_id=self.config.participant["scenario_args"]["federation_id"] + ) + payload = node_done_req.model_dump() + data = json.dumps(payload) + # data = json.dumps({"idx": self.config.participant["device_args"]["idx"], + # "deployment": self.config.participant["scenario_args"]["deployment"], + # "name": self.config.participant["scenario_args"]["name"], + # "federation_id": self.config.participant["scenario_args"]["federation_id"]}) headers = { "Content-Type": "application/json", "User-Agent": f"NEBULA Participant {self.config.participant['device_args']['idx']}", @@ -263,11 +274,13 @@ async def __report_status_to_controller(self): - Delays for 5 seconds upon general exceptions to avoid rapid retry loops. """ try: + node_updt_req = NodeUpdateRequest(config=self.config.participant) + payload = node_updt_req.model_dump() async with ( aiohttp.ClientSession() as session, session.post( self.url, - data=json.dumps(self.config.participant), + data=json.dumps(payload), headers={ "Content-Type": "application/json", "User-Agent": f"NEBULA Participant {self.config.participant['device_args']['idx']}", diff --git a/nebula/addons/reputation/reputation.py b/nebula/addons/reputation/reputation.py index bca072cea..9377b2d42 100644 --- a/nebula/addons/reputation/reputation.py +++ b/nebula/addons/reputation/reputation.py @@ -3,7 +3,7 @@ import time import numpy as np import torch - +from nebula.core.addonmanager import NebulaAddon from datetime import datetime from typing import TYPE_CHECKING from nebula.addons.functions import print_msg_box @@ -54,7 +54,7 @@ def __init__( self.similarity = [] -class Reputation: +class Reputation(NebulaAddon): """ Class to define and manage the reputation of a participant in the network. @@ -114,7 +114,7 @@ def __init__(self, engine: "Engine", config: "Config"): def _configure_constants(self): """Configure system constants from config or use defaults.""" - reputation_config = self._config.participant.get("defense_args", {}).get("reputation", {}) + reputation_config = self._config.participant.get("addons", {}).get("reputation", {}) constants_config = reputation_config.get("constants", {}) self.REPUTATION_THRESHOLD = constants_config.get("reputation_threshold", self.REPUTATION_THRESHOLD) @@ -168,7 +168,7 @@ def _initialize_data_structures(self): def _load_configuration(self): """Load and validate reputation configuration.""" - reputation_config = self._config.participant["defense_args"]["reputation"] + reputation_config = self._config.participant["addons"]["reputation"] self._enabled = reputation_config["enabled"] self._metrics = reputation_config["metrics"] self._initial_reputation = float(reputation_config["initial_reputation"]) @@ -316,7 +316,7 @@ def save_data( except Exception: logging.exception(f"Error saving data for type {type_data} and neighbor {nei}") - async def setup(self): + async def start(self): """Set up the reputation system by subscribing to relevant events.""" if self._enabled: await EventManager.get_instance().subscribe_node_event(RoundStartEvent, self.on_round_start) @@ -340,6 +340,9 @@ async def setup(self): ) await EventManager.get_instance().subscribe_node_event(DuplicatedMessageEvent, self.recollect_duplicated_number_message) + async def stop(): + pass + async def init_reputation( self, federation_nodes=None, round_num=None, last_feedback_round=None, init_reputation=None ): @@ -1963,7 +1966,7 @@ async def recollect_similarity(self, ure: UpdateReceivedEvent): if not (self._enabled and self._is_metric_enabled("model_similarity")): return - if not self._engine.config.participant["adaptive_args"]["model_similarity"]: + if not self._engine.config.participant["addons"]["reputation"]["adaptive_args"] and not self._engine.config.participant["addons"]["reputation"]["adaptive_args"]["model_similarity"]: return if nei == self._addr: diff --git a/nebula/config/config.py b/nebula/config/config.py index d5f3a813f..1e0a5eaa7 100755 --- a/nebula/config/config.py +++ b/nebula/config/config.py @@ -208,14 +208,14 @@ def add_neighbor_from_config(self, addr): if not neighbors: self.participant["network_args"]["neighbors"] = [addr] - self.participant["mobility_args"]["neighbors_distance"][addr] = None + self.participant["addons"]["mobility"]["neighbors_distance"][addr] = None else: if addr not in neighbors: self.participant["network_args"]["neighbors"].append(addr) - self.participant["mobility_args"]["neighbors_distance"][addr] = None + self.participant["addons"]["mobility"]["neighbors_distance"][addr] = None def update_nodes_distance(self, distances: dict): - self.participant["mobility_args"]["neighbors_distance"] = {node: dist for node, (dist, _) in distances.items()} + self.participant["addons"]["mobility"]["neighbors_distance"] = {node: dist for node, (dist, _) in distances.items()} def update_neighbors_from_config(self, current_connections, dest_addr): final_neighbors = [n for n in current_connections if n != dest_addr] @@ -224,10 +224,10 @@ def update_neighbors_from_config(self, current_connections, dest_addr): self.participant["network_args"]["neighbors"] = final_neighbors # Update neighbors location - self.participant["mobility_args"]["neighbors_distance"] = { - n: self.participant["mobility_args"]["neighbors_distance"][n] + self.participant["addons"]["mobility"]["neighbors_distance"] = { + n: self.participant["addons"]["mobility"]["neighbors_distance"][n] for n in final_neighbors - if n in self.participant["mobility_args"]["neighbors_distance"] + if n in self.participant["addons"]["mobility"]["neighbors_distance"] } logging.info(f"Final neighbors: {final_neighbors} (config updated)") @@ -241,8 +241,8 @@ def remove_neighbor_from_config(self, addr): neighbors.remove(addr) self.participant["network_args"]["neighbors"] = neighbors - if addr in self.participant["mobility_args"]["neighbors_distance"]: - del self.participant["mobility_args"]["neighbors_distance"][addr] + if addr in self.participant["addons"]["mobility"]["neighbors_distance"]: + del self.participant["addons"]["mobility"]["neighbors_distance"][addr] def reload_config_file(self): diff --git a/nebula/controller/database.py b/nebula/controller/database.py index 407ce3908..ed4ead816 100755 --- a/nebula/controller/database.py +++ b/nebula/controller/database.py @@ -194,6 +194,9 @@ async def list_nodes_by_scenario_name(scenario_name): except Exception as e: logging.error(f"Error occurred while listing nodes by scenario name: {e}") return None + finally: + if conn: + await conn.close() async def update_node_record( @@ -321,7 +324,7 @@ async def get_all_scenarios_and_check_completed(username, role, sort_by="start_t if sort_by not in allowed_sort_fields: sort_by = "start_time" # Safe default value - # Building the ORDER BY clause + # Building the ORDER BY clause (same as get_all_scenarios) if sort_by == "start_time": order_by_clause = """ ORDER BY diff --git a/nebula/controller/federation/__init__.py b/nebula/controller/federation/__init__.py new file mode 100755 index 000000000..e69de29bb diff --git a/nebula/controller/federation/controllers/__init__.py b/nebula/controller/federation/controllers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/nebula/controller/federation/controllers/docker_federation_controller.py b/nebula/controller/federation/controllers/docker_federation_controller.py new file mode 100644 index 000000000..64ea123a4 --- /dev/null +++ b/nebula/controller/federation/controllers/docker_federation_controller.py @@ -0,0 +1,623 @@ +import asyncio +import glob +import json +import os +import shutil +from nebula.utils import DockerUtils, APIUtils +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 typing import Dict +from fastapi import Request +from nebula.config.config import Config +from nebula.core.utils.certificate import generate_ca_certificate +from nebula.core.utils.locker import Locker + +class NebulaFederationDocker(): + def __init__(self): + self.scenario_name = "" + self.participants_alive = 0 + self.round_per_participant = {} + self.additionals_participants = {} + self.additionals_deployables = [] + self.config = Config(entity="FederationController") + self.network_name = "" + self.base_network_name = "" + self.base = "" + self.last_index_deployed: int = 0 + 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) + + async def get_additionals_to_be_deployed(self, config) -> list: + async with self.federation_deployment_lock: + if not self.additionals_participants: + return False + + participant_idx = int(config["device_args"]["idx"]) + participant_round = int(config["federation_args"]["round"]) + self.round_per_participant[participant_idx] = participant_round + self.federation_round = min(self.round_per_participant.values()) + + self.additionals_deployables = [ + idx + for idx, round in self.additionals_participants.items() + if self.federation_round >= round + ] + + additionals_deployables = self.additionals_deployables.copy() + for idx in additionals_deployables: + self.additionals_participants.pop(idx) + return additionals_deployables + + async def is_experiment_finish(self): + async with self.participants_alive_lock: + self.participants_alive -= 1 + if self.participants_alive <= 0: + return True + else: + return False + +class DockerFederationController(FederationController): + + def __init__(self, hub_url, logger): + super().__init__(hub_url, logger) + self.root_path = "" + self.host_platform = "" + self.config_dir = "" + self.log_dir = "" + self.cert_dir = "" + self.advanced_analytics = "" + self.url = "" + self._nebula_federations_pool: dict[str, NebulaFederationDocker] = {} + self._federations_dict_lock = Locker("federations_dict_lock", async_lock=True) + + @property + def nfp(self): + """Nebula Federations Pool""" + return self._nebula_federations_pool + + """ ############################### + # ENDPOINT CALLBACKS # + ############################### + """ + + async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str): + #TODO maintain files on memory, not read them again + federation = await self._add_nebula_federation_to_pool(federation_id, user) + id = "" + if federation: + scenario_builder = ScenarioBuilder(federation_id) + 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) + self._start_initial_nodes(scenario_builder, federation) + id = scenario_builder.get_scenario_name() + try: + nebula_federation = self.nfp[federation_id] + nebula_federation.scenario_name = id + except Exception as e: + self.logger.info(f"ERROR: federation ID: ({federation_id}) not found on pool..") + return None + else: + self.logger.info(f"ERROR: federation ID: ({federation_id}) already exists..") + return id + + async def stop_scenario(self, federation_id: str): + """ + Remove all participant containers and the scenario network. + Reads ALL scenario.metadata and removes all listed containers and the network, then deletes the metadata file. + Also forcibly stops and removes any containers still attached to the network before removing it. + """ + federation_scenario_name = await self._remove_nebula_federation_from_pool(federation_id) + if not federation_scenario_name: + return False + + # Try multiple possible config directory locations. This depends on where the user called the function from. + possible_config_dirs = [ + os.environ.get("NEBULA_CONFIG_DIR"), + "/nebula/app/config", + "./app/config", + os.path.join(os.getcwd(), "app", "config"), + os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "app", "config"), + ] + + config_dir = None + for dir_path in possible_config_dirs: + if dir_path and os.path.exists(dir_path): + config_dir = dir_path + break + + if not config_dir: + self.logger.info("No valid config directory found, skipping cleanup") + return + + scenario_dirs = [] + self.logger.info(f"Config directory: {config_dir}") + if os.path.exists(config_dir): + for item in os.listdir(config_dir): + scenario_path = os.path.join(config_dir, item) + if os.path.isdir(scenario_path): + metadata_file = os.path.join(scenario_path, "scenario.metadata") + if os.path.exists(metadata_file): + scenario_dirs.append(scenario_path) + + self.logger.info(f"Removing scenario containers for {scenario_dirs}") + if not scenario_dirs: + self.logger.info("No active scenarios found to clean up") + return + + client = docker.from_env() + + for scenario_dir in scenario_dirs: + if scenario_dir != federation_scenario_name: + continue + + metadata_path = os.path.join(scenario_dir, "scenario.metadata") + if not os.path.exists(metadata_path): + self.logger.info(f"Skipping {scenario_dir} - no scenario.metadata found") + continue + + with open(metadata_path) as f: + meta = json.load(f) + + # Remove containers listed in metadata + for name in meta.get("containers", []): + try: + container = client.containers.get(name) + container.remove(force=True) + self.logger.info(f"Removed scenario container {name}") + except Exception as e: + self.logger.info(f"Could not remove scenario container {name}: {e}") + + # Remove network, but first forcibly remove any containers still attached + network_name = meta.get("network") + if network_name: + try: + network = client.networks.get(network_name) + attached_containers = network.attrs.get("Containers") or {} + for container_id in attached_containers: + try: + c = client.containers.get(container_id) + c.remove(force=True) + self.logger.info(f"Force-removed container {c.name} attached to {network_name}") + except Exception as e: + self.logger.info(f"Could not force-remove container {container_id}: {e}") + network.remove() + self.logger.info(f"Removed scenario network {network_name}") + except Exception as e: + self.logger.info(f"Could not remove scenario network {network_name}: {e}") + + # Remove metadata file + try: + os.remove(metadata_path) + except Exception as e: + self.logger.info(f"Could not remove scenario.metadata: {e}") + + if scenario_dir == federation_scenario_name: + break + + return True #TODO care about cases + + async def update_nodes(self, federation_id: str, node_update_request: NodeUpdateRequest): + config = node_update_request.config + scenario_name = config["scenario_args"]["name"] + fed_id = config["scenario_args"]["federation_id"] + + try: + nebula_federation = self.nfp[fed_id] + self.logger.info(f"Update received from node on federation ID: ({fed_id})") + last_fed_round = nebula_federation.federation_round + additionals = await nebula_federation.get_additionals_to_be_deployed(config) # It modifies if neccesary the federation round + if additionals: + current_fed_round = nebula_federation.federation_round + adds_deployed = set() + if current_fed_round != last_fed_round: + self.logger.info(f"Federation Round updating for ID: ({fed_id}), current value: {current_fed_round}") + for index in additionals: + if index in adds_deployed: + continue + + for idx, node in enumerate(nebula_federation.config.participants): + if index == idx: + if index in additionals: + self.logger.info(f"Deploying additional participant: {index}") + deployed_successfully = self._start_node(nebula_federation.scenario_name, node, nebula_federation.network_name, nebula_federation.base_network_name, nebula_federation.base, nebula_federation.last_index_deployed, nebula_federation) + if deployed_successfully: + self.logger.info(f"Deployment successfully for additional participant: {index}") + nebula_federation.last_index_deployed += 1 + #additionals.remove(index) + adds_deployed.add(index) + payload = node_update_request.model_dump() + asyncio.create_task(self._send_to_hub("update", payload, federation_id=fed_id)) + return {"message": "Node updated successfully in Federation Controller"} + except Exception as e: + self.logger.info(f"ERROR: federation ID: ({fed_id}), {e}") + return {"message": "Node updated failed in Federation Controller"} + + async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest): + nebula_federation = self.nfp[federation_id] + self.logger.info(f"Node-Done received from node on federation ID: ({federation_id})") + + if await nebula_federation.is_experiment_finish(): + payload = node_done_request.model_dump() + self.logger.info(f"All nodes have finished on federation ID: ({federation_id}), reporting to hub..") + await self._remove_nebula_federation_from_pool(federation_id) + asyncio.create_task(self._send_to_hub("finish", payload, federation_id=federation_id)) + + payload = node_done_request.model_dump() + asyncio.create_task(self._send_to_hub("done", payload, federation_id=federation_id)) + return {"message": "Nodes done received successfully"} + + """ ############################### + # FUNCTIONALITIES # + ############################### + """ + + async def _add_nebula_federation_to_pool(self, federation_id: str, user: str): + fed = None + async with self._federations_dict_lock: + if not federation_id in self.nfp: + fed = NebulaFederationDocker() + self.nfp[federation_id] = fed + self.logger.info(f"SUCCESS: new ID: ({federation_id}) added to the pool") + else: + self.logger.info(f"ERROR: trying to add ({federation_id}) to federations pool..") + return fed + + async def _remove_nebula_federation_from_pool(self, federation_id: str): + async with self._federations_dict_lock: + if federation_id in self.nfp: + federation = self.nfp.pop(federation_id) + self.logger.info(f"SUCCESS: Federation ID: ({federation_id}) removed from pool") + return federation.scenario_name + else: + self.logger.info(f"ERROR: trying to remove ({federation_id}) from federations pool..") + return "" + + async def _update_federation_on_pool(self, federation_id: str, user: str, nf: NebulaFederationDocker): + updated = False + async with self._federations_dict_lock: + if not federation_id in self.nfp: + self.nfp[federation_id] = nf + updated = True + self.logger.info(f"UPDATED: federation: ({federation_id}) successfully updated") + else: + 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="" ): + 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}") + await APIUtils.post(url_request, payload) + except Exception as e: + self.logger.info(f"Failed to send update to Hub: {e}") + + async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federation: NebulaFederationDocker): + # 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() + + 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.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")) + + 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(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(self.cert_dir, 0o777) + + # Save the scenario configuration + scenario_file = os.path.join(self.config_dir, "scenario.json") + with open(scenario_file, "w") as f: + json.dump(scenario_data, f, sort_keys=False, indent=2) + + os.chmod(scenario_file, 0o777) + + # Save management settings + settings = { + "scenario_name": scenario_name, + "root_path": self.root_path, + "config_dir": self.config_dir, + "log_dir": self.log_dir, + "cert_dir": self.cert_dir, + "env": None, + } + + settings_file = os.path.join(self.config_dir, "settings.json") + with open(settings_file, "w") as f: + json.dump(settings, f, sort_keys=False, indent=2) + + os.chmod(settings_file, 0o777) + + # Attacks assigment and mobility + self.logger.info("πŸ”§ Building general configuration") + sb.build_general_configuration() + self.logger.info("βœ… Building general configuration done") + + # Create participant configs and .json + for index, (_, node) in enumerate(sb.get_federation_nodes().items()): + 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") + self.logger.info(f"Filename: {participant_file}") + os.makedirs(os.path.dirname(participant_file), exist_ok=True) + except Exception as e: + self.logger.info(f"ERROR while creating files: {e}") + + try: + participant_config = sb.build_scenario_config_for_node(index, node) + #self.logger.info(f"dictionary: {participant_config}") + except Exception as e: + self.logger.info(f"ERROR while building configuration for node: {e}") + + try: + with open(participant_file, "w") as f: + json.dump(participant_config, f, sort_keys=False, indent=2) + os.chmod(participant_file, 0o777) + except Exception as e: + self.logger.info(f"ERROR while dumping configuration into files: {e}") + + self.logger.info("βœ… Initializing Scenario Builder done") + + 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.sort() + if len(participant_files) == 0: + raise ValueError("No participant files found in config folder") + + federation.config.set_participants_config(participant_files) + n_nodes = len(participant_files) + #self.logger.info(f"Number of nodes: {n_nodes}") + + sb.create_topology_manager(federation.config) + + # Update participants configuration + is_start_node = False + config_participants = [] + + additional_participants = sb.get_additional_nodes() + additional_nodes = len(additional_participants) if additional_participants else 0 + #self.logger.info(f"######## nodes: {n_nodes} + additionals: {additional_nodes} ######") + + participant_files.sort(key=lambda x: int(x.split("_")[-1].split(".")[0])) + + # Initial participants + 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: + 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) + 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: + 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") + + config_participants.append(( + participant_config["network_args"]["ip"], + participant_config["network_args"]["port"], + participant_config["device_args"]["role"], + )) + if participant_config["device_args"]["start"]: + if not is_start_node: + is_start_node = True + else: + raise ValueError("Only one node can be start node") + + self.logger.info("βœ… Building preload configuration for initial nodes done") + + 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) + + # Additional participants + self.logger.info("πŸ”§ Building preload configuration for additional nodes...") + additional_participants_files = [] + if additional_participants: + last_participant_file = participant_files[-1] + 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" + shutil.copy(last_participant_file, additional_participant_file) + + with open(additional_participant_file) as f: + participant_config = json.load(f) + + self.logger.info(f"Configuration | additional nodes | participant: {n_nodes + i}") + sb.build_preload_additional_node_configuration(last_participant_index, i, participant_config) + + with open(additional_participant_file, "w") as f: + json.dump(participant_config, f, sort_keys=False, indent=2) + + additional_participants_files.append(additional_participant_file) + + if additional_participants_files: + federation.config.add_participants_config(additional_participants_files) + + if additional_participants: + n_nodes += len(additional_participants) + + self.logger.info("βœ… Building preload configuration for additional nodes done") + self.logger.info("βœ… Loading Scenario configuration done") + + # Build dataset + dataset = sb.configure_dataset(self.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") + + def _get_network_name(self, suffix: str) -> str: + """ + Generate a standardized network name using tags. + Args: + suffix (str): Suffix for the network (default: 'net-base'). + Returns: + str: The composed network name. + """ + return f"{self.env_tag}_{self.prefix_tag}_{self.user_tag}_{suffix}" + + def _get_participant_container_name(self, scenario_name, idx: int) -> str: + """ + Generate a standardized container name for a participant using tags. + Args: + idx (int): The participant index. + Returns: + str: The composed container name. + """ + return f"{self.env_tag}_{self.prefix_tag}_{self.user_tag}_{scenario_name}_participant{idx}" + + 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.base_network_name = self._get_network_name("net-base") + + # Create the Docker network + federation.base = DockerUtils.create_docker_network(federation.network_name) + + federation.config.participants.sort(key=lambda x: x["device_args"]["idx"]) + federation.last_index_deployed = 2 + for idx, node in enumerate(federation.config.participants): + + if node["deployment_args"]["additional"]: + federation.additionals_participants[idx] = int(node["deployment_args"]["deployment_round"]) + federation.participants_alive += 1 + self.logger.info(f"Participant {idx} is additional. Round of deployment: {int(node['deployment_args']['deployment_round'])}") + else: + # 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) + if deployed_successfully: + federation.last_index_deployed += 1 + federation.participants_alive += 1 + + def _start_node(self, scenario_name, node, network_name, base_network_name, base, i, federation: NebulaFederationDocker): + success = True + client = docker.from_env() + + federation.config.participants.sort(key=lambda x: x["device_args"]["idx"]) + container_ids = [] + container_names = [] # Track names for metadata + + image = "nebula-core" + name = self._get_participant_container_name(scenario_name, node["device_args"]["idx"]) + if node["device_args"]["accelerator"] == "gpu": + environment = { + "NVIDIA_DISABLE_REQUIRE": True, + "NEBULA_LOGS_DIR": "/nebula/app/logs/", + "NEBULA_CONFIG_DIR": "/nebula/app/config/", + } + host_config = client.api.create_host_config( + binds=[f"{self.root_path}:/nebula", "/var/run/docker.sock:/var/run/docker.sock"], + privileged=True, + device_requests=[docker.types.DeviceRequest(driver="nvidia", count=-1, capabilities=[["gpu"]])], + extra_hosts={"host.docker.internal": "host-gateway"}, + ) + else: + environment = {"NEBULA_LOGS_DIR": "/nebula/app/logs/", "NEBULA_CONFIG_DIR": "/nebula/app/config/"} + host_config = client.api.create_host_config( + binds=[f"{self.root_path}:/nebula", "/var/run/docker.sock:/var/run/docker.sock"], + privileged=True, + device_requests=[], + extra_hosts={"host.docker.internal": "host-gateway"}, + ) + volumes = ["/nebula", "/var/run/docker.sock"] + start_command = "sleep 10" if node["device_args"]["start"] else "sleep 0" + command = [ + "/bin/bash", + "-c", + f"{start_command} && ifconfig && echo '{base}.1 host.docker.internal' >> /etc/hosts && python /nebula/nebula/core/node.py /nebula/app/config/{scenario_name}/participant_{node['device_args']['idx']}.json", + ] + networking_config = client.api.create_networking_config({ + network_name: client.api.create_endpoint_config( + ipv4_address=f"{base}.{i}", + ), + 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["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" + node["security_args"]["keyfile"] = f"/nebula/app/certs/participant_{node['device_args']['idx']}_key.pem" + node["security_args"]["cafile"] = "/nebula/app/certs/ca_cert.pem" + node = json.loads(json.dumps(node).replace("192.168.50.", f"{base}.")) # TODO change this + try: + existing = client.containers.get(name) + self.logger.info(f"Container {name} already exists. Deployment may fail or cause conflicts.") + success = False + 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: + json.dump(node, f, indent=4) + try: + container_id = client.api.create_container( + image=image, + name=name, + detach=True, + volumes=volumes, + environment=environment, + command=command, + host_config=host_config, + networking_config=networking_config, + ) + except Exception as e: + success = False + self.logger.info(f"Creating container {name}: {e}") + try: + client.api.start(container_id) + container_ids.append(container_id) + container_names.append(name) + self.logger.info(f"Adding name: {name} for metadata") + except Exception as e: + success = False + self.logger.info(f"Starting participant {name} error: {e}") + + # 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: + if i == 2: + json.dump(scenario_metadata, f, indent=2) + else: + with open(os.path.join(self.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: + json.dump(metadata, f, indent=2) + + return success \ No newline at end of file diff --git a/nebula/controller/federation/controllers/physicall_federation_controller.py b/nebula/controller/federation/controllers/physicall_federation_controller.py new file mode 100644 index 000000000..f60c2a8e1 --- /dev/null +++ b/nebula/controller/federation/controllers/physicall_federation_controller.py @@ -0,0 +1,4 @@ +from nebula.controller.federation.federation_controller import FederationController + +class PhysicalFederationController(FederationController): + pass \ No newline at end of file diff --git a/nebula/controller/federation/controllers/processes_federation_controller.py b/nebula/controller/federation/controllers/processes_federation_controller.py new file mode 100644 index 000000000..cd5c36121 --- /dev/null +++ b/nebula/controller/federation/controllers/processes_federation_controller.py @@ -0,0 +1,542 @@ +import asyncio +import glob +import json +import os +import shutil +from nebula.utils import APIUtils +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 typing import Dict +from fastapi import Request +from nebula.config.config import Config +from nebula.core.utils.certificate import generate_ca_certificate +from nebula.core.utils.locker import Locker + +class NebulaFederationProcesses(): + def __init__(self): + self.scenario_name = "" + self.participants_alive = 0 + self.round_per_participant = {} + self.additionals_participants = {} + self.additionals_deployables = [] + self.config = Config(entity="FederationController") + self.network_name = "" + self.base_network_name = "" + self.base = "" + self.last_index_deployed: int = 0 + 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) + + async def get_additionals_to_be_deployed(self, config) -> list: + async with self.federation_deployment_lock: + if not self.additionals_participants: + return False + + participant_idx = int(config["device_args"]["idx"]) + participant_round = int(config["federation_args"]["round"]) + self.round_per_participant[participant_idx] = participant_round + self.federation_round = min(self.round_per_participant.values()) + + self.additionals_deployables = [ + idx + for idx, round in self.additionals_participants.items() + if self.federation_round >= round + ] + + additionals_deployables = self.additionals_deployables.copy() + for idx in additionals_deployables: + self.additionals_participants.pop(idx) + return additionals_deployables + + async def is_experiment_finish(self): + async with self.participants_alive_lock: + self.participants_alive -= 1 + if self.participants_alive <= 0: + return True + else: + return False + +class ProcessesFederationController(FederationController): + def __init__(self, hub_url, logger): + super().__init__(hub_url, logger) + self.root_path = "" + self.host_platform = "" + self.config_dir = "" + self.log_dir = "" + self.cert_dir = "" + self.advanced_analytics = "" + self.url = "" + + self._nebula_federations_pool: dict[tuple[str,str], NebulaFederationProcesses] = {} + self._federations_dict_lock = Locker("federations_dict_lock", async_lock=True) + + @property + def nfp(self): + """Nebula Federations Pool""" + return self._nebula_federations_pool + + """ ############################### + # ENDPOINT CALLBACKS # + ############################### + """ + + async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str): + #TODO maintain files on memory, not read them again + federation = await self._add_nebula_federation_to_pool(federation_id, user) + id = "" + if federation: + scenario_builder = ScenarioBuilder(federation_id) + 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) + self._start_initial_nodes(scenario_builder, federation) + id = scenario_builder.get_scenario_name() + try: + nebula_federation = self.nfp[federation_id] + nebula_federation.scenario_name = id + except Exception as e: + self.logger.info(f"ERROR: federation ID: ({federation_id}) not found on pool..") + return None + else: + self.logger.info(f"ERROR: federation ID: ({federation_id}) already exists..") + return id + + async def stop_scenario(self, federation_id: str = ""): + """ + Stop running participant nodes by removing the scenario command files. + + This method deletes the 'current_scenario_commands.sh' (or '.ps1' on Windows) + file associated with a scenario. Removing this file signals the nodes to stop + by terminating their processes. + + Args: + scenario_name (str, optional): The name of the scenario to stop. If None, + all scenarios' command files will be removed. + + Notes: + - If the environment variable NEBULA_CONFIG_DIR is not set, a default + configuration directory path is used. + - Supports both Linux/macOS ('.sh') and Windows ('.ps1') script files. + - Any errors during file removal are logged with the traceback. + """ + federation_name = await self._remove_nebula_federation_from_pool(federation_id) + if not federation_name: + return False + + # When stopping the nodes, we need to remove the current_scenario_commands.sh file -> it will cause the nodes to stop using PIDs + try: + nebula_config_dir = os.environ.get("NEBULA_CONFIG_DIR") + if not nebula_config_dir: + current_dir = os.path.dirname(__file__) + nebula_base_dir = os.path.abspath(os.path.join(current_dir, "..", "..")) + nebula_config_dir = os.path.join(nebula_base_dir, "app", "config") + self.logger.info(f"NEBULA_CONFIG_DIR not found. Using default path: {nebula_config_dir}") + if federation_id: + if os.environ.get("NEBULA_HOST_PLATFORM") == "windows": + scenario_commands_file = os.path.join( + nebula_config_dir, federation_name, "current_scenario_commands.ps1" + ) + else: + scenario_commands_file = os.path.join( + nebula_config_dir, federation_name, "current_scenario_commands.sh" + ) + if os.path.exists(scenario_commands_file): + os.remove(scenario_commands_file) + else: + if os.environ.get("NEBULA_HOST_PLATFORM") == "windows": + files = glob.glob( + os.path.join(nebula_config_dir, "**/current_scenario_commands.ps1"), recursive=True + ) + else: + files = glob.glob( + os.path.join(nebula_config_dir, "**/current_scenario_commands.sh"), recursive=True + ) + for file in files: + os.remove(file) + return True + except Exception as e: + self.logger.exception(f"Error while removing current_scenario_commands.sh file: {e}") + return False + + async def update_nodes(self, federation_id: str, node_update_request: NodeUpdateRequest): + config = node_update_request.config + fed_id = config["scenario_args"]["federation_id"] + scenario_name = config["scenario_args"]["name"] + + try: + nebula_federation = self.nfp[fed_id] + self.logger.info(f"Update received from node on federation ID: ({fed_id})") + last_fed_round = nebula_federation.federation_round + additionals = await nebula_federation.get_additionals_to_be_deployed(config) # It modifies if neccesary the federation round + if additionals: + current_fed_round = nebula_federation.federation_round + adds_deployed = set() + if current_fed_round != last_fed_round: + self.logger.info(f"Federation Round updating for ID: ({fed_id}), current value: {current_fed_round}") + for index in additionals: + if index in adds_deployed: + continue + + for idx, node in enumerate(nebula_federation.config.participants): + if index == idx: + if index in additionals: + self.logger.info(f"Deploying additional participant: {index}") + #TODO additionals not working + self._start_node(node, nebula_federation.network_name, nebula_federation.base_network_name, nebula_federation.base, nebula_federation.last_index_deployed, nebula_federation, additional=True) + nebula_federation.last_index_deployed += 1 + additionals.remove(index) + adds_deployed.add(index) + payload = node_update_request.model_dump() + asyncio.create_task(self._send_to_hub("update", payload, federation_id=fed_id)) + return {"message": "Node updated successfully in Federation Controller"} + except Exception as e: + self.logger.info(f"ERROR: federation ID: ({fed_id}) not found on pool..") + return {"message": "Node updated failed in Federation Controller, ID not found.."} + + async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest): + nebula_federation = self.nfp[federation_id] + self.logger.info(f"Node-Done received from node on federation ID: ({federation_id})") + + if await nebula_federation.is_experiment_finish(): + payload = node_done_request.model_dump() + self.logger.info(f"All nodes have finished on federation ID: ({federation_id}), reporting to hub..") + await self._remove_nebula_federation_from_pool(federation_id) + asyncio.create_task(self._send_to_hub("finish", payload, federation_id=federation_id)) + + payload = node_done_request.model_dump() + asyncio.create_task(self._send_to_hub("done", payload, federation_id=federation_id)) + return {"message": "Nodes done received successfully"} + + """ ############################### + # FUNCTIONALITIES # + ############################### + """ + + async def _add_nebula_federation_to_pool(self, federation_id: str, user: str): + fed = None + async with self._federations_dict_lock: + if not federation_id in self.nfp: + fed = NebulaFederationProcesses() + self.nfp[federation_id] = fed + self.logger.info(f"SUCCESS: new ID: ({federation_id}) added to the pool") + else: + self.logger.info(f"ERROR: trying to add ({federation_id}) to federations pool..") + return fed + + async def _remove_nebula_federation_from_pool(self, federation_id: str): + async with self._federations_dict_lock: + if federation_id in self.nfp: + federation = self.nfp.pop(federation_id) + self.logger.info(f"SUCCESS: Federation ID: ({federation_id}) removed from pool") + return federation.scenario_name + else: + self.logger.info(f"ERROR: trying to remove ({federation_id}) from federations pool..") + return "" + + async def _send_to_hub(self, path, payload, scenario_name="", federation_id="" ): + 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}") + await APIUtils.post(url_request, payload) + except Exception as e: + self.logger.info(f"Failed to send update to Hub: {e}") + + async def _initialize_scenario(self, sb: ScenarioBuilder, scenario_data, federation: NebulaFederationProcesses): + # 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() + + 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.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")) + + 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(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(self.cert_dir, 0o777) + + # Save the scenario configuration + scenario_file = os.path.join(self.config_dir, "scenario.json") + with open(scenario_file, "w") as f: + json.dump(scenario_data, f, sort_keys=False, indent=2) + + os.chmod(scenario_file, 0o777) + + # Save management settings + settings = { + "scenario_name": scenario_name, + "root_path": self.root_path, + "config_dir": self.config_dir, + "log_dir": self.log_dir, + "cert_dir": self.cert_dir, + "env": None, + } + + settings_file = os.path.join(self.config_dir, "settings.json") + with open(settings_file, "w") as f: + json.dump(settings, f, sort_keys=False, indent=2) + + os.chmod(settings_file, 0o777) + + # Attacks assigment and mobility + self.logger.info("πŸ”§ Building general configuration") + sb.build_general_configuration() + self.logger.info("βœ… Building general configuration done") + + # Create participant configs and .json + for index, (_, node) in enumerate(sb.get_federation_nodes().items()): + 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") + self.logger.info(f"Filename: {participant_file}") + os.makedirs(os.path.dirname(participant_file), exist_ok=True) + except Exception as e: + self.logger.info(f"ERROR while creating files: {e}") + + try: + participant_config = sb.build_scenario_config_for_node(index, node) + #self.logger.info(f"dictionary: {participant_config}") + except Exception as e: + self.logger.info(f"ERROR while building configuration for node: {e}") + + try: + with open(participant_file, "w") as f: + json.dump(participant_config, f, sort_keys=False, indent=2) + os.chmod(participant_file, 0o777) + except Exception as e: + self.logger.info(f"ERROR while dumping configuration into files: {e}") + + self.logger.info("βœ… Initializing Scenario Builder done") + + 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.sort() + if len(participant_files) == 0: + raise ValueError("No participant files found in config folder") + + federation.config.set_participants_config(participant_files) + n_nodes = len(participant_files) + self.logger.info(f"Number of nodes: {n_nodes}") + + sb.create_topology_manager(federation.config) + + # Update participants configuration + is_start_node = False + config_participants = [] + + additional_participants = sb.get_additional_nodes() + additional_nodes = len(additional_participants) if additional_participants else 0 + + participant_files.sort(key=lambda x: int(x.split("_")[-1].split(".")[0])) + + # Initial participants + 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: + 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) + 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: + 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") + + config_participants.append(( + participant_config["network_args"]["ip"], + participant_config["network_args"]["port"], + participant_config["device_args"]["role"], + )) + if participant_config["device_args"]["start"]: + if not is_start_node: + is_start_node = True + else: + raise ValueError("Only one node can be start node") + + self.logger.info("βœ… Building preload configuration for initial nodes done") + + 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) + + # Additional participants + self.logger.info("πŸ”§ Building preload configuration for additional nodes...") + additional_participants_files = [] + if additional_participants: + last_participant_file = participant_files[-1] + 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" + shutil.copy(last_participant_file, additional_participant_file) + + with open(additional_participant_file) as f: + participant_config = json.load(f) + + self.logger.info(f"Configuration | additional nodes | participant: {n_nodes + i}") + sb.build_preload_additional_node_configuration(last_participant_index, i, participant_config) + + with open(additional_participant_file, "w") as f: + json.dump(participant_config, f, sort_keys=False, indent=2) + + additional_participants_files.append(additional_participant_file) + + if additional_participants_files: + federation.config.add_participants_config(additional_participants_files) + + if additional_participants: + n_nodes += len(additional_participants) + + self.logger.info("βœ… Building preload configuration for additional nodes done") + self.logger.info("βœ… Loading Scenario configuration done") + + # Build dataset + dataset = sb.configure_dataset(self.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") + + def _start_initial_nodes(self, sb: ScenarioBuilder, federation: NebulaFederationProcesses): + self.logger.info("Starting nodes as processes...") + self.logger.info(f"Number of participants: {len(federation.config.participants)}") + federation.config.participants.sort(key=lambda x: x["device_args"]["idx"]) + federation.last_index_deployed = 2 + + commands = "" + commands = self._build_initial_commands() + if not commands: + self.logger.info("ERROR: Cannot create commands file, abort..") + return + + for idx, node in enumerate(federation.config.participants): + if node["deployment_args"]["additional"]: + federation.additionals_participants[idx] = int(node["deployment_args"]["deployment_round"]) + federation.participants_alive += 1 + self.logger.info(f"Participant {idx} is additional. Round of deployment: {int(node['deployment_args']['deployment_round'])}") + else: + # deploy initial nodes + self.logger.info(f"Deployment starting for participant {idx}") + federation.round_per_participant[idx] = 0 + node_command = self._start_node(sb, node, federation.network_name, federation.base_network_name, federation.base, federation.last_index_deployed, federation) + commands += node_command + if node_command: + federation.last_index_deployed += 1 + federation.participants_alive += 1 + + if federation.config.participants and commands: + self._write_commands_on_file(commands) + else: + self.logger.info("ERROR: No commands on a proccesses deployment..") + + def _start_node(self, sb: ScenarioBuilder, node, network_name, base_network_name, base, i, federation: NebulaFederationProcesses, additional=False): + 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["scenario_args"]["controller"] = self.url + node["scenario_args"]["deployment"] = sb.get_deployment() + node["security_args"]["certfile"] = os.path.join( + self.root_path, "app", "certs", f"participant_{node['device_args']['idx']}_cert.pem" + ) + node["security_args"]["keyfile"] = os.path.join( + self.root_path, "app", "certs", f"participant_{node['device_args']['idx']}_key.pem" + ) + 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: + json.dump(node, f, indent=4) + + self.logger.info(f"Configuration file created successfully: {node_idx}") + commands = "" + try: + if self.host_platform == "windows": + if node["device_args"]["start"]: + commands += "Start-Sleep -Seconds 10\n" + 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' + # 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 + Add-Content -Path $PID_FILE -Value $process.Id + """ + else: + if node["device_args"]["start"]: + commands += "sleep 10\n" + 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 += "echo $! >> $PID_FILE\n\n" + except Exception as e: + raise Exception(f"Error starting nodes as processes: {e}") + + return commands + + def _build_initial_commands(self): + commands = "" + try: + if self.host_platform == "windows": + commands = """ + $ParentDir = Split-Path -Parent $PSScriptRoot + $PID_FILE = "$PSScriptRoot\\current_scenario_pids.txt" + New-Item -Path $PID_FILE -Force -ItemType File + + """ + else: + commands = '#!/bin/bash\n\nPID_FILE="$(dirname "$0")/current_scenario_pids.txt"\n\n> $PID_FILE\n\n' + except Exception as e: + raise Exception(f"Error starting nodes as processes: {e}") + return commands + + def _write_commands_on_file(self, commands: str): + 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: + #self.logger.info(f"Process commands: {commands}") + f.write(commands) + os.chmod(f"{self.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: + #self.logger.info(f"Process commands: {commands}") + f.write(commands) + os.chmod(f"{self.config_dir}/current_scenario_commands.sh", 0o755) + except Exception as e: + raise Exception(f"Error starting nodes as processes: {e}") \ No newline at end of file diff --git a/nebula/controller/federation/factory_federation_controller.py b/nebula/controller/federation/factory_federation_controller.py new file mode 100644 index 000000000..1cdfc86b2 --- /dev/null +++ b/nebula/controller/federation/factory_federation_controller.py @@ -0,0 +1,15 @@ +from nebula.controller.federation.federation_controller import FederationController + +def federation_controller_factory(mode: str, wa_controller_url: str, logger) -> FederationController: + from nebula.controller.federation.controllers.docker_federation_controller import DockerFederationController + from nebula.controller.federation.controllers.processes_federation_controller import ProcessesFederationController + from nebula.controller.federation.controllers.physicall_federation_controller import PhysicalFederationController + + if mode == "docker": + return DockerFederationController(wa_controller_url, logger) + elif mode == "physical": + return PhysicalFederationController(wa_controller_url, logger) + elif mode == "process": + return ProcessesFederationController(wa_controller_url, logger) + else: + raise ValueError("Unknown federation mode") \ No newline at end of file diff --git a/nebula/controller/federation/federation_api.py b/nebula/controller/federation/federation_api.py new file mode 100644 index 000000000..746696c4c --- /dev/null +++ b/nebula/controller/federation/federation_api.py @@ -0,0 +1,113 @@ +import argparse +import os +import logging +from fastapi import FastAPI, Body, Path, Request +from fastapi.concurrency import asynccontextmanager +from typing import Dict +from typing import Annotated +from functools import wraps +from fastapi import HTTPException +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 + +fed_controllers: Dict[str, FederationController] = {} + +@asynccontextmanager +async def lifespan(app: FastAPI): + log_path = os.environ.get("NEBULA_FEDERATION_CONTROLLER_LOG") + + # Configure and register the logger under the name "controller" + LoggerUtils.configure_logger(name="Federation-Controller", log_file=log_path) + + # Retrieve the logger by name + logger = logging.getLogger("Federation-Controller") + logger.info("Logger initialized for Federation Controller") + + # Create all controller types + hub_port = os.environ.get("NEBULA_CONTROLLER_PORT") + controller_host = os.environ.get("NEBULA_CONTROLLER_HOST") + hub_url = f"http://{controller_host}:{hub_port}" + + #["docker", "processes", "physical"] + for exp_type in ["docker", "process"]: + fed_controllers[exp_type] = federation_controller_factory(exp_type, hub_url, logger) + logger.info(f"{exp_type} Federation controller created.") + + yield + +app = FastAPI(lifespan=lifespan) + +@app.get("/") +async def read_root(): + """ + Root endpoint of the NEBULA Controller API. + + Returns: + dict: A welcome message indicating the API is accessible. + """ + logger = logging.getLogger("Federation-Controller") + logger.info("Test curl succesfull") + return {"message": "Welcome to the NEBULA Federation Controller API"} + +@app.post(Routes.RUN) +async def run_scenario(run_scenario_request: RunScenarioRequest): + global fed_controllers + experiment_type = run_scenario_request.scenario_data["deployment"] + logger = logging.getLogger("Federation-Controller") + logger.info(f"[API]: run experiment request for deployment type: {experiment_type}") + controller = fed_controllers.get(experiment_type, None) + if controller: + return await controller.run_scenario(run_scenario_request.federation_id, run_scenario_request.scenario_data, run_scenario_request.user) + else: + return {"message": "Experiment type not allowed"} + +@app.post(Routes.STOP) +async def stop_scenario(stop_scenario_request: StopScenarioRequest): + global fed_controllers + experiment_type = stop_scenario_request.experiment_type + controller = fed_controllers.get(experiment_type, None) + logger = logging.getLogger("Federation-Controller") + logger.info(f"[API]: stop experiment request for federation ID: {stop_scenario_request.federation_id}") + if controller: + return await controller.stop_scenario(stop_scenario_request.federation_id) + else: + return {"message": "Experiment type not allowed"} + +@app.post(Routes.UPDATE) +async def update_nodes( + federation_id: str, + node_update_request: NodeUpdateRequest, +): + global fed_controllers + experiment_type = node_update_request.config["scenario_args"]["deployment"] + controller = fed_controllers.get(experiment_type, None) + if controller: + return await controller.update_nodes(federation_id, node_update_request) + else: + return {"message": "Experiment type not allowed on response for update message.."} + +@app.post(Routes.DONE) +async def node_done( + federation_id: str, + node_done_request: NodeDoneRequest, +): + global fed_controllers + experiment_type = node_done_request.deployment + controller = fed_controllers.get(experiment_type, None) + if controller: + return await controller.node_done(federation_id, node_done_request) + else: + return {"message": "Experiment type not allowed on responde for Node done message.."} + +if __name__ == "__main__": + # Parse args from command line + parser = argparse.ArgumentParser() + parser.add_argument("--port", type=int, default=5051, help="Port to run the Federation controller on.") + args = parser.parse_args() + + import uvicorn + uvicorn.run(app, host="0.0.0.0", port=args.port) + + \ No newline at end of file diff --git a/nebula/controller/federation/federation_controller.py b/nebula/controller/federation/federation_controller.py new file mode 100644 index 000000000..99bdda0bc --- /dev/null +++ b/nebula/controller/federation/federation_controller.py @@ -0,0 +1,35 @@ +from abc import ABC, abstractmethod +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 +import logging + +class NebulaFederation(ABC): + pass + +class FederationController(ABC): + + def __init__(self, hub_url, logger): + self._logger: logging.Logger = logger + self._hub_url = hub_url + + @property + def logger(self): + return self._logger + + @abstractmethod + async def run_scenario(self, federation_id: str, scenario_data: Dict, user: str): + pass + + @abstractmethod + async def stop_scenario(self, federation_id: str): + pass + + @abstractmethod + async def update_nodes(self, federation_id: str, node_update_request: NodeUpdateRequest): + pass + + abstractmethod + async def node_done(self, federation_id: str, node_done_request: NodeDoneRequest): + pass \ No newline at end of file diff --git a/nebula/controller/federation/scenario_builder.py b/nebula/controller/federation/scenario_builder.py new file mode 100644 index 000000000..4e7780563 --- /dev/null +++ b/nebula/controller/federation/scenario_builder.py @@ -0,0 +1,895 @@ +import logging +from datetime import datetime +import hashlib +import math +from collections import defaultdict +from nebula.addons.topologymanager import TopologyManager +from nebula.config.config import Config +from nebula.core.utils.certificate import generate_certificate +from nebula.core.datasets.nebuladataset import NebulaDataset, factory_nebuladataset, factory_dataset_setup + +class ScenarioBuilder(): + def __init__(self, federation_id): + 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 + + @property + def sd(self): + """Scenario data dict""" + return self._scenario_data + + @property + def tm(self): + """Topology Manager""" + return self._topology_manager + + def get_scenario_name(self): + return self._scenario_name + + def set_scenario_data(self, scenario_data: dict): + self._scenario_data = scenario_data + federation_name = self.sd["federation"] + self._scenario_name = f"nebula_{federation_name}_{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}" + + def set_config_setup(self, setup: dict): + self._config_setup = setup + + def get_federation_nodes(self) -> dict: + return self.sd["nodes"] + + def get_additional_nodes(self): + return self.sd["additional_participants"] + + def get_dataset_name(self) -> str: + return self.sd["dataset"] + + def get_deployment(self) -> str: + return self.sd["deployment"] + + """ ############################### + # SCENARIO CONFIG NODE # + ############################### + """ + def build_general_configuration(self): + try: + self.sd["nodes"] = self._configure_nodes_attacks() + + if self.sd.get("mobility", None): + mobile_participants_percent = int(self.sd["mobile_participants_percent"]) + self.sd["nodes"] = self._mobility_assign(self.sd["nodes"], mobile_participants_percent) + else: + self.sd["nodes"] = self._mobility_assign(self.sd["nodes"], 0) + except Exception as e: + self.logger.info(f"ERROR: {e}") + + def _configure_nodes_attacks(self): + self.logger.info("Configurating node attacks...") + poisoned_node_percent = self.sd["attack_params"].get("poisoned_node_percent", 0) + poisoned_sample_percent = self.sd["attack_params"].get("poisoned_sample_percent", 0) + poisoned_noise_percent = self.sd["attack_params"].get("poisoned_noise_percent", 0) + + nodes = self.attack_node_assign( + self.sd.get("nodes"), + self.sd.get("federation"), + int(poisoned_node_percent), + int(poisoned_sample_percent), + int(poisoned_noise_percent), + self.sd.get("attack_params"), + ) + + self.logger.info("Configurating node attacks done") + return nodes + + def attack_node_assign( + self, + nodes, + federation, + poisoned_node_percent, + poisoned_sample_percent, + poisoned_noise_percent, + attack_params, + ): + """ + Assign and configure attack parameters to nodes within a federated learning network. + + This method: + - Validates input attack parameters and percentages. + - Determines which nodes will be marked as malicious based on the specified + poisoned node percentage and attack type. + - Assigns attack roles and parameters to selected nodes. + - Supports multiple attack types such as Label Flipping, Sample Poisoning, + Model Poisoning, GLL Neuron Inversion, Swapping Weights, Delayer, and Flooding. + - Ensures proper validation and setting of attack-specific parameters, including + targeting, noise types, delays, intervals, and attack rounds. + - Updates nodes' malicious status, reputation, and attack parameters accordingly. + + Args: + nodes (dict): Dictionary of nodes with their current attributes. + federation (str): Type of federated learning framework (e.g., "DFL"). + poisoned_node_percent (float): Percentage of nodes to be poisoned (0-100). + poisoned_sample_percent (float): Percentage of samples to be poisoned (0-100). + poisoned_noise_percent (float): Percentage of noise to apply in poisoning (0-100). + attack_params (dict): Dictionary containing attack type and associated parameters. + + Returns: + dict: Updated nodes dictionary with assigned malicious roles and attack parameters. + + Raises: + ValueError: If any input parameter is invalid or attack type is unrecognized. + """ + import random + + # Validate input parameters + def validate_percentage(value, name): + """ + Validate that a given value is a float percentage between 0 and 100. + + Args: + value: The value to validate, expected to be convertible to float. + name (str): Name of the parameter, used for error messages. + + Returns: + float: The validated percentage value. + + Raises: + ValueError: If the value is not a float or not within the range [0, 100]. + """ + try: + value = float(value) + if not 0 <= value <= 100: + raise ValueError(f"{name} must be between 0 and 100") + return value + except (TypeError, ValueError) as e: + raise ValueError(f"Invalid {name}: {e!s}") + + def validate_positive_int(value, name): + """ + Validate that a given value is a positive integer (including zero). + + Args: + value: The value to validate, expected to be convertible to int. + name (str): Name of the parameter, used for error messages. + + Returns: + int: The validated positive integer value. + + Raises: + ValueError: If the value is not an integer or is negative. + """ + try: + value = int(value) + if value < 0: + raise ValueError(f"{name} must be positive") + return value + except (TypeError, ValueError) as e: + raise ValueError(f"Invalid {name}: {e!s}") + + # Validate attack type + valid_attacks = { + "No Attack", + "Label Flipping", + "Sample Poisoning", + "Model Poisoning", + "GLL Neuron Inversion", + "Swapping Weights", + "Delayer", + "Flooding", + } + + # Get attack type from attack_params + if attack_params and "attacks" in attack_params: + attack = attack_params["attacks"] + + # Handle attack parameter which can be either a string or None + if attack is None: + attack = "No Attack" + elif not isinstance(attack, str): + raise ValueError(f"Invalid attack type: {attack}. Expected string or None.") + + if attack not in valid_attacks: + raise ValueError(f"Invalid attack type: {attack}. Must be one of {valid_attacks}") + + # Get attack parameters from attack_params + poisoned_node_percent = attack_params.get("poisoned_node_percent", poisoned_node_percent) + poisoned_sample_percent = attack_params.get("poisoned_sample_percent", poisoned_sample_percent) + poisoned_noise_percent = attack_params.get("poisoned_noise_percent", poisoned_noise_percent) + + # Validate percentage parameters + poisoned_node_percent = validate_percentage(poisoned_node_percent, "poisoned_node_percent") + poisoned_sample_percent = validate_percentage(poisoned_sample_percent, "poisoned_sample_percent") + poisoned_noise_percent = validate_percentage(poisoned_noise_percent, "poisoned_noise_percent") + + nodes_index = [] + # Get the nodes index + if federation == "DFL": + nodes_index = list(nodes.keys()) + else: + for node in nodes: + if nodes[node]["role"] != "server": + nodes_index.append(node) + + self.logger.info(f"Nodes index: {nodes_index}") + self.logger.info(f"Attack type: {attack}") + self.logger.info(f"Poisoned node percent: {poisoned_node_percent}") + + mal_nodes_defined = any(nodes[node]["malicious"] for node in nodes) + self.logger.info(f"Malicious nodes already defined: {mal_nodes_defined}") + + attacked_nodes = [] + + if not mal_nodes_defined and attack != "No Attack": + n_nodes = len(nodes_index) + # Number of attacked nodes, round up + num_attacked = int(math.ceil(poisoned_node_percent / 100 * n_nodes)) + if num_attacked > n_nodes: + num_attacked = n_nodes + + # Get the index of attacked nodes + attacked_nodes = random.sample(nodes_index, num_attacked) + self.logger.info(f"Number of nodes to attack: {num_attacked}") + self.logger.info(f"Attacked nodes: {attacked_nodes}") + + # Assign the role of each node + for node in nodes: + node_att = "No Attack" + malicious = False + #node_reputation = self.reputation.copy() if self.reputation else None + + if node in attacked_nodes or nodes[node]["malicious"]: + malicious = True + node_reputation = None + node_att = attack + self.logger.info(f"Node {node} marked as malicious with attack {attack}") + + # Initialize attack parameters with defaults + node_attack_params = attack_params.copy() if attack_params else {} + + # Set attack-specific parameters + if attack == "Label Flipping": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + node_attack_params["poisoned_sample_percent"] = poisoned_sample_percent + node_attack_params["targeted"] = attack_params.get("targeted", False) + if node_attack_params["targeted"]: + node_attack_params["target_label"] = validate_positive_int( + attack_params.get("target_label", 4), "target_label" + ) + node_attack_params["target_changed_label"] = validate_positive_int( + attack_params.get("target_changed_label", 7), "target_changed_label" + ) + + elif attack == "Sample Poisoning": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + node_attack_params["poisoned_sample_percent"] = poisoned_sample_percent + node_attack_params["poisoned_noise_percent"] = poisoned_noise_percent + node_attack_params["noise_type"] = attack_params.get("noise_type", "Gaussian") + node_attack_params["targeted"] = attack_params.get("targeted", False) + if node_attack_params["targeted"]: + node_attack_params["target_label"] = validate_positive_int( + attack_params.get("target_label", 4), "target_label" + ) + + elif attack == "Model Poisoning": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + node_attack_params["poisoned_noise_percent"] = poisoned_noise_percent + node_attack_params["noise_type"] = attack_params.get("noise_type", "Gaussian") + + elif attack == "GLL Neuron Inversion": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + + elif attack == "Swapping Weights": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + node_attack_params["layer_idx"] = validate_positive_int( + attack_params.get("layer_idx", 0), "layer_idx" + ) + + elif attack == "Delayer": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + node_attack_params["delay"] = validate_positive_int(attack_params.get("delay", 10), "delay") + node_attack_params["target_percentage"] = validate_percentage( + attack_params.get("target_percentage", 100), "target_percentage" + ) + node_attack_params["selection_interval"] = validate_positive_int( + attack_params.get("selection_interval", 1), "selection_interval" + ) + + elif attack == "Flooding": + node_attack_params["poisoned_node_percent"] = poisoned_node_percent + node_attack_params["flooding_factor"] = validate_positive_int( + attack_params.get("flooding_factor", 100), "flooding_factor" + ) + node_attack_params["target_percentage"] = validate_percentage( + attack_params.get("target_percentage", 100), "target_percentage" + ) + node_attack_params["selection_interval"] = validate_positive_int( + attack_params.get("selection_interval", 1), "selection_interval" + ) + + # Add common attack parameters + node_attack_params["round_start_attack"] = validate_positive_int( + attack_params.get("round_start_attack", 1), "round_start_attack" + ) + node_attack_params["round_stop_attack"] = validate_positive_int( + attack_params.get("round_stop_attack", 10), "round_stop_attack" + ) + node_attack_params["attack_interval"] = validate_positive_int( + attack_params.get("attack_interval", 1), "attack_interval" + ) + + # Validate round parameters + if node_attack_params["round_start_attack"] >= node_attack_params["round_stop_attack"]: + raise ValueError("round_start_attack must be less than round_stop_attack") + + node_attack_params["attacks"] = node_att + nodes[node]["malicious"] = True + nodes[node]["attack_params"] = node_attack_params + nodes[node]["fake_behavior"] = nodes[node]["role"] + nodes[node]["role"] = "malicious" + # else: + # nodes[node]["attack_params"] = {"attacks": "No Attack"} + + if nodes[node].get("attack_params", None): + self.logger.info( + f"Node {node} final configuration - malicious: {nodes[node]['malicious']}, attack: {nodes[node]['attack_params']['attacks']}" + ) + else: + self.logger.info( + f"Node {node} final configuration - malicious: {nodes[node]['malicious']}" + ) + + return nodes + + def _mobility_assign(self, nodes, mobile_participants_percent): + """ + Assign mobility status to a subset of nodes based on a specified percentage. + + This method: + - Calculates the number of mobile nodes by applying the given percentage. + - Randomly selects nodes to be marked as mobile. + - Updates each node's "mobility" attribute to True or False accordingly. + + Args: + nodes (dict): Dictionary of nodes with their current attributes. + mobile_participants_percent (float): Percentage of nodes to be assigned mobility (0-100). + + Returns: + dict: Updated nodes dictionary with mobility status assigned. + """ + import random + + # Number of mobile nodes, round down + num_mobile = math.floor(mobile_participants_percent / 100 * len(nodes)) + if num_mobile > len(nodes): + num_mobile = len(nodes) + + # Get the index of mobile nodes + mobile_nodes = random.sample(list(nodes.keys()), num_mobile) + + # Assign the role of each node + for node in nodes: + node_mob = False + if node in mobile_nodes: + node_mob = True + nodes[node]["mobility"] = node_mob + return nodes + + + """ ############################### + # SCENARIO CONFIG NODE # + ############################### + """ + + def build_scenario_config_for_node(self, index, node) -> dict: + self.logger.info(f"Start building the scenario configuration for participant {index}") + + def recursive_defaultdict(): + return defaultdict(recursive_defaultdict) + + def dictify(d): + if isinstance(d, defaultdict): + return {k: dictify(v) for k, v in d.items()} + return d + + participant_config = recursive_defaultdict() + + addons_config = defaultdict() + #participant_config["addons"] = dict() + + # General configuration + participant_config["scenario_args"]["name"] = self._scenario_name + participant_config["scenario_args"]["start_time"] = datetime.now().strftime("%d/%m/%Y %H:%M:%S") + participant_config["scenario_args"]["federation_id"] = self._federation_id + participant_config["deployment_args"]["additional"] = False + + node_config = node #self.sd["nodes"][index] + participant_config["network_args"]["ip"] = node_config["ip"] + if self.sd["deployment"] == "physical": + participant_config["network_args"]["port"] = 8000 + else: + participant_config["network_args"]["port"] = int(node_config["port"]) + + participant_config["network_args"]["simulation"] = self.sd["network_simulation"] + participant_config["device_args"]["idx"] = node_config["id"] + participant_config["device_args"]["start"] = node_config["start"] + participant_config["device_args"]["role"] = node_config["role"] + participant_config["device_args"]["proxy"] = node_config["proxy"] + participant_config["device_args"]["malicious"] = node_config["malicious"] + participant_config["scenario_args"]["rounds"] = int(self.sd["rounds"]) + participant_config["scenario_args"]["random_seed"] = 42 + participant_config["federation_args"]["round"] = 0 + participant_config["data_args"]["dataset"] = self.sd["dataset"] + participant_config["data_args"]["iid"] = self.sd["iid"] + participant_config["data_args"]["num_workers"] = 0 + participant_config["data_args"]["partition_selection"] = self.sd["partition_selection"] + participant_config["data_args"]["partition_parameter"] = self.sd["partition_parameter"] + participant_config["model_args"]["model"] = self.sd["model"] + participant_config["training_args"]["epochs"] = int(self.sd["epochs"]) + participant_config["training_args"]["trainer"] = "lightning" + participant_config["device_args"]["accelerator"] = self.sd["accelerator"] + participant_config["device_args"]["gpu_id"] = self.sd["gpu_id"] + participant_config["device_args"]["logging"] = self.sd["logginglevel"] + participant_config["aggregator_args"]["algorithm"] = self.sd["agg_algorithm"] + participant_config["aggregator_args"]["aggregation_timeout"] = 60 + + participant_config["message_args"]= self._configure_message_args() + participant_config["reporter_args"]= self._configure_reporter_args() + participant_config["forwarder_args"]= self._configure_forwarder_args() + participant_config["propagator_args"]= self._configure_propagator_args() + participant_config["misc_args"]= self._configure_misc_args() + + # Addons configuration + + # Trustworthiness + try: + if self.sd.get("with_trustworthiness", None): + addons_config["trustworthiness"] = self._configure_trustworthiness() + except Exception as e: + self.logger.info(f"ERROR: Cannot build trustworthiness configuration - {e}") + + # Reputation + try: + if self.sd.get("reputation", None) and self.sd["reputation"]["enabled"] and not node_config["role"] == "malicious": + addons_config["reputation"] = self._configure_reputation() + except Exception as e: + self.logger.info(f"ERROR: Cannot build reputation configuration - {e}") + + # Network simulation + try: + network_args: dict = (self.sd.get("network_args"), None) + if network_args and isinstance(network_args, dict) and network_args.get("enabled", None): + addons_config["network_simulation"] = self._configure_network_simulation() + except Exception as e: + self.logger.info(f"ERROR: Cannot build network simulation configuration - {e}") + + # Attacks + try: + if node_config["role"] == "malicious": + addons_config["adversarial_args"] = self._configure_malicious_role(node_config) + except Exception as e: + self.logger.info(f"ERROR: Cannot build role configuration - {e}") + + # Mobility + try: + if self.sd.get("mobility", None): + addons_config["mobility"] = self._configure_mobility_args() + except Exception as e: + self.logger.info(f"ERROR: Cannot build mobility configuration - {e}") + + # Situational awareness module + try: + if self._situational_awareness_needed(): + addons_config["situational_awareness"] = self._configure_situational_awareness(index) + except Exception as e: + self.logger.info(f"ERROR: Cannot build situational awareness configuration - {e}") + + # Addon addition to the configuration + participant_config["addons"] = addons_config + + try: + config = dictify(participant_config) + except Exception as e: + self.logger.info(f"ERROR: Translating into dictionary - {e}") + + return config + + def _configure_message_args(self): + return { + "max_local_messages": 10000, + "compression": "zlib" + } + + def _configure_reporter_args(self): + return { + "grace_time_reporter": 10, + "report_frequency": 5, + "report_status_data_queue": True + } + + def _configure_forwarder_args(self): + return { + "forwarder_interval": 1, + "forward_messages_interval": 0, + "number_forwarded_messages": 100 + } + + def _configure_propagator_args(self): + return { + "propagate_interval": 3, + "propagate_model_interval": 0, + "propagation_early_stop": 3, + "history_size": 20 + } + + def _configure_misc_args(self): + return { + "grace_time_connection": 10, + "grace_time_start_federation": 10 + } + + def _configure_mobility_args(self): + return { + "enabled": True, + "mobility_type": self.sd["mobility_type"], + "topology_type": self.sd["topology"], + "radius_federation": self.sd["radius_federation"], + "scheme_mobility": self.sd["scheme_mobility"], + "round_frequency": self.sd["round_frequency"], + "grace_time_mobility": 60, + "change_geo_interval": 5 + } + + def _configure_malicious_role(self, node_config: dict): + return { + "fake_behavior": node_config["fake_behavior"], + "attack_params": node_config["attack_params"] + } + + def _configure_trustworthiness(self) -> dict: + trust_config = { + "robustness_pillar": self.sd["robustness_pillar"], + "resilience_to_attacks": self.sd["resilience_to_attacks"], + "algorithm_robustness": self.sd["algorithm_robustness"], + "client_reliability": self.sd["client_reliability"], + "privacy_pillar": self.sd["privacy_pillar"], + "technique": self.sd["technique"], + "uncertainty": self.sd["uncertainty"], + "indistinguishability": self.sd["indistinguishability"], + "fairness_pillar": self.sd["fairness_pillar"], + "selection_fairness": self.sd["selection_fairness"], + "performance_fairness": self.sd["performance_fairness"], + "class_distribution": self.sd["class_distribution"], + "explainability_pillar": self.sd["explainability_pillar"], + "interpretability": self.sd["interpretability"], + "post_hoc_methods": self.sd["post_hoc_methods"], + "accountability_pillar": self.sd["accountability_pillar"], + "factsheet_completeness": self.sd["factsheet_completeness"], + "architectural_soundness_pillar": self.sd["architectural_soundness_pillar"], + "client_management": self.sd["client_management"], + "optimization": self.sd["optimization"], + "sustainability_pillar": self.sd["sustainability_pillar"], + "energy_source": self.sd["energy_source"], + "hardware_efficiency": self.sd["hardware_efficiency"], + "federation_complexity": self.sd["federation_complexity"], + "scenario": self.sd, + } + return trust_config + + def _configure_reputation(self) -> dict: + rep = self.sd.get("reputation") + rep["adaptive_args"] = True + return rep + + def _configure_network_simulation(self) -> dict: + network_parameters = {} + network_generation = dict(self.sd["network_args"]).pop("network_type") + enabled = dict(self.sd["network_args"]).pop("enabled") + type = dict(self.sd["network_args"]).pop("type") + addrs = "" + + for node in self.sd["nodes"]: + ip = self.sd["nodes"][node]["ip"] + port = self.sd["nodes"][node]["port"] + addrs = addrs + " " + f"{ip}:{port}" + + network_configuration = { + "interface": "eth0", + "verbose": False, + "preset": network_generation, + "federation": addrs + } + + network_parameters = { + "enabled": enabled, + "type": type, + "network_config": network_configuration + } + + return network_parameters + + def _situational_awareness_needed(self): + enabled = False + arrivals_dep = self.sd.get("arrivals_departures_args", None) + if arrivals_dep: + enabled = arrivals_dep["enabled"] + with_sa = self.sd.get("with_sa", None) + additionals = self.sd.get("additional_participants", None) + mob = self.sd.get("mobility", None) + + return with_sa or enabled or arrivals_dep or additionals or mob + + def _configure_situational_awareness(self, index) -> dict: + try: + scheduled_isolation = self._configure_arrivals_departures(index) + except Exception as e: + self.logger.info(f"ERROR: cannot configure arrival departures section - {e}") + + snp = self.sd.get("sar_neighbor_policy", None) + topology_management = snp if (snp != "") else self.sd["topology"] + + situational_awareness_config = { + "strict_topology": self.sd["strict_topology"], + "sa_discovery": { + "candidate_selector": topology_management, + "model_handler": self.sd["sad_model_handler"], + "verbose": True, + }, + "sa_reasoner": { + "arbitration_policy": self.sd["sar_arbitration_policy"], + "verbose": True, + "sar_components": { + "sa_network": True, + "sa_training": self.sd["sar_training"] + }, + "sa_network": { + "neighbor_policy": topology_management, + "scheduled_isolation" : scheduled_isolation, + "verbose": True + }, + "sa_training": { + "training_policy": self.sd["sar_training_policy"], + "verbose": True + }, + }, + } + return situational_awareness_config + + def _configure_arrivals_departures(self, index) -> dict: + arrival_dep_section = self.sd.get("arrivals_departures_args", None) + if not arrival_dep_section or (arrival_dep_section and not self.sd["arrivals_departures_args"]["enabled"]): + return {"enabled": False} + + config = {"enabled": True} + departures: list = self.sd["arrivals_departures_args"]["departures"] + index_departure_config: dict = departures[index] + if index_departure_config["round_start"] != "": + config["round_start"] = index_departure_config["round_start"] + config["duration"] = index_departure_config["duration"] if index_departure_config["duration"] != "" else None + else: + config = {"enabled": False} + + return config + + """ ############################### + # PRELOAD CONFIG # + ############################### + """ + + def build_preload_initial_node_configuration(self, index, participant_config: dict, log_dir, config_dir, cert_dir, advanced_analytics): + try: + participant_config["scenario_args"]["federation"] = self.sd["federation"] + n_nodes = len(self.sd["nodes"].keys()) + n_additionals = len(self.sd["additional_participants"]) + participant_config["scenario_args"]["n_nodes"] = n_nodes + n_additionals + + participant_config["network_args"]["neighbors"] = self.tm.get_neighbors_string(index) + + participant_config["device_args"]["idx"] = index + participant_config["device_args"]["uid"] = hashlib.sha1( + ( + str(participant_config["network_args"]["ip"]) + + str(participant_config["network_args"]["port"]) + + str(participant_config["scenario_args"]["name"]) + ).encode() + ).hexdigest() + except Exception as e: + self.logger.info(f"ERROR while setting up general stuff") + + try: + if participant_config.get("addons", None) and participant_config["addons"].get("mobility", None): + if participant_config["addons"]["mobility"].get("random_geo", None): + ( + participant_config["addons"]["mobility"]["latitude"], + participant_config["addons"]["mobility"]["longitude"], + ) = TopologyManager.get_coordinates(random_geo=True) + else: + participant_config["addons"]["mobility"]["latitude"] = self.sd["latitude"] + participant_config["addons"]["mobility"]["longitude"] = self.sd["longitude"] + except Exception as e: + self.logger.info(f"ERROR while setting up mobility parameters - {e}") + + try: + participant_config["tracking_args"] = {} + participant_config["security_args"] = {} + + # If not, use the given coordinates in the frontend + participant_config["tracking_args"]["local_tracking"] = "default" + participant_config["tracking_args"]["log_dir"] = log_dir + participant_config["tracking_args"]["config_dir"] = config_dir + # Generate node certificate + keyfile_path, certificate_path = generate_certificate( + dir_path=cert_dir, + node_id=f"participant_{index}", + ip=participant_config["network_args"]["ip"], + ) + participant_config["security_args"]["certfile"] = certificate_path + participant_config["security_args"]["keyfile"] = keyfile_path + except Exception as e: + self.logger.info(f"ERROR while setting up tracking args and certificates") + + def build_preload_additional_node_configuration(self, last_participant_index, index, participant_config): + n_nodes = len(self.sd["nodes"].keys()) + n_additionals = len(self.sd["additional_participants"]) + last_ip = participant_config["network_args"]["ip"] + participant_config["scenario_args"]["n_nodes"] = n_nodes + n_additionals # self.n_nodes + i + 1 + participant_config["device_args"]["idx"] = last_participant_index + index + participant_config["network_args"]["neighbors"] = "" + participant_config["network_args"]["ip"] = ( + participant_config["network_args"]["ip"].rsplit(".", 1)[0] + + "." + + str(int(participant_config["network_args"]["ip"].rsplit(".", 1)[1]) + index + 1) + ) + participant_config["device_args"]["uid"] = hashlib.sha1( + ( + str(participant_config["network_args"]["ip"]) + + str(participant_config["network_args"]["port"]) + + str(self._scenario_name) + ).encode() + ).hexdigest() + participant_config["deployment_args"]["additional"] = True + + deployment_round = self.sd["additional_participants"][index]["time_start"] + participant_config["deployment_args"]["deployment_round"] = deployment_round + + # used for late creation nodes + + """ ############################### + # TOPOLOGY MANAGER # + ############################### + """ + + def create_topology_manager(self, config: Config): + try: + self._topology_manager = ( + self._create_topology(config, matrix=self.sd["matrix"]) if self.sd["matrix"] else self._create_topology(config) + ) + except Exception as e: + self.logger.info(f"ERROR: cannot create topology manager - {e}") + + def _create_topology(self, config: Config, matrix=None): + """ + Create and return a network topology manager based on the scenario's topology settings or a given adjacency matrix. + + Supports multiple topology types: + - Random: Generates an ErdΕ‘s-RΓ©nyi random graph with specified connection probability. + - Matrix: Uses a provided adjacency matrix to define the topology. + - Fully: Creates a fully connected network. + - Ring: Creates a ring-structured network with partial connectivity. + - Star: Creates a centralized star topology (only for CFL federation). + + The method assigns IP and port information to nodes and returns the configured TopologyManager instance. + + Args: + matrix (optional): Adjacency matrix to define custom topology. If provided, overrides scenario topology. + + Raises: + ValueError: If an unknown topology type is specified in the scenario. + + Returns: + TopologyManager: Configured topology manager with nodes assigned. + """ + import numpy as np + + n_nodes = len(self.sd["nodes"].keys()) + if self.sd["topology"] == "Random": + # Create network topology using topology manager (random) + probability = float(self.sd["random_topology_probability"]) + logging.info( + f"Creating random network topology using erdos_renyi_graph: nodes={n_nodes}, probability={probability}" + ) + topologymanager = TopologyManager( + scenario_name=self._scenario_name, + n_nodes=n_nodes, + b_symmetric=True, + undirected_neighbor_num=3, + ) + topologymanager.generate_random_topology(probability) + elif matrix is not None: + if n_nodes > 2: + topologymanager = TopologyManager( + topology=np.array(matrix), + scenario_name=self._scenario_name, + n_nodes=n_nodes, + b_symmetric=True, + undirected_neighbor_num=n_nodes - 1, + ) + else: + topologymanager = TopologyManager( + topology=np.array(matrix), + scenario_name=self._scenario_name, + n_nodes=n_nodes, + b_symmetric=True, + undirected_neighbor_num=2, + ) + elif self.sd["topology"] == "Fully": + # Create a fully connected network + topologymanager = TopologyManager( + scenario_name=self._scenario_name, + n_nodes=n_nodes, + b_symmetric=True, + undirected_neighbor_num=n_nodes - 1, + ) + topologymanager.generate_topology() + elif self.sd["topology"] == "Ring": + # Create a partially connected network (ring-structured network) + topologymanager = TopologyManager(scenario_name=self._scenario_name, n_nodes=n_nodes, b_symmetric=True) + topologymanager.generate_ring_topology(increase_convergence=True) + elif self.sd["topology"] == "Star" and self.sd["federation"] == "CFL": + # Create a centralized network + topologymanager = TopologyManager(scenario_name=self._scenario_name, n_nodes=n_nodes, b_symmetric=True) + topologymanager.generate_server_topology() + else: + top = self.sd["topology"] + raise ValueError(f"Unknown topology type: {top}") + + # Assign nodes to topology + nodes_ip_port = [] + config.participants.sort(key=lambda x: int(x["device_args"]["idx"])) + for i, node in enumerate(config.participants): + nodes_ip_port.append(( + node["network_args"]["ip"], + node["network_args"]["port"], + "undefined", + )) + + topologymanager.add_nodes(nodes_ip_port) + return topologymanager + + def visualize_topology(self, config_participants, path, plot): + try: + self.tm.update_nodes(config_participants) + self.tm.draw_graph(path=path, plot=plot) + except Exception as e: + self.logger.info(f"ERROR: cannot visualize topology - {e}") + + """ ############################### + # DATASET CONFIGURATION # + ############################### + """ + + def configure_dataset(self, config_dir) -> NebulaDataset: + try: + dataset_name = self.get_dataset_name() + dataset = factory_nebuladataset( + dataset_name, + **self._configure_dataset_config(dataset_name, config_dir) + ) + except Exception as e: + self.logger.info(f"ERROR: cannot configure dataset - {e}") + return dataset + + def _configure_dataset_config(self, dataset_name, config_dir): + num_classes = factory_dataset_setup(dataset_name) + n_nodes = len(self.sd["nodes"].keys()) + n_nodes += len(self.sd["additional_participants"]) + return { + "num_classes": num_classes, + "partitions_number": n_nodes, + "iid": self.sd["iid"], + "partition": self.sd["partition_selection"], + "partition_parameter": self.sd["partition_parameter"], + "seed": 42, + "config_dir": config_dir, + } \ No newline at end of file diff --git a/nebula/controller/federation/utils_requests.py b/nebula/controller/federation/utils_requests.py new file mode 100644 index 000000000..e501b3377 --- /dev/null +++ b/nebula/controller/federation/utils_requests.py @@ -0,0 +1,46 @@ +from pydantic import BaseModel +from typing import Dict, Any + +class RunScenarioRequest(BaseModel): + scenario_data: Dict[str, Any] + user: str + federation_id: str + +class StopScenarioRequest(BaseModel): + experiment_type: str + federation_id: str + +class NodeUpdateRequest(BaseModel): + config: Dict[str, Any] = {} + +class NodeDoneRequest(BaseModel): + idx: int + deployment: str + name: str + federation_id: str + +class Routes: + INIT = "/init" + RUN = "/scenarios/run" + STOP = "/scenarios/stop" + UPDATE = "/nodes/{federation_id}/update" + DONE = "/nodes/{federation_id}/done" + FINISH = "/scenarios/{federation_id}/finish" + +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 + 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}") + + \ No newline at end of file diff --git a/nebula/controller/scenarios.py b/nebula/controller/scenarios.py index 0d8f2286e..64c5d84a2 100644 --- a/nebula/controller/scenarios.py +++ b/nebula/controller/scenarios.py @@ -569,6 +569,26 @@ def to_json(scenario_obj): # Convert the dictionary to a JSON string return json.dumps(scenario_dict, indent=2) # Using indent for pretty-printing + @staticmethod + def to_json(scenario_obj): + """ + Converts a Scenario object to a JSON string. + + Args: + scenario_obj (Scenario): An instance of the Scenario class. + + Returns: + str: A JSON string representation of the Scenario object. + """ + if not isinstance(scenario_obj, Scenario): + raise TypeError("Input must be an instance of the Scenario class.") + + # Get all attributes of the Scenario object + scenario_dict = scenario_obj.__dict__ + + # Convert the dictionary to a JSON string + return json.dumps(scenario_dict, indent=2) # Using indent for pretty-printing + # Class to manage the current scenario class ScenarioManagement: @@ -993,6 +1013,8 @@ async def load_configurations_and_start_nodes( with open(additional_participant_file) as f: participant_config = json.load(f) + + logging.info(f"Configuration | additional nodes | participant: {self.n_nodes + i + 1}") last_ip = participant_config["network_args"]["ip"] diff --git a/nebula/controller/start_services.sh b/nebula/controller/start_services.sh index fa0eaf031..c0ecb234b 100644 --- a/nebula/controller/start_services.sh +++ b/nebula/controller/start_services.sh @@ -11,7 +11,15 @@ echo "path $(pwd)" # Start Gunicorn NEBULA_SOCK=nebula.sock -echo "Starting Gunicorn..." -uvicorn nebula.controller.controller:app --host 0.0.0.0 --port $NEBULA_CONTROLLER_PORT --log-level debug --proxy-headers --forwarded-allow-ips "*" & +echo "NEBULA_PRODUCTION: $NEBULA_PRODUCTION" +if [ "$NEBULA_PRODUCTION" = "False" ]; then + echo "Starting Gunicorn in dev mode..." + uvicorn nebula.controller.web_app_controller:app --host 0.0.0.0 --port $NEBULA_CONTROLLER_PORT --log-level debug --proxy-headers --forwarded-allow-ips "*" & + uvicorn nebula.controller.federation.federation_api:app --host 0.0.0.0 --port $NEBULA_FEDERATION_CONTROLLER_PORT --log-level debug --proxy-headers --forwarded-allow-ips "*" & +else + echo "Starting Gunicorn in production mode..." + uvicorn nebula.controller.web_app_controller:app --host 0.0.0.0 --port $NEBULA_CONTROLLER_PORT --log-level info --proxy-headers --forwarded-allow-ips "*" & + uvicorn nebula.controller.federation.federation_api:app --host 0.0.0.0 --port $NEBULA_FEDERATION_CONTROLLER_PORT --log-level debug --proxy-headers --forwarded-allow-ips "*" & +fi tail -f /dev/null diff --git a/nebula/controller/controller.py b/nebula/controller/web_app_controller.py similarity index 90% rename from nebula/controller/controller.py rename to nebula/controller/web_app_controller.py index 7e59ae7ec..7835f2a1a 100755 --- a/nebula/controller/controller.py +++ b/nebula/controller/web_app_controller.py @@ -24,7 +24,8 @@ scenario_set_status_to_finished, ) from nebula.controller.http_helpers import remote_get, remote_post_form -from nebula.utils import DockerUtils +from nebula.utils import DockerUtils, APIUtils +from nebula.controller.federation.utils_requests import RunScenarioRequest, StopScenarioRequest, factory_requests_path # Setup controller logger @@ -106,6 +107,7 @@ def configure_logger(controller_log): handler.setFormatter(logging.Formatter("[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s")) logger.addHandler(handler) +id_counter = 1 @asynccontextmanager async def lifespan(app: FastAPI): @@ -316,41 +318,53 @@ async def run_scenario( """ import subprocess - + global id_counter from nebula.controller.scenarios import ScenarioManagement + 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}" + factory_requests_path("init") + url_run_scenario = f"http://{fed_controller_host}:{fed_controller_port}" + factory_requests_path("run") + #init_fed_req = InitFederationRequest(experiment_type="docker") + run_scenario_req = RunScenarioRequest(scenario_data=scenario_data, federation_id=f"id_nebula_{id_counter}", user=user) #TODO ID per experiment + id_counter += 1 + #await APIUtils.post(url_init_fed_controller, init_fed_req.model_dump()) + await APIUtils.post(url_run_scenario, run_scenario_req.model_dump()) + except Exception as e: + logging.info(e) validate_physical_fields(scenario_data) db_scenario = copy.deepcopy(scenario_data) # Manager for the actual scenario - scenarioManagement = ScenarioManagement(scenario_data, user) - - await update_scenario( - scenario_name=scenarioManagement.scenario_name, - start_time=scenarioManagement.start_date_scenario, - end_time="", - scenario=scenario_data, - status="running", - role=role, - username=user, - ) + #scenarioManagement = ScenarioManagement(scenario_data, user) + + # await update_scenario( + # scenario_name=scenarioManagement.scenario_name, + # start_time=scenarioManagement.start_date_scenario, + # end_time="", + # scenario=scenario_data, + # status="running", + # role=role, + # username=user, + # ) # Run the actual scenario - try: - if scenarioManagement.scenario.mobility: - additional_participants = scenario_data["additional_participants"] - schema_additional_participants = scenario_data["schema_additional_participants"] - await scenarioManagement.load_configurations_and_start_nodes( - additional_participants, schema_additional_participants - ) - else: - await scenarioManagement.load_configurations_and_start_nodes() - except subprocess.CalledProcessError as e: - logging.exception(f"Error docker-compose up: {e}") - return - - return scenarioManagement.scenario_name + # try: + # if scenarioManagement.scenario.mobility: + # additional_participants = scenario_data["additional_participants"] + # schema_additional_participants = scenario_data["schema_additional_participants"] + # await scenarioManagement.load_configurations_and_start_nodes( + # additional_participants, schema_additional_participants + # ) + # else: + # await scenarioManagement.load_configurations_and_start_nodes() + # except subprocess.CalledProcessError as e: + # logging.exception(f"Error docker-compose up: {e}") + # return + + return ""#scenarioManagement.scenario_name @app.post("/scenarios/stop") @@ -378,17 +392,26 @@ async def stop_scenario( Note: This function does not currently trigger statistics generation. """ - from nebula.controller.scenarios import ScenarioManagement - - ScenarioManagement.cleanup_scenario_containers() + 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}" + factory_requests_path("stop") + stop_scenario_req = StopScenarioRequest(federation_id="id_nebula") try: - if all: - await scenario_set_all_status_to_finished() - else: - await scenario_set_status_to_finished(scenario_name) + await APIUtils.post(url_stop_scenario, stop_scenario_req.model_dump()) except Exception as e: - logging.exception(f"Error setting scenario {scenario_name} to finished: {e}") - raise HTTPException(status_code=500, detail="Internal server error") + 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("/scenarios/remove") @@ -638,20 +661,20 @@ async def update_nodes( timestamp = datetime.datetime.now() # Update the node in database await update_node_record( - str(config["device_args"]["uid"]), - str(config["device_args"]["idx"]), - str(config["network_args"]["ip"]), - str(config["network_args"]["port"]), - str(config["device_args"]["role"]), - config["network_args"]["neighbors"], - str(config["mobility_args"]["latitude"]), - str(config["mobility_args"]["longitude"]), + str(config["data"]["device_args"]["uid"]), + str(config["data"]["device_args"]["idx"]), + str(config["data"]["network_args"]["ip"]), + str(config["data"]["network_args"]["port"]), + str(config["data"]["device_args"]["role"]), + config["data"]["network_args"]["neighbors"], + str(config["data"]["addons"]["mobility"]["latitude"]), + str(config["data"]["addons"]["mobility"]["longitude"]), str(timestamp), - str(config["scenario_args"]["federation"]), - str(config["federation_args"]["round"]), - str(config["scenario_args"]["name"]), - str(config["tracking_args"]["run_hash"]), - str(config["device_args"]["malicious"]), + str(config["data"]["data"]["scenario_args"]["federation"]), + str(config["data"]["federation_args"]["round"]), + str(config["data"]["scenario_args"]["name"]), + str(config["data"]["tracking_args"]["run_hash"]), + str(config["data"]["device_args"]["malicious"]), ) except Exception as e: logging.exception(f"Error updating nodes: {e}") diff --git a/nebula/core/addonmanager.py b/nebula/core/addonmanager.py index 46fbfd60a..f3bf3a505 100644 --- a/nebula/core/addonmanager.py +++ b/nebula/core/addonmanager.py @@ -1,5 +1,6 @@ import logging from typing import TYPE_CHECKING +from abc import ABC, abstractmethod from nebula.addons.functions import print_msg_box from nebula.addons.gps.gpsmodule import factory_gpsmodule @@ -10,6 +11,15 @@ if TYPE_CHECKING: from nebula.core.engine import Engine +class NebulaAddon(ABC): + @abstractmethod + async def start(): + raise NotImplementedError + + @abstractmethod + async def stop(): + raise NotImplementedError + class AddondManager: """ @@ -51,24 +61,43 @@ async def deploy_additional_services(self): - Services are only launched if the corresponding configuration flags are set. """ print_msg_box(msg="Deploying Additional Services", indent=2, title="Addons Manager") - if self._config.participant["trustworthiness"]: - from nebula.addons.trustworthiness.trustworthiness import Trustworthiness - - trustworthiness = Trustworthiness(self._engine, self._config) - self._addons.append(trustworthiness) - - if self._config.participant["mobility_args"]["mobility"]: - mobility = Mobility(self._config, verbose=False) - self._addons.append(mobility) - - update_interval = 5 - gps = factory_gpsmodule("nebula", self._config, self._engine.addr, update_interval, verbose=False) - self._addons.append(gps) - - if self._config.participant["network_args"]["simulation"]: - refresh_conditions_interval = 5 - network_simulation = factory_network_simulator("nebula", refresh_conditions_interval, "eth0", verbose=False) - self._addons.append(network_simulation) + for addon, addon_config in self._config.participant["addons"].items(): + if addon == "mobility": + mobility = Mobility(self._config, verbose=False) + self._addons.append(mobility) + update_interval = 5 + gps = factory_gpsmodule("nebula", self._config, self._engine.addr, update_interval, verbose=False) + self._addons.append(gps) + elif addon == "trustworthiness": + from nebula.addons.trustworthiness.trustworthiness import Trustworthiness + trustworthiness = Trustworthiness(self._engine, self._config) + self._addons.append(trustworthiness) + elif addon == "network_simulation": + #TODO review parameters for network simulation + type_of_network = self._config.participant["network_args"]["network_simulation"]["type"] + network_config = self._config.participant["network_args"]["network_simulation"]["network_config"] + network_simulation = factory_network_simulator(type_of_network, network_config) + self._addons.append(network_simulation) + #TODO update config access + + # if self._config.participant["trustworthiness"]: + # from nebula.addons.trustworthiness.trustworthiness import Trustworthiness + + # trustworthiness = Trustworthiness(self._engine, self._config) + # self._addons.append(trustworthiness) + + # if self._config.participant["mobility_args"]["mobility"]: + # mobility = Mobility(self._config, verbose=False) + # self._addons.append(mobility) + + # update_interval = 5 + # gps = factory_gpsmodule("nebula", self._config, self._engine.addr, update_interval, verbose=False) + # self._addons.append(gps) + + # if self._config.participant["network_args"]["simulation"]: + # refresh_conditions_interval = 5 + # network_simulation = factory_network_simulator("nebula", refresh_conditions_interval, "eth0", verbose=False) + # self._addons.append(network_simulation) for add in self._addons: await add.start() diff --git a/nebula/core/datasets/nebuladataset.py b/nebula/core/datasets/nebuladataset.py index 0c2e03d8a..a3d1cbe92 100755 --- a/nebula/core/datasets/nebuladataset.py +++ b/nebula/core/datasets/nebuladataset.py @@ -1299,3 +1299,17 @@ def factory_nebuladataset(dataset, **config) -> NebulaDataset: if not cs: raise ValueError(f"Dataset {dataset} not supported") return cs(**config) + +def factory_dataset_setup(dataset) -> dict: + options = { + "MNIST": 10, + "FashionMNIST": 10, + "EMNIST": 47, + "CIFAR10": 10, + "CIFAR100": 100, + } + + num_classes = options.get(dataset, None) + if not num_classes: + raise ValueError(f"Dataset {dataset} not supported") + return num_classes diff --git a/nebula/core/engine.py b/nebula/core/engine.py index fc2fef4f6..6195f6926 100644 --- a/nebula/core/engine.py +++ b/nebula/core/engine.py @@ -115,7 +115,6 @@ def __init__( self._aggregator = create_aggregator(config=self.config, engine=self) self._secure_neighbors = [] - self._is_malicious = self.config.participant["adversarial_args"]["attack_params"]["attacks"] != "No Attack" role = config.participant["device_args"]["role"] self._role_behavior: RoleBehavior = factory_role_behavior(role, self, config) @@ -132,7 +131,6 @@ def __init__( msg += f"\nIID: {self.config.participant['data_args']['iid']}" msg += f"\nModel: {model.__class__.__name__}" msg += f"\nAggregation algorithm: {self._aggregator.__class__.__name__}" - msg += f"\nNode behavior: {'malicious' if self._is_malicious else 'benign'}" print_msg_box(msg=msg, indent=2, title="Scenario information") print_msg_box( msg=f"Logging type: {self._trainer.logger.__class__.__name__}", @@ -160,12 +158,12 @@ def __init__( self._addon_manager = AddondManager(self, self.config) # Additional Components - if "situational_awareness" in self.config.participant: + if "situational_awareness" in self.config.participant["addons"]: self._situational_awareness = SituationalAwareness(self.config, self) else: self._situational_awareness = None - if self.config.participant["defense_args"]["reputation"]["enabled"]: + if dict(self.config.participant["addons"]).get("reputation", None): self._reputation = Reputation(engine=self, config=self.config) @property @@ -619,10 +617,10 @@ async def deploy_components(self): the federated learning process starts. """ await self.aggregator.init() - if "situational_awareness" in self.config.participant: - await self.sa.init() - if self.config.participant["defense_args"]["reputation"]["enabled"]: - await self._reputation.setup() + if "situational_awareness" in self.config.participant["addons"]: + await self.sa.start() + if "reputation" in self.config.participant["addons"]: + await self._reputation.start() await self._reporter.start() await self._addon_manager.deploy_additional_services() diff --git a/nebula/core/network/communications.py b/nebula/core/network/communications.py index e0b1c17a5..5b022b61e 100755 --- a/nebula/core/network/communications.py +++ b/nebula/core/network/communications.py @@ -88,7 +88,8 @@ def __init__(self, engine: "Engine"): ) self.receive_messages_lock = Locker(name="receive_messages_lock", async_lock=True) - self._discoverer = Discoverer(addr=self.addr, config=self.config) + #self._discoverer = Discoverer(addr=self.addr, config=self.config) + self._discoverer = None # self._health = Health(addr=self.addr, config=self.config) self._health = None self._forwarder = Forwarder(config=self.config) diff --git a/nebula/core/node.py b/nebula/core/node.py index adc83fe1d..a8df622c2 100755 --- a/nebula/core/node.py +++ b/nebula/core/node.py @@ -77,7 +77,7 @@ async def main(config: Config): model_name = config.participant["model_args"]["model"] idx = config.participant["device_args"]["idx"] - additional_node_status = config.participant["mobility_args"]["additional_node"]["status"] + additional_node_status = config.participant["deployment_args"]["additional"] # Adjust the total number of nodes and the index of the current node for CFL, as it doesn't require a specific partition for the server (not used for training) if config.participant["scenario_args"]["federation"] == "CFL": @@ -179,11 +179,6 @@ def randomize_value(value, variability): config_keys = [ ["reporter_args", "report_frequency"], - ["discoverer_args", "discovery_frequency"], - ["health_args", "health_interval"], - ["health_args", "grace_time_health"], - ["health_args", "check_alive_interval"], - ["health_args", "send_alive_interval"], ["forwarder_args", "forwarder_interval"], ["forwarder_args", "forward_messages_interval"], ] @@ -209,9 +204,10 @@ def randomize_value(value, variability): await node.deploy_federation() if additional_node_status: - time = config.participant["mobility_args"]["additional_node"]["time_start"] - logging.info(f"Waiting time to start finding federation: {time}") - await asyncio.sleep(int(config.participant["mobility_args"]["additional_node"]["time_start"])) + # time = config.participant["addons"]["mobility"]["additional_node"]["time_start"] + # logging.info(f"Waiting time to start finding federation: {time}") + # await asyncio.sleep(int(config.participant["addons"]["mobility"]["additional_node"]["time_start"])) + #await asyncio.sleep(120) #TODO REMOVE await node._aditional_node_start() if node.cm is not None: diff --git a/nebula/core/noderole.py b/nebula/core/noderole.py index 9bd258fef..d1b227e6e 100644 --- a/nebula/core/noderole.py +++ b/nebula/core/noderole.py @@ -190,7 +190,7 @@ def __init__(self, engine: Engine, config: Config): self.attack = create_attack(self._engine) logging.info("Attack behavior created") self.aggregator_bening = self._engine._aggregator - benign_role = self._config.participant["adversarial_args"]["fake_behavior"] + benign_role = self._config.participant["addons"]["adversarial_args"]["fake_behavior"] self._fake_role_behavior = factory_role_behavior(benign_role, self._engine, self._config) self._role = factory_node_role("malicious") @@ -206,7 +206,7 @@ async def extended_learning_cycle(self): try: await self.attack.attack() except Exception: - attack_name = self._config.participant["adversarial_args"]["attacks"] + attack_name = self._config.participant["addons"]["adversarial_args"]["attacks"] logging.exception(f"Attack {attack_name} failed") await self._fake_role_behavior.extended_learning_cycle() diff --git a/nebula/core/situationalawareness/awareness/sareasoner.py b/nebula/core/situationalawareness/awareness/sareasoner.py index 40e6de94d..b7f76e25c 100644 --- a/nebula/core/situationalawareness/awareness/sareasoner.py +++ b/nebula/core/situationalawareness/awareness/sareasoner.py @@ -85,9 +85,10 @@ def __init__( title="SA Reasoner", ) logging.info("🌐 Initializing SAReasoner") - self._config = copy.deepcopy(config.participant) + self._is_additional_node = config.participant["deployment_args"]["additional"] + self._config = copy.deepcopy(config.participant["addons"]) self._addr = config.participant["network_args"]["addr"] - self._topology = config.participant["mobility_args"]["topology_type"] + self._topology = config.participant["addons"]["mobility"]["topology_type"] self._situational_awareness_network: SANetwork | None = None self._situational_awareness_training = None self._restructure_process_lock = Locker(name="restructure_process_lock", async_lock=True) @@ -96,11 +97,11 @@ def __init__( self._suggestion_buffer = SuggestionBuffer(self._arbitrator_notification, verbose=True) self._communciation_manager = CommunicationsManager.get_instance() self._sys_monitor = SystemMonitor() - arb_pol = config.participant["situational_awareness"]["sa_reasoner"]["arbitration_policy"] + arb_pol = config.participant["addons"]["situational_awareness"]["sa_reasoner"]["arbitration_policy"] self._arbitatrion_policy = factory_arbitration_policy(arb_pol, True) self._sa_components: dict[str, SAMComponent] = {} self._sa_discovery: ISADiscovery | None = None - self._verbose = config.participant["situational_awareness"]["sa_reasoner"]["verbose"] + self._verbose = config.participant["addons"]["situational_awareness"]["sa_reasoner"]["verbose"] @property def san(self) -> SANetwork | None: @@ -146,7 +147,8 @@ def is_additional_participant(self): Returns: bool: True if the node is marked as an additional participant, False otherwise. """ - return self._config["mobility_args"]["additional_node"]["status"] + return self._is_additional_node + """ ############################### # REESTRUCTURE TOPOLOGY # diff --git a/nebula/core/situationalawareness/awareness/suggestionbuffer.py b/nebula/core/situationalawareness/awareness/suggestionbuffer.py index 98cae49b2..c1d354133 100644 --- a/nebula/core/situationalawareness/awareness/suggestionbuffer.py +++ b/nebula/core/situationalawareness/awareness/suggestionbuffer.py @@ -5,7 +5,7 @@ from nebula.core.situationalawareness.awareness.sautils.sacommand import SACommand from nebula.core.situationalawareness.awareness.sautils.samoduleagent import SAModuleAgent from nebula.core.utils.locker import Locker -from nebula.utils import logging +import logging class SuggestionBuffer: diff --git a/nebula/core/situationalawareness/situationalawareness.py b/nebula/core/situationalawareness/situationalawareness.py index 6a5dbcbd6..2fee2d7de 100644 --- a/nebula/core/situationalawareness/situationalawareness.py +++ b/nebula/core/situationalawareness/situationalawareness.py @@ -1,6 +1,6 @@ import asyncio from abc import ABC, abstractmethod - +from nebula.core.addonmanager import NebulaAddon from nebula.addons.functions import print_msg_box @@ -156,7 +156,7 @@ def factory_sa_reasoner(sa_reasoner, config) -> ISAReasoner: raise Exception(f"SA Reasoner service {sa_reasoner} not found.") -class SituationalAwareness: +class SituationalAwareness(NebulaAddon): """ High-level coordinator for Situational Awareness in the DFL federation. @@ -178,16 +178,16 @@ def __init__(self, config, engine): title="Situational Awareness module", ) self._config = config - selector = self._config.participant["situational_awareness"]["sa_discovery"]["candidate_selector"] + selector = self._config.participant["addons"]["situational_awareness"]["sa_discovery"]["candidate_selector"] selector = selector.lower() - model_handler = config.participant["situational_awareness"]["sa_discovery"]["model_handler"] + model_handler = config.participant["addons"]["situational_awareness"]["sa_discovery"]["model_handler"] self._sad = factory_sa_discovery( "nebula", - self._config.participant["mobility_args"]["additional_node"]["status"], + self._config.participant["deployment_args"]["additional"], selector, model_handler, engine=engine, - verbose=config.participant["situational_awareness"]["sa_discovery"]["verbose"], + verbose=config.participant["addons"]["situational_awareness"]["sa_discovery"]["verbose"], ) self._sareasoner = factory_sa_reasoner( "nebula", @@ -214,7 +214,7 @@ def sar(self): """ return self._sareasoner - async def init(self): + async def start(self): """ Initialize both discovery and reasoner components, linking them together. """ diff --git a/nebula/utils.py b/nebula/utils.py index 60819ed1a..34b0c9621 100644 --- a/nebula/utils.py +++ b/nebula/utils.py @@ -2,8 +2,16 @@ import os import socket +import aiohttp import docker +import re +from typing import Optional + +from fastapi import HTTPException +from aiohttp import ClientConnectorError +from aiohttp.client_exceptions import ClientError +import asyncio class FileUtils: """ @@ -96,7 +104,6 @@ def find_free_port(cls, start_port=49152, end_port=65535): return port return None - class DockerUtils: """ Utility class for Docker operations such as creating networks, @@ -202,3 +209,150 @@ def check_docker_by_prefix(cls, prefix): logging.exception("Error interacting with Docker") except Exception: logging.exception("Unexpected error") + + +class LoggerUtils: + + @staticmethod + def configure_logger( + name: Optional[str] = None, + log_file: Optional[str] = None, + level: int = logging.INFO, + console: bool = True, + strip_ansi: bool = True, + file_mode: str = "w", + log_format: str = "[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s", + date_format: str = "%Y-%m-%d %H:%M:%S", + ) -> logging.Logger: + """ + Configure and return a logger with optional console and file output. + + Args: + name (str): Logger name. If None, the root logger is used. + log_file (str): Path to the log file. + level (int): Logging level (DEBUG, INFO, etc). + console (bool): If True, output is also printed to the console. + strip_ansi (bool): Placeholder for future ANSI stripping support. + file_mode (str): File mode for the log file ('a' for append, 'w' for overwrite). + log_format (str): Format for log messages. + date_format (str): Format for timestamps. + + Returns: + logging.Logger: Configured logger instance. + """ + logger = logging.getLogger(name) + logger.setLevel(level) + + # Prevent duplicate handler setup + if getattr(logger, "_is_configured", False): + return logger + + formatter = logging.Formatter(fmt=log_format, datefmt=date_format) + + if log_file: + os.makedirs(os.path.dirname(log_file), exist_ok=True) + fh = logging.FileHandler(log_file, mode=file_mode) + fh.setLevel(level) + fh.setFormatter(formatter) + logger.addHandler(fh) + + if console: + ch = logging.StreamHandler() + ch.setLevel(level) + ch.setFormatter(formatter) + logger.addHandler(ch) + + # Mark this logger as configured to avoid re-adding handlers + logger._is_configured = True + logger.propagate = False + + return logger + +class APIUtils(): + + @staticmethod + async def retry_with_backoff(func, *args, max_retries=5, initial_delay=1): + """ + Retry a function with exponential backoff. + + Args: + func: The async function to retry + *args: Arguments to pass to the function + max_retries: Maximum number of retry attempts + initial_delay: Initial delay between retries in seconds + + Returns: + The result of the function if successful + + Raises: + The last exception if all retries fail + """ + delay = initial_delay + last_exception = None + + for attempt in range(max_retries): + try: + return await func(*args) + except (ClientConnectorError, ClientError) as e: + last_exception = e + if attempt < max_retries - 1: + logging.warning(f"Connection attempt {attempt + 1} failed: {str(e)}. Retrying in {delay} seconds...") + await asyncio.sleep(delay) + delay *= 2 # Exponential backoff + else: + logging.error(f"All {max_retries} connection attempts failed") + raise last_exception + + @staticmethod + async def get(url): + """ + Fetch JSON data from a remote controller endpoint via asynchronous HTTP GET. + + Parameters: + url (str): The full URL of the controller API endpoint. + + Returns: + Any: Parsed JSON response when the HTTP status code is 200. + + Raises: + HTTPException: If the response status is not 200, raises with the response status code and an error detail. + """ + + async def _get(): + async with aiohttp.ClientSession() as session: + async with session.get(url) as response: + if response.status == 200: + return await response.json() + else: + raise HTTPException(status_code=response.status, detail="Error fetching data") + + return await APIUtils.retry_with_backoff(_get) + + @staticmethod + async def post(url, data=None): + """ + Asynchronously send a JSON payload via HTTP POST to a controller endpoint and parse the response. + + Parameters: + url (str): The full URL of the controller API endpoint. + data (Any, optional): JSON-serializable payload to include in the POST request (default: None). + + Returns: + Any: Parsed JSON response when the HTTP status code is 200. + + Raises: + HTTPException: If the response status is not 200, with the status code and an error detail. + """ + + async def _post(): + async with aiohttp.ClientSession() as session: + async with session.post(url, json=data) as response: + if response.status == 200: + return await response.json() + else: + detail = await response.text() + raise HTTPException(status_code=response.status, detail=detail) + + return await APIUtils.retry_with_backoff(_post) + + \ No newline at end of file