linting

Author: liustve

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

@@ -3,16 +3,15 @@
 # Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 import os
 import re
-from logging import NOTSET, CRITICAL, Logger, getLogger
+from logging import NOTSET, Logger, getLogger
 from typing import ClassVar, Dict, List, Type, Union
 
 from importlib_metadata import version
 from typing_extensions import override
 
 from amazon.opentelemetry.distro._aws_attribute_keys import AWS_LOCAL_SERVICE
-from amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor import AwsBatchLogRecordProcessor
-from amazon.opentelemetry.distro._utils import is_agent_observability_enabled
 from amazon.opentelemetry.distro._aws_resource_attribute_configurator import get_service_attribute
+from amazon.opentelemetry.distro._utils import is_agent_observability_enabled
 from amazon.opentelemetry.distro.always_record_sampler import AlwaysRecordSampler
 from amazon.opentelemetry.distro.attribute_propagating_span_processor_builder import (
     AttributePropagatingSpanProcessorBuilder,
@@ -23,13 +22,14 @@
     AwsMetricAttributesSpanExporterBuilder,
 )
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
+from amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor import AwsBatchLogRecordProcessor
 from amazon.opentelemetry.distro.exporter.otlp.aws.logs.otlp_aws_logs_exporter import OTLPAwsLogExporter
 from amazon.opentelemetry.distro.exporter.otlp.aws.traces.otlp_aws_span_exporter import OTLPAwsSpanExporter
 from amazon.opentelemetry.distro.otlp_udp_exporter import OTLPUdpSpanExporter
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from amazon.opentelemetry.distro.scope_based_exporter import ScopeBasedPeriodicExportingMetricReader
 from amazon.opentelemetry.distro.scope_based_filtering_view import ScopeBasedRetainingView
-from opentelemetry._logs import set_logger_provider, get_logger_provider
+from opentelemetry._logs import get_logger_provider, set_logger_provider
 from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPHttpOTLPMetricExporter
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
@@ -166,7 +166,7 @@ def _initialize_components():
     sampler = _custom_import_sampler(sampler_name, resource)
 
     logging_enabled = os.getenv(_OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED, "false")
-    
+
     if logging_enabled.strip().lower() == "true":
         _init_logging(log_exporters, resource)
 
@@ -177,7 +177,6 @@ def _initialize_components():
         resource=resource,
     )
     _init_metrics(metric_exporters, resource)
-    
 
 
 def _init_logging(
@@ -187,7 +186,7 @@ def _init_logging(
 
     # Provides a default OTLP log exporter when the environment is not set.
     # This is the behavior for the logs exporters for other languages.
-    if not exporters and os.environ.get(OTEL_LOGS_EXPORTER) == None:
+    if not exporters and os.environ.get(OTEL_LOGS_EXPORTER) is None:
         exporters = {"otlp": OTLPLogExporter}
 
     provider = LoggerProvider(resource=resource)
@@ -373,10 +372,7 @@ def _customize_span_exporter(span_exporter: SpanExporter, resource: Resource) ->
         if isinstance(span_exporter, OTLPSpanExporter):
             if is_agent_observability_enabled():
 
-                span_exporter = OTLPAwsSpanExporter(
-                    endpoint=traces_endpoint,
-                    logs_provider=get_logger_provider()
-                )
+                span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint, logs_provider=get_logger_provider())
             else:
                 span_exporter = OTLPAwsSpanExporter(endpoint=traces_endpoint)
 
@@ -643,4 +639,4 @@ def create_exporter(self):
                 endpoint=application_signals_endpoint, preferred_temporality=temporality_dict
             )
 
-        raise RuntimeError(f"Unsupported AWS Application Signals export protocol: {protocol} ")
+        raise RuntimeError(f"Unsupported AWS Application Signals export protocol: {protocol} ")

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

@@ -1,8 +1,4 @@
-import json
 import logging
