feat: Enhance telemetry integration with OpenTelemetry and Azure Monitor for improved event tracking

Harmanpreet-Microsoft · Harmanpreet-Microsoft · commit e3c5d1eb21d7 · 2026-03-10T18:06:08.000+05:30
diff --git a/src/backend/api/api_routes.py b/src/backend/api/api_routes.py
@@ -14,9 +14,6 @@
 from api.status_updates import app_connection_manager, close_connection
 
 # Third-party
-# Azure Monitor OpenTelemetry integration is currently causing issues with OpenAI calls in process_batch_async, needs further investigation, commenting out for now
-# from azure.monitor.opentelemetry import configure_azure_monitor
-
 from common.logger.app_logger import AppLogger
 from common.services.batch_service import BatchService
 
@@ -40,21 +37,6 @@
 router = APIRouter()
 logger = AppLogger("APIRoutes")
 
-# Check if the Application Insights Instrumentation Key is set in the environment variables
-instrumentation_key = os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
-if instrumentation_key:
-    # Configure Application Insights if the Instrumentation Key is found
-    # commenting below line as configure_azure_monitor is causing issues with OpenAI calls in process_batch_async, needs further investigation
-    # configure_azure_monitor(connection_string=instrumentation_key)
-    logging.info(
-        "Application Insights configured with the provided Instrumentation Key"
-    )
-else:
-    # Log a warning if the Instrumentation Key is not found
-    logging.warning(
-        "No Application Insights Instrumentation Key found. Skipping configuration"
-    )
-
 
 def record_exception_to_trace(e):
     """Record exception to the current OpenTelemetry trace span."""
diff --git a/src/backend/api/event_utils.py b/src/backend/api/event_utils.py
@@ -3,17 +3,41 @@
 import os
 
 # Third-party
-from azure.monitor.events.extension import track_event
+from opentelemetry import trace
+from opentelemetry.trace import Status, StatusCode
 
 from dotenv import load_dotenv
 
 load_dotenv()
 
 
 def track_event_if_configured(event_name: str, event_data: dict):
+    """Track a custom event using OpenTelemetry.
+    
+    This creates a span with the event name and adds the event data as attributes.
+    The span will appear in Application Insights as a dependency with the event data.
+    """
     instrumentation_key = os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
     if instrumentation_key:
-        track_event(event_name, event_data)
+        try:
+            tracer = trace.get_tracer(__name__)
+            with tracer.start_as_current_span(f"event:{event_name}") as span:
+                # Set span kind to internal for custom events
+                span.set_attribute("event.name", event_name)
+                span.set_attribute("event.type", "custom")
+                
+                # Add all event data as span attributes
+                for key, value in event_data.items():
+                    # Convert value to string to ensure it's serializable
+                    span.set_attribute(f"event.{key}", str(value))
+                
+                # Add event to the span (appears in Application Insights)
+                span.add_event(event_name, attributes=event_data)
+                span.set_status(Status(StatusCode.OK))
+                
+                logging.debug(f"Tracked event: {event_name} with data: {event_data}")
+        except Exception as e:
+            logging.error(f"Failed to track event {event_name}: {e}")
     else:
         logging.warning(
             f"Skipping track_event for {event_name} as Application Insights is not configured"
diff --git a/src/backend/app.py b/src/backend/app.py
@@ -5,6 +5,9 @@
 
 from api.api_routes import router as backend_router
 
+from azure.monitor.opentelemetry import configure_azure_monitor
+from azure.monitor.opentelemetry.exporter import AzureMonitorLogExporter, AzureMonitorTraceExporter
+
 from common.config.config import app_config
 from common.logger.app_logger import AppLogger
 
@@ -15,6 +18,14 @@
 
 from helper.azure_credential_utils import get_azure_credential
 
+from opentelemetry import trace
+from opentelemetry._logs import set_logger_provider
+from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
 from semantic_kernel.agents.azure_ai.azure_ai_agent import AzureAIAgent  # pylint: disable=E0611
 
 from sql_agents.agent_manager import clear_sql_agents, set_sql_agents
@@ -46,6 +57,11 @@
 for logger_name in AZURE_LOGGING_PACKAGES:
     logging.getLogger(logger_name).setLevel(getattr(logging, AZURE_PACKAGE_LOGGING_LEVEL, logging.WARNING))
 
+# Suppress noisy OpenTelemetry and Azure Monitor logs
+# logging.getLogger("opentelemetry.sdk").setLevel(logging.ERROR)
+# logging.getLogger("azure.core.pipeline.policies.http_logging_policy").setLevel(logging.WARNING)
+# logging.getLogger("azure.monitor.opentelemetry.exporter.export._base").setLevel(logging.WARNING)
+
 logger = AppLogger("app")
 
 # Global variables for agents
@@ -119,6 +135,60 @@ def create_app() -> FastAPI:
         allow_headers=["*"],
     )
 
+    # Configure Azure Monitor and instrument FastAPI for OpenTelemetry
+    # This must happen AFTER app creation but BEFORE route registration
+    instrumentation_key = os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
+    if instrumentation_key:
+        # SOLUTION: Use manual telemetry setup instead of configure_azure_monitor
+        # This gives us precise control over what gets instrumented, avoiding interference
+        # with Semantic Kernel's async generators while still tracking Azure SDK calls
+        
+        # Set up Azure Monitor exporter for traces
+        azure_trace_exporter = AzureMonitorTraceExporter(connection_string=instrumentation_key)
+        
+        # Create a tracer provider and add the Azure Monitor exporter
+        tracer_provider = TracerProvider()
+        tracer_provider.add_span_processor(BatchSpanProcessor(azure_trace_exporter))
+        
+        # Set the global tracer provider
+        trace.set_tracer_provider(tracer_provider)
+
+        # Set up Azure Monitor exporter for logs (appears in traces table)
+        azure_log_exporter = AzureMonitorLogExporter(connection_string=instrumentation_key)
+        
+        # Create a logger provider and add the Azure Monitor exporter
+        logger_provider = LoggerProvider()
+        logger_provider.add_log_record_processor(BatchLogRecordProcessor(azure_log_exporter))
+        set_logger_provider(logger_provider)
+        
+        # Attach OpenTelemetry handler to Python's root logger
+        handler = LoggingHandler(logger_provider=logger_provider)
+        logging.getLogger().addHandler(handler)
+
+        # Instrument ONLY FastAPI for HTTP request/response tracing
+        # This is safe because it only wraps HTTP handlers, not internal async operations
+        FastAPIInstrumentor.instrument_app(
+            app,
+            excluded_urls="socket,ws",  # Exclude WebSocket URLs to reduce noise
+            tracer_provider=tracer_provider
+        )
+        
+        # Optional: Add manual spans in your code for Azure SDK operations using:
+        # from opentelemetry import trace
+        # tracer = trace.get_tracer(__name__)
+        # with tracer.start_as_current_span("operation_name"):
+        #     # your Azure SDK call here
+        
+        logger.logger.info("Application Insights configured with selective instrumentation")
+        logger.logger.info("✓ FastAPI HTTP tracing enabled")
+        logger.logger.info("✓ Python logging export to Application Insights enabled")
+        logger.logger.info("✓ Manual span support enabled for Azure SDK operations")
+        logger.logger.info("✓ Custom events via OpenTelemetry enabled")
+        logger.logger.info("✓ Semantic Kernel async generators unaffected")
+    else:
+        logger.logger.warning("No Application Insights connection string found. Telemetry disabled.")
+
+
     # Include routers with /api prefix
     app.include_router(backend_router, prefix="/api", tags=["backend"])
     # app.include_router(agents_router, prefix="/api/agents", tags=["agents"])
