11# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
22# SPDX-License-Identifier: Apache-2.0
33
4- from typing import Dict , Optional
4+ import gzip
5+ import logging
6+ from io import BytesIO
7+ from time import sleep
8+ from typing import Dict , Optional , Sequence
9+
10+ import requests
511
612from amazon .opentelemetry .distro .exporter .otlp .aws .common .aws_auth_session import AwsAuthSession
13+ from opentelemetry .exporter .otlp .proto .common ._log_encoder import encode_logs
714from opentelemetry .exporter .otlp .proto .http import Compression
8- from opentelemetry .exporter .otlp .proto .http ._log_exporter import OTLPLogExporter
15+ from opentelemetry .exporter .otlp .proto .http ._log_exporter import OTLPLogExporter , _create_exp_backoff_generator
16+ from opentelemetry .sdk ._logs import (
17+ LogData ,
18+ )
19+ from opentelemetry .sdk ._logs .export import (
20+ LogExportResult ,
21+ )
22+
23+ _logger = logging .getLogger (__name__ )
924
1025
1126class OTLPAwsLogExporter (OTLPLogExporter ):
27+ _LARGE_LOG_HEADER = {"x-aws-log-semantics" : "otel" }
28+ _RETRY_AFTER_HEADER = "Retry-After" # https://opentelemetry.io/docs/specs/otlp/#otlphttp-throttling
29+
1230 def __init__ (
1331 self ,
1432 endpoint : Optional [str ] = None ,
@@ -18,6 +36,7 @@ def __init__(
1836 headers : Optional [Dict [str , str ]] = None ,
1937 timeout : Optional [int ] = None ,
2038 ):
39+ self ._gen_ai_flag = False
2140 self ._aws_region = None
2241
2342 if endpoint :
@@ -34,3 +53,133 @@ def __init__(
3453 compression = Compression .Gzip ,
3554 session = AwsAuthSession (aws_region = self ._aws_region , service = "logs" ),
3655 )
56+
57+ # https://github.com/open-telemetry/opentelemetry-python/blob/main/exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py#L167
58+ def export (self , batch : Sequence [LogData ]) -> LogExportResult :
59+ """
60+ Exports the given batch of OTLP log data.
61+ Behaviors of how this export will work -
62+
63+ 1. Always compresses the serialized data into gzip before sending.
64+
65+ 2. If self._gen_ai_flag is enabled, the log data is > 1 MB a
66+ and the assumption is that the log is a normalized gen.ai LogEvent.
67+ - inject the 'x-aws-log-semantics' flag into the header.
68+
69+ 3. Retry behavior is now the following:
70+ - if the response contains a status code that is retryable and the response contains Retry-After in its
71+ headers, the serialized data will be exported after that set delay
72+
73+ - if the response does not contain that Retry-After header, default back to the current iteration of the
74+ exponential backoff delay
75+ """
76+
77+ if self ._shutdown :
78+ _logger .warning ("Exporter already shutdown, ignoring batch" )
79+ return LogExportResult .FAILURE
80+
81+ serialized_data = encode_logs (batch ).SerializeToString ()
82+
83+ gzip_data = BytesIO ()
84+ with gzip .GzipFile (fileobj = gzip_data , mode = "w" ) as gzip_stream :
85+ gzip_stream .write (serialized_data )
86+
87+ data = gzip_data .getvalue ()
88+
89+ backoff = _create_exp_backoff_generator (max_value = self ._MAX_RETRY_TIMEOUT )
90+
91+ while True :
92+ resp = self ._send (data )
93+
94+ if resp .ok :
95+ return LogExportResult .SUCCESS
96+
97+ if not self ._retryable (resp ):
98+ _logger .error (
99+ "Failed to export logs batch code: %s, reason: %s" ,
100+ resp .status_code ,
101+ resp .text ,
102+ )
103+ self ._gen_ai_flag = False
104+ return LogExportResult .FAILURE
105+
106+ # https://opentelemetry.io/docs/specs/otlp/#otlphttp-throttling
107+ maybe_retry_after = resp .headers .get (self ._RETRY_AFTER_HEADER , None )
108+
109+ # Set the next retry delay to the value of the Retry-After response in the headers.
110+ # If Retry-After is not present in the headers, default to the next iteration of the
111+ # exponential backoff strategy.
112+
113+ delay = self ._parse_retryable_header (maybe_retry_after )
114+
115+ if delay == - 1 :
116+ delay = next (backoff , self ._MAX_RETRY_TIMEOUT )
117+
118+ if delay == self ._MAX_RETRY_TIMEOUT :
119+ _logger .error (
120+ "Transient error %s encountered while exporting logs batch. "
121+ "No Retry-After header found and all backoff retries exhausted. "
122+ "Logs will not be exported." ,
123+ resp .reason ,
124+ )
125+ self ._gen_ai_flag = False
126+ return LogExportResult .FAILURE
127+
128+ _logger .warning (
129+ "Transient error %s encountered while exporting logs batch, retrying in %ss." ,
130+ resp .reason ,
131+ delay ,
132+ )
133+
134+ sleep (delay )
135+
136+ def set_gen_ai_flag (self ):
137+ """
138+ Sets the gen_ai flag to true to signal injecting the LLO flag to the headers of the export request.
139+ """
140+ self ._gen_ai_flag = True
141+
142+ def _send (self , serialized_data : bytes ):
143+ try :
144+ return self ._session .post (
145+ url = self ._endpoint ,
146+ headers = self ._LARGE_LOG_HEADER if self ._gen_ai_flag else None ,
147+ data = serialized_data ,
148+ verify = self ._certificate_file ,
149+ timeout = self ._timeout ,
150+ cert = self ._client_cert ,
151+ )
152+ except ConnectionError :
153+ return self ._session .post (
154+ url = self ._endpoint ,
155+ headers = self ._LARGE_LOG_HEADER if self ._gen_ai_flag else None ,
156+ data = serialized_data ,
157+ verify = self ._certificate_file ,
158+ timeout = self ._timeout ,
159+ cert = self ._client_cert ,
160+ )
161+
162+ @staticmethod
163+ def _retryable (resp : requests .Response ) -> bool :
164+ """
165+ Is it a retryable response?
166+ """
167+ if resp .status_code in (429 , 503 ):
168+ return True
169+
170+ return OTLPLogExporter ._retryable (resp )
171+
172+ @staticmethod
173+ def _parse_retryable_header (retry_header : Optional [str ]) -> float :
174+ """
175+ Converts the given retryable header into a delay in seconds, returns -1 if there's no header
176+ or error with the parsing
177+ """
178+ if not retry_header :
179+ return - 1
180+
181+ try :
182+ val = float (retry_header )
183+ return val if val >= 0 else - 1
184+ except ValueError :
185+ return - 1
0 commit comments