SIENTIAPDE-1646

Refactor worker task queue management and update README - Introduced runtime-scoped task queues for workflows, replacing legacy queue names. - Updated worker implementation to utilize `sientia_do.temporal.worker.prepare_worker`. - Added `RUNTIME` environment variable to configure task queue suffixes. - Enhanced README documentation to reflect changes in task queue structure and worker setup.
2026-05-19 17:07:18 -03:00
parent d856150e24
commit 2ccda3e440
5 changed files with 64 additions and 92 deletions
--- a/README.md
+++ b/README.md
@@ -126,10 +126,15 @@ Laborious uses a Temporal-based architecture with strong separation of concerns
 ### Key Components

 #### **Worker (`laborious/worker/worker.py`)**
- Temporal client setup, worker lifecycle, task queues
+- Temporal client setup, four workers via `sientia_do.temporal.worker.prepare_worker`
+- Runtime-scoped task queues: `{workflow}-{RUNTIME}-queue` for all workflows
 - Metrics server initialization, notification handler setup
 - Graceful shutdown and autoscaling-friendly behavior

+**Breaking (schedulers):** drift and simple_metrics queues are no longer `drift-queue` /
+`simple_metrics-queue`. Use `drift-{RUNTIME}-queue` and `simple_metrics-{RUNTIME}-queue`
+matching the worker pod `RUNTIME` env (same as `predictions_batch` / `minimal_retrain`).
+
 #### **Workflows (`laborious/workflows/`)**
 - `predictions_batch.py`: Batch prediction entry point
 - `sub_workflows/prediction_process.py`: Core prediction pipeline
@@ -802,6 +807,7 @@ See [OPC UA Communication](#opc-ua-communication) for semantics, concurrency, an
 |----------|-------------|---------|----------|
 | `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes |
 | `TEMPORAL_NAMESPACE` | Temporal namespace | `laborious` | No |
+| `RUNTIME` | Task queue suffix for all workflows (`{workflow}-{RUNTIME}-queue`) | _(none)_ | Yes |
 | `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes |
 | `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes |
 | `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes |
@@ -1097,8 +1103,7 @@ laborious/
 │       ├── prediction_process.py # Core prediction workflow
 │       └── format_and_export_prediction.py # Export workflow
 ├── worker/                  # Worker implementation
-│   ├── worker.py           # Main worker orchestrator
-│   └── prepare_worker.py   # Worker factory with autoscaling config
+│   └── worker.py           # Main worker orchestrator (uses sientia_do prepare_worker)
 ├── utils/                   # Utility functions
 │   ├── connectors_config.py # Environment-driven config builders
 │   ├── models/             # Data models
@@ -1179,7 +1184,9 @@ export LOG_LEVEL=DEBUG
 ### Scaling Considerations

 - **Horizontal Scaling**: Deploy multiple worker instances
- **Task Queue Distribution**: Use multiple task queues for different workflow types
+- **Task Queue Distribution**: One worker pod per `RUNTIME`; queues are
+  `predictions_batch-{RUNTIME}-queue`, `minimal_retrain-{RUNTIME}-queue`,
+  `drift-{RUNTIME}-queue`, `simple_metrics-{RUNTIME}-queue`
 - **Database Performance**: Optimize indexes and connection pooling
 - **MLFlow Performance**: Configure appropriate model serving resources

--- a/laborious/worker/prepare_worker.py
+++ b/laborious/worker/prepare_worker.py
@@ -1,73 +0,0 @@
-import os
-import re
-from collections.abc import Sequence
-from typing import Any
-
-from sientia_do.observability.logger import Logger
-from temporalio.client import Client
-from temporalio.worker import PollerBehaviorAutoscaling, Worker
-
-parameters = [
-    ('MAX_CONCURRENT_WORKFLOW_TASKS', '200'),
-    ('MAX_CONCURRENT_ACTIVITIES', '200'),
-    ('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'),
-    ('MAX_CACHED_WORKFLOWS', '200'),
-    ('WORKFLOW_POLLER_BEHAVIOUR_MINIMUM', '10'),
-    ('WORKFLOW_POLLER_BEHAVIOUR_INITIAL', '100'),
-    ('WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM', '200'),
-    ('ACTIVITY_POLLER_BEHAVIOUR_MINIMUM', '10'),
-    ('ACTIVITY_POLLER_BEHAVIOUR_INITIAL', '100'),
-    ('ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM', '200'),
-]
-
-
-def camel_to_snake(text: str) -> str:
-    """Convert camelCase or PascalCase to snake_case."""
-    text = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text)
-    text = re.sub('([a-z0-9])([A-Z])', r'\1_\2', text)
-    return text.lower()
-
-
-def prepare_worker(
-    main_workflow: type,
-    other_workflows: Sequence[type],
-    activities: Sequence[Any],
-    temporal_client: Client,
-    logger: Logger,
-) -> Worker:
-    main_workflow_name = main_workflow.__name__.upper()
-
-    queue_name = f'{camel_to_snake(main_workflow.__name__)}-queue'
-
-    local_workflow_parameters = {}
-
-    for parameter in parameters:
-        local_workflow_parameters[parameter[0]] = int(
-            os.getenv(main_workflow_name + '_' + parameter[0], parameter[1])
-        )
-
-    logger.info(f'Preparing worker for {main_workflow_name} with queue {queue_name}')
-    logger.info(f'Worker runtime config: {local_workflow_parameters}')
-
-    return Worker(
-        temporal_client,
-        task_queue=queue_name,
-        workflows=[main_workflow, *other_workflows],
-        activities=[*activities],
-        max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'],
-        max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'],
-        max_concurrent_local_activities=local_workflow_parameters[
-            'MAX_CONCURRENT_LOCAL_ACTIVITIES'
-        ],
-        max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'],
-        workflow_task_poller_behavior=PollerBehaviorAutoscaling(
-            minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MINIMUM'],
-            initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_INITIAL'],
-            maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM'],
-        ),
-        activity_task_poller_behavior=PollerBehaviorAutoscaling(
-            minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MINIMUM'],
-            initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_INITIAL'],
-            maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM'],
-        ),
-    )
--- a/laborious/worker/worker.py
+++ b/laborious/worker/worker.py
@@ -5,12 +5,14 @@ This module provides the main worker implementation for the Sientia DataOps Labo
 It orchestrates Temporal workers, manages task queues, and handles the lifecycle of
 prediction and retraining workflows.

