From 042c7aab33c7f5ead9c88054c3bbff65eefcd1d3 Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Thu, 4 Jun 2026 20:36:25 -0300 Subject: [PATCH] feat(background-execution): emit job lifecycle metrics from runner --- .../services/background_execution/runner.py | 58 ++++++++++++++++++- .../background_execution/test_runner.py | 31 ++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/src/backend/base/langflow/services/background_execution/runner.py b/src/backend/base/langflow/services/background_execution/runner.py index 3fbf0775b6..cf54ad17c0 100644 --- a/src/backend/base/langflow/services/background_execution/runner.py +++ b/src/backend/base/langflow/services/background_execution/runner.py @@ -18,13 +18,16 @@ from __future__ import annotations import asyncio import contextlib import json +import time from collections.abc import AsyncIterator, Callable from typing import TYPE_CHECKING, Any from lfx.log.logger import logger +from langflow.services.background_execution import metrics as bg_metrics from langflow.services.background_execution.live_bus import LiveFrame from langflow.services.database.models.jobs.model import JobStatus, SignalType +from langflow.services.deps import get_settings_service if TYPE_CHECKING: from uuid import UUID @@ -63,6 +66,15 @@ class JobRunner: # (scripted tests that don't care about liveness) the heartbeat is off. self._owner = owner self._heartbeat_interval_s = max(heartbeat_interval_s, 0.1) + # Metric label for the backend this run executes on, derived once. + # "scaled" when a redis job queue drains jobs in separate workers, else + # "default" (in-process executor). Best-effort: a missing settings + # service (scripted tests) must not break runner construction. + backend = "default" + with contextlib.suppress(Exception): + if get_settings_service().settings.background_backend_is_scaled: + backend = "scaled" + self._backend = backend async def run(self, *, job_id: UUID, source_kwargs: dict[str, Any]) -> None: """Execute one background job to a terminal state.""" @@ -81,8 +93,18 @@ class JobRunner: else: await self._drive(job_id=job_id, source_kwargs=source_kwargs) - heartbeat_task = self._start_heartbeat(job_id) + bg_metrics.emit_job_started(backend=self._backend) + # Monotonic start: the job row has no started_at/finished_at, so we + # measure run duration off a monotonic clock captured here and read at + # the terminal point. Monotonic avoids wall-clock skew. + started_monotonic = time.monotonic() + + # Start the heartbeat INSIDE the try so any failure here still lands in + # the finally: emit_job_started already fired, so its terminal partner + # (_emit_terminal_metrics) must always run to keep started/terminal paired. + heartbeat_task: asyncio.Task | None = None try: + heartbeat_task = self._start_heartbeat(job_id) await self._jobs.execute_with_status(job_id, _wrapped) except asyncio.CancelledError as exc: # A cooperative STOP that we raised ourselves ends the run cleanly @@ -118,6 +140,40 @@ class JobRunner: with contextlib.suppress(Exception): await asyncio.shield(self._finalize_terminal_event(job_id)) await self._bus.close(str(job_id)) + # One terminal emit per run, off the authoritative final status (read + # after the stop reconcile + finalize so a late-stop CANCELLED counts + # as cancelled, not completed). emit_* helpers swallow their own + # errors, so no extra guard is needed here. + await self._emit_terminal_metrics(job_id, time.monotonic() - started_monotonic) + + async def _emit_terminal_metrics(self, job_id: UUID, duration_seconds: float) -> None: + """Emit the completed/failed counter + duration histogram once per run. + + Reads the final job status and maps it to the metric outcome/reason: + COMPLETED -> completed; FAILED -> failed(reason=error); TIMED_OUT -> + failed(reason=timeout); CANCELLED -> failed(reason=cancelled). A missing + job row (best-effort fetch failure) is skipped silently. + + A non-terminal status read here (e.g. a status-write failure left the row + IN_PROGRESS, or the run was orphaned) is an INTENTIONAL no-op: that run's + terminal metric is emitted by the reconciliation path (Task 5), not here. + """ + job = None + with contextlib.suppress(Exception): + job = await self._jobs.get_job_by_job_id(job_id) + if job is None: + return + if job.status == JobStatus.COMPLETED: + bg_metrics.emit_job_completed(backend=self._backend) + bg_metrics.emit_job_duration(seconds=duration_seconds, outcome="completed", backend=self._backend) + elif job.status in (JobStatus.FAILED, JobStatus.TIMED_OUT, JobStatus.CANCELLED): + reason = { + JobStatus.FAILED: "error", + JobStatus.TIMED_OUT: "timeout", + JobStatus.CANCELLED: "cancelled", + }[job.status] + bg_metrics.emit_job_failed(reason=reason, backend=self._backend) + bg_metrics.emit_job_duration(seconds=duration_seconds, outcome="failed", backend=self._backend) async def _finalize_terminal_event(self, job_id: UUID) -> None: """Backfill the error blob + terminal event for TIMED_OUT / CANCELLED. diff --git a/src/backend/tests/unit/services/background_execution/test_runner.py b/src/backend/tests/unit/services/background_execution/test_runner.py index 6a4a73a4dd..33ba70fd76 100644 --- a/src/backend/tests/unit/services/background_execution/test_runner.py +++ b/src/backend/tests/unit/services/background_execution/test_runner.py @@ -184,6 +184,37 @@ async def test_late_stop_after_completion_clears_result(active_user): assert (job.error or {}).get("type") == "cancelled" +async def test_runner_lifecycle_metric_emit_does_not_break_run(active_user): + """Emit wiring must reach a real terminal state without raising. + + The runner emits started/completed/failed counters + a duration histogram at + its lifecycle points. With the real telemetry service available (via the + ``client`` fixture), driving a job to COMPLETED must still reach the terminal + state and must not raise out of the emit calls. Counter VALUES are not + asserted here (OTel instruments are write-only); the contract is "emit does + not change runner behavior". The runner derives a backend label at __init__. + """ + job_service = get_job_service() + job_id = await _make_job(uuid4(), active_user.id) + + async def source(**_kwargs) -> AsyncIterator[tuple[bytes, str]]: + yield _frame("build_start", {}) + yield _frame("end_vertex", {"id": "n1"}) + yield _frame("end", {}) + + bus = InMemoryLiveBus() + adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t")) + runner = JobRunner(job_service=job_service, live_bus=bus, adapter=adapter, frame_source=source) + # backend label is derived once at construction and is one of the small enum. + assert runner._backend in ("default", "scaled") + + # No exception may propagate from the emit calls wired into run(). + await runner.run(job_id=job_id, source_kwargs={}) + + job = await job_service.get_job_by_job_id(job_id) + assert job.status == JobStatus.COMPLETED + + async def test_runner_stops_at_signal_boundary(active_user): job_service = get_job_service() job_id = await _make_job(uuid4(), active_user.id)