push

Author: hassiebp

langfuse/otel/__init__.py

@@ -115,7 +115,6 @@ def __init__(
                 timeout=timeout,
                 environment=environment,
                 release=release,
-                debug=debug,
             ).tracer
             if self.tracing_enabled
             else otel_trace_api.NoOpTracer()

langfuse/otel/_span_processor/__init__.py

@@ -0,0 +1,70 @@
+import base64
+from typing import Optional
+
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.trace import ReadableSpan
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
+from langfuse.otel._logger import logger
+from langfuse.otel._utils import span_formatter
+from langfuse.otel.constants import LANGFUSE_TRACER_NAME
+from langfuse.version import __version__ as langfuse_version
+
+
+class LangfuseSpanProcessor(BatchSpanProcessor):
+    def __init__(
+        self,
+        *,
+        public_key: str,
+        secret_key: str,
+        host: str,
+        timeout: Optional[int] = None,
+    ):
+        self.public_key = public_key
+
+        langfuse_span_exporter = OTLPSpanExporter(
+            endpoint=f"{host}/api/public/otel/v1/traces",
+            headers={
+                "Authorization": "Basic "
+                + base64.b64encode(f"{public_key}:{secret_key}".encode("utf-8")).decode(
+                    "ascii"
+                ),
+                "x_langfuse_sdk_name": "python",
+                "x_langfuse_sdk_version": langfuse_version,
+                "x_langfuse_public_key": public_key,
+            },
+            timeout=timeout,
+        )
+
+        super().__init__(span_exporter=langfuse_span_exporter)
+
+    def on_end(self, span: ReadableSpan) -> None:
+        # Only export spans that belong to the scoped project
+        # This is important to not send spans to wrong project in multi-project setups
+        if self._is_langfuse_span(span) and not self._is_langfuse_project_span(span):
+            logger.debug(
+                f"Skipping span from different project (current processor is for project '{self.public_key}'): {span_formatter(span)}"
+            )
+            return
+
+        logger.debug(f"Processing span:\n{span_formatter(span)}")
+
+        super().on_end(span)
+
+    @staticmethod
+    def _is_langfuse_span(span: ReadableSpan) -> bool:
+        return (
+            span.instrumentation_scope is not None
+            and LANGFUSE_TRACER_NAME in span.instrumentation_scope.name
+        )
+
+    def _is_langfuse_project_span(self, span: ReadableSpan) -> bool:
+        if not LangfuseSpanProcessor._is_langfuse_span(span):
+            return False
+
+        if span.instrumentation_scope is not None:
+            public_key_in_span = span.instrumentation_scope.name.split(":")[-1]
+
+            return public_key_in_span == self.public_key
+
+        return False

langfuse/otel/tracer.py

@@ -1,20 +1,13 @@
-import base64
 import os
 import threading
-from typing import Optional, cast
+from typing import Dict, Optional, cast
 
 from opentelemetry import trace as otel_trace_api
-from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import (
-    BatchSpanProcessor,
-    ConsoleSpanExporter,
-    SimpleSpanProcessor,
-)
 
 from langfuse.environment import get_common_release_envs
-from langfuse.otel._utils import span_formatter
+from langfuse.otel._span_processor import LangfuseSpanProcessor
 from langfuse.otel.attributes import LangfuseSpanAttributes
 from langfuse.otel.constants import LANGFUSE_TRACER_NAME
 from langfuse.otel.environment_variables import (
@@ -28,7 +21,7 @@
 class LangfuseTracer:
     """Singleton that provides access to the OTEL tracer."""
 
-    _instance: Optional["LangfuseTracer"] = None
+    _instances: Dict[str, Optional["LangfuseTracer"]] = {}
     _lock = threading.Lock()
 
     def __new__(
@@ -40,74 +33,59 @@ def __new__(
         timeout: Optional[int] = None,
         environment: Optional[str] = None,
         release: Optional[str] = None,
-        debug: Optional[bool] = False,
     ):
-        if cls._instance:
-            return cls._instance
+        if public_key in cls._instances:
+            return cls._instances[public_key]
 
         with cls._lock:
-            if not cls._instance:
-                cls._instance = super(LangfuseTracer, cls).__new__(cls)
-
-                cls._instance._otel_tracer = None
-                cls._instance._initialize(
+            if public_key not in cls._instances:
+                instance = super(LangfuseTracer, cls).__new__(cls)
+                instance._otel_tracer = None
+                instance._initialize_instance(
                     public_key=public_key,
                     secret_key=secret_key,
                     host=host,
                     timeout=timeout,
                     environment=environment,
                     release=release,
-                    debug=debug,
                 )
 
-            return cls._instance
+                cls._instances[public_key] = instance
+
+            return cls._instances[public_key]
 
-    def _initialize(
+    def _initialize_instance(
         self,
         *,
         public_key: str,
         secret_key: str,
         host: str,
+        timeout: Optional[int] = None,
         environment: Optional[str] = None,
         release: Optional[str] = None,
-        timeout: Optional[int] = None,
-        debug: Optional[bool] = False,
     ):
         tracer_provider = _init_tracer_provider(
             environment=environment, release=release
         )
 
-        if debug:
-            console_span_exporter = ConsoleSpanExporter(formatter=span_formatter)
-            console_span_processor = SimpleSpanProcessor(
-                span_exporter=console_span_exporter
-            )
-            tracer_provider.add_span_processor(console_span_processor)
-
-        langfuse_exporter = OTLPSpanExporter(
-            endpoint=f"{host}/api/public/otel/v1/traces",
-            headers={
-                "Authorization": "Basic "
-                + base64.b64encode(f"{public_key}:{secret_key}".encode("utf-8")).decode(
-                    "ascii"
-                ),
-                "x_langfuse_sdk_name": "python",
-                "x_langfuse_sdk_version": langfuse_version,
-                "x_langfuse_public_key": public_key,
-            },
+        langfuse_processor = LangfuseSpanProcessor(
+            public_key=public_key,
+            secret_key=secret_key,
+            host=host,
             timeout=timeout,
         )
-        langfuse_processor = BatchSpanProcessor(span_exporter=langfuse_exporter)
         tracer_provider.add_span_processor(langfuse_processor)
 
-        self.name = LANGFUSE_TRACER_NAME
+        tracer_provider = otel_trace_api.get_tracer_provider()
+        self.name = f"{LANGFUSE_TRACER_NAME}:{public_key}"
         self._otel_tracer = tracer_provider.get_tracer(self.name, langfuse_version)
 
     @property
     def tracer(self):
         return self._otel_tracer
 
-    def get_current_span(self):
+    @staticmethod
+    def get_current_span():
         return otel_trace_api.get_current_span()