-import os
-import threading
-from time import sleep
 from typing import Mapping, Sequence
 
 from amazon.opentelemetry.distro.exporter.otlp.aws.common.constants import (
@@ -48,8 +44,8 @@ def __init__(
     def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
         """
         Overrides the batching behavior of upstream's export method. Preserves existing batching behavior but
-        will intermediarly export small log batches if the size of the data in the batch is at or above AWS CloudWatch's maximum request size limit
-        of 1 MB.
+        will intermediarly export small log batches if the size of the data in the batch is at or
+        above AWS CloudWatch's maximum request size limit of 1 MB.
 
         - Data size of exported batches will ALWAYS be <= 1 MB except for the case below:
         - If the data size of an exported batch is ever > 1 MB then the batch size is guaranteed to be 1
@@ -97,8 +93,8 @@ def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
 
     def _get_size_of_log(self, log_data: LogData) -> int:
         """
-        Estimates the size of a given LogData based on the size of the body + a buffer amount representing a rough guess of other data present
-        in the log.
+        Estimates the size of a given LogData based on the size of the body + a buffer
+        amount representing a rough guess of other data present in the log.
         """
         size = BASE_LOG_BUFFER_BYTE_SIZE
         body = log_data.log_record.body

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

@@ -2,11 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import gzip
-import itertools
 import logging
 from io import BytesIO
 from time import sleep
-from typing import Dict, Iterator, Mapping, Optional, Sequence
+from typing import Dict, Mapping, Optional, Sequence
 
 import requests
 
@@ -26,6 +25,7 @@
 
 
 class OTLPAwsLogExporter(OTLPLogExporter):
+    COUNT = 0
     _LARGE_LOG_HEADER = {"x-aws-log-semantics": "otel"}
     _RETRY_AFTER_HEADER = "Retry-After"  # https://opentelemetry.io/docs/specs/otlp/#otlphttp-throttling
 
@@ -62,7 +62,8 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
 
         print(f"Exporting batch of {len(batch)} logs")
         print("TOTAL DATA SIZE " + str(sum(self._get_size_of_log(logz) for logz in batch)))
-        print("GEN_AI_FLAG " + str(self._gen_ai_flag))
+        self.COUNT += len(batch)
+        print("COUNT " + str(self.COUNT))
 
         """
         Exports the given batch of OTLP log data.
@@ -74,8 +75,8 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
             - inject the 'x-aws-log-semantics' flag into the header.
 
         3. Retry behavior is now the following:
-            - if the response contains a status code that is retryable and the response contains Retry-After in its headers,
-              the serialized data will be exported after that set delay
+            - if the response contains a status code that is retryable and the response contains Retry-After in its
+              headers, the serialized data will be exported after that set delay
 
             - if the response does not contain that Retry-After header, default back to the current iteration of the
               exponential backoff delay
@@ -93,11 +94,18 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
 
         data = gzip_data.getvalue()
 
-        backoff = _create_exp_backoff_generator(self._MAX_RETRY_TIMEOUT)
+        backoff = _create_exp_backoff_generator(max_value=self._MAX_RETRY_TIMEOUT)
 
         while True:
             resp = self._send(data)
 
+            print(f"Response status: {resp.status_code}")
+            print(f"Response headers: {resp.headers}")
+            try:
+                print(f"Response body: {resp.text}")
+            except:
+                print("Could not print response body")
+
             if resp.ok:
                 return LogExportResult.SUCCESS
 
@@ -111,13 +119,18 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
                 return LogExportResult.FAILURE
 
             # https://opentelemetry.io/docs/specs/otlp/#otlphttp-throttling
-            retry_after = resp.headers.get(self._RETRY_AFTER_HEADER, None)
+            maybe_retry_after = resp.headers.get(self._RETRY_AFTER_HEADER, None)
 
             # Set the next retry delay to the value of the Retry-After response in the headers.
-            # If Retry-After is not present in the headers, default to the next iteration of the exponential backoff strategy.
-            delay = next(backoff, -1) if retry_after == None else self._parse_retryable_header(retry_after)
+            # If Retry-After is not present in the headers, default to the next iteration of the
+            # exponential backoff strategy.
+
+            delay = self._parse_retryable_header(maybe_retry_after)
 
             if delay == -1:
+                delay = next(backoff, self._MAX_RETRY_TIMEOUT)
+
+            if delay == self._MAX_RETRY_TIMEOUT:
                 _logger.error(
                     "Transient error %s encountered while exporting logs batch. "
                     "No Retry-After header found and all backoff retries exhausted. "
@@ -134,7 +147,7 @@ def export(self, batch: Sequence[LogData]) -> LogExportResult:
             )
 
             sleep(delay)
-    
+
     def set_gen_ai_flag(self):
         """
         Sets the gen_ai flag to true to signal injecting the LLO flag to the headers of the export request.
@@ -171,10 +184,15 @@ def _retryable(resp: requests.Response) -> bool:
 
         return OTLPLogExporter._retryable(resp)
 
-    def _parse_retryable_header(self, retry_header: str) -> float:
+    def _parse_retryable_header(self, retry_header: Optional[str]) -> float:
         """
-        Converts the given retryable header into a delay in seconds, returns -1 if there's an error with the parsing
+        Converts the given retryable header into a delay in seconds, returns -1 if there's no header
+        or error with the parsing
         """
+
+        if not retry_header:
+            return -1
+
         try:
             return float(retry_header)
         except ValueError: