Merge branch 'release-1.11.0' into fix/message-history-order-direction

Author: erichare

src/backend/base/langflow/services/job_queue/service.py

@@ -129,6 +129,9 @@ class JobQueueService(Service):
 
     name = "job_queue_service"
 
+    # Backend label used on OpenTelemetry metrics. Subclasses override.
+    _backend_label: str = "memory"
+
     def __init__(self) -> None:
         """Initialize the JobQueueService.
 
@@ -142,6 +145,44 @@ def __init__(self) -> None:
         self._closed = False
         self.ready = False
         self.CLEANUP_GRACE_PERIOD = 300  # 5 minutes before cleaning up marked tasks
+        # Cached OpenTelemetry handle (resolved lazily on first emit).
+        # _otel_resolved is True after the first lookup attempt, including failures,
+        # so we don't retry on every bump. _otel remains None when lookup fails.
+        self._otel: Any = None
+        self._otel_resolved: bool = False
+
+    def _get_otel(self) -> Any:
+        """Best-effort lookup of the OpenTelemetry singleton. Cached after first call.
+
+        Returns the ``ot`` handle, or ``None`` if telemetry is unavailable. Never raises —
+        instrumentation must not break the queue.
+        """
+        if self._otel_resolved:
+            return self._otel
+        self._otel_resolved = True
+        try:
+            from langflow.services.deps import get_telemetry_service
+
+            self._otel = get_telemetry_service().ot
+        except Exception:  # noqa: BLE001 - telemetry must not crash the queue
+            self._otel = None
+        return self._otel
+
+    def _emit_otel_counter(self, metric_name: str, labels: dict[str, str], value: float = 1.0) -> None:
+        """Best-effort OpenTelemetry counter bump. Silent on failure."""
+        ot = self._get_otel()
+        if ot is None:
+            return
+        with contextlib.suppress(Exception):
+            ot.increment_counter(metric_name, labels, value)
+
+    def _emit_otel_up_down(self, metric_name: str, value: float, labels: dict[str, str]) -> None:
+        """Best-effort OpenTelemetry up-down counter delta. Silent on failure."""
+        ot = self._get_otel()
+        if ot is None:
+            return
+        with contextlib.suppress(Exception):
+            ot.up_down_counter(metric_name, value, labels)
 
     def is_started(self) -> bool:
         """Check if the JobQueueService has started.
@@ -241,6 +282,7 @@ def create_queue(self, job_id: str) -> tuple[asyncio.Queue, EventManager]:
 
         # Register the queue without an active task.
         self._queues[job_id] = (main_queue, event_manager, None, None)
+        self._emit_otel_up_down("langflow_job_queue_active_jobs", 1, {"backend": self._backend_label})
         logger.debug(f"Queue and event manager successfully created for job_id {job_id}")
         return main_queue, event_manager
 
@@ -436,7 +478,8 @@ async def cleanup_job(self, job_id: str) -> None:
 
         await logger.adebug(f"Removed {items_cleared} items from queue for job_id {job_id}")
         # Remove the job entry from the registry
-        self._queues.pop(job_id, None)
+        if self._queues.pop(job_id, None) is not None:
+            self._emit_otel_up_down("langflow_job_queue_active_jobs", -1, {"backend": self._backend_label})
         self._job_owners.pop(job_id, None)
         self._public_jobs.discard(job_id)
         await logger.adebug(f"Cleanup successful for job_id {job_id}: resources have been released.")
@@ -810,6 +853,8 @@ class RedisJobQueueService(JobQueueService):
     ACTIVITY_PREFIX = _ACTIVITY_PREFIX
     PUBLIC_JOB_PREFIX = _PUBLIC_JOB_PREFIX
 
