feature federation ID on all API requests

Author: AlejandroAvilesSerrano

nebula/addons/reporter.py

@@ -54,7 +54,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,7 +170,7 @@ 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"
+        url = f"http://{self.config.participant['scenario_args']['controller']}/nodes/{self.config.participant["scenario_args"]["federation_id"]}/done"
         data = json.dumps({"idx": self.config.participant["device_args"]["idx"],
                            "deployment": self.config.participant["scenario_args"]["deployment"],
                            "federation_id": self.config.participant["scenario_args"]["federation_id"]})

nebula/controller/federation/controllers/docker_federation_controller.py

@@ -230,7 +230,7 @@ async def update_nodes(self, scenario_name: str, request: Request):
                                     adds_deployed.add(index)
             request_body = await request.json()
             payload = {"scenario_name": scenario_name, "data": request_body}
-            asyncio.create_task(self._send_to_hub("update", payload, scenario_name))
+            asyncio.create_task(self._send_to_hub("update", payload, fed_id))
             return {"message": "Node updated successfully in Federation Controller"}
         except Exception as e:
             self.logger.info(f"ERROR: federation ID: ({fed_id}), {e}")
@@ -246,10 +246,10 @@ async def node_done(self, scenario_name: str, request: Request):
             payload = {"federation_id": federation_id, "scenario_name": scenario_name, "data": request_body}
             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, scenario_name))
+            asyncio.create_task(self._send_to_hub("finish", payload, federation_id))
 
         payload = {"scenario_name": scenario_name, "data": request_body}
-        asyncio.create_task(self._send_to_hub("done", payload, scenario_name))
+        asyncio.create_task(self._send_to_hub("done", payload, federation_id))
         return {"message": "Nodes done received successfully"}
 
     """                                             ###############################

nebula/controller/federation/controllers/processes_federation_controller.py

@@ -190,7 +190,7 @@ async def update_nodes(self, scenario_name: str, request: Request):
                                     adds_deployed.add(index)
             request_body = await request.json()
             payload = {"scenario_name": scenario_name, "data": request_body}
-            asyncio.create_task(self._send_to_hub("update", payload, scenario_name))
+            asyncio.create_task(self._send_to_hub("update", payload, 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..")
@@ -206,10 +206,10 @@ async def node_done(self, scenario_name: str, request: Request):
             payload = {"federation_id": federation_id, "scenario_name": scenario_name, "data": request_body}
             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, scenario_name))
+            asyncio.create_task(self._send_to_hub("finish", payload, federation_id))
 
         payload = {"scenario_name": scenario_name, "data": request_body}
-        asyncio.create_task(self._send_to_hub("done", payload, scenario_name))
+        asyncio.create_task(self._send_to_hub("done", payload, federation_id))
         return {"message": "Nodes done received successfully"}
 
     """                                             ###############################

nebula/controller/federation/federation_api.py

@@ -10,7 +10,7 @@
 from nebula.utils import LoggerUtils
 from nebula.controller.federation.federation_controller import FederationController 
 from nebula.controller.federation.factory_federation_controller import federation_controller_factory
-from nebula.controller.federation.utils_requests import RunScenarioRequest, StopScenarioRequest
+from nebula.controller.federation.utils_requests import RunScenarioRequest, StopScenarioRequest, Routes
 
 fed_controllers: Dict[str, FederationController] = {}
 
@@ -51,7 +51,7 @@ async def read_root():
     logger.info("Test curl succesfull")
     return {"message": "Welcome to the NEBULA Federation Controller API"}
 
-@app.post("/scenarios/run")
+@app.post(Routes.RUN)
 async def run_scenario(run_scenario_request: RunScenarioRequest):
     global fed_controllers
     experiment_type = run_scenario_request.scenario_data["deployment"]
@@ -63,7 +63,7 @@ async def run_scenario(run_scenario_request: RunScenarioRequest):
     else:
         return {"message": "Experyment type not allowed"}
 
-@app.post("/scenarios/stop")
+@app.post(Routes.STOP)
 async def stop_scenario(stop_scenario_request: StopScenarioRequest):
     global fed_controllers
     experiment_type = stop_scenario_request.experiment_type
@@ -75,7 +75,7 @@ async def stop_scenario(stop_scenario_request: StopScenarioRequest):
     else:
         return {"message": "Experyment type not allowed"}
 
-@app.post("/nodes/{scenario_name}/update")
+@app.post(Routes.UPDATE)
 async def update_nodes(
     scenario_name: Annotated[
         str,
@@ -92,7 +92,7 @@ async def update_nodes(
     else:
         return {"message": "Experyment type not allowed on response for update message.."}
 
-@app.post("/nodes/{scenario_name}/done")
+@app.post(Routes.DONE)
 async def update_nodes(
     scenario_name: Annotated[
         str,

nebula/controller/federation/utils_requests.py

@@ -10,19 +10,27 @@ class StopScenarioRequest(BaseModel):
     experiment_type: 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 "/scenarios/run"
+        return Routes.RUN
     elif resource == "stop":
-        return "/scenarios/stop"
+        return Routes.STOP
     elif resource == "update":
-        return f"/nodes/{scenario_name}/update"
+        return Routes.UPDATE.format(federation_id=federation_id)
     elif resource == "done":
-        return f"/nodes/{scenario_name}/done"
+        return Routes.DONE.format(federation_id=federation_id)
     elif resource == "finish":
-        return f"/scenarios/{federation_id}/finish"
+        return Routes.FINISH.format(federation_id=federation_id)
     else:
         raise Exception(f"resource not found: {resource}")