tegwick
26934e25b9
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f
134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
"""Temporal worker entrypoint for activity-core.
|
|
|
|
Starts the ``orchestrator-tq`` worker for RunActivityWorkflow and its activities.
|
|
|
|
T23: Calls sync_schedules before entering the worker run loop to ensure
|
|
all cron ActivityDefinitions have live Temporal Schedules.
|
|
|
|
T31: Exposes Prometheus metrics via the Temporal SDK runtime on :9090/metrics.
|
|
|
|
Run with:
|
|
TEMPORAL_HOST=localhost:7233 \
|
|
ACTCORE_DB_URL=postgresql+asyncpg://actcore:actcore@localhost:5433/actcore \
|
|
python -m activity_core.worker
|
|
|
|
Environment variables:
|
|
TEMPORAL_HOST Temporal frontend address (default: localhost:7233)
|
|
TEMPORAL_NAMESPACE Temporal namespace (default: default)
|
|
ACTCORE_DB_URL App DB connection string (required)
|
|
PROMETHEUS_BIND_ADDR Prometheus metrics bind (default: 0.0.0.0:9090)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import signal
|
|
|
|
from temporalio.client import Client
|
|
from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
|
|
from temporalio.worker import Worker
|
|
|
|
from activity_core.activities import (
|
|
apply_sbom_catchup,
|
|
emit_tasks,
|
|
execute_bounded_operation,
|
|
evaluate_instructions,
|
|
evaluate_rules,
|
|
init_session_factory,
|
|
load_activity_definition,
|
|
log_run,
|
|
persist_instruction_reports,
|
|
persist_ops_evidence,
|
|
resolve_context,
|
|
)
|
|
from activity_core.db import make_engine
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
from activity_core.sync_service import run_sync
|
|
from activity_core.workflows import RunActivityWorkflow
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TEMPORAL_HOST = os.environ.get("TEMPORAL_HOST", "localhost:7233")
|
|
TEMPORAL_NAMESPACE = os.environ.get("TEMPORAL_NAMESPACE", "default")
|
|
PROMETHEUS_BIND_ADDR = os.environ.get("PROMETHEUS_BIND_ADDR", "0.0.0.0:9090")
|
|
|
|
ORCHESTRATOR_TASK_QUEUE = "orchestrator-tq"
|
|
|
|
|
|
async def run() -> None:
|
|
db_url = os.environ.get("ACTCORE_DB_URL")
|
|
if not db_url:
|
|
raise RuntimeError("ACTCORE_DB_URL is required")
|
|
init_session_factory(db_url)
|
|
|
|
# T31: Configure the Temporal SDK runtime to emit metrics in Prometheus format.
|
|
runtime = Runtime(
|
|
telemetry=TelemetryConfig(
|
|
metrics=PrometheusConfig(bind_address=PROMETHEUS_BIND_ADDR)
|
|
)
|
|
)
|
|
|
|
client = await Client.connect(
|
|
TEMPORAL_HOST, namespace=TEMPORAL_NAMESPACE, runtime=runtime
|
|
)
|
|
|
|
logger.info("Syncing ActivityDefinitions and Temporal Schedules...")
|
|
sync_engine = make_engine(db_url)
|
|
session_factory = async_sessionmaker(sync_engine, expire_on_commit=False)
|
|
try:
|
|
sync_result = await run_sync(
|
|
session_factory=session_factory,
|
|
temporal_client=client,
|
|
definitions=True,
|
|
schedules=True,
|
|
event_types=False,
|
|
)
|
|
for error in sync_result["errors"]:
|
|
logger.error(
|
|
"startup sync %s failed — %s: %s",
|
|
error["stage"],
|
|
error["type"],
|
|
error["message"],
|
|
)
|
|
finally:
|
|
await sync_engine.dispose()
|
|
|
|
orchestrator_worker = Worker(
|
|
client,
|
|
task_queue=ORCHESTRATOR_TASK_QUEUE,
|
|
workflows=[RunActivityWorkflow],
|
|
activities=[
|
|
load_activity_definition,
|
|
resolve_context,
|
|
apply_sbom_catchup,
|
|
execute_bounded_operation,
|
|
log_run,
|
|
evaluate_rules,
|
|
evaluate_instructions,
|
|
persist_instruction_reports,
|
|
persist_ops_evidence,
|
|
emit_tasks,
|
|
],
|
|
)
|
|
|
|
loop = asyncio.get_running_loop()
|
|
stop = asyncio.Event()
|
|
loop.add_signal_handler(signal.SIGTERM, stop.set)
|
|
loop.add_signal_handler(signal.SIGINT, stop.set)
|
|
|
|
async with orchestrator_worker:
|
|
logger.info(
|
|
"Workers running — queues: %r (namespace=%r)",
|
|
[ORCHESTRATOR_TASK_QUEUE],
|
|
TEMPORAL_NAMESPACE,
|
|
)
|
|
await stop.wait()
|
|
logger.info("Shutdown signal received — draining workers")
|
|
logger.info("Workers stopped cleanly")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
asyncio.run(run())
|