+    _backend_label: str = "redis"
+
     def __init__(
         self,
         host: str = "localhost",
@@ -880,6 +925,16 @@ def cross_worker_cancel_enabled(self) -> bool:
         """
         return self._cancel_channel_enabled
 
+    def _bump_cancel_stat(self, key: str, value: int = 1) -> None:
+        """Increment a cancel_stat counter and emit the matching OTel counter event.
+
+        Single point of mutation for ``self._cancel_stats`` so ``/monitor/job_queue``
+        and Prometheus stay in lockstep. The OTel emit is best-effort and never
+        raises (see :meth:`_emit_otel_counter` on the base class).
+        """
+        self._cancel_stats[key] += value
+        self._emit_otel_counter("langflow_job_queue_cancel_events_total", {"event_type": key}, float(value))
+
     def _stream_key(self, job_id: str) -> str:
         return f"{self.STREAM_PREFIX}{job_id}"
 
@@ -1208,7 +1263,7 @@ async def touch_activity(self, job_id: str) -> None:
         except asyncio.CancelledError:
             raise
         except Exception as exc:  # noqa: BLE001
-            self._cancel_stats["activity_touch_errors"] += 1
+            self._bump_cancel_stat("activity_touch_errors")
             await logger.adebug(f"touch_activity SET failed for {job_id}: {exc}")
 
     def _owner_refresh_interval_s(self) -> float:
@@ -1248,7 +1303,7 @@ async def _check_pending_cancel_marker(self, job_id: str) -> None:
         try:
             if await self._client.exists(marker_key):
                 await self._client.delete(marker_key)
-                self._cancel_stats["marker_hit"] += 1
+                self._bump_cancel_stat("marker_hit")
                 await self._handle_cancel(job_id, source="marker")
         except Exception as exc:  # noqa: BLE001
             await logger.awarning(f"Pending cancel marker check failed for {job_id}: {exc}")
@@ -1306,7 +1361,7 @@ async def _run_polling_watchdog(self) -> None:
                 try:
                     raw = await self._client.get(self._activity_key(job_id))
                 except Exception as exc:  # noqa: BLE001
-                    self._cancel_stats["activity_get_errors"] += 1
+                    self._bump_cancel_stat("activity_get_errors")
                     await logger.adebug(f"polling watchdog: GET failed for {job_id}: {exc}")
                     continue
                 # Default to "infinitely stale" so a missed elif below cannot
@@ -1324,7 +1379,7 @@ async def _run_polling_watchdog(self) -> None:
                     try:
                         last = float(raw.decode() if isinstance(raw, bytes) else raw)
                     except (ValueError, TypeError) as exc:
-                        self._cancel_stats["activity_parse_errors"] += 1
+                        self._bump_cancel_stat("activity_parse_errors")
                         await logger.awarning(
                             f"polling watchdog: ignoring malformed activity value for {job_id}: {exc}"
                         )
@@ -1335,7 +1390,7 @@ async def _run_polling_watchdog(self) -> None:
                 # Stale → cancel this job. Local cancel on owned jobs skips the
                 # pubsub round-trip and stays correct even during a dispatcher
                 # reconnect window.
-                self._cancel_stats["polling_watchdog_kills"] += 1
+                self._bump_cancel_stat("polling_watchdog_kills")
                 await logger.ainfo(f"polling watchdog: reclaiming abandoned job {job_id} (age={age:.1f}s)")
                 await self._handle_cancel(job_id, source="watchdog")
                 with contextlib.suppress(Exception):
@@ -1383,7 +1438,7 @@ async def _run_cancel_dispatcher(self) -> None:
                     job_id = channel_str[len(self.CANCEL_CHANNEL_PREFIX) :]
                     await self._handle_cancel(job_id, source="pubsub")
                 # listen() returned cleanly — treat as a soft disconnect and reconnect.
-                self._cancel_stats["dispatcher_reconnects"] += 1
+                self._bump_cancel_stat("dispatcher_reconnects")
                 await logger.awarning(f"Cancel dispatcher pubsub.listen() ended; reconnecting in {backoff:.1f}s")
             except asyncio.CancelledError:
                 with contextlib.suppress(Exception):
@@ -1393,15 +1448,15 @@ async def _run_cancel_dispatcher(self) -> None:
             except (ConnectionError, TimeoutError, OSError) as exc:
                 # Expected transient failure: Redis dropped the pubsub, network
                 # blip, broker restart. Reconnect quietly via the backoff loop.
-                self._cancel_stats["dispatcher_reconnects"] += 1
+                self._bump_cancel_stat("dispatcher_reconnects")
                 await logger.awarning(f"Cancel dispatcher disconnect (retrying in {backoff:.1f}s): {exc!r}")
             except Exception as exc:  # noqa: BLE001
                 # Unexpected exception — likely a bug in dispatch logic, NOT a
                 # Redis problem. Surface at error level with traceback so it
                 # reaches Sentry / log aggregation, then still reconnect so a
                 # one-off bug doesn't kill cross-worker cancel permanently.
-                self._cancel_stats["dispatcher_reconnects"] += 1
-                self._cancel_stats["dispatcher_internal_errors"] += 1
+                self._bump_cancel_stat("dispatcher_reconnects")
+                self._bump_cancel_stat("dispatcher_internal_errors")
                 await logger.aerror(
                     f"Cancel dispatcher internal error (retrying in {backoff:.1f}s): {exc!r}",
                     exc_info=True,
@@ -1418,7 +1473,7 @@ async def _run_cancel_dispatcher(self) -> None:
 
     async def _on_cancel_dispatcher_connection_reconnect(self, _connection: Any) -> None:
         """Record redis-py reconnects that happen inside an active PubSub."""
-        self._cancel_stats["dispatcher_reconnects"] += 1
+        self._bump_cancel_stat("dispatcher_reconnects")
         with contextlib.suppress(Exception):
             await logger.awarning("Cancel dispatcher pubsub connection reconnected transparently")
 
@@ -1458,10 +1513,10 @@ async def _apply_cancel(self, job_id: str, *, source: str, wait_for_cleanup: boo
         """
         entry = self._queues.get(job_id)
         if entry is None:
-            self._cancel_stats["dispatched_foreign"] += 1
+            self._bump_cancel_stat("dispatched_foreign")
             await logger.adebug(f"Cancel for {job_id} ignored on this worker (not owner); source={source}")
             return
-        self._cancel_stats["dispatched_owned"] += 1
+        self._bump_cancel_stat("dispatched_owned")
         await logger.ainfo(f"Cancel applied to {job_id} (source={source})")
         main_queue, _, task, _ = entry
         if task is not None and not task.done():
@@ -1569,9 +1624,9 @@ async def signal_cancel(self, job_id: str) -> int:
             await self._client.set(self._cancel_marker_key(job_id), "1", ex=self._cancel_marker_ttl)
             receivers = int(await self._client.publish(self._cancel_channel(job_id), "1"))
         except Exception:
-            self._cancel_stats["publish_errors"] += 1
+            self._bump_cancel_stat("publish_errors")
             raise
-        self._cancel_stats["published"] += 1
+        self._bump_cancel_stat("published")
         await logger.ainfo(f"signal_cancel: job_id={job_id} receivers={receivers}")
         return receivers
 

src/backend/base/langflow/services/telemetry/opentelemetry.py

@@ -140,6 +140,25 @@ def _register_metric(self) -> None:
             metric_type=MetricType.COUNTER,
             labels={"flow_id": mandatory_label},
         )
+        self._add_metric(
+            name="langflow_job_queue_cancel_events_total",
+            description=(
+                "Job queue cancel-channel and watchdog events. event_type is one of: "
+                "published, marker_hit, dispatched_owned, dispatched_foreign, publish_errors, "
+                "dispatcher_reconnects, dispatcher_internal_errors, polling_watchdog_kills, "
+                "activity_touch_errors, activity_get_errors, activity_parse_errors."
+            ),
+            unit="",
+            metric_type=MetricType.COUNTER,
+            labels={"event_type": mandatory_label},
+        )
+        self._add_metric(
+            name="langflow_job_queue_active_jobs",
+            description="Active jobs tracked by the job queue on this worker.",
+            unit="",
+            metric_type=MetricType.UP_DOWN_COUNTER,
+            labels={"backend": mandatory_label},
+        )
 
     def __init__(self, *, prometheus_enabled: bool = True):
         # Only initialize once
@@ -154,8 +173,10 @@ def __init__(self, *, prometheus_enabled: bool = True):
             # Get existing meter provider if any
             existing_provider = metrics.get_meter_provider()
 
-            # Check if FastAPI instrumentation is already set up
-            if hasattr(existing_provider, "get_meter") and existing_provider.get_meter("http.server"):
+            # Reuse a concrete SDK provider installed by another integration. The
+            # default API proxy also returns meters, but it has no readers and must
+            # be replaced so Prometheus can collect Langflow metrics.
+            if isinstance(existing_provider, MeterProvider):
                 self._meter_provider = existing_provider
             else:
                 resource = Resource.create({"service.name": "langflow"})

src/backend/tests/unit/components/processing/test_operations_component.py

@@ -9,13 +9,16 @@
 
 import pandas as pd
 import pytest
+from lfx.components.processing.data_operations import DataOperationsComponent
+from lfx.components.processing.dataframe_operations import DataFrameOperationsComponent
 from lfx.components.processing.operations import (
     JSON_OPERATIONS,
     OPERATIONS_BY_TYPE,
     TABLE_OPERATIONS,
     TEXT_OPERATIONS,
     OperationsComponent,
 )
+from lfx.components.processing.text_operations import TextOperations
 from lfx.schema import Data
 from lfx.schema.dataframe import DataFrame
 from lfx.schema.dotdict import dotdict
@@ -44,6 +47,14 @@ def file_names_mapping(self):
         return []
 
 
+@pytest.mark.parametrize(
+    "legacy_component",
+    [DataOperationsComponent, DataFrameOperationsComponent, TextOperations],
+)
+def test_legacy_operations_components_point_to_unified_replacement(legacy_component):
+    assert legacy_component.replacement == ["processing.Operations"]
+
+
 class TestOperationCatalog:
     """The unified picker exposes every operation across all three data types.
 

src/backend/tests/unit/test_redis_job_queue_service.py

@@ -2904,3 +2904,119 @@ async def test_initialize_services_fails_fast_when_redis_queue_unreachable(monke
     finally:
         manager.factories.clear()
         manager.services.clear()
+
+
+# ---------------------------------------------------------------------------
+# OpenTelemetry instrumentation
+# ---------------------------------------------------------------------------
+
+
+class _OtelRecorder:
+    """Capture OTel emissions so tests can assert what the queue exported.
+
+    Substituted for the real OT singleton via ``service._otel``. Mirrors the
+    surface area used by ``_emit_otel_counter`` and ``_emit_otel_up_down``.
+    """
+
+    def __init__(self) -> None:
+        self.counters: list[tuple[str, dict[str, str], float]] = []
+        self.up_downs: list[tuple[str, float, dict[str, str]]] = []
+
+    def increment_counter(self, name: str, labels: dict[str, str], value: float = 1.0) -> None:
+        self.counters.append((name, dict(labels), value))
+
+    def up_down_counter(self, name: str, value: float, labels: dict[str, str]) -> None:
+        self.up_downs.append((name, value, dict(labels)))
+
+
+def _attach_recorder(service: Any) -> _OtelRecorder:
+    """Bypass the lazy OTel resolver and inject a recorder."""
+    recorder = _OtelRecorder()
+    service._otel = recorder
+    service._otel_resolved = True
+    return recorder
+
+
+@pytest.mark.asyncio
+async def test_bump_cancel_stat_updates_dict_and_otel_counter():
+    service, _client = await _make_service(cancel_channel_enabled=False)
+    recorder = _attach_recorder(service)
+    try:
+        service._bump_cancel_stat("published")
+        service._bump_cancel_stat("marker_hit", value=2)
+
+        # Dict mirror still works for /monitor/job_queue.
+        assert service._cancel_stats["published"] == 1
+        assert service._cancel_stats["marker_hit"] == 2
+
+        # OTel counter received both bumps with event_type label.
+        assert (
+            "langflow_job_queue_cancel_events_total",
+            {"event_type": "published"},
+            1.0,
+        ) in recorder.counters
+        assert (
+            "langflow_job_queue_cancel_events_total",
+            {"event_type": "marker_hit"},
+            2.0,
+        ) in recorder.counters
+    finally:
+        await _stop_service(service)
+
+
+@pytest.mark.asyncio
+async def test_create_and_cleanup_move_active_jobs_up_down_counter():
+    service, _client = await _make_service(cancel_channel_enabled=False)
+    recorder = _attach_recorder(service)
+    try:
+        job_id = uuid.uuid4().hex
+        service.create_queue(job_id)
+        await service.cleanup_job(job_id)
+
+        deltas = [
+            (value, labels) for name, value, labels in recorder.up_downs if name == "langflow_job_queue_active_jobs"
+        ]
+        assert (1, {"backend": "redis"}) in deltas
+        assert (-1, {"backend": "redis"}) in deltas
+    finally:
+        await _stop_service(service)
+
+
+@pytest.mark.asyncio
+async def test_otel_emit_is_silent_when_telemetry_unavailable():
+    """A broken OT handle must never propagate out of the emit helpers."""
+    service, _client = await _make_service(cancel_channel_enabled=False)
+    try:
+        explosion = "telemetry exploded"
+
+        class _BrokenOt:
+            def increment_counter(self, *_a, **_kw):
+                raise RuntimeError(explosion)
+
+            def up_down_counter(self, *_a, **_kw):
+                raise RuntimeError(explosion)
+
+        service._otel = _BrokenOt()
+        service._otel_resolved = True
+
+        # These must not raise even though OT itself does.
+        service._bump_cancel_stat("published")
+        service._emit_otel_up_down("langflow_job_queue_active_jobs", 1, {"backend": "redis"})
+        # Dict mirror still updated despite OT failure.
+        assert service._cancel_stats["published"] == 1
+    finally:
+        await _stop_service(service)
+
+
+@pytest.mark.asyncio
+async def test_all_cancel_stat_keys_route_through_helper():
+    """Every key initialized in _cancel_stats must be a valid argument to _bump_cancel_stat."""
+    service, _client = await _make_service(cancel_channel_enabled=False)
+    recorder = _attach_recorder(service)
+    try:
+        for key in list(service._cancel_stats):
+            service._bump_cancel_stat(key)
+        emitted_event_types = {labels["event_type"] for _, labels, _ in recorder.counters}
+        assert emitted_event_types == set(service._cancel_stats.keys())
+    finally:
+        await _stop_service(service)