mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-25 16:09:31 +08:00
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.
This commit is contained in:
ogabrielluiz
@ -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
|
||||
|
||||
@ -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},
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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"})
|
||||
|
||||
@ -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())
|
||||
|
||||
Reference in New Issue
Block a user