emit llo to log pipeline

Author: yiyuan-he

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/llo_handler.py

@@ -1,17 +1,28 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+import logging
+import os
 import re
 from typing import Dict, Any, List, Optional, Sequence
 
 from opentelemetry.attributes import BoundedAttributes
 from opentelemetry.sdk.trace import ReadableSpan, Event
+from opentelemetry.sdk._logs import LoggerProvider, LogRecord
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry._logs import get_logger
+from opentelemetry._logs.severity import SeverityNumber
+from opentelemetry.trace import TraceFlags
+
+from amazon.opentelemetry.distro.otlp_aws_logs_exporter import OTLPAwsLogExporter
+
+_logger = logging.getLogger(__name__)
 
 
 class LLOHandler:
     """
     Utility class for handling Large Language Model Output (LLO) attributes.
-    This class identifies LLO attributes and determines whether they should be
-    processed or filtered out from telemetry data.
+    This class identifies LLO attributes, emits them as log records,
+    and filters them out from telemetry data.
     """
 
     def __init__(self):
@@ -35,6 +46,27 @@ def __init__(self):
             r"^llm.output_messages\.\d+\.message.content$",
         ]
 
+        # Set up logger for LLO data
+        self._setup_logger()
+
+    def _setup_logger(self):
+        """
+        Set up the logger with OTLP AWS Logs Exporter
+        """
+        # Set up the logs exporter
+        logs_endpoint = os.getenv("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT")
+        if logs_endpoint:
+            self._logs_exporter = OTLPAwsLogExporter(endpoint=logs_endpoint)
+            self._logger_provider = LoggerProvider()
+            self._logger_provider.add_log_record_processor(
+                BatchLogRecordProcessor(self._logs_exporter)
+            )
+            self._logger = get_logger("llo_logger", logger_provider=self._logger_provider)
+            _logger.debug(f"Initialized LLO logger with AWS OTLP Logs exporter at {logs_endpoint}")
+        else:
+            self._logger = None
+            _logger.warning("No OTEL_EXPORTER_OTLP_LOGS_ENDPOINT specified, LLO attributes will be filtered but not emitted as logs")
+
     def is_llo_attribute(self, key: str) -> bool:
         """
         Determine if an attribute is LLO based on its key.
@@ -67,13 +99,85 @@ def filter_attributes(self, attributes: Dict[str, Any]) -> Dict[str, Any]:
                 filtered_attributes[key] = value
         return filtered_attributes
 
+    def emit_llo_attributes(self, span: ReadableSpan, attributes: Dict[str, Any], 
+                            event_name: Optional[str] = None, timestamp: Optional[int] = None) -> None:
+        """
+        Extract and emit LLO attributes as log records.
+
+        Args:
+            span: The span containing the LLO attributes
+            attributes: Dictionary of attributes to check for LLO attributes
+            event_name: Optional name of the event (if attributes are from an event)
+            timestamp: Optional timestamp to use (default: span start time)
+        """
+        if not self._logger:
+            return
+
+        try:
+            # Use span timestamp by default
+            ts = timestamp or span.start_time
+
+            for key, value in attributes.items():
+                # Only process LLO attributes
+                if not self.is_llo_attribute(key):
+                    continue
+
+                # Create a structured body similar to OpenAI format
+                # This matches the example log format provided
+                body = {
+                    "attribute_key": key,
+                    "content": value
+                }
+
+                # Add event information to body if available
+                if event_name:
+                    body["event_name"] = event_name
+
+                # Add span name to body
+                body["span_name"] = span.name
+
+                # Build attributes dict for the log record
+                log_attributes = {
+                    "event.name": f"llo.attribute.{key.split('.')[-1]}",  # Extract last part of attribute key
+                }
+
+                # Add additional context from span if available
+                for context_key in ["gen_ai.system", "gen_ai.operation.name", "gen_ai.request.model"]:
+                    if context_key in span.attributes:
+                        log_attributes[context_key] = span.attributes[context_key]
+
+                # Emit the log record directly using the logger
+                self._logger.emit(
+                    LogRecord(
+                        timestamp=ts,
+                        observed_timestamp=ts,
+                        trace_id=span.context.trace_id,
+                        span_id=span.context.span_id,
+                        trace_flags=TraceFlags(0x01),
+                        severity_number=SeverityNumber.INFO,
+                        severity_text=None,  # Set to None as in the example
+                        body=body,  # Use the structured body
+                        attributes=log_attributes
+                    )
+                )
+
+                _logger.debug(f"Emitted LLO log record for attribute: {key}")
+
+        except Exception as e:
+            _logger.error(f"Error emitting LLO log records: {e}", exc_info=True)
+
     def update_span_attributes(self, span: ReadableSpan) -> None:
         """
-        Update span attributes by filtering out LLO attributes.
+        Update span attributes by:
+        1. Emitting LLO attributes as log records (if logger is configured)
+        2. Filtering out LLO attributes from the span
 
         Args:
             span: The span to update
         """
+        # Emit LLO attributes as log records
+        self.emit_llo_attributes(span, span.attributes)
+
         # Filter out LLO attributes
         updated_attributes = self.filter_attributes(span.attributes)
 
@@ -90,7 +194,9 @@ def update_span_attributes(self, span: ReadableSpan) -> None:
 
     def process_span_events(self, span: ReadableSpan) -> None:
         """
-        Process events within a span by filtering out LLO attributes from event attributes.
+        Process events within a span by:
+        1. Emitting LLO attributes as log records (if logger is configured)
+        2. Filtering out LLO attributes from event attributes
 
         Args:
             span: The span containing events to process
@@ -106,6 +212,14 @@ def process_span_events(self, span: ReadableSpan) -> None:
                 updated_events.append(event)  # Keep the original event
                 continue
 
+            # Emit LLO attributes as log records
+            self.emit_llo_attributes(
+                span, 
+                event.attributes, 
+                event_name=event.name, 
+                timestamp=event.timestamp
+            )
+
             # Filter out LLO attributes from event
             updated_event_attributes = self.filter_attributes(event.attributes)
 
@@ -135,8 +249,9 @@ def process_span_events(self, span: ReadableSpan) -> None:
 
     def process_spans(self, spans: Sequence[ReadableSpan]) -> List[ReadableSpan]:
         """
-        Process a list of spans by filtering out LLO attributes from both
-        span attributes and event attributes.
+        Process a list of spans by:
+        1. Emitting LLO attributes as log records (if logger is configured)
+        2. Filtering out LLO attributes from both span attributes and event attributes
 
         Args:
             spans: List of spans to process