Issue #85

CHANGELOG.md

@@ -32,6 +32,7 @@
 - Trigger orchestrator at critical moments ([Issue](https://github.com/orgs/amosproj/projects/79/views/2?pane=issue&itemId=114047181&issue=amosproj%7Camos2025ss01-embark-orchestration-framework%7C59))
 - Use celery for IP range scanning and update config UI ([Issue](https://github.com/orgs/amosproj/projects/79/views/2?pane=issue&itemId=118038092&issue=amosproj%7Camos2025ss01-embark-orchestration-framework%7C91))
 - `process_update_queue` call once the analysis is finished ([Issue](https://github.com/orgs/amosproj/projects/79/views/2?pane=issue&itemId=114650622&issue=amosproj%7Camos2025ss01-embark-orchestration-framework%7C70))
+- Handle unresponsive worker nodes ([Issue](https://github.com/orgs/amosproj/projects/79/views/2?pane=issue&itemId=115550958&issue=amosproj%7Camos2025ss01-embark-orchestration-framework%7C85))
 - Wiki entries for API ([Issue](https://github.com/orgs/amosproj/projects/79/views/2?pane=issue&itemId=114744149&issue=amosproj%7Camos2025ss01-embark-orchestration-framework%7C71))
 - Indicate currently installed dependency versions ([Issue](https://github.com/orgs/amosproj/projects/79/views/2?pane=issue&itemId=118051293&issue=amosproj%7Camos2025ss01-embark-orchestration-framework%7C94))
 

dev-tools/debug-server-start.sh

@@ -137,7 +137,7 @@ python3 ./manage.py runapscheduler | tee -a ../logs/scheduler.log &
 # systemctl start embark.service
 
 # Start celery worker
-celery -A embark worker --beat --scheduler django -l DEBUG --logfile=../logs/celery.log &
+celery -A embark worker --beat --scheduler django -l INFO --logfile=../logs/celery.log &
 CELERY_PID=$!
 trap 'kill ${CELERY_PID} 2>/dev/null; exit' SIGINT SIGTERM EXIT
 

embark/embark/logreader.py

@@ -295,9 +295,14 @@ def read_loop(self):
                 pass
 
         while not self.finish:
+            emba_log_path = f"{self.analysis.path_to_logs}/emba.log"
+            if not pathlib.Path(emba_log_path).exists():
+                time.sleep(5)
+                continue
+
             # look for new events in log
-            logger.debug("looking for events in %s", f"{self.analysis.path_to_logs}/emba.log")
-            got_event = self.inotify_events(f"{self.analysis.path_to_logs}/emba.log")
+            logger.debug("looking for events in %s", emba_log_path)
+            got_event = self.inotify_events(emba_log_path)
 
             for eve in got_event:
                 for flag in flags.from_mask(eve.mask):
@@ -307,7 +312,7 @@ def read_loop(self):
                     # Act on file change
                     elif flag is flags.MODIFY:
                         # get the actual difference
-                        tmp = self.get_diff(f"{self.analysis.path_to_logs}/emba.log")
+                        tmp = self.get_diff(emba_log_path)
                         logger.debug("Got diff-output: %s", tmp)
                         # send changes to frontend
                         self.input_processing(tmp)

embark/embark/settings/deploy.py

@@ -321,6 +321,7 @@
 WORKER_FIRMWARE_DIR = "/root/firmware/"
 WORKER_EMBA_LOGS = "/root/emba_logs/"
 WORKER_UPDATE_QUEUE_SIZE = 50
+WORKER_REACHABLE_TIMEOUT = 10
 
 # Celery task queue
 CELERY_BROKER_URL = f"redis://{REDIS_HOST}:{REDIS_PORT}/0"

embark/embark/settings/dev.py

@@ -288,6 +288,7 @@
 WORKER_FIRMWARE_DIR = "/root/firmware/"
 WORKER_EMBA_LOGS = "/root/emba_logs/"
 WORKER_UPDATE_QUEUE_SIZE = 50
+WORKER_REACHABLE_TIMEOUT = 10
 
 # Celery task queue
 CELERY_BROKER_URL = f"redis://{REDIS_HOST}:{REDIS_PORT}/0"

embark/workers/apps.py

@@ -19,3 +19,28 @@ def reset_orchestrator(**kwargs):
             orchestrator = get_orchestrator()
             orchestrator.reset()
         post_migrate.connect(reset_orchestrator)
+
+        def reset_dependency_users(**kwargs):
+            """
+            Reset the 'used_by' field for all DependencyState instances.
+            An unexpected shutdown may leave this field populated with workers
+            that no longer use the dependency.
+            """
+            from workers.models import DependencyState  # pylint: disable=import-outside-toplevel
+            states = DependencyState.objects.all()
+            for state in states:
+                state.used_by.clear()
+                state.save()
+        post_migrate.connect(reset_dependency_users)
+
+        def reset_update_queues(**kwargs):
+            """
+            Reset the update queues for all workers.
+            An unexpected shutdown may leave the update queues
+            populated with updates that are no longer relevant.
+            """
+            from workers.models import Worker, WorkerUpdate  # pylint: disable=import-outside-toplevel
+            workers = Worker.objects.all()
+            for worker in workers:
+                WorkerUpdate.objects.filter(worker__id=worker.id).delete()
+        post_migrate.connect(reset_update_queues)

embark/workers/models.py

@@ -69,6 +69,7 @@ class ConfigStatus(models.TextChoices):  # pylint: disable=too-many-ancestors
     reachable = models.BooleanField(default=False)
     status = models.CharField(max_length=1, choices=ConfigStatus, default=ConfigStatus.UNCONFIGURED)
     analysis_id = models.UUIDField(blank=True, null=True, help_text="ID of the analysis currently running on this worker")
+    last_reached = models.DateTimeField(auto_now_add=True)
 
     dependency_version = models.OneToOneField(
         WorkerDependencyVersion,
@@ -91,15 +92,16 @@ def clean(self):
                     except ValueError as value_error:
                         raise ValidationError({"configuration": f"Invalid IP range: {value_error}"}) from value_error
 
-    def ssh_connect(self):
+    def ssh_connect(self, timeout=30):
         """
         Tries to establish an ssh connection with each configuration and returns the first successful connection
+        :param timeout: max ssh connect timeout
         """
         ssh_client = new_autoadd_client()
 
         for configuration in self.configurations.all():
             try:
-                ssh_client.connect(self.ip_address, username=configuration.ssh_user, password=configuration.ssh_password)
+                ssh_client.connect(self.ip_address, username=configuration.ssh_user, password=configuration.ssh_password, timeout=timeout)
 
                 # save the ssh user and password so they can later be used in commands
                 ssh_client.ssh_user = configuration.ssh_user
@@ -108,6 +110,9 @@ def ssh_connect(self):
             except (paramiko.SSHException, socket.error):
                 continue
 
+        if ssh_client.get_transport() is None or not ssh_client.get_transport().is_active():
+            raise paramiko.SSHException("Failed to connect to worker with any configuration.")
+
         return ssh_client
 
 

embark/workers/orchestrator.py

@@ -291,19 +291,22 @@ def add_worker(self, worker: Worker):
     def _remove_worker(self, worker: Worker, check: bool = True):
         """
         Remove a worker from the orchestrator. The worker is removed from either the free or busy workers list.
+        We assume that any analysis running on the worker has been completed or cancelled before calling this method.
 
         :param worker: The worker to be removed
         :param check: If True, throws an error
         :raises ValueError: If the worker does not exist in the orchestrator and check = True
         """
+        worker.analysis_id = None
+        worker.save()
+
         if worker.ip_address in self.free_workers:
             del self.free_workers[worker.ip_address]
         elif worker.ip_address in self.busy_workers:
             del self.busy_workers[worker.ip_address]
         else:
             if check:
-                raise ValueError(f"Worker with IP {worker.ip_address} does not exist.")
-
+                raise ValueError(f"Worker with IP {worker.ip_address} not registered in the orchestrator.")
             return
 
         logger.info("Worker: %s removed from orchestrator", worker.name)