LCORE-1792: Add OpenTelemetry instrumentation for POST /v1/query endpoint

Implements comprehensive OpenTelemetry (OTEL) tracing instrumentation for the POST
/v1/query endpoint to enable distributed tracing and observability.

**Instrumented Components**

1. Query Endpoint Handler (`src/app/endpoints/query.py`)
2. Quota Check (`src/utils/quota_utils.py`)
3. Shield Moderation (`src/utils/shields.py`)
4. RAG Retrieval (`src/utils/vector_search.py`)
5. LLM Inference (`src/utils/agents/query.py`)
6. Tool Execution (`src/utils/agents/query.py`)

**Span Hierarchy**

```
query.handle_request (root span)
├── quota.check
├── shield.moderate
├── rag.retrieve
└── llm.inference
    └── tool.execution (attributes only)
```

Author: anik120

src/app/endpoints/query.py

@@ -4,6 +4,7 @@
 from typing import Annotated, Any
 
 from fastapi import APIRouter, Depends, Request
+from opentelemetry import trace
 
 from authentication import get_auth_dependency
 from authentication.interface import AuthTuple
@@ -38,6 +39,13 @@
 )
 from utils.mcp_headers import McpHeaders, mcp_headers_dependency
 from utils.mcp_oauth_probe import check_mcp_auth
