Feature/new federation controller (#54)

* postgres docker created

* fix postgres db endpoints

* redis docker added

* feature federation controller class en API endpoints

* feature federation controller api running

* feature federation API:
- Scenario Builder
- Dataset factories

* feature scenario building

* fix attack assigment on ScenarioBuilder

* feature scenario building complete on federation controller

* feature scenario building fix and complete.
Fix references on backend

* feature additionals late deployment logic

* feature federation API request parameters using pydantic. Directories changed

* fix participants deployment

* feature  eputation and attacks  configuration

* feature additionals participant deployment

* feature docker federation controller integration

* feature federationID on docker controller

* feature process deployment and completion of comunication hub-controller

* fix remove docker containers

* feature process deployment

* fix minor details

* fix draw graph

* remove comments

* feature federation ID on all API requests

* update API requests

* feature node update-done format

---------

Co-authored-by: FerTV <fernando.torres.vega@gmail.com>

Author: AlejandroAvilesSerrano

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({

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",

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")

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):

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:

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(

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']}",

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:

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):

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