diff --git a/src/backend/common/telemetry/__init__.py b/src/backend/common/telemetry/__init__.py
@@ -0,0 +1,17 @@
+"""Telemetry utilities for Application Insights integration."""
+
+from common.telemetry.telemetry_helper import (
+    add_span_attributes,
+    get_tracer,
+    trace_context,
+    trace_operation,
+    trace_sync_context,
+)
+
+__all__ = [
+    "trace_operation",
+    "trace_context",
+    "trace_sync_context",
+    "get_tracer",
+    "add_span_attributes",
+]
diff --git a/src/backend/common/telemetry/telemetry_helper.py b/src/backend/common/telemetry/telemetry_helper.py
@@ -0,0 +1,162 @@
+"""Helper utilities for adding telemetry spans to Azure SDK operations.
+
+This module provides decorators and context managers for adding OpenTelemetry
+spans to Azure SDK calls (CosmosDB, Blob Storage, etc.) without interfering
+with Semantic Kernel's async generators.
+
+Example usage:
+    from common.telemetry.telemetry_helper import trace_operation
+    
+    @trace_operation("cosmosdb_query")
+    async def query_items(self, query: str):
+        # Your CosmosDB query here
+        pass
+"""
+
+import functools
+from contextlib import asynccontextmanager, contextmanager
+from typing import Any, Optional
+
+from opentelemetry import trace
+from opentelemetry.trace import Status, StatusCode
+
+
+def get_tracer(name: str = __name__):
+    """Get a tracer instance for the given name."""
+    return trace.get_tracer(name)
+
+
+def trace_operation(operation_name: str, attributes: Optional[dict] = None):
+    """Decorator to add telemetry span to a function or method.
+    
+    Args:
+        operation_name: Name of the operation for the span
+        attributes: Optional dictionary of attributes to add to the span
+        
+    Example:
+        @trace_operation("batch_processing", {"service": "sql_agents"})
+        async def process_batch(batch_id: str):
+            # Your code here
+            pass
+    """
+    def decorator(func):
+        @functools.wraps(func)
+        async def async_wrapper(*args, **kwargs):
+            tracer = get_tracer(func.__module__)
+            with tracer.start_as_current_span(operation_name) as span:
+                # Add custom attributes if provided
+                if attributes:
+                    for key, value in attributes.items():
+                        span.set_attribute(key, str(value))
+                
+                # Add function arguments as attributes (optional, for debugging)
+                span.set_attribute("function", func.__name__)
+                
+                try:
+                    result = await func(*args, **kwargs)
+                    span.set_status(Status(StatusCode.OK))
+                    return result
+                except Exception as e:
+                    span.record_exception(e)
+                    span.set_status(Status(StatusCode.ERROR, str(e)))
+                    raise
+        
+        @functools.wraps(func)
+        def sync_wrapper(*args, **kwargs):
+            tracer = get_tracer(func.__module__)
+            with tracer.start_as_current_span(operation_name) as span:
+                if attributes:
+                    for key, value in attributes.items():
+                        span.set_attribute(key, str(value))
+                
+                span.set_attribute("function", func.__name__)
+                
+                try:
+                    result = func(*args, **kwargs)
+                    span.set_status(Status(StatusCode.OK))
+                    return result
+                except Exception as e:
+                    span.record_exception(e)
+                    span.set_status(Status(StatusCode.ERROR, str(e)))
+                    raise
+        
+        # Return appropriate wrapper based on function type
+        if asyncio.iscoroutinefunction(func):
+            return async_wrapper
+        else:
+            return sync_wrapper
+    
+    return decorator
+
+
+@asynccontextmanager
+async def trace_context(operation_name: str, attributes: Optional[dict] = None):
+    """Async context manager for adding telemetry span to a code block.
+    
+    Args:
+        operation_name: Name of the operation for the span
+        attributes: Optional dictionary of attributes to add to the span
+        
+    Example:
+        async with trace_context("cosmosdb_batch_query", {"batch_id": batch_id}):
+            results = await database.query_items(query)
+            # Your code here
+    """
+    tracer = get_tracer()
+    with tracer.start_as_current_span(operation_name) as span:
+        if attributes:
+            for key, value in attributes.items():
+                span.set_attribute(key, str(value))
+        
+        try:
+            yield span
+            span.set_status(Status(StatusCode.OK))
+        except Exception as e:
+            span.record_exception(e)
+            span.set_status(Status(StatusCode.ERROR, str(e)))
+            raise
+
+
+@contextmanager
+def trace_sync_context(operation_name: str, attributes: Optional[dict] = None):
+    """Sync context manager for adding telemetry span to a code block.
+    
+    Args:
+        operation_name: Name of the operation for the span
+        attributes: Optional dictionary of attributes to add to the span
+        
+    Example:
+        with trace_sync_context("blob_upload", {"file_name": file_name}):
+            blob_client.upload_blob(data)
+    """
+    tracer = get_tracer()
+    with tracer.start_as_current_span(operation_name) as span:
+        if attributes:
+            for key, value in attributes.items():
+                span.set_attribute(key, str(value))
+        
+        try:
+            yield span
+            span.set_status(Status(StatusCode.OK))
+        except Exception as e:
+            span.record_exception(e)
+            span.set_status(Status(StatusCode.ERROR, str(e)))
+            raise
+
+
+def add_span_attributes(attributes: dict):
+    """Add attributes to the current span.
+    
+    Args:
+        attributes: Dictionary of attributes to add
+        
+    Example:
+        add_span_attributes({"user_id": user_id, "batch_id": batch_id})
+    """
+    span = trace.get_current_span()
+    if span and span.is_recording():
+        for key, value in attributes.items():
+            span.set_attribute(key, str(value))
+
+
+import asyncio
diff --git a/src/backend/requirements.txt b/src/backend/requirements.txt
@@ -49,5 +49,6 @@ opentelemetry-sdk==1.39.0
 opentelemetry-api==1.39.0
 opentelemetry-semantic-conventions==0.60b0
 opentelemetry-instrumentation==0.60b0
+opentelemetry-instrumentation-fastapi==0.60b0
 azure-monitor-opentelemetry==1.8.6
 azure-ai-projects==1.0.0
diff --git a/src/backend/sql_agents/process_batch.py b/src/backend/sql_agents/process_batch.py