+from utils.otel_tracing import (
+    SpanAttributes,
+    SpanEvents,
+    add_span_event,
+    anonymize_value,
+    set_span_attributes,
+)
 from utils.query import (
     consume_query_tokens,
     prepare_input,
@@ -56,6 +64,7 @@
 from utils.vector_search import build_rag_context
 
 logger = get_logger(__name__)
+tracer = trace.get_tracer(__name__)
 router = APIRouter(tags=["query"])
 
 query_response: dict[int | str, dict[str, Any]] = {
@@ -113,11 +122,51 @@ async def query_endpoint_handler(
     - 500: Internal Server Error - Configuration not loaded or other server errors
     - 503: Service Unavailable - Unable to connect to OGX backend
     """
+    with tracer.start_as_current_span("query.handle_request") as root_span:
+        return await _handle_query_with_tracing(
+            request, query_request, auth, mcp_headers, root_span
+        )
+
+
+async def _handle_query_with_tracing(
+    request: Request,
+    query_request: QueryRequest,
+    auth: AuthTuple,
+    mcp_headers: McpHeaders,
+    root_span: trace.Span,
+) -> QueryResponse:
+    """Handle query request with OTEL tracing instrumentation.
+
+    Parameters:
+        request: The incoming HTTP request.
+        query_request: Request payload containing query and optional parameters.
+        auth: Authentication tuple (user_id, username, skip_check, token).
+        mcp_headers: Headers to be passed to MCP servers.
+        root_span: OpenTelemetry root span for this request.
+
+    Returns:
+        QueryResponse containing conversation ID, LLM response, and metadata.
+
+    Raises:
+        HTTPException: On authentication, authorization, quota, or model errors.
+    """
     check_configuration_loaded(configuration)
 
     started_at = datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
     user_id, _, _skip_userid_check, token = auth
 
+    # Set initial span attributes
+    set_span_attributes(
+        root_span,
+        {
+            SpanAttributes.USER_ID: anonymize_value(user_id),
+            SpanAttributes.INPUT: anonymize_value(query_request.query),
+            SpanAttributes.REQUEST_ATTACHMENTS_COUNT: (
+                len(query_request.attachments) if query_request.attachments else 0
+            ),
+        },
+    )
+
     # Check MCP Auth
     await check_mcp_auth(configuration, mcp_headers, token, request.headers)
 
@@ -136,6 +185,9 @@ async def query_endpoint_handler(
     if query_request.attachments:
         validate_attachments_metadata(query_request.attachments)
 
+    # Validation completed
+    add_span_event(root_span, SpanEvents.VALIDATION_COMPLETED)
+
     # Retrieve conversation if conversation_id is provided
     user_conversation = None
     if query_request.conversation_id:
@@ -279,6 +331,22 @@ async def query_endpoint_handler(
     )
 
     logger.info("Building final response")
+
+    # Set final span attributes
+    set_span_attributes(
+        root_span,
+        {
+            SpanAttributes.SESSION_ID: conversation_id,
+            SpanAttributes.LLM_USAGE_INPUT_TOKENS: turn_summary.token_usage.input_tokens,
+            SpanAttributes.LLM_USAGE_OUTPUT_TOKENS: turn_summary.token_usage.output_tokens,
+            SpanAttributes.OUTPUT: anonymize_value(turn_summary.llm_response),
+        },
+    )
+
+    # Emit final events
+    add_span_event(root_span, SpanEvents.LLM_RESPONSE_COMPLETED)
+    add_span_event(root_span, SpanEvents.TURN_PERSISTED)
+
     return QueryResponse(
         conversation_id=conversation_id,
         response=turn_summary.llm_response,

src/constants.py

@@ -259,6 +259,10 @@
 # Special RAG ID that activates the OKP provider when listed in rag.inline or rag.tool
 OKP_RAG_ID: Final[str] = "okp"
 
+# OpenTelemetry anonymization configuration
+# Environment variable for HMAC secret used to anonymize sensitive trace data
+OTEL_ANONYMIZATION_SECRET_ENV_VAR: Final[str] = "OTEL_ANONYMIZATION_SECRET"
+
 # Logging configuration constants
 # Environment variable name for configurable log level
 LIGHTSPEED_STACK_LOG_LEVEL_ENV_VAR: Final[str] = "LIGHTSPEED_STACK_LOG_LEVEL"

src/utils/agents/query.py

@@ -7,6 +7,7 @@
 
 from fastapi import HTTPException
 from ogx_client import APIConnectionError, APIStatusError, AsyncOgxClient
+from opentelemetry import trace
 from pydantic_ai.exceptions import (
     AgentRunError,
 )
@@ -36,6 +37,12 @@
     process_native_tool_result,
 )
 from utils.conversations import append_turn_items_to_conversation
+from utils.otel_tracing import (
+    SpanAttributes,
+    SpanEvents,
+    add_span_event,
+    set_span_attributes,
+)
 from utils.pydantic_ai_helpers import build_agent
 from utils.query import (
     build_multimodal_input,
@@ -45,6 +52,7 @@
 from utils.token_counter import TokenCounter
 
 logger = get_logger(__name__)
+tracer = trace.get_tracer(__name__)
 
 AgentInferenceError: TypeAlias = (
     AgentRunError | APIStatusError | APIConnectionError | RuntimeError
@@ -180,20 +188,41 @@ def build_turn_summary_from_agent_run(
         turn_summary=TurnSummary(),
     )
 
+    # Track tool calls for OTEL instrumentation
+    tool_call_names: list[str] = []
+
     for message in run_result.new_messages():
         if isinstance(message, ModelResponse):
             if message.text:
                 state.turn_summary.llm_response = message.text
             for tool_call_part in message.tool_calls:
                 process_function_tool_call(state, tool_call_part)
+                tool_call_names.append(tool_call_part.tool_name)
             for call_part, return_part in message.native_tool_calls:
                 process_native_tool_call(state, call_part)
                 process_native_tool_result(state, return_part)
+                tool_call_names.append(call_part.tool_name)
         elif isinstance(message, ModelRequest):
             for request_part in message.parts:
                 if isinstance(request_part, ToolReturnPart):
                     process_function_tool_result(state, request_part)
 
+    # Add tool execution attributes to current span (parent llm.inference span)
+    current_span = trace.get_current_span()
+    if current_span.is_recording() and tool_call_names:
+        set_span_attributes(
+            current_span,
+            {
+                SpanAttributes.TOOL_CALLS_COUNT: len(tool_call_names),
+                SpanAttributes.TOOL_CALLS_NAMES: tool_call_names,
+            },
+        )
+        add_span_event(
+            current_span,
+            SpanEvents.TOOL_EXECUTION_COMPLETED,
+            {"tool.calls": ", ".join(tool_call_names)},
+        )
+
     state.turn_summary.id = run_result.response.provider_response_id or ""
     state.turn_summary.token_usage = extract_agent_token_usage(
         run_result.usage,
@@ -231,44 +260,83 @@ async def retrieve_agent_response(
     Raises:
         HTTPException: On moderation is not applicable; on agent or provider failure.
     """
-    if moderation_result.decision == "blocked":
-        await append_turn_items_to_conversation(
-            client,
-            responses_params.conversation,
-            responses_params.input,
-            [moderation_result.refusal_response],
+    with tracer.start_as_current_span("llm.inference") as span:
+        # Extract provider and model from model_id
+        provider_id, model_id = extract_provider_and_model_from_model_id(
+            responses_params.model
         )
-        return TurnSummary(
-            id=moderation_result.moderation_id,
-            llm_response=moderation_result.message,
-        )
-    try:
-        agent = build_agent(
-            client,
-            responses_params,
-            configuration,
-            shields=shield_ids,
-            no_tools=no_tools,
+
+        # Set LLM attributes
+        set_span_attributes(
+            span,
+            {
+                SpanAttributes.LLM_MODEL_ID: model_id,
+                SpanAttributes.LLM_PROVIDER_ID: provider_id,
+            },
         )
-        logger.debug("Starting agent non-streaming response processing")
-        if image_attachments:
-            prompt = build_multimodal_input(
-                cast(str, responses_params.input),
-                image_attachments,
+
+        if moderation_result.decision == "blocked":
+            await append_turn_items_to_conversation(
+                client,
+                responses_params.conversation,
+                responses_params.input,
+                [moderation_result.refusal_response],
             )
-        else:
-            prompt = cast(str, responses_params.input)
-        run_result = await agent.run(prompt)
-    except (AgentRunError, APIStatusError, APIConnectionError, RuntimeError) as exc:
-        response = map_agent_inference_error(exc, responses_params.model)
-        raise HTTPException(**response.model_dump()) from exc
-
-    vector_store_ids = extract_vector_store_ids_from_tools(responses_params.tools)
-    rag_id_mapping = configuration.rag_id_mapping
-    return build_turn_summary_from_agent_run(
-        run_result,
-        model_id=responses_params.model,
-        endpoint_path=endpoint_path,
-        vector_store_ids=vector_store_ids,
-        rag_id_mapping=rag_id_mapping,
-    )
+            return TurnSummary(
+                id=moderation_result.moderation_id,
+                llm_response=moderation_result.message,
+            )
+
+        # Emit inference started event
+        add_span_event(span, SpanEvents.LLM_INFERENCE_STARTED)
+
+        try:
+            agent = build_agent(
+                client,
+                responses_params,
+                configuration,
+                shields=shield_ids,
+                no_tools=no_tools,
+            )
+            logger.debug("Starting agent non-streaming response processing")
+            if image_attachments:
+                prompt = build_multimodal_input(
+                    cast(str, responses_params.input),
+                    image_attachments,
+                )
+            else:
+                prompt = cast(str, responses_params.input)
+            run_result = await agent.run(prompt)
+        except (
+            AgentRunError,
+            APIStatusError,
+            APIConnectionError,
+            RuntimeError,
+        ) as exc:
+            response = map_agent_inference_error(exc, responses_params.model)
+            raise HTTPException(**response.model_dump()) from exc
+
+        # Set token usage attributes
+        if run_result.usage:
+            set_span_attributes(
+                span,
+                {
+                    SpanAttributes.LLM_USAGE_INPUT_TOKENS: run_result.usage.input_tokens,
+                    SpanAttributes.LLM_USAGE_OUTPUT_TOKENS: run_result.usage.output_tokens,
+                },
+            )
+
+        vector_store_ids = extract_vector_store_ids_from_tools(responses_params.tools)
+        rag_id_mapping = configuration.rag_id_mapping
+        turn_summary = build_turn_summary_from_agent_run(
+            run_result,
+            model_id=responses_params.model,
+            endpoint_path=endpoint_path,
+            vector_store_ids=vector_store_ids,
+            rag_id_mapping=rag_id_mapping,
+        )
+
+        # Emit inference completed event after successful summary build
+        add_span_event(span, SpanEvents.LLM_INFERENCE_COMPLETED)
+
+        return turn_summary