From 77e355ae92a1d36a43238dc8c8cd1606bfc094fe Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Fri, 5 Jun 2026 14:25:16 -0300 Subject: [PATCH] feat(background-execution): DB-derived worker_online/busy/idle gauges Replace langflow_bg_alive_workers with three registry-derived gauges (online/busy/idle) the collector reads from worker_registry via WorkerRegistryService.count_by_state each tick, and prune stale rows past retention once per tick (best-effort). online_window is 3x the worker registry interval; both interval and retention are threaded from settings at the collector construction site. --- .../background_execution/metrics_collector.py | 175 +++++++---- .../services/telemetry/opentelemetry.py | 18 +- .../test_metrics_collector.py | 294 +++++++++++++----- .../test_metrics_collector_lifespan.py | 7 +- .../services/telemetry/test_opentelemetry.py | 6 +- .../telemetry/test_prometheus_exposition.py | 4 +- 6 files changed, 354 insertions(+), 150 deletions(-) diff --git a/src/backend/base/langflow/services/background_execution/metrics_collector.py b/src/backend/base/langflow/services/background_execution/metrics_collector.py index 22a4e3ec96..c4ff79fef0 100644 --- a/src/backend/base/langflow/services/background_execution/metrics_collector.py +++ b/src/backend/base/langflow/services/background_execution/metrics_collector.py @@ -9,9 +9,10 @@ writes — Task 7 wires these into the OTel gauges. timezone-aware column, but SQLite hands it back naive; we normalize to aware UTC before subtracting so the math is valid on both SQLite and Postgres. -The "alive" worker definition reuses ``JobService.is_lease_stale`` semantics so -it matches the watchdog exactly: a heartbeat is FRESH while its age is within -the lease window (``age <= lease_window``), stale once older. +The worker online/busy/idle gauges are derived from the durable +``worker_registry`` table (``WorkerRegistryService.count_by_state``): a row is +online while ``last_heartbeat >= now - online_window`` (online_window = 3x the +worker's registry interval). The collector also prunes rows past retention. """ from __future__ import annotations @@ -26,7 +27,8 @@ from lfx.log.logger import logger from sqlmodel import col, func, select from langflow.services.background_execution.metrics import current_backend -from langflow.services.database.models.jobs.model import Job, JobStatus +from langflow.services.background_execution.worker_registry import WorkerRegistryService +from langflow.services.database.models.jobs.model import Job, JobEvent, JobStatus from langflow.services.deps import get_telemetry_service, session_scope if TYPE_CHECKING: @@ -38,20 +40,41 @@ if TYPE_CHECKING: NONTERMINAL_STATUSES = (JobStatus.QUEUED, JobStatus.IN_PROGRESS) -async def count_nonterminal_jobs(session) -> dict[str, int]: - """Return counts of non-terminal jobs keyed by status string. +def _has_job_events(): + """EXISTS subquery: the job has at least one ``job_events`` row. - ``{"queued": 3, "in_progress": 1}`` via a ``GROUP BY status`` aggregate - filtered to the non-terminal set. Statuses with zero rows are omitted (the - collector loop fills in the canonical set with 0 when it sets gauges). + Distinguishes a TRUE background job from a Memory-Base "run job". Every flow + build writes a second ``job`` row keyed by the graph run_id (``build.py``, + predates the bg work); it is ``type=workflow`` too, so a naive status query + double-counts it. The background runner uniquely appends ``job_events`` rows, + while run jobs never have any (they go straight IN_PROGRESS->terminal via + ``execute_with_status`` and never QUEUED). So a background job is + ``status == QUEUED`` OR EXISTS(job_events). ``job_id`` is indexed + (UNIQUE(job_id, seq)), so the EXISTS is cheap and dialect-agnostic. """ - stmt = select(Job.status, func.count()).where(col(Job.status).in_(NONTERMINAL_STATUSES)).group_by(Job.status) - result = await session.exec(stmt) + return select(JobEvent.id).where(col(JobEvent.job_id) == Job.job_id).exists() + + +async def count_nonterminal_jobs(session) -> dict[str, int]: + """Return counts of non-terminal background jobs keyed by status string. + + ``{"queued": 3, "in_progress": 1}``. QUEUED is bg-only (run jobs are never + queued), so it counts as-is. IN_PROGRESS adds the EXISTS(job_events) filter + because a run job is briefly IN_PROGRESS with no events — without the filter + it would inflate the in_progress count. A status with zero rows is omitted + (the collector loop fills in the canonical set with 0 when it sets gauges). + """ + queued = (await session.exec(select(func.count()).select_from(Job).where(Job.status == JobStatus.QUEUED))).one() + in_progress = ( + await session.exec( + select(func.count()).select_from(Job).where(Job.status == JobStatus.IN_PROGRESS).where(_has_job_events()) + ) + ).one() counts: dict[str, int] = {} - for status, count in result.all(): - # ``status`` is a JobStatus enum; ``.value`` is the wire string. - key = status.value if isinstance(status, JobStatus) else str(status) - counts[key] = int(count) + if int(queued): + counts[JobStatus.QUEUED.value] = int(queued) + if int(in_progress): + counts[JobStatus.IN_PROGRESS.value] = int(in_progress) return counts @@ -75,38 +98,15 @@ async def oldest_queued_seconds(session, now: datetime) -> float: return max(age, 0.0) -async def alive_worker_count(session, now: datetime, lease_window: float) -> int: - """Count distinct heartbeat owners whose lease is still fresh. +async def worker_state_counts(session, now: datetime, online_window: timedelta) -> dict[str, int]: + """Aggregate online/busy/idle worker counts from the durable ``worker_registry``. - Fresh means ``heartbeat_at >= now - lease_window`` — the exact inverse of - ``JobService.is_lease_stale`` (stale once ``age > lease_window``), so a - worker counted "alive" here is one the watchdog would NOT reconcile. The - heartbeat lease lives in ``job_metadata`` (``owner`` + ``heartbeat_at`` ISO - string), stamped only on IN_PROGRESS rows by ``JobService.heartbeat``. - - Parsing the ISO ``heartbeat_at`` in Python (rather than a cross-dialect - JSON-string SQL comparison) keeps the boundary identical to is_lease_stale - on both SQLite and Postgres. + Thin wrapper over ``WorkerRegistryService.count_by_state`` so the collector + derives its fleet gauges from the same source of truth the worker loop writes. + A row is online when ``last_heartbeat >= now - online_window``; busy/idle split + online rows by state. ``now`` is injected for deterministic freshness math. """ - stmt = select(Job.job_metadata).where(Job.status == JobStatus.IN_PROGRESS) - result = await session.exec(stmt) - alive_owners: set[str] = set() - for row in result.all(): - meta = row or {} - owner = meta.get("owner") - raw = meta.get("heartbeat_at") - if not owner or not raw: - continue - try: - hb = datetime.fromisoformat(raw) - except (TypeError, ValueError): - continue - if hb.tzinfo is None: - hb = hb.replace(tzinfo=timezone.utc) - age = (now - hb).total_seconds() - if age <= lease_window: - alive_owners.add(owner) - return len(alive_owners) + return await WorkerRegistryService().count_by_state(session, now=now, online_window=online_window) def _error_type_expr(session): @@ -136,25 +136,32 @@ async def terminal_counts(session) -> dict[str, int]: Returns ``{"started", "completed", "failed_error", "failed_worker_lost", "timed_out", "cancelled"}``: - * ``started`` — every job that has begun: ``status != QUEUED`` (IN_PROGRESS - plus every terminal state). + * ``started`` — every background job that has begun: ``status != QUEUED`` + AND EXISTS(job_events) (IN_PROGRESS plus every terminal state). * ``completed`` — ``status == COMPLETED``. * FAILED jobs split by ``error->>'type'``: ``failed_worker_lost`` is FAILED AND ``type == 'worker_lost'``; ``failed_error`` is every other FAILED row. * ``timed_out`` / ``cancelled`` — the matching terminal statuses. - The worker_lost split uses a dialect-aware JSON extract (see - ``_error_type_expr``) so it stays a single bounded SQL aggregate on both - SQLite and Postgres. + Every non-queued count adds the EXISTS(job_events) filter so a Memory-Base + "run job" (build.py's second workflow row, which never appends job_events) + is excluded — without it the counts double (see ``_has_job_events``). The + worker_lost split uses a dialect-aware JSON extract (see ``_error_type_expr``) + so it stays a single bounded SQL aggregate on both SQLite and Postgres. """ - started_stmt = select(func.count()).select_from(Job).where(Job.status != JobStatus.QUEUED) - completed_stmt = select(func.count()).select_from(Job).where(Job.status == JobStatus.COMPLETED) - timed_out_stmt = select(func.count()).select_from(Job).where(Job.status == JobStatus.TIMED_OUT) - cancelled_stmt = select(func.count()).select_from(Job).where(Job.status == JobStatus.CANCELLED) + has_events = _has_job_events() + started_stmt = select(func.count()).select_from(Job).where(Job.status != JobStatus.QUEUED).where(has_events) + completed_stmt = select(func.count()).select_from(Job).where(Job.status == JobStatus.COMPLETED).where(has_events) + timed_out_stmt = select(func.count()).select_from(Job).where(Job.status == JobStatus.TIMED_OUT).where(has_events) + cancelled_stmt = select(func.count()).select_from(Job).where(Job.status == JobStatus.CANCELLED).where(has_events) error_type = _error_type_expr(session) worker_lost_stmt = ( - select(func.count()).select_from(Job).where(Job.status == JobStatus.FAILED).where(error_type == "worker_lost") + select(func.count()) + .select_from(Job) + .where(Job.status == JobStatus.FAILED) + .where(has_events) + .where(error_type == "worker_lost") ) started = (await session.exec(started_stmt)).one() @@ -166,7 +173,9 @@ async def terminal_counts(session) -> dict[str, int]: # with a NULL error or no ``type`` key, so count FAILED-not-worker_lost as the # complement of worker_lost to capture those rows too. failed_total = ( - await session.exec(select(func.count()).select_from(Job).where(Job.status == JobStatus.FAILED)) + await session.exec( + select(func.count()).select_from(Job).where(Job.status == JobStatus.FAILED).where(has_events) + ) ).one() return { @@ -196,6 +205,9 @@ async def duration_percentiles(session, now: datetime, window_seconds: float) -> boundary rows), so SQLite gets a naive-UTC cutoff that matches the stored format. Naive SQLite datetimes are still normalized to aware UTC before subtracting, the same way ``oldest_queued_seconds`` does. + + Only TRUE background jobs are considered (EXISTS(job_events)) so a run job's + near-instant duration does not skew p50/p95 — see ``_has_job_events``. """ cutoff = now - timedelta(seconds=window_seconds) # Bind the cutoff in the form the stored column uses: aware for postgres, @@ -206,6 +218,7 @@ async def duration_percentiles(session, now: datetime, window_seconds: float) -> select(Job.created_timestamp, Job.finished_timestamp) .where(col(Job.finished_timestamp).is_not(None)) .where(col(Job.finished_timestamp) >= sql_cutoff) + .where(_has_job_events()) ) result = await session.exec(stmt) durations: list[float] = [] @@ -242,26 +255,41 @@ class BackgroundMetricsCollector: Best-effort: a failing tick logs a warning and returns without raising so the loop keeps running — observability must never crash the service. - ``lease_window`` defaults to ``45.0`` to match the runtime - ``background_lease_ttl_s`` setting that ``sweep_orphans`` / ``requeue_lost`` - are actually called with, so "alive workers" agrees with what the watchdog - reconciles. Task 8 passes ``settings.background_lease_ttl_s`` explicitly. + ``registry_interval`` is the worker's ``background_worker_registry_interval_s``; + the online window is ``3 * registry_interval`` (a worker is online while it has + beat within three of its own intervals). ``registry_retention_s`` is how long a + stale row is kept before the per-tick prune removes it. Both default from the + settings at the construction site (``maybe_start_metrics_collector``). """ - def __init__(self, *, interval: float, lease_window: float = 45.0, duration_window_seconds: float = 300.0): + def __init__( + self, + *, + interval: float, + registry_interval: float = 10.0, + registry_retention_s: float = 3600.0, + duration_window_seconds: float = 300.0, + ): self.interval = interval - self.lease_window = lease_window + self.registry_interval = registry_interval + self.registry_retention_s = registry_retention_s self.duration_window_seconds = duration_window_seconds self._stopped = False self._task: asyncio.Task | None = None - async def collect_once(self, session) -> None: - """Run the three queries and push the gauges. Never raises.""" + async def collect_once(self, session, *, now: datetime | None = None) -> None: + """Run the queries and push the gauges. Never raises. + + ``now`` defaults to the wall clock so the loop owns the tick's clock; tests + inject an explicit ``now`` for deterministic freshness/retention math. + """ try: - now = datetime.now(timezone.utc) + if now is None: + now = datetime.now(timezone.utc) counts = await count_nonterminal_jobs(session) oldest = await oldest_queued_seconds(session, now) - alive = await alive_worker_count(session, now, self.lease_window) + online_window = timedelta(seconds=3 * self.registry_interval) + worker_counts = await worker_state_counts(session, now, online_window) backend = current_backend() ot = get_telemetry_service().ot @@ -275,7 +303,9 @@ class BackgroundMetricsCollector: {"status": status.value, "backend": backend}, ) ot.update_gauge("langflow_bg_oldest_queued_seconds", oldest, {"backend": backend}) - ot.update_gauge("langflow_bg_alive_workers", alive, {"backend": backend}) + ot.update_gauge("langflow_bg_workers_online", worker_counts["online"], {"backend": backend}) + ot.update_gauge("langflow_bg_workers_busy", worker_counts["busy"], {"backend": backend}) + ot.update_gauge("langflow_bg_workers_idle", worker_counts["idle"], {"backend": backend}) # Cumulative all-time throughput/outcome counts, set as observable # counters (last-absolute-value-wins per label-set). The worker runs @@ -303,6 +333,14 @@ class BackgroundMetricsCollector: p50, p95 = await duration_percentiles(session, now, self.duration_window_seconds) ot.update_gauge("langflow_bg_job_duration_p50_seconds", p50, {"backend": backend}) ot.update_gauge("langflow_bg_job_duration_p95_seconds", p95, {"backend": backend}) + + # Best-effort prune of crashed-worker rows past retention. Guarded on its + # own so a prune failure (e.g. a write contention) does not drop the gauges + # already set this tick — the next tick retries. + try: + await WorkerRegistryService().prune_stale(session, now=now, retention_s=self.registry_retention_s) + except Exception as exc: # noqa: BLE001 - prune is best-effort, never breaks the tick + logger.warning(f"bg worker_registry prune skipped: {exc}") except Exception as exc: # noqa: BLE001 - observability must never crash the loop logger.warning(f"bg metrics collection tick skipped: {exc}") @@ -346,7 +384,8 @@ async def maybe_start_metrics_collector(app: FastAPI, settings: Any, *, promethe try: collector = BackgroundMetricsCollector( interval=settings.background_metrics_interval, - lease_window=settings.background_lease_ttl_s, + registry_interval=settings.background_worker_registry_interval_s, + registry_retention_s=settings.background_worker_registry_retention_s, ) collector.start() app.state.background_metrics_collector = collector diff --git a/src/backend/base/langflow/services/telemetry/opentelemetry.py b/src/backend/base/langflow/services/telemetry/opentelemetry.py index 9dae212b8f..7abea44aef 100644 --- a/src/backend/base/langflow/services/telemetry/opentelemetry.py +++ b/src/backend/base/langflow/services/telemetry/opentelemetry.py @@ -180,8 +180,22 @@ class OpenTelemetry(metaclass=ThreadSafeSingletonMetaUsingWeakref): labels={"backend": mandatory_label}, ) self._add_metric( - name="langflow_bg_alive_workers", - description="Distinct background workers heartbeating within the lease window", + name="langflow_bg_workers_online", + description="Background workers heartbeating within the online window", + unit="", + metric_type=MetricType.OBSERVABLE_GAUGE, + labels={"backend": mandatory_label}, + ) + self._add_metric( + name="langflow_bg_workers_busy", + description="Online background workers currently running a job", + unit="", + metric_type=MetricType.OBSERVABLE_GAUGE, + labels={"backend": mandatory_label}, + ) + self._add_metric( + name="langflow_bg_workers_idle", + description="Online background workers currently idle", unit="", metric_type=MetricType.OBSERVABLE_GAUGE, labels={"backend": mandatory_label}, diff --git a/src/backend/tests/unit/services/background_execution/test_metrics_collector.py b/src/backend/tests/unit/services/background_execution/test_metrics_collector.py index 6d7c6c8a7c..f4be930dd0 100644 --- a/src/backend/tests/unit/services/background_execution/test_metrics_collector.py +++ b/src/backend/tests/unit/services/background_execution/test_metrics_collector.py @@ -5,10 +5,10 @@ They run against the REAL test DB (the ``client`` fixture; SQLite locally, Postgres in CI) with NO mocking. ``now`` is injected so the time math is deterministic and never races the wall clock. -The "alive" definition MUST match the watchdog's lease-staleness boundary -(``JobService.is_lease_stale``): a heartbeat is FRESH while its age is within -the lease window, so ``alive_worker_count`` counts owners with -``heartbeat_at >= now - lease_window``. +The worker online/busy/idle gauges are derived from the durable +``worker_registry`` table via ``WorkerRegistryService.count_by_state``: a row is +online while ``last_heartbeat >= now - online_window`` (online_window = 3x the +worker's registry interval), and busy/idle split the online rows by state. """ from __future__ import annotations @@ -20,13 +20,13 @@ import pytest from langflow.services.background_execution.metrics import current_backend from langflow.services.background_execution.metrics_collector import ( BackgroundMetricsCollector, - alive_worker_count, count_nonterminal_jobs, duration_percentiles, oldest_queued_seconds, terminal_counts, ) from langflow.services.database.models.jobs.model import JobStatus +from langflow.services.database.models.worker_registry.model import WorkerRegistry, WorkerState from langflow.services.deps import get_telemetry_service, session_scope from langflow.services.jobs.service import JobService @@ -34,20 +34,26 @@ pytestmark = pytest.mark.usefixtures("client") async def test_count_nonterminal_jobs_excludes_terminal(): - """Only non-terminal statuses are counted, keyed by status string.""" + """Only non-terminal background statuses are counted, keyed by status string.""" service = JobService() queued_a = uuid4() queued_b = uuid4() in_progress = uuid4() completed = uuid4() + run_in_progress = uuid4() await service.create_job(job_id=queued_a, flow_id=uuid4(), user_id=uuid4()) await service.create_job(job_id=queued_b, flow_id=uuid4(), user_id=uuid4()) await service.create_job(job_id=in_progress, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(in_progress, "run_started", {}) await service.update_job_status(in_progress, JobStatus.IN_PROGRESS) + # A RUN job briefly IN_PROGRESS with NO job_events — must be excluded. + await service.create_job(job_id=run_in_progress, flow_id=uuid4(), user_id=uuid4()) + await service.update_job_status(run_in_progress, JobStatus.IN_PROGRESS) + # Terminal: must be excluded from the non-terminal aggregate. await service.create_job(job_id=completed, flow_id=uuid4(), user_id=uuid4()) await service.update_job_status(completed, JobStatus.COMPLETED, finished_timestamp=True) @@ -97,61 +103,6 @@ async def test_oldest_queued_seconds_zero_when_none_queued(): assert age == 0.0 -async def test_alive_worker_count_excludes_stale_heartbeats(): - """Distinct owners with a FRESH heartbeat (within the lease window) are alive. - - Two owners heartbeat fresh; one owner's heartbeat is older than the window. - The stale owner is excluded, matching ``is_lease_stale``. - """ - service = JobService() - lease_window = 30.0 - - fresh_a = uuid4() - fresh_b = uuid4() - stale = uuid4() - - await service.create_job(job_id=fresh_a, flow_id=uuid4(), user_id=uuid4()) - await service.update_job_status(fresh_a, JobStatus.IN_PROGRESS) - await service.heartbeat(fresh_a, owner="worker-A") - - await service.create_job(job_id=fresh_b, flow_id=uuid4(), user_id=uuid4()) - await service.update_job_status(fresh_b, JobStatus.IN_PROGRESS) - await service.heartbeat(fresh_b, owner="worker-B") - - await service.create_job(job_id=stale, flow_id=uuid4(), user_id=uuid4()) - await service.update_job_status(stale, JobStatus.IN_PROGRESS) - old = (datetime.now(timezone.utc) - timedelta(seconds=300)).isoformat() - await service.update_job_metadata(stale, {"owner": "worker-dead", "heartbeat_at": old}) - - now = datetime.now(timezone.utc) - async with session_scope() as session: - alive = await alive_worker_count(session, now, lease_window) - - assert alive == 2 - - -async def test_alive_worker_count_distinct_owner(): - """Multiple fresh heartbeats from the SAME owner count once.""" - service = JobService() - lease_window = 30.0 - - job_one = uuid4() - job_two = uuid4() - await service.create_job(job_id=job_one, flow_id=uuid4(), user_id=uuid4()) - await service.update_job_status(job_one, JobStatus.IN_PROGRESS) - await service.heartbeat(job_one, owner="worker-A") - - await service.create_job(job_id=job_two, flow_id=uuid4(), user_id=uuid4()) - await service.update_job_status(job_two, JobStatus.IN_PROGRESS) - await service.heartbeat(job_two, owner="worker-A") - - now = datetime.now(timezone.utc) - async with session_scope() as session: - alive = await alive_worker_count(session, now, lease_window) - - assert alive == 1 - - def _gauge_value(metric_name: str, labels: dict[str, str]) -> float: """Read a gauge value straight off the real OTel ObservableGaugeWrapper. @@ -177,8 +128,8 @@ async def test_collect_once_sets_gauges(): await service.create_job(job_id=queued_b, flow_id=uuid4(), user_id=uuid4()) await service.create_job(job_id=in_progress, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(in_progress, "run_started", {}) await service.update_job_status(in_progress, JobStatus.IN_PROGRESS) - await service.heartbeat(in_progress, owner="worker-A") await service.create_job(job_id=completed, flow_id=uuid4(), user_id=uuid4()) await service.update_job_status(completed, JobStatus.COMPLETED, finished_timestamp=True) @@ -190,7 +141,6 @@ async def test_collect_once_sets_gauges(): assert _gauge_value("langflow_bg_jobs", {"status": "queued", "backend": backend}) == 2 assert _gauge_value("langflow_bg_jobs", {"status": "in_progress", "backend": backend}) == 1 assert _gauge_value("langflow_bg_oldest_queued_seconds", {"backend": backend}) > 0 - assert _gauge_value("langflow_bg_alive_workers", {"backend": backend}) == 1 async def test_collect_once_zero_fills_dropped_status(): @@ -253,16 +203,31 @@ async def test_run_stop_lifecycle(): async def _seed_failed(service: JobService, *, error: dict): - """Create a job, flip it FAILED, and stamp the given error blob.""" + """Create a BACKGROUND job (with a job_events row), flip FAILED, stamp error.""" job_id = uuid4() await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) + # A job_events row is what marks this as a TRUE background job (vs a run job). + await service.append_event(job_id, "run_started", {}) await service.update_job_status(job_id, JobStatus.FAILED, finished_timestamp=True) await service.set_error(job_id, error) return job_id async def _seed_terminal(service: JobService, status: JobStatus): - """Create a job and flip it to the given terminal status.""" + """Create a BACKGROUND job (with a job_events row) and flip it terminal.""" + job_id = uuid4() + await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(job_id, "run_started", {}) + await service.update_job_status(job_id, status, finished_timestamp=True) + return job_id + + +async def _seed_run_job(service: JobService, status: JobStatus = JobStatus.COMPLETED): + """Create a Memory-Base RUN job: created directly, terminal, NO job_events. + + Mimics build.py's second workflow row — it must be excluded by the collector's + EXISTS(job_events) filter. + """ job_id = uuid4() await service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4()) await service.update_job_status(job_id, status, finished_timestamp=True) @@ -271,7 +236,7 @@ async def _seed_terminal(service: JobService, status: JobStatus): async def test_terminal_counts_splits_outcomes(): """terminal_counts returns the per-outcome split; started excludes only QUEUED.""" - from langflow.services.database.models.jobs.model import Job + from langflow.services.database.models.jobs.model import Job, JobEvent service = JobService() @@ -279,6 +244,7 @@ async def test_terminal_counts_splits_outcomes(): await service.create_job(job_id=uuid4(), flow_id=uuid4(), user_id=uuid4()) in_progress = uuid4() await service.create_job(job_id=in_progress, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(in_progress, "run_started", {}) await service.update_job_status(in_progress, JobStatus.IN_PROGRESS) # 2 COMPLETED. @@ -294,14 +260,21 @@ async def test_terminal_counts_splits_outcomes(): await _seed_terminal(service, JobStatus.TIMED_OUT) await _seed_terminal(service, JobStatus.CANCELLED) + # RUN jobs (no job_events) in several terminal states — must be excluded. + await _seed_run_job(service, JobStatus.COMPLETED) + await _seed_run_job(service, JobStatus.FAILED) + async with session_scope() as session: tc = await terminal_counts(session) # Absolute totals in a shared test DB are not isolated across tests, so - # assert against the actual current rows by status to keep this robust. + # assert against the actual current BACKGROUND rows (EXISTS(job_events)) + # by status to keep this robust and to exclude run jobs. from sqlmodel import col, func, select + has_events = select(JobEvent.id).where(col(JobEvent.job_id) == Job.job_id).exists() + async def _count(*statuses): - stmt = select(func.count()).select_from(Job).where(col(Job.status).in_(statuses)) + stmt = select(func.count()).select_from(Job).where(col(Job.status).in_(statuses)).where(has_events) return int((await session.exec(stmt)).one()) total = await _count(*list(JobStatus)) @@ -315,13 +288,38 @@ async def test_terminal_counts_splits_outcomes(): assert tc["completed"] == completed assert tc["timed_out"] == timed_out assert tc["cancelled"] == cancelled - # The split must partition all FAILED rows. + # The split must partition all FAILED background rows. assert tc["failed_worker_lost"] + tc["failed_error"] == failed # We seeded exactly two worker_lost rows; any pre-existing FAILED rows from # other tests carry a different/absent type and land in failed_error. assert tc["failed_worker_lost"] >= 2 +async def test_terminal_counts_excludes_run_jobs(): + """A COMPLETED run job (no job_events) must NOT count toward completed. + + This is the 2x-inflation guard: build.py writes a second workflow row per + flow build that goes straight to COMPLETED with no job_events. The collector + must count only the TRUE background job (EXISTS(job_events)). + """ + service = JobService() + + async with session_scope() as session: + before = (await terminal_counts(session))["completed"] + + # One real background completion (+1) and three run-job completions (+0). + await _seed_terminal(service, JobStatus.COMPLETED) + await _seed_run_job(service, JobStatus.COMPLETED) + await _seed_run_job(service, JobStatus.COMPLETED) + await _seed_run_job(service, JobStatus.COMPLETED) + + async with session_scope() as session: + after = (await terminal_counts(session))["completed"] + + # Only the single background job moved the needle; the three run jobs did not. + assert after - before == 1 + + async def test_duration_percentiles_deterministic(): """p50/p95 over jobs finished within the window match known durations.""" from langflow.services.database.models.jobs.model import Job @@ -329,12 +327,14 @@ async def test_duration_percentiles_deterministic(): service = JobService() now = datetime.now(timezone.utc) - # Seed five COMPLETED jobs with known durations 10..50s, all finished now. + # Seed five COMPLETED BACKGROUND jobs (with job_events) with known durations + # 10..50s, all finished now. durations = [10, 20, 30, 40, 50] job_ids = [] for _ in durations: jid = uuid4() await service.create_job(job_id=jid, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(jid, "run_started", {}) await service.update_job_status(jid, JobStatus.COMPLETED, finished_timestamp=True) job_ids.append(jid) @@ -376,8 +376,13 @@ async def test_duration_percentiles_excludes_jobs_outside_window(): recent = uuid4() for jid in (old, recent): await service.create_job(job_id=jid, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(jid, "run_started", {}) await service.update_job_status(jid, JobStatus.COMPLETED, finished_timestamp=True) + # A RUN job (no job_events) finished AT base with a 5000s duration — inside + # the window but must be excluded by EXISTS(job_events), so it never skews p95. + run_job = await _seed_run_job(service, JobStatus.COMPLETED) + async with session_scope() as session: old_job = await session.get(Job, old) old_job.finished_timestamp = base - timedelta(hours=1) @@ -387,11 +392,16 @@ async def test_duration_percentiles_excludes_jobs_outside_window(): recent_job.finished_timestamp = base recent_job.created_timestamp = base - timedelta(seconds=7) session.add(recent_job) + run_row = await session.get(Job, run_job) + run_row.finished_timestamp = base + run_row.created_timestamp = base - timedelta(seconds=5000) + session.add(run_row) await session.flush() # 60s window back from ``base``: the recent job (finished AT base) is in; the - # old job (finished an hour earlier) is out. Only the 7s sample remains, so - # p50 == p95 == 7 and the 999s duration never leaks in. + # old job (finished an hour earlier) is out by the SQL cutoff; the run job is + # out by EXISTS(job_events). Only the 7s sample remains, so p50 == p95 == 7 + # and neither the 999s nor the 5000s duration leaks in. async with session_scope() as session: p50, p95 = await duration_percentiles(session, base, window_seconds=60.0) @@ -439,10 +449,11 @@ async def test_collect_once_sets_counters_and_duration_gauges(): await _seed_terminal(service, JobStatus.CANCELLED), ] - # A COMPLETED job with a known 12s duration finished now so the duration - # gauges are non-zero with a tight window. + # A COMPLETED BACKGROUND job (with a job_events row) with a known 12s + # duration finished now so the duration gauges are non-zero with a tight window. dur_job = uuid4() await service.create_job(job_id=dur_job, flow_id=uuid4(), user_id=uuid4()) + await service.append_event(dur_job, "run_started", {}) await service.update_job_status(dur_job, JobStatus.COMPLETED, finished_timestamp=True) async with session_scope() as session: # Push every other seeded job's finish well outside the duration window @@ -490,3 +501,138 @@ async def test_collect_once_sets_counters_and_duration_gauges(): # wires p50/p95 from a real finished-job sample). assert _gauge_value("langflow_bg_job_duration_p50_seconds", {"backend": backend}) > 0.0 assert _gauge_value("langflow_bg_job_duration_p95_seconds", {"backend": backend}) > 0.0 + + +async def test_collect_once_sets_worker_gauges_from_registry(): + """A tick derives online/busy/idle from worker_registry; stale rows count toward none. + + online_window = 3 * registry_interval = 30s. Seed a fresh IDLE, a fresh BUSY, + and a STALE row (heartbeat older than 30s). ``now`` is injected so the freshness + boundary is deterministic and never races the wall clock. + """ + backend = current_backend() + now = datetime(2026, 6, 5, 12, 0, 0, tzinfo=timezone.utc) + suffix = uuid4().hex + + async with session_scope() as session: + session.add( + WorkerRegistry( + owner=f"fresh-idle-{suffix}", + pid=1, + host="h", + started_at=now - timedelta(minutes=5), + last_heartbeat=now, + state=WorkerState.IDLE, + current_job_id=None, + ) + ) + session.add( + WorkerRegistry( + owner=f"fresh-busy-{suffix}", + pid=2, + host="h", + started_at=now - timedelta(minutes=5), + last_heartbeat=now - timedelta(seconds=10), + state=WorkerState.BUSY, + current_job_id=uuid4(), + ) + ) + # Stale: heartbeat 60s ago, past the 30s online window — excluded everywhere, + # but within the 1h retention so prune does not delete it this tick. + session.add( + WorkerRegistry( + owner=f"stale-{suffix}", + pid=3, + host="h", + started_at=now - timedelta(minutes=5), + last_heartbeat=now - timedelta(seconds=60), + state=WorkerState.IDLE, + current_job_id=None, + ) + ) + await session.flush() + + # registry_interval=10 -> online_window 30s; retention 1h leaves the stale row. + collector = BackgroundMetricsCollector( + interval=15.0, + registry_interval=10.0, + registry_retention_s=3600.0, + ) + async with session_scope() as session: + await collector.collect_once(session, now=now) + + assert _gauge_value("langflow_bg_workers_online", {"backend": backend}) == 2 + assert _gauge_value("langflow_bg_workers_busy", {"backend": backend}) == 1 + assert _gauge_value("langflow_bg_workers_idle", {"backend": backend}) == 1 + + # The stale-but-within-retention row is still present (prune did not remove it). + async with session_scope() as session: + assert await session.get(WorkerRegistry, f"stale-{suffix}") is not None + + +async def test_collect_once_prunes_rows_past_retention(): + """A tick prunes registry rows older than retention while keeping recent ones. + + With retention=3600s, a row whose heartbeat is 5h old is deleted; one whose + heartbeat is 10s old survives. ``now`` is injected so the cutoff is exact. + """ + now = datetime(2026, 6, 5, 12, 0, 0, tzinfo=timezone.utc) + suffix = uuid4().hex + keep = f"keep-{suffix}" + drop = f"drop-{suffix}" + + async with session_scope() as session: + session.add( + WorkerRegistry( + owner=keep, + pid=1, + host="h", + started_at=now - timedelta(hours=6), + last_heartbeat=now - timedelta(seconds=10), + state=WorkerState.IDLE, + current_job_id=None, + ) + ) + session.add( + WorkerRegistry( + owner=drop, + pid=2, + host="h", + started_at=now - timedelta(hours=6), + last_heartbeat=now - timedelta(hours=5), + state=WorkerState.IDLE, + current_job_id=None, + ) + ) + await session.flush() + + collector = BackgroundMetricsCollector( + interval=15.0, + registry_interval=10.0, + registry_retention_s=3600.0, + ) + async with session_scope() as session: + await collector.collect_once(session, now=now) + + async with session_scope() as session: + assert await session.get(WorkerRegistry, keep) is not None + assert await session.get(WorkerRegistry, drop) is None + + +async def test_collect_once_no_longer_emits_alive_workers(): + """The replaced langflow_bg_alive_workers gauge is gone; the rest still emit.""" + backend = current_backend() + now = datetime(2026, 6, 5, 12, 0, 0, tzinfo=timezone.utc) + + collector = BackgroundMetricsCollector(interval=15.0, registry_interval=10.0) + async with session_scope() as session: + await collector.collect_once(session, now=now) + + ot = get_telemetry_service().ot + # The metric is no longer registered, so update_gauge would raise on it. + assert "langflow_bg_alive_workers" not in ot._metrics + assert "langflow_bg_alive_workers" not in ot._metrics_registry + + # The rest of the collector's gauges still emit on this tick. + assert "langflow_bg_workers_online" in ot._metrics + assert _gauge_value("langflow_bg_oldest_queued_seconds", {"backend": backend}) >= 0.0 diff --git a/src/backend/tests/unit/services/background_execution/test_metrics_collector_lifespan.py b/src/backend/tests/unit/services/background_execution/test_metrics_collector_lifespan.py index eecc8e1e22..ee40f40eea 100644 --- a/src/backend/tests/unit/services/background_execution/test_metrics_collector_lifespan.py +++ b/src/backend/tests/unit/services/background_execution/test_metrics_collector_lifespan.py @@ -62,8 +62,11 @@ async def test_started_when_port_bound_and_enabled(): # The loop task is live: created and not yet finished. assert collector._task is not None assert not collector._task.done() - # The lease window is wired from settings so "alive" agrees with the watchdog. - assert collector.lease_window == _settings(prometheus_enabled=True).background_lease_ttl_s + # The registry interval/retention are wired from settings so the online window + # (3x interval) and the per-tick prune match the worker's own registry config. + settings = _settings(prometheus_enabled=True) + assert collector.registry_interval == settings.background_worker_registry_interval_s + assert collector.registry_retention_s == settings.background_worker_registry_retention_s await stop_metrics_collector(app) # Stop actually ended the loop: task cleared and the underlying task finished. diff --git a/src/backend/tests/unit/services/telemetry/test_opentelemetry.py b/src/backend/tests/unit/services/telemetry/test_opentelemetry.py index 04334a0b1e..e1f6ebc59a 100644 --- a/src/backend/tests/unit/services/telemetry/test_opentelemetry.py +++ b/src/backend/tests/unit/services/telemetry/test_opentelemetry.py @@ -10,7 +10,9 @@ def ot(): BG_METRICS = { "langflow_bg_jobs": MetricType.OBSERVABLE_GAUGE, "langflow_bg_oldest_queued_seconds": MetricType.OBSERVABLE_GAUGE, - "langflow_bg_alive_workers": MetricType.OBSERVABLE_GAUGE, + "langflow_bg_workers_online": MetricType.OBSERVABLE_GAUGE, + "langflow_bg_workers_busy": MetricType.OBSERVABLE_GAUGE, + "langflow_bg_workers_idle": MetricType.OBSERVABLE_GAUGE, "langflow_bg_jobs_started_total": MetricType.OBSERVABLE_COUNTER, "langflow_bg_jobs_completed_total": MetricType.OBSERVABLE_COUNTER, "langflow_bg_jobs_failed_total": MetricType.OBSERVABLE_COUNTER, @@ -36,4 +38,4 @@ def test_set_observable_counter_sets_readable_value(ot): def test_set_observable_counter_on_non_observable_counter_raises(ot): with pytest.raises(TypeError): - ot.set_observable_counter("langflow_bg_alive_workers", 1, {"backend": "scaled"}) + ot.set_observable_counter("langflow_bg_workers_online", 1, {"backend": "scaled"}) diff --git a/src/backend/tests/unit/services/telemetry/test_prometheus_exposition.py b/src/backend/tests/unit/services/telemetry/test_prometheus_exposition.py index 6709ef8ac2..ce9ebf4662 100644 --- a/src/backend/tests/unit/services/telemetry/test_prometheus_exposition.py +++ b/src/backend/tests/unit/services/telemetry/test_prometheus_exposition.py @@ -22,11 +22,11 @@ def test_otel_metrics_reach_prometheus_exposition(): async def main(): ts = get_telemetry_service() ts.ot.set_observable_counter("langflow_bg_jobs_started_total", 1, {"backend": "scaled"}) - ts.ot.update_gauge("langflow_bg_alive_workers", 2, {"backend": "scaled"}) + ts.ot.update_gauge("langflow_bg_workers_online", 2, {"backend": "scaled"}) from prometheus_client import generate_latest out = generate_latest().decode() assert "langflow_bg_jobs_started_total" in out, "counter missing from exposition" - assert "langflow_bg_alive_workers" in out, "gauge missing from exposition" + assert "langflow_bg_workers_online" in out, "gauge missing from exposition" print("EXPOSITION_OK") asyncio.run(main())