From 07fd1e29faae0bb7f68ba2d0afe84bbfc7170594 Mon Sep 17 00:00:00 2001 From: Sam Thornton Date: Thu, 28 May 2026 14:13:41 -0600 Subject: [PATCH] Replace fragile workflow status heuristic with Temporal describe API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The status endpoint used asyncio.wait_for with a 1-second timeout to guess whether a workflow was running or completed. This created abandoned coroutines and couldn't distinguish failed/cancelled/timed-out states — unreliable for the compose pipeline polling unattended for hours. Switch to Temporal's describe() API which returns a definitive status enum (RUNNING, COMPLETED, FAILED, etc.) plus start/close times. Also adds compose-pipeline support: - Deterministic workflow IDs (rh-matcher-compose-v{major}) for single-version triggers so Temporal deduplicates concurrent runs - 409 Conflict response when a workflow is already running - Admin UI status panel with live polling after triggering a workflow --- apollo/server/routes/admin_workflows.py | 93 +++++++++--- apollo/server/routes/api_workflows.py | 45 ++++-- apollo/server/services/workflow_service.py | 141 ++++++++++-------- apollo/server/templates/admin_workflows.jinja | 114 +++++++++++++- 4 files changed, 291 insertions(+), 102 deletions(-) diff --git a/apollo/server/routes/admin_workflows.py b/apollo/server/routes/admin_workflows.py index cfb26ec..08dfa9b 100644 --- a/apollo/server/routes/admin_workflows.py +++ b/apollo/server/routes/admin_workflows.py @@ -7,6 +7,8 @@ from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from typing import List, Optional +from temporalio.client import RPCError + from apollo.server.utils import templates, admin_user_scheme from apollo.server.services.workflow_service import WorkflowService from apollo.server.services.database_service import DatabaseService @@ -23,6 +25,9 @@ async def admin_workflows(request: Request, user: User = Depends(admin_user_sche env_info = await db_service.get_environment_info() index_state = await db_service.get_last_indexed_at() + tracked_wf_id = request.session.pop("tracked_workflow_id", None) + tracked_wf_name = request.session.pop("tracked_workflow_name", None) + return templates.TemplateResponse( "admin_workflows.jinja", { "request": request, @@ -32,6 +37,8 @@ async def admin_workflows(request: Request, user: User = Depends(admin_user_sche "reset_allowed": env_info["reset_allowed"], "last_indexed_at": index_state.get("last_indexed_at_iso"), "last_indexed_exists": index_state.get("exists", False), + "tracked_workflow_id": tracked_wf_id, + "tracked_workflow_name": tracked_wf_name, } ) @@ -44,30 +51,50 @@ async def trigger_rh_matcher( ): """Trigger RHMatcherWorkflow from admin web interface""" try: - # Convert string versions to integers if provided int_versions = None if major_versions: int_versions = [int(v) for v in major_versions if v.isdigit()] - + service = WorkflowService() workflow_id = await service.trigger_rh_matcher_workflow(int_versions) - - Logger().info(f"Admin user {user.email} triggered RhMatcherWorkflow {workflow_id} with major_versions: {int_versions}") - - # Store success message in session - request.session["workflow_message"] = f"RhMatcherWorkflow triggered successfully: {workflow_id}" + + Logger().info( + f"Admin user {user.email} triggered RhMatcherWorkflow " + f"{workflow_id} with major_versions: {int_versions}" + ) + + request.session["workflow_message"] = ( + f"RhMatcherWorkflow triggered successfully: {workflow_id}" + ) request.session["workflow_type"] = "success" - + request.session["tracked_workflow_id"] = workflow_id + request.session["tracked_workflow_name"] = "RH Matcher" + + except RPCError as e: + if "already running" in str(e).lower(): + request.session["workflow_message"] = ( + "A workflow for this version is already running" + ) + request.session["workflow_type"] = "error" + else: + Logger().error(f"Temporal error triggering workflow: {e}") + request.session["workflow_message"] = ( + f"Workflow service error: {e}" + ) + request.session["workflow_type"] = "error" + except ValueError as e: - Logger().error(f"Validation error triggering RhMatcher workflow: {str(e)}") - request.session["workflow_message"] = f"Error: {str(e)}" + Logger().error(f"Validation error triggering workflow: {e}") + request.session["workflow_message"] = f"Error: {e}" request.session["workflow_type"] = "error" - + except Exception as e: - Logger().error(f"Error triggering RhMatcher workflow: {str(e)}") - request.session["workflow_message"] = f"Error triggering workflow: {str(e)}" + Logger().error(f"Error triggering RhMatcher workflow: {e}") + request.session["workflow_message"] = ( + f"Error triggering workflow: {e}" + ) request.session["workflow_type"] = "error" - + return RedirectResponse(url="/admin/workflows", status_code=303) @@ -80,21 +107,41 @@ async def trigger_poll_rhcsaf( try: service = WorkflowService() workflow_id = await service.trigger_poll_rhcsaf_workflow() - - Logger().info(f"Admin user {user.email} triggered PollRHCSAFAdvisoriesWorkflow {workflow_id}") - - # Store success message in session - request.session["workflow_message"] = f"PollRHCSAFAdvisoriesWorkflow triggered successfully: {workflow_id}" + + Logger().info( + f"Admin user {user.email} triggered " + f"PollRHCSAFAdvisoriesWorkflow {workflow_id}" + ) + + request.session["workflow_message"] = ( + f"PollRHCSAFAdvisoriesWorkflow triggered successfully: " + f"{workflow_id}" + ) request.session["workflow_type"] = "success" - + request.session["tracked_workflow_id"] = workflow_id + request.session["tracked_workflow_name"] = "Poll RHCSAF" + except Exception as e: - Logger().error(f"Error triggering PollRHCSAF workflow: {str(e)}") - request.session["workflow_message"] = f"Error triggering workflow: {str(e)}" + Logger().error(f"Error triggering PollRHCSAF workflow: {e}") + request.session["workflow_message"] = ( + f"Error triggering workflow: {e}" + ) request.session["workflow_type"] = "error" - + return RedirectResponse(url="/admin/workflows", status_code=303) +@router.get("/workflows/{workflow_id}/status") +async def admin_workflow_status( + workflow_id: str, + user: User = Depends(admin_user_scheme), +): + """Return workflow status as JSON for the admin polling UI.""" + service = WorkflowService() + status_info = await service.get_workflow_status(workflow_id) + return JSONResponse(status_info) + + @router.post("/workflows/update-index-timestamp") async def update_index_timestamp( request: Request, diff --git a/apollo/server/routes/api_workflows.py b/apollo/server/routes/api_workflows.py index e236f7b..2cfc704 100644 --- a/apollo/server/routes/api_workflows.py +++ b/apollo/server/routes/api_workflows.py @@ -4,6 +4,8 @@ from fastapi import APIRouter, Depends, HTTPException, status from typing import List +from temporalio.client import RPCError + from apollo.server.models.workflow import ( ProductListResponse, ProductInfo, WorkflowTriggerRequest, WorkflowTriggerResponse, WorkflowStatusResponse, WorkflowListResponse @@ -51,40 +53,53 @@ async def trigger_rh_matcher_workflow( """ try: service = WorkflowService() - workflow_id = await service.trigger_rh_matcher_workflow(request.major_versions) - + workflow_id = await service.trigger_rh_matcher_workflow( + request.major_versions, + ) + logger = Logger() - logger.info(f"User {user.email} triggered RhMatcherWorkflow {workflow_id} with major_versions: {request.major_versions}") - + logger.info( + f"User {user.email} triggered RhMatcherWorkflow " + f"{workflow_id} with major_versions: {request.major_versions}" + ) + return WorkflowTriggerResponse( workflow_id=workflow_id, status="started", message="RhMatcherWorkflow triggered successfully", - filtered_major_versions=request.major_versions + filtered_major_versions=request.major_versions, ) - - except ValueError as e: - # Handle validation errors (invalid major versions) + + except RPCError as e: + if "already running" in str(e).lower(): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="A workflow for this version is already running", + ) logger = Logger() - logger.error(f"Validation error triggering workflow: {str(e)}") + logger.error(f"Temporal RPC error triggering workflow: {e}") + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Workflow service unavailable", + ) + except ValueError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=str(e) + detail=str(e), ) except RuntimeError as e: - # Handle Temporal client errors logger = Logger() - logger.error(f"Runtime error triggering workflow: {str(e)}") + logger.error(f"Runtime error triggering workflow: {e}") raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="Workflow service unavailable" + detail="Workflow service unavailable", ) except Exception as e: logger = Logger() - logger.error(f"Error triggering workflow: {str(e)}") + logger.error(f"Error triggering workflow: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Failed to trigger workflow" + detail="Failed to trigger workflow", ) diff --git a/apollo/server/services/workflow_service.py b/apollo/server/services/workflow_service.py index 56d5103..dcc8ac6 100644 --- a/apollo/server/services/workflow_service.py +++ b/apollo/server/services/workflow_service.py @@ -5,7 +5,8 @@ from typing import Optional, List, Dict, Any from datetime import datetime -from temporalio.client import WorkflowHandle +from temporalio.client import RPCError, WorkflowHandle +from temporalio.common import WorkflowIDReusePolicy from apollo.db import SupportedProduct from apollo.rpmworker.rh_matcher_workflows import RhMatcherWorkflow, RhMatcherWorkflowInput @@ -48,40 +49,53 @@ async def get_available_products(self) -> List[Dict[str, Any]]: async def trigger_rh_matcher_workflow(self, major_versions: Optional[List[int]] = None) -> str: """ - Trigger RhMatcherWorkflow with optional major version filtering - + Trigger RhMatcherWorkflow with optional major version filtering. + + Uses a deterministic workflow ID when a single major version is + specified so Temporal deduplicates concurrent triggers for the + same version. Raises ``WorkflowAlreadyRunning`` if a workflow + for that version is already in progress. + Args: - major_versions: Optional list of Rocky Linux major versions to process (e.g., [8, 9, 10]) - If None, processes all versions (backward compatibility) - + major_versions: Optional list of Rocky Linux major versions + to process (e.g., [8, 9, 10]). If None, processes all + versions (backward compatibility). + Returns: - Workflow ID for tracking + Workflow ID for tracking. """ temporal_client = await self._get_temporal_client() - + if not temporal_client or not temporal_client.client: raise RuntimeError("Temporal client not initialized") - - # Validate major versions if provided + if major_versions: await self._validate_major_versions(major_versions) - - # Create workflow input - workflow_input = RhMatcherWorkflowInput(major_versions=major_versions) if major_versions else None - - # Generate unique workflow ID - workflow_id = f"rh-matcher-{uuid.uuid4()}" - - self.logger.info(f"Starting RhMatcherWorkflow with ID: {workflow_id}, major_versions: {major_versions}") - - # Start the workflow - workflow_handle: WorkflowHandle = await temporal_client.client.start_workflow( + + workflow_input = ( + RhMatcherWorkflowInput(major_versions=major_versions) + if major_versions + else None + ) + + if major_versions and len(major_versions) == 1: + workflow_id = f"rh-matcher-compose-v{major_versions[0]}" + else: + workflow_id = f"rh-matcher-{uuid.uuid4()}" + + self.logger.info( + f"Starting RhMatcherWorkflow with ID: {workflow_id}, " + f"major_versions: {major_versions}" + ) + + await temporal_client.client.start_workflow( RhMatcherWorkflow.run, workflow_input, id=workflow_id, task_queue=TASK_QUEUE, + id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE, ) - + return workflow_id async def trigger_poll_rhcsaf_workflow(self) -> str: @@ -112,64 +126,65 @@ async def trigger_poll_rhcsaf_workflow(self) -> str: async def get_workflow_status(self, workflow_id: str) -> Dict[str, Any]: """ - Get status of a specific workflow - + Get status of a specific workflow via Temporal's describe API. + Args: - workflow_id: The workflow ID to check - + workflow_id: The workflow ID to check. + Returns: - Dictionary containing workflow status information + Dictionary containing workflow status information. """ temporal_client = await self._get_temporal_client() - + if not temporal_client or not temporal_client.client: raise RuntimeError("Temporal client not initialized") - + try: - workflow_handle = temporal_client.client.get_workflow_handle(workflow_id) - - # Check workflow status without blocking - try: - # Try to get result with timeout to avoid blocking - import asyncio - result = await asyncio.wait_for(workflow_handle.result(), timeout=1.0) - status = "completed" - result_data = result - except asyncio.TimeoutError: - # Workflow is still running - status = "running" - result_data = None - except Exception as e: - # Check if workflow failed - if "workflow execution already completed" in str(e).lower(): - # Try to get the result without timeout - try: - result = await workflow_handle.result() - status = "completed" - result_data = result - except Exception: - status = "failed" - result_data = str(e) - else: - status = "error" - result_data = str(e) - + handle = temporal_client.client.get_workflow_handle(workflow_id) + description = await handle.describe() + execution_status = description.status.name + + result_data = None + if execution_status == "COMPLETED": + result_data = await handle.result() + return { "workflow_id": workflow_id, - "status": status, + "status": execution_status.lower(), "result": result_data, "execution_info": { "workflow_id": workflow_id, - "run_id": workflow_handle.run_id if hasattr(workflow_handle, 'run_id') else None - } + "run_id": description.run_id, + "start_time": ( + description.start_time.isoformat() + if description.start_time + else None + ), + "close_time": ( + description.close_time.isoformat() + if description.close_time + else None + ), + }, + } + + except RPCError as e: + self.logger.error( + f"Temporal RPC error for workflow {workflow_id}: {e}" + ) + return { + "workflow_id": workflow_id, + "status": "error", + "error": str(e), } - except Exception as e: - self.logger.error(f"Error getting workflow status for {workflow_id}: {str(e)}") + self.logger.error( + f"Error getting workflow status for {workflow_id}: {e}" + ) return { "workflow_id": workflow_id, "status": "error", - "error": str(e) + "error": str(e), } async def list_recent_workflows(self, limit: int = 50) -> List[Dict[str, Any]]: diff --git a/apollo/server/templates/admin_workflows.jinja b/apollo/server/templates/admin_workflows.jinja index b3f7916..be8119d 100644 --- a/apollo/server/templates/admin_workflows.jinja +++ b/apollo/server/templates/admin_workflows.jinja @@ -27,8 +27,47 @@ {% endif %} + + {% if tracked_workflow_id %} +
+
+
+

Workflow Status: {{ tracked_workflow_name }}

+

+ Workflow ID: {{ tracked_workflow_id }} +

+
+ + + + + + + + + + + + + + + + + + + +
Statuschecking...
Start Time--
Close Time--
Run ID--
+
+

+ Polling every 15 seconds... +

+
+
+
+ {% endif %} + -
+

RH Matcher Workflow

@@ -210,6 +249,79 @@