-The worker supports multiple task queues:
- predictions_batch-queue: Handles batch prediction workflows (heavy workload)
-  Includes activities for MLFlow, data quality gates, OPC export, PI Web API export, and PostgreSQL
- minimal_retrain-queue: Handles model retraining workflows
- drift-queue: Handles drift detection workflows
- simple_metrics-queue: Handles simple metrics calculation workflows
+The worker supports multiple runtime-scoped task queues (via ``sientia_do.temporal.worker.prepare_worker``):
+- predictions_batch-{runtime}-queue: Batch prediction workflows (heavy workload)
+- minimal_retrain-{runtime}-queue: Model retraining workflows
+- drift-{runtime}-queue: Drift detection workflows
+- simple_metrics-{runtime}-queue: Simple metrics workflows
+
+``RUNTIME`` must be set; it is passed to every ``prepare_worker`` call. Schedulers must use the
+same queue names (breaking change vs legacy ``drift-queue`` / ``simple_metrics-queue``).

 Key Features:
 - Resource-based scaling with WorkerTuner (CPU and memory aware)
@@ -21,6 +23,7 @@ Key Features:
 - Multiple worker instances for different workflow types

 Environment Variables:
+- RUNTIME: Required non-empty string; suffix for all task queue names
 - TEMPORAL_HOST: Temporal server address (default: localhost:7233)
 - TEMPORAL_NAMESPACE: Temporal namespace (default: laborious)
 - POD_ID: Kubernetes pod identifier for metrics
@@ -40,6 +43,7 @@ with workflow.unsafe.imports_passed_through():
    from prometheus_client import start_http_server
    from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
    from sientia_do.observability.logger import get_logger
+    from sientia_do.temporal.worker.prepare_worker import prepare_worker
    from sientia_do.utils.connectors_config import (
        build_api_config,
        build_mongodb_config,
@@ -53,7 +57,6 @@ with workflow.unsafe.imports_passed_through():
        build_mlflow_config,
        build_opc_config,
    )
-    from laborious.worker.prepare_worker import prepare_worker
    from laborious.workflows.drift import Drift
    from laborious.workflows.minimal_retrain import MinimalRetrain
    from laborious.workflows.predictions_batch import PredictionsBatch
@@ -100,10 +103,21 @@ async def main():

    logger.custom_info(f'Starting Worker with POD_ID: {POD_ID}', metadata)

-    logger.custom_info('Starting prometheus client...', metadata)
+    runtime = os.getenv('RUNTIME', '').strip()
+    if not runtime:
+        logger.custom_critical(
+            'RUNTIME environment variable is required and must be non-empty',
+            metadata,
+        )
+        metrics.APP_UP.labels(pod_id=POD_ID).set(0)
+        sys.exit(1)
+
+    metadata_runtime = {**metadata, 'runtime': runtime}
+
+    logger.custom_info('Starting prometheus client...', metadata_runtime)
    start_prometheus_server()

-    logger.custom_info('Starting Notification Handler...', metadata)
+    logger.custom_info('Starting Notification Handler...', metadata_runtime)

    mongo_config = build_mongodb_config()
    notification_handler = NotificationHandler(
@@ -113,7 +127,7 @@ async def main():
        project_name=os.getenv('PROJECT_NAME', 'laborious'),
    )

-    logger.custom_info('Starting Activities...', metadata)
+    logger.custom_info('Starting Activities...', metadata_runtime)

    activities = Activities(
        postgres_config=build_postgres_config(),
@@ -125,10 +139,13 @@ async def main():
        notification_handler=notification_handler,
    )

-    logger.custom_info('Initializing OPC...', metadata)
+    logger.custom_info('Initializing OPC...', metadata_runtime)
    await activities.init_opc()

-    logger.custom_info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
+    logger.custom_info(
+        f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...',
+        metadata_runtime,
+    )

    new_runtime = Runtime(
        telemetry=TelemetryConfig(
@@ -136,7 +153,7 @@ async def main():
        )
    )

-    logger.custom_info(f'Starting Temporal Client at {host}...', metadata)
+    logger.custom_info(f'Starting Temporal Client at {host}...', metadata_runtime)

    temporal_client = await client.Client.connect(
        target_host=host,
@@ -144,7 +161,7 @@ async def main():
        runtime=new_runtime,
    )

-    logger.custom_info('Starting Workers...', metadata)
+    logger.custom_info(f'Starting Workers (runtime={runtime})...', metadata_runtime)

    workers = [
        prepare_worker(
@@ -159,6 +176,7 @@ async def main():
                activities.export_data_to_postgres,
            ],
            logger=logger,
+            runtime=runtime,
        ),
        prepare_worker(
            temporal_client=temporal_client,
@@ -170,6 +188,7 @@ async def main():
                activities.export_data_to_postgres,
            ],
            logger=logger,
+            runtime=runtime,
        ),
        prepare_worker(
            temporal_client=temporal_client,
@@ -182,6 +201,7 @@ async def main():
                activities.export_data_to_postgres,
            ],
            logger=logger,
+            runtime=runtime,
        ),
        prepare_worker(
            temporal_client=temporal_client,
@@ -211,6 +231,7 @@ async def main():
                activities.write_pi_web_api_data,
            ],
            logger=logger,
+            runtime=runtime,
        ),
    ]

@@ -218,7 +239,7 @@ async def main():
    for w in workers:
        handlers.append(w.run())

-    logger.custom_info('Workers started successfully', metadata)
+    logger.custom_info('Workers started successfully', metadata_runtime)

    exit_code = 0
    try:
--- a/tests/laborious/worker/test_runtime_task_queues.py
+++ b/tests/laborious/worker/test_runtime_task_queues.py
@@ -0,0 +1,14 @@
+from sientia_do.temporal.worker.prepare_worker import build_queue_name
+
+from laborious.workflows.drift import Drift
+from laborious.workflows.minimal_retrain import MinimalRetrain
+from laborious.workflows.predictions_batch import PredictionsBatch
+from laborious.workflows.simple_metrics import SimpleMetrics
+
+
+def test_runtime_scoped_queue_names():
+    runtime = 'prod-a'
+    assert build_queue_name(PredictionsBatch.__name__, runtime) == 'predictions_batch-prod-a-queue'
+    assert build_queue_name(MinimalRetrain.__name__, runtime) == 'minimal_retrain-prod-a-queue'
+    assert build_queue_name(Drift.__name__, runtime) == 'drift-prod-a-queue'
+    assert build_queue_name(SimpleMetrics.__name__, runtime) == 'simple_metrics-prod-a-queue'
--- a/values.yaml
+++ b/values.yaml
@@ -209,6 +209,9 @@ env:
    value: "temporal-frontend.temporal.svc.cluster.local:7233"
  - name: TEMPORAL_NAMESPACE
    value: "laborious"
+  # Suffix for all Temporal task queues: {workflow}-{RUNTIME}-queue
+  - name: RUNTIME
+    value: "legacy"

  - name: MONGODB_USERNAME
    value: "root"