Merge pull request #65 from ks6088ts-labs/feature/issue-64_otel

support otel

Author: ks6088ts

.env.template

@@ -59,3 +59,7 @@ CSV_LOADER_DATA_DIR_PATH="./data"
 
 ## PDF Loader Settings
 PDF_LOADER_DATA_DIR_PATH="./data"
+
+## OpenTelemetry Settings
+OTEL_SERVICE_NAME="template-langgraph"
+OTEL_COLLECTOR_ENDPOINT="http://localhost:4317"

docker-compose.yml

@@ -31,3 +31,19 @@ services:
       POSTGRES_PASSWORD: password
       POSTGRES_DB: db
     restart: always
+  jaeger:
+    image: jaegertracing/all-in-one:1.72.0
+    container_name: jaeger
+    environment:
+      - SPAN_STORAGE_TYPE=elasticsearch
+      - ES_SERVER_URLS=http://elasticsearch:9200
+      - LOG_LEVEL=debug
+    ports:
+      - "16686:16686"
+      - "4317:4317"
+      - "4318:4318"
+      - "5778:5778"
+      - "9411:9411"
+    depends_on:
+      - elasticsearch
+    restart: unless-stopped

pyproject.toml

@@ -19,6 +19,9 @@ dependencies = [
     "langgraph>=0.6.2",
     "langgraph-supervisor>=0.0.29",
     "openai>=1.98.0",
+    "opentelemetry-api>=1.36.0",
+    "opentelemetry-exporter-otlp>=1.36.0",
+    "opentelemetry-sdk>=1.36.0",
     "psycopg2-binary>=2.9.10",
     "pydantic-settings>=2.9.1",
     "pypdf>=5.9.0",

scripts/otel_operator.py

@@ -0,0 +1,56 @@
+import logging
+import time
+
+import typer
+from dotenv import load_dotenv
+
+from template_langgraph.loggers import get_logger
+from template_langgraph.utilities.otel_helpers import OtelWrapper
+
+# Initialize the Typer application
+app = typer.Typer(
+    add_completion=False,
+    help="OTEL operator CLI",
+)
+
+# Set up logging
+logger = get_logger(__name__)
+
+
+@app.command()
+def run(
+    query: str = typer.Option(
+        "What is the weather like today?",
+        "--query",
+        "-q",
+        help="Query to run against the Ollama model",
+    ),
+    verbose: bool = typer.Option(
+        False,
+        "--verbose",
+        "-v",
+        help="Enable verbose output",
+    ),
+):
+    # Set up logging
+    if verbose:
+        logger.setLevel(logging.DEBUG)
+    otel_wrapper = OtelWrapper()
+    otel_wrapper.initialize()
+
+    logger.info("Running...")
+    tracer = otel_wrapper.get_tracer(name=__name__)
+    with tracer.start_as_current_span("otel_operator_run"):
+        logger.info(f"Query: {query}")
+        # Simulate some work
+        response = {"content": "It's sunny!"}
+        time.sleep(1)  # Simulate processing time
+        logger.info(f"Response: {response['content']}")
+
+
+if __name__ == "__main__":
+    load_dotenv(
+        override=True,
+        verbose=True,
+    )
+    app()

template_langgraph/utilities/otel_helpers.py

@@ -0,0 +1,54 @@
+from functools import lru_cache
+
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    otel_service_name: str = "<OTEL_SERVICE_NAME>"
+    otel_collector_endpoint: str = "<OTEL_COLLECTOR_ENDPOINT>"
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_ignore_empty=True,
+        extra="ignore",
+    )
+
+
+@lru_cache
+def get_otel_settings() -> Settings:
+    """Get OpenTelemetry settings."""
+    return Settings()
+
+
+class OtelWrapper:
+    def __init__(
+        self,
+        settings: Settings = None,
+    ):
+        if settings is None:
+            settings = get_otel_settings()
+        self.settings = settings
+
+    def initialize(self):
+        provider = TracerProvider(
+            resource=Resource(
+                attributes={
+                    "service.name": self.settings.otel_service_name,
+                }
+            )
+        )
+        otlp_exporter = OTLPSpanExporter(
+            endpoint=self.settings.otel_collector_endpoint,
+        )
+        provider.add_span_processor(
+            span_processor=BatchSpanProcessor(otlp_exporter),
+        )
+        trace.set_tracer_provider(provider)
+
+    def get_tracer(self, name: str):
+        return trace.get_tracer(name)