push

Author: hassiebp

langfuse/otel/__init__.py

@@ -7,6 +7,7 @@
 import httpx
 from opentelemetry import trace
 from opentelemetry import trace as otel_trace_api
+from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
 
 from langfuse.api.client import AsyncFernLangfuse, FernLangfuse
@@ -43,15 +44,15 @@ def __init__(
         httpx_client: Optional[httpx.Client] = None,
         debug: bool = False,
         tracing_enabled: Optional[bool] = True,
-        # threads: Optional[int] = None,
-        # flush_at: Optional[int] = None,
-        # flush_interval: Optional[float] = None,
-        # max_retries: Optional[int] = None,
-        # sample_rate: Optional[float] = None,
-        # mask: Optional[MaskFunction] = None,
-        # sdk_integration: Optional[str] = "default",
+        flush_at: Optional[int] = None,
+        flush_interval: Optional[float] = None,
         environment: Optional[str] = None,
         release: Optional[str] = None,
+        # sample_rate: Optional[float] = None, # TODO: Implement sampling
+        # mask: Optional[MaskFunction] = None, # TODO: implement masking
+        # sdk_integration: Optional[str] = "default", -> TO BE DEPRECATED
+        # threads: Optional[int] = None, -> TO BE DEPRECATED
+        # max_retries: Optional[int] = None, -> TO BE DEPRECATED
     ):
         debug = debug if debug else (os.getenv(LANGFUSE_DEBUG, "False") == "True")
 
@@ -115,6 +116,8 @@ def __init__(
                 timeout=timeout,
                 environment=environment,
                 release=release,
+                flush_at=flush_at,
+                flush_interval=flush_interval,
             ).tracer
             if self.tracing_enabled
             else otel_trace_api.NoOpTracer()
@@ -271,7 +274,7 @@ def _create_span_with_parent_context(
             ) as span:
                 yield span
 
-    @staticmethod
+    @staticmethod  # TODO: reconsider marking methods as static as changing object method later is breaking change
     def get_current_span():
         return otel_trace_api.get_current_span()
 
@@ -496,8 +499,16 @@ def update_finished_span():
 
     @staticmethod
     def flush():
-        pass
+        tracer_provider = cast(TracerProvider, otel_trace_api.get_tracer_provider())
+        if isinstance(tracer_provider, otel_trace_api.ProxyTracerProvider):
+            return
+
+        tracer_provider.force_flush()
 
     @staticmethod
     def shutdown():
-        pass
+        tracer_provider = cast(TracerProvider, otel_trace_api.get_tracer_provider())
+        if isinstance(tracer_provider, otel_trace_api.ProxyTracerProvider):
+            return
+
+        tracer_provider.force_flush()

langfuse/otel/_span_processor/__init__.py

@@ -19,6 +19,8 @@ def __init__(
         secret_key: str,
         host: str,
         timeout: Optional[int] = None,
+        flush_at: Optional[int] = None,
+        flush_interval: Optional[float] = None,
     ):
         self.public_key = public_key
 
@@ -36,7 +38,12 @@ def __init__(
             timeout=timeout,
         )
 
-        super().__init__(span_exporter=langfuse_span_exporter)
+        super().__init__(
+            span_exporter=langfuse_span_exporter,
+            export_timeout_millis=timeout * 1_000 if timeout else None,
+            max_export_batch_size=flush_at,
+            schedule_delay_millis=flush_interval,
+        )
 
     def on_end(self, span: ReadableSpan) -> None:
         # Only export spans that belong to the scoped project

langfuse/otel/tracer.py

@@ -21,7 +21,7 @@
 class LangfuseTracer:
     """Singleton that provides access to the OTEL tracer."""
 
-    _instances: Dict[str, Optional["LangfuseTracer"]] = {}
+    _instances: Dict[str, "LangfuseTracer"] = {}
     _lock = threading.Lock()
 
     def __new__(
@@ -30,10 +30,12 @@ def __new__(
         public_key: str,
         secret_key: str,
         host: str,
-        timeout: Optional[int] = None,
         environment: Optional[str] = None,
         release: Optional[str] = None,
-    ):
+        timeout: Optional[int] = None,
+        flush_at: Optional[int] = None,
+        flush_interval: Optional[float] = None,
+    ) -> "LangfuseTracer":
         if public_key in cls._instances:
             return cls._instances[public_key]
 
@@ -48,6 +50,8 @@ def __new__(
                     timeout=timeout,
                     environment=environment,
                     release=release,
+                    flush_at=flush_at,
+                    flush_interval=flush_interval,
                 )
 
                 cls._instances[public_key] = instance
@@ -60,9 +64,11 @@ def _initialize_instance(
         public_key: str,
         secret_key: str,
         host: str,
-        timeout: Optional[int] = None,
         environment: Optional[str] = None,
         release: Optional[str] = None,
+        timeout: Optional[int] = None,
+        flush_at: Optional[int] = None,
+        flush_interval: Optional[float] = None,
     ):
         tracer_provider = _init_tracer_provider(
             environment=environment, release=release
@@ -73,6 +79,8 @@ def _initialize_instance(
             secret_key=secret_key,
             host=host,
             timeout=timeout,
+            flush_at=flush_at,
+            flush_interval=flush_interval,
         )
         tracer_provider.add_span_processor(langfuse_processor)