add logs pipeline

Author: liustve

aws-opentelemetry-distro/pyproject.toml

@@ -24,62 +24,62 @@ classifiers = [
 ]
 
 dependencies = [
-  "opentelemetry-api == 1.27.0",
-  "opentelemetry-sdk == 1.27.0",
-  "opentelemetry-exporter-otlp-proto-grpc == 1.27.0",
-  "opentelemetry-exporter-otlp-proto-http == 1.27.0",
-  "opentelemetry-propagator-b3 == 1.27.0",
-  "opentelemetry-propagator-jaeger == 1.27.0",
-  "opentelemetry-exporter-otlp-proto-common == 1.27.0",
+  "opentelemetry-api == 1.33.1",
+  "opentelemetry-sdk == 1.33.1",
+  "opentelemetry-exporter-otlp-proto-grpc == 1.33.1",
+  "opentelemetry-exporter-otlp-proto-http == 1.33.1",
+  "opentelemetry-propagator-b3 == 1.33.1",
+  "opentelemetry-propagator-jaeger == 1.33.1",
+  "opentelemetry-exporter-otlp-proto-common == 1.33.1",
   "opentelemetry-sdk-extension-aws == 2.0.2",
   "opentelemetry-propagator-aws-xray == 1.0.1",
-  "opentelemetry-distro == 0.48b0",
-  "opentelemetry-processor-baggage == 0.48b0",
-  "opentelemetry-propagator-ot-trace == 0.48b0",
-  "opentelemetry-instrumentation == 0.48b0",
-  "opentelemetry-instrumentation-aws-lambda == 0.48b0",
-  "opentelemetry-instrumentation-aio-pika == 0.48b0",
-  "opentelemetry-instrumentation-aiohttp-client == 0.48b0",
-  "opentelemetry-instrumentation-aiopg == 0.48b0",
-  "opentelemetry-instrumentation-asgi == 0.48b0",
-  "opentelemetry-instrumentation-asyncpg == 0.48b0",
-  "opentelemetry-instrumentation-boto == 0.48b0",
-  "opentelemetry-instrumentation-boto3sqs == 0.48b0",
-  "opentelemetry-instrumentation-botocore == 0.48b0",
-  "opentelemetry-instrumentation-celery == 0.48b0",
-  "opentelemetry-instrumentation-confluent-kafka == 0.48b0",
-  "opentelemetry-instrumentation-dbapi == 0.48b0",
-  "opentelemetry-instrumentation-django == 0.48b0",
-  "opentelemetry-instrumentation-elasticsearch == 0.48b0",
-  "opentelemetry-instrumentation-falcon == 0.48b0",
-  "opentelemetry-instrumentation-fastapi == 0.48b0",
-  "opentelemetry-instrumentation-flask == 0.48b0",
-  "opentelemetry-instrumentation-grpc == 0.48b0",
-  "opentelemetry-instrumentation-httpx == 0.48b0",
-  "opentelemetry-instrumentation-jinja2 == 0.48b0",
-  "opentelemetry-instrumentation-kafka-python == 0.48b0",
-  "opentelemetry-instrumentation-logging == 0.48b0",
-  "opentelemetry-instrumentation-mysql == 0.48b0",
-  "opentelemetry-instrumentation-mysqlclient == 0.48b0",
-  "opentelemetry-instrumentation-pika == 0.48b0",
-  "opentelemetry-instrumentation-psycopg2 == 0.48b0",
-  "opentelemetry-instrumentation-pymemcache == 0.48b0",
-  "opentelemetry-instrumentation-pymongo == 0.48b0",
-  "opentelemetry-instrumentation-pymysql == 0.48b0",
-  "opentelemetry-instrumentation-pyramid == 0.48b0",
-  "opentelemetry-instrumentation-redis == 0.48b0",
-  "opentelemetry-instrumentation-remoulade == 0.48b0",
-  "opentelemetry-instrumentation-requests == 0.48b0",
-  "opentelemetry-instrumentation-sqlalchemy == 0.48b0",
-  "opentelemetry-instrumentation-sqlite3 == 0.48b0",
-  "opentelemetry-instrumentation-starlette == 0.48b0",
-  "opentelemetry-instrumentation-system-metrics == 0.48b0",
-  "opentelemetry-instrumentation-tornado == 0.48b0",
-  "opentelemetry-instrumentation-tortoiseorm == 0.48b0",
-  "opentelemetry-instrumentation-urllib == 0.48b0",
-  "opentelemetry-instrumentation-urllib3 == 0.48b0",
-  "opentelemetry-instrumentation-wsgi == 0.48b0",
-  "opentelemetry-instrumentation-cassandra == 0.48b0",
+  "opentelemetry-distro == 0.54b1",
+  "opentelemetry-processor-baggage == 0.54b1",
+  "opentelemetry-propagator-ot-trace == 0.54b1",
+  "opentelemetry-instrumentation == 0.54b1",
+  "opentelemetry-instrumentation-aws-lambda == 0.54b1",
+  "opentelemetry-instrumentation-aio-pika == 0.54b1",
+  "opentelemetry-instrumentation-aiohttp-client == 0.54b1",
+  "opentelemetry-instrumentation-aiopg == 0.54b1",
+  "opentelemetry-instrumentation-asgi == 0.54b1",
+  "opentelemetry-instrumentation-asyncpg == 0.54b1",
+  "opentelemetry-instrumentation-boto == 0.54b1",
+  "opentelemetry-instrumentation-boto3sqs == 0.54b1",
+  "opentelemetry-instrumentation-botocore == 0.54b1",
+  "opentelemetry-instrumentation-celery == 0.54b1",
+  "opentelemetry-instrumentation-confluent-kafka == 0.54b1",
+  "opentelemetry-instrumentation-dbapi == 0.54b1",
+  "opentelemetry-instrumentation-django == 0.54b1",
+  "opentelemetry-instrumentation-elasticsearch == 0.54b1",
+  "opentelemetry-instrumentation-falcon == 0.54b1",
+  "opentelemetry-instrumentation-fastapi == 0.54b1",
+  "opentelemetry-instrumentation-flask == 0.54b1",
+  "opentelemetry-instrumentation-grpc == 0.54b1",
+  "opentelemetry-instrumentation-httpx == 0.54b1",
+  "opentelemetry-instrumentation-jinja2 == 0.54b1",
+  "opentelemetry-instrumentation-kafka-python == 0.54b1",
+  "opentelemetry-instrumentation-logging == 0.54b1",
+  "opentelemetry-instrumentation-mysql == 0.54b1",
+  "opentelemetry-instrumentation-mysqlclient == 0.54b1",
+  "opentelemetry-instrumentation-pika == 0.54b1",
+  "opentelemetry-instrumentation-psycopg2 == 0.54b1",
+  "opentelemetry-instrumentation-pymemcache == 0.54b1",
+  "opentelemetry-instrumentation-pymongo == 0.54b1",
+  "opentelemetry-instrumentation-pymysql == 0.54b1",
+  "opentelemetry-instrumentation-pyramid == 0.54b1",
+  "opentelemetry-instrumentation-redis == 0.54b1",
+  "opentelemetry-instrumentation-remoulade == 0.54b1",
+  "opentelemetry-instrumentation-requests == 0.54b1",
+  "opentelemetry-instrumentation-sqlalchemy == 0.54b1",
+  "opentelemetry-instrumentation-sqlite3 == 0.54b1",
+  "opentelemetry-instrumentation-starlette == 0.54b1",
+  "opentelemetry-instrumentation-system-metrics == 0.54b1",
+  "opentelemetry-instrumentation-tornado == 0.54b1",
+  "opentelemetry-instrumentation-tortoiseorm == 0.54b1",
+  "opentelemetry-instrumentation-urllib == 0.54b1",
+  "opentelemetry-instrumentation-urllib3 == 0.54b1",
+  "opentelemetry-instrumentation-wsgi == 0.54b1",
+  "opentelemetry-instrumentation-cassandra == 0.54b1",
 ]
 
 [project.optional-dependencies]

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

@@ -35,7 +35,6 @@
 )
 from amazon.opentelemetry.distro._aws_resource_attribute_configurator import get_service_attribute
 from amazon.opentelemetry.distro._aws_span_processing_util import (
-    GEN_AI_REQUEST_MODEL,
     LOCAL_ROOT,
     MAX_KEYWORD_LENGTH,
     SQL_KEYWORD_PATTERN,
@@ -60,6 +59,7 @@
 from amazon.opentelemetry.distro.sqs_url_parser import SqsUrlParser
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import BoundedAttributes, ReadableSpan
+from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import GEN_AI_REQUEST_MODEL
 from opentelemetry.semconv.trace import SpanAttributes
 
 # Pertinent OTEL attribute keys

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

@@ -26,16 +26,6 @@
 # Max keyword length supported by parsing into remote_operation from DB_STATEMENT
 MAX_KEYWORD_LENGTH = 27
 
-# TODO: Use Semantic Conventions once upgrade to 0.47b0
-GEN_AI_REQUEST_MODEL: str = "gen_ai.request.model"
-GEN_AI_SYSTEM: str = "gen_ai.system"
-GEN_AI_REQUEST_MAX_TOKENS: str = "gen_ai.request.max_tokens"
-GEN_AI_REQUEST_TEMPERATURE: str = "gen_ai.request.temperature"
-GEN_AI_REQUEST_TOP_P: str = "gen_ai.request.top_p"
-GEN_AI_RESPONSE_FINISH_REASONS: str = "gen_ai.response.finish_reasons"
-GEN_AI_USAGE_INPUT_TOKENS: str = "gen_ai.usage.input_tokens"
-GEN_AI_USAGE_OUTPUT_TOKENS: str = "gen_ai.usage.output_tokens"
-
 
 # Get dialect keywords retrieved from dialect_keywords.json file.
 # Only meant to be invoked by SQL_KEYWORD_PATTERN and unit tests

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

@@ -2,10 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import os
-import sys
+from importlib.metadata import PackageNotFoundError, version
 from logging import Logger, getLogger
 
-import pkg_resources
+from packaging.requirements import Requirement
 
 _logger: Logger = getLogger(__name__)
 
@@ -14,15 +14,21 @@
 
 def is_installed(req: str) -> bool:
     """Is the given required package installed?"""
-
-    if req in sys.modules and sys.modules[req] is not None:
-        return True
+    req = Requirement(req)
 
     try:
-        pkg_resources.get_distribution(req)
-    except Exception as exc:  # pylint: disable=broad-except
+        dist_version = version(req.name)
+    except PackageNotFoundError as exc:
         _logger.debug("Skipping instrumentation patch: package %s, exception: %s", req, exc)
         return False
+
+    if not list(req.specifier.filter([dist_version])):
+        _logger.debug(
+            "instrumentation for package %s is available but version %s is installed. Skipping.",
+            req,
+            dist_version,
+        )
+        return False
     return True
 
 

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

@@ -180,11 +180,11 @@ def _init_logging(
     resource: Resource = None,
 ):
 
-    # Provides a default OTLP log exporter when none is specified.
+    # Provides a default OTLP log exporter when it's not set.
     # This is the behavior for the logs exporters for other languages.
-    logs_exporter = os.environ.get("OTEL_LOGS_EXPORTER")
+    logs_exporters = os.environ.get("OTEL_LOGS_EXPORTER")
 
-    if not exporters and logs_exporter and (logs_exporter.lower() != "none"):
+    if not exporters and logs_exporters and logs_exporters.lower() != "none":
         exporters = {"otlp": OTLPLogExporter}
 
     provider = LoggerProvider(resource=resource)

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/logs/aws_batch_log_record_processor.py

@@ -1,13 +1,11 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import logging
-from typing import Mapping, Optional, Sequence, cast
+from typing import List, Mapping, Optional, Sequence, cast
 
 from amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter import OTLPAwsLogExporter
-from opentelemetry.context import (
-    _SUPPRESS_INSTRUMENTATION_KEY,
-    attach,
-    detach,
-    set_value,
-)
+from opentelemetry.context import _SUPPRESS_INSTRUMENTATION_KEY, attach, detach, set_value
 from opentelemetry.sdk._logs import LogData
 from opentelemetry.sdk._logs._internal.export import BatchLogExportStrategy
 from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
@@ -18,7 +16,7 @@
 
 class AwsBatchLogRecordProcessor(BatchLogRecordProcessor):
     _BASE_LOG_BUFFER_BYTE_SIZE = (
-        2000  # Buffer size in bytes to account for log metadata not included in the body size calculation
+        1000  # Buffer size in bytes to account for log metadata not included in the body or attribute size calculation
     )
     _MAX_LOG_REQUEST_BYTE_SIZE = (
         1048576  # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-OTLPEndpoint.html
@@ -66,7 +64,7 @@ def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
 
                     for _ in range(batch_length):
                         log_data: LogData = self._queue.pop()
-                        log_size = self._BASE_LOG_BUFFER_BYTE_SIZE + self._get_any_value_size(log_data.log_record.body)
+                        log_size = self._estimate_log_size(log_data)
 
                         if batch and (batch_data_size + log_size > self._MAX_LOG_REQUEST_BYTE_SIZE):
                             # if batch_data_size > MAX_LOG_REQUEST_BYTE_SIZE then len(batch) == 1
@@ -88,64 +86,74 @@ def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
                                 self._exporter.set_gen_ai_log_flag()
 
                         self._exporter.export(batch)
-                except Exception as e:  # pylint: disable=broad-exception-caught
-                    _logger.exception("Exception while exporting logs: " + str(e))
+                except Exception as exception:  # pylint: disable=broad-exception-caught
+                    _logger.exception("Exception while exporting logs: " + str(exception))
                 detach(token)
 
-    def _get_any_value_size(self, val: AnyValue, depth: int = 3) -> int:
+    def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:
         """
-        Only used to indicate whether we should export a batch log size of 1 or not.
-        Calculates the size in bytes of an AnyValue object.
-        Will processs complex AnyValue structures up to the specified depth limit.
-        If the depth limit of the AnyValue structure is exceeded, returns 0.
+        Estimates the size in bytes of a log by calculating the size of its body and its attributes
+        and adding a buffer amount to account for other log metadata information.
+        Will process complex log structures up to the specified depth limit.
+        If the depth limit of the log structure is exceeded, returns truncates calculation
+        to everything up to that point.
 
         Args:
-            val: The AnyValue object to calculate size for
+            log: The Log object to calculate size for
             depth: Maximum depth to traverse in nested structures (default: 3)
 
         Returns:
-            int: Total size of the AnyValue object in bytes
+            int: The estimated size of the log object in bytes
         """
-        # Use a stack to prevent excessive recursive calls.
-        stack = [(val, 0)]
-        size: int = 0
-
-        while stack:
-            # small optimization. We can stop calculating the size once it reaches the 1 MB limit.
-            if size >= self._MAX_LOG_REQUEST_BYTE_SIZE:
-                return size
-
-            next_val, current_depth = stack.pop()
-
-            if isinstance(next_val, (str, bytes)):
-                size += len(next_val)
-                continue
-
-            if isinstance(next_val, bool):
-                size += 4 if next_val else 5
-                continue
-
-            if isinstance(next_val, (float, int)):
-                size += len(str(next_val))
-                continue
-
-            if current_depth <= depth:
-                if isinstance(next_val, Sequence):
-                    for content in next_val:
-                        stack.append((cast(AnyValue, content), current_depth + 1))
-
-                if isinstance(next_val, Mapping):
-                    for key, content in next_val.items():
-                        size += len(key)
-                        stack.append((content, current_depth + 1))
-            else:
-                _logger.debug("Max log depth exceeded. Log data size will not be accurately calculated.")
-                return 0
+
+        # Use a queue to prevent excessive recursive calls.
+        # We calculate based on the size of the log record body and attributes for the log.
+        queue: List[tuple[AnyValue, int]] = [(log.log_record.body, 0), (log.log_record.attributes, -1)]
+
+        size: int = self._BASE_LOG_BUFFER_BYTE_SIZE
+
+        while queue:
+            new_queue: List[tuple[AnyValue, int]] = []
+
+            for data in queue:
+                # small optimization, can stop calculating the size once it reaches the 1 MB limit.
+                if size >= self._MAX_LOG_REQUEST_BYTE_SIZE:
+                    return size
+
+                next_val, current_depth = data
+
+                if isinstance(next_val, (str, bytes)):
+                    size += len(next_val)
+                    continue
+
+                if isinstance(next_val, bool):
+                    size += 4 if next_val else 5
+                    continue
+
+                if isinstance(next_val, (float, int)):
+                    size += len(str(next_val))
+                    continue
+
+                if current_depth <= depth:
+                    if isinstance(next_val, Sequence):
+                        for content in next_val:
+                            new_queue.append((cast(AnyValue, content), current_depth + 1))
+
+                    if isinstance(next_val, Mapping):
+                        for key, content in next_val.items():
+                            size += len(key)
+                            new_queue.append((content, current_depth + 1))
+                else:
+                    _logger.debug(
+                        f"Max log depth of {depth} exceeded. Log data size will not be accurately calculated."
+                    )
+
+            queue = new_queue
 
         return size
 
     @staticmethod
-    def _is_gen_ai_log(log_data: LogData) -> bool:
+    def _is_gen_ai_log(log: LogData) -> bool:
         """
         Is the log a Gen AI log event?
         """
@@ -157,4 +165,4 @@ def _is_gen_ai_log(log_data: LogData) -> bool:
             "openlit.otel.tracing",
         }
 
-        return log_data.instrumentation_scope.name in gen_ai_instrumentations
+        return log.instrumentation_scope.name in gen_ai_instrumentations