Add trigger restart as last step, to ensure tree changes picked up immediately

Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>

Author: nuclearcat

deployment_manager.py

@@ -440,3 +440,62 @@ async def docker_workaround(self) -> Dict[str, Any]:
             )
 
         return {"success": True, "results": results}
+
+    async def restart_trigger_service(self) -> Dict[str, Any]:
+        """
+        Restart the trigger service in the pipeline
+        Returns: {"success": bool, "error": str, "details": dict}
+        """
+        result = {
+            "success": False,
+            "error": None,
+            "details": {"start_time": datetime.utcnow().isoformat()},
+        }
+
+        try:
+            # Set environment variable for pipeline settings
+            env = os.environ.copy()
+            env["SETTINGS"] = PIPELINE_SETTINGS_PATH
+
+            # Restart the trigger service using docker-compose
+            cmd = ["docker-compose"]
+
+            # Add compose files if configured
+            if self.compose_files:
+                cmd.extend(self.compose_files)
+
+            # Add restart command for trigger service
+            cmd.extend(["restart", "trigger"])
+
+            print(f"Restarting trigger service with command: {' '.join(cmd)}")
+
+            process = await asyncio.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                env=env,
+                cwd=self.pipeline_path,
+            )
+
+            stdout, stderr = await process.communicate()
+
+            if process.returncode == 0:
+                result["success"] = True
+                result["details"]["stdout"] = stdout.decode("utf-8", errors="ignore")
+                result["details"]["stderr"] = stderr.decode("utf-8", errors="ignore")
+                result["details"]["end_time"] = datetime.utcnow().isoformat()
+                print("Successfully restarted trigger service")
+            else:
+                result["error"] = (
+                    f"Failed to restart trigger service: {stderr.decode('utf-8', errors='ignore')}"
+                )
+                result["details"]["stdout"] = stdout.decode("utf-8", errors="ignore")
+                result["details"]["stderr"] = stderr.decode("utf-8", errors="ignore")
+                print(f"Failed to restart trigger service: {result['error']}")
+
+        except Exception as e:
+            result["error"] = f"Exception restarting trigger service: {str(e)}"
+            result["details"]["exception"] = str(e)
+            print(f"Exception restarting trigger service: {e}")
+
+        return result

models.py

@@ -32,6 +32,7 @@ class StagingStepType(str, enum.Enum):
     KERNEL_TREE_UPDATE = "kernel_tree_update"
     API_PIPELINE_UPDATE = "api_pipeline_update"
     MONITORING_SETUP = "monitoring_setup"
+    TRIGGER_RESTART = "trigger_restart"
 
 
 class StagingRunStatus(str, enum.Enum):

orchestrator.py

@@ -137,11 +137,15 @@ async def initialize_staging_steps(self, staging_run: StagingRun, db: Session):
 
         steps.extend(
             [
-                {"type": StagingStepType.KERNEL_TREE_UPDATE, "order": order_counter},
                 {
                     "type": StagingStepType.API_PIPELINE_UPDATE,
+                    "order": order_counter,
+                },
+                {
+                    "type": StagingStepType.KERNEL_TREE_UPDATE,
                     "order": order_counter + 1,
                 },
+                {"type": StagingStepType.TRIGGER_RESTART, "order": order_counter + 2},
             ]
         )
 
@@ -221,6 +225,8 @@ async def process_step(
             await self.process_kernel_tree_step(staging_run, step, db)
         elif step.step_type == StagingStepType.API_PIPELINE_UPDATE:
             await self.process_api_pipeline_step(staging_run, step, db)
+        elif step.step_type == StagingStepType.TRIGGER_RESTART:
+            await self.process_trigger_restart_step(staging_run, step, db)
 
     async def process_github_workflow_step(
         self, staging_run: StagingRun, step: StagingRunStep, db: Session
@@ -441,6 +447,36 @@ async def process_api_pipeline_step(
                 step.end_time = datetime.utcnow()
                 db.commit()
 
+    async def process_trigger_restart_step(
+        self, staging_run: StagingRun, step: StagingRunStep, db: Session
+    ):
+        """Process trigger restart step - restart the trigger service in pipeline"""
+        if step.status == StagingStepStatus.PENDING:
+            step.status = StagingStepStatus.RUNNING
+            step.start_time = datetime.utcnow()
+            staging_run.current_step = "trigger_restart"
+            db.commit()
+
+            try:
+                # Restart the trigger service in the pipeline
+                result = await self.deployment_manager.restart_trigger_service()
+
+                step.end_time = datetime.utcnow()
+                if result["success"]:
+                    step.status = StagingStepStatus.COMPLETED
+                    step.details = json.dumps(result)
+                else:
+                    step.status = StagingStepStatus.FAILED
+                    step.error_message = result.get("error", "Unknown error")
+
+                db.commit()
+
+            except Exception as e:
+                step.status = StagingStepStatus.FAILED
+                step.error_message = str(e)
+                step.end_time = datetime.utcnow()
+                db.commit()
+
     async def complete_staging_run(
         self, staging_run: StagingRun, db: Session, success: bool
     ):
@@ -508,6 +544,8 @@ async def recover_stuck_steps(self, staging_run: StagingRun, db: Session):
                         timeout_minutes = 15  # Git operations
                     elif step.step_type == StagingStepType.SELF_UPDATE:
                         timeout_minutes = 10  # Quick git pull
+                    elif step.step_type == StagingStepType.TRIGGER_RESTART:
+                        timeout_minutes = 5  # Quick docker restart
 
                     if running_duration.total_seconds() > (timeout_minutes * 60):
                         print(
@@ -576,6 +614,7 @@ async def startup_recovery(self):
                             StagingStepType.API_PIPELINE_UPDATE,
                             StagingStepType.KERNEL_TREE_UPDATE,
                             StagingStepType.SELF_UPDATE,
+                            StagingStepType.TRIGGER_RESTART,
                         ]:
                             print(
                                 f"Recovering stuck local step: {step.step_type.value}"