emit llo span data to logs pipeline

Author: yiyuan-he

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

@@ -3,16 +3,24 @@
 import logging
 from typing import Dict, Optional, Sequence
 
+import os
 import re
 import requests
 
 from amazon.opentelemetry.distro._utils import is_installed
 from amazon.opentelemetry.distro.llo_sender_client import LLOSenderClient
+from amazon.opentelemetry.distro.otlp_aws_logs_exporter import OTLPAwsLogExporter
 from opentelemetry.attributes import BoundedAttributes
 from opentelemetry.exporter.otlp.proto.http import Compression
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
-from opentelemetry.sdk.trace import ReadableSpan, Event
+from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExportResult
+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
+
 
 AWS_SERVICE = "xray"
 _logger = logging.getLogger(__name__)
@@ -42,6 +50,11 @@ def __init__(
         self._aws_region = None
         self._has_required_dependencies = False
         self._llo_sender_client = LLOSenderClient()
+        # set up logger for LLO data
+        self._logs_exporter = OTLPAwsLogExporter(endpoint=os.getenv("OTEL_EXPORTER_OTLP_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)
         # Requires botocore to be installed to sign the headers. However,
         # some users might not need to use this exporter. In order not conflict
         # with existing behavior, we check for botocore before initializing this exporter.
@@ -94,11 +107,30 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                         "span_name": span.name
                     }
 
-                    # Get S3 pointer from LLOSenderClient
-                    s3_pointer = self._llo_sender_client.upload(value, metadata)
+                    # Emit a log record for LLO data
+                    llo_attributes = {
+                        "llo.attribute_name": metadata.get('attribute_name', 'unknown'),
+                        "llo.content": value,
+                        "llo.span_name": metadata.get('span_name', ''),
+                        "service.name": "llm-app",
+
+                    }
+
+                    self._logger.emit(
+                        LogRecord(
+                            timestamp=span.start_time,
+                            observed_timestamp=span.start_time,
+                            trace_id=span.context.trace_id,
+                            span_id=span.context.span_id,
+                            trace_flags=TraceFlags(0x01),
+                            severity_text="INFO",
+                            severity_number=SeverityNumber.INFO,
+                            body=f"LLO attribute: {metadata.get('attribute_name')}",
+                            attributes=llo_attributes
+                        )
+                    )
 
-                    # Store the S3 pointer instead of original value to trim span
-                    updated_attributes[key] = s3_pointer
+                    # We drop the LLO attribute (don't add to updated_attributes)
                 else:
                     # Keep original value if it is not LLO
                     updated_attributes[key] = value
@@ -114,62 +146,15 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
             else:
                 span._attributes = updated_attributes
 
-            # Process span events
-            if span.events:
-                updated_events = []
-
-                for event in span.events:
-                    # Check if this event has any attributes to process
-                    if not event.attributes:
-                        updated_events.append(event) # Keep the original event
-                        continue
-
-                    # Process event attributes for LLO content
-                    updated_event_attributes = {}
-                    need_to_update = False
-
-                    for key, value in event.attributes.items():
-                        if self._should_offload(key):
-                            metadata = {
-                                "trace_id": format(span.context.trace_id, 'x'),
-                                "span_id": format(span.context.span_id, 'x'),
-                                "attribute_name": key,
-                                "event_name": event.name,
-                                "event_time": str(event.timestamp)
-                            }
-
-                            s3_pointer = self._llo_sender_client.upload(value, metadata)
-                            updated_event_attributes[key] = s3_pointer
-                            need_to_update = True
-                        else:
-                            updated_event_attributes[key] = value
-
-                    if need_to_update:
-                        # Create new Event with the updated attributes
-                        limit = None
-                        if isinstance(event.attributes, BoundedAttributes):
-                            limit = event.attributes.maxlen
-
-                        updated_event = Event(
-                            name=event.name,
-                            attributes=updated_event_attributes,
-                            timestamp=event.timestamp,
-                            limit=limit
-                        )
-
-                        updated_events.append(updated_event)
-                    else:
-                        # Keep the original event
-                        updated_events.append(event)
-
-                # Update the span's events with processed events
-                span._events = updated_events
-
+            # Add modified span to list
             modified_spans.append(span)
 
+            # TODO: Process span events
+
         # Export the modified spans
         return super().export(modified_spans)
 
+
     def _should_offload(self, key):
         """Determine if LLO based on the attribute key. Strict matching is enforced as to not introduce unintended behavior."""
         exact_match_patterns = [