Merge pull request #5 from mdrideout/2025_10_11_dependency-updates

Add telemetry flush support and update dependencies

Author: mdrideout

examples/base/src/base/main.py

@@ -12,7 +12,7 @@ async def main():
     load_dotenv()
 
     # Setup OpenTelemetry before anything else happens
-    init_otel(service_name="Junjo Base Example")
+    exporter = init_otel(service_name="Junjo Base Example")
 
     # Subscribe to state changes
     def on_state_change(new_state: SampleWorkflowState):
@@ -27,6 +27,11 @@ def on_state_change(new_state: SampleWorkflowState):
     await unsubscribe()
 
     print("Done executing the base example workflow.")
+
+    # Flush telemetry before exit
+    if exporter is not None:
+        print("Flushing telemetry...")
+        exporter.flush()
     return
 
 if __name__ == "__main__":

examples/base/src/base/otel_config.py

@@ -8,15 +8,15 @@
 from opentelemetry.sdk.trace import TracerProvider
 
 
-def init_otel(service_name: str):
+def init_otel(service_name: str) -> JunjoServerOtelExporter | None:
     """Configure OpenTelemetry for this application."""
 
     # Load the JUNJO_SERVER_API_KEY from the environment variable
     JUNJO_SERVER_API_KEY = os.getenv("JUNJO_SERVER_API_KEY")
     if JUNJO_SERVER_API_KEY is None:
         print("JUNJO_SERVER_API_KEY environment variable is not set. "
                          "Generate a new API key in the Junjo Server UI.")
-        return
+        return None
 
     # Configure OpenTelemetry for this application
     # Create the OpenTelemetry Resource to identify this service
@@ -51,3 +51,5 @@ def init_otel(service_name: str):
     # Instrument OpenInference Libraries
     # Google genai
     GoogleGenAIInstrumentor().instrument(tracer_provider=tracer_provider)
+
+    return junjo_server_exporter

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "junjo"
-version = "0.60.2"
+version = "0.61.0"
 description = "A graph workflow execution library for building agentic AI workflows."
 readme = "README.md"
 requires-python = ">=3.11"
@@ -21,11 +21,11 @@ classifiers = [
 ]
 
 dependencies = [
-    "grpcio>=1.70.0",
+    "grpcio>=1.75.1",
     "jsonpatch>=1.33",
     "nanoid>=2.0.0",
-    "opentelemetry-exporter-otlp-proto-grpc>=1.30.0",
-    "opentelemetry-sdk>=1.30.0",
+    "opentelemetry-exporter-otlp-proto-grpc>=1.37.0",
+    "opentelemetry-sdk>=1.37.0",
     "pydantic>=2.10.6",
 ]
 

src/junjo/telemetry/junjo_server_otel_exporter.py

@@ -49,7 +49,10 @@ def __init__(
 
         # Set OTLP Span Exporter for Junjo Server
         oltp_exporter = OTLPSpanExporter(
-            endpoint=self._endpoint, insecure=self._insecure, headers=exporter_headers
+            endpoint=self._endpoint,
+            insecure=self._insecure,
+            headers=exporter_headers,
+            timeout=120
         )
         self._span_processor = BatchSpanProcessor(oltp_exporter)
 
@@ -69,3 +72,37 @@ def span_processor(self):
     @property
     def metric_reader(self):
         return self._metric_reader
+
+    def flush(self, timeout_millis: float = 120000) -> bool:
+        """
+        Flush all pending telemetry manually.
+
+        This method blocks until all telemetry is exported or the timeout is reached.
+        It leverages the existing retry/timeout logic in the underlying gRPC exporters.
+        It can be used to force a flush of all pending telemetry before the application exits.
+
+        Args:
+            timeout_millis: Maximum time to wait for flush in milliseconds.
+                           Defaults to 120000ms (120 seconds) to match the
+                           exporter timeout and allow for retries.
+
+        Returns:
+            True if all telemetry was flushed successfully, False otherwise.
+        """
+        success = True
+
+        # Flush span processor
+        try:
+            if not self._span_processor.force_flush(int(timeout_millis)):
+                success = False
+        except Exception:
+            success = False
+
+        # Flush metric reader
+        try:
+            if not self._metric_reader.force_flush(timeout_millis):
+                success = False
+        except Exception:
+            success = False
+
+        return success