feat: opentelemetry and prometheus (#2543)

* opentelemetry and prometheus

* set env override in create_app

* add prometheus_client to pyproject

* update top level poetry.lock

src/backend/base/langflow/main.py

@@ -1,3 +1,4 @@
+import os
 import asyncio
 import warnings
 from contextlib import asynccontextmanager
@@ -14,6 +15,7 @@
 from pydantic import PydanticDeprecatedSince20
 from rich import print as rprint
 from starlette.middleware.base import BaseHTTPMiddleware
+from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 
 from langflow.api import router, health_check_router
 from langflow.initial_setup.setup import (
@@ -137,9 +139,27 @@ async def flatten_query_string_lists(request: Request, call_next):
 
         return await call_next(request)
 
+    settings = get_settings_service().settings
+    if prome_port_str := os.environ.get("LANGFLOW_PROMETHEUS_PORT"):
+        # set here for create_app() entry point
+        prome_port = int(prome_port_str)
+        if prome_port > 0 or prome_port < 65535:
+            rprint(f"[bold green]Starting Prometheus server on port {prome_port}...[/bold green]")
+            settings.prometheus_enabled = True
+            settings.prometheus_port = prome_port
+        else:
+            raise ValueError(f"Invalid port number {prome_port_str}")
+
+    if settings.prometheus_enabled:
+        from prometheus_client import start_http_server
+
+        start_http_server(settings.prometheus_port)
+
     app.include_router(router)
     app.include_router(health_check_router)
 
+    FastAPIInstrumentor.instrument_app(app)
+
     return app
 
 

src/backend/base/langflow/services/settings/base.py

@@ -77,6 +77,11 @@ class Settings(BaseSettings):
     variable_store: str = "db"
     """The store can be 'db' or 'kubernetes'."""
 
+    prometheus_enabled: bool = False
+    """If set to True, Langflow will expose Prometheus metrics."""
+    prometheus_port: int = 9090
+    """The port on which Langflow will expose Prometheus metrics. 9090 is the default port."""
+
     remove_api_keys: bool = False
     components_path: List[str] = []
     langchain_cache: str = "InMemoryCache"

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

@@ -0,0 +1,30 @@
+from opentelemetry import metrics
+from opentelemetry.exporter.prometheus import PrometheusMetricReader
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.resources import Resource
+
+
+class OpenTelemetry:
+    def __init__(self, prometheus_enabled: bool = True):
+        resource = Resource.create({"service.name": "langflow"})
+        meter_provider = MeterProvider(resource=resource)
+        self.prometheus_enabled = prometheus_enabled
+        if prometheus_enabled:
+            reader = PrometheusMetricReader()
+            meter_provider = MeterProvider(resource=resource, metric_readers=[reader])
+
+        metrics.set_meter_provider(meter_provider)
+        self.meter = meter_provider.get_meter("langflow")
+
+        self._register_metrics()
+
+    def _register_metrics(self):
+        pass
+        """
+        metrics can be registered in this function
+        self.counter = self.meter.create_counter(
+            name = "requests",
+            unit = "bytes",
+            description="The number of requests",
+        )
+        """

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

@@ -8,6 +8,7 @@
 import httpx
 from loguru import logger
 from pydantic import BaseModel
+from langflow.services.telemetry.opentelemetry import OpenTelemetry
 
 from langflow.services.base import Service
 from langflow.services.telemetry.schema import (
@@ -35,6 +36,8 @@ def __init__(self, settings_service: "SettingsService"):
         self.running = False
         self.package = get_version_info()["package"]
 
+        self.ot = OpenTelemetry(prometheus_enabled=settings_service.settings.prometheus_enabled)
+
         # Check for do-not-track settings
         self.do_not_track = (
             os.getenv("DO_NOT_TRACK", "False").lower() == "true" or settings_service.settings.do_not_track