Skip to content

Commit f1dda55

Browse files
artemziborevartemziborev
authored
Merge branch 'main' into refactor-flask-spanattributes
2 parents 7fbec0f + 4e42ed6 commit f1dda55

File tree

16 files changed

+664
-51
lines changed

16 files changed

+664
-51
lines changed

‎CHANGELOG.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
1313

1414
### Fixed
1515

16+
- `opentelemetry-instrumentation-system-metrics`: fix loading on Google Cloud Run
17+
([#3533](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3533))
1618
- `opentelemetry-instrumentation-fastapi`: fix wrapping of middlewares
1719
([#3012](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3012))
20+
- `opentelemetry-instrumentation-urllib3`: proper bucket boundaries in stable semconv http duration metrics
21+
([#3518](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3518))
22+
- `opentelemetry-instrumentation-urllib`: proper bucket boundaries in stable semconv http duration metrics
23+
([#3519](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3519))
1824
- `opentelemetry-instrumentation-falcon`: proper bucket boundaries in stable semconv http duration
1925
([#3525](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3525))
2026
- `opentelemetry-instrumentation-wsgi`: add explicit http duration buckets for stable semconv
@@ -27,6 +33,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
2733
([#3524](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3524))
2834
- `opentelemetry-instrumentation-grpc`: support non-list interceptors
2935
([#3520](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3520))
36+
- `opentelemetry-instrumentation-botocore` Ensure spans end on early stream closure for Bedrock Streaming APIs
37+
([#3481](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3481))
3038

3139
### Breaking changes
3240

‎instrumentation/opentelemetry-instrumentation-boto/src/opentelemetry/instrumentation/boto/__init__.py‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,10 @@
5252
from opentelemetry.instrumentation.boto.version import __version__
5353
from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
5454
from opentelemetry.instrumentation.utils import unwrap
55-
from opentelemetry.semconv.trace import SpanAttributes
55+
from opentelemetry.semconv._incubating.attributes.http_attributes import (
56+
HTTP_METHOD,
57+
HTTP_STATUS_CODE,
58+
)
5659
from opentelemetry.trace import SpanKind, get_tracer
5760

5861
logger = logging.getLogger(__name__)
@@ -158,12 +161,8 @@ def _common_request( # pylint: disable=too-many-locals
158161
for key, value in meta.items():
159162
span.set_attribute(key, value)
160163

161-
span.set_attribute(
162-
SpanAttributes.HTTP_STATUS_CODE, getattr(result, "status")
163-
)
164-
span.set_attribute(
165-
SpanAttributes.HTTP_METHOD, getattr(result, "_method")
166-
)
164+
span.set_attribute(HTTP_STATUS_CODE, getattr(result, "status"))
165+
span.set_attribute(HTTP_METHOD, getattr(result, "_method"))
167166

168167
return result
169168

‎instrumentation/opentelemetry-instrumentation-boto/tests/test_boto_instrumentation.py‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -28,13 +28,16 @@
2828
)
2929

3030
from opentelemetry.instrumentation.boto import BotoInstrumentor
31-
from opentelemetry.semconv.trace import SpanAttributes
31+
from opentelemetry.semconv._incubating.attributes.http_attributes import (
32+
HTTP_METHOD,
33+
HTTP_STATUS_CODE,
34+
)
3235
from opentelemetry.test.test_base import TestBase
3336

3437

3538
def assert_span_http_status_code(span, code):
3639
"""Assert on the span's 'http.status_code' tag"""
37-
tag = span.attributes[SpanAttributes.HTTP_STATUS_CODE]
40+
tag = span.attributes[HTTP_STATUS_CODE]
3841
assert tag == code, f"{tag} != {code}"
3942

4043

@@ -60,7 +63,7 @@ def test_ec2_client(self):
6063
span = spans[0]
6164
self.assertEqual(span.attributes["aws.operation"], "DescribeInstances")
6265
assert_span_http_status_code(span, 200)
63-
self.assertEqual(span.attributes[SpanAttributes.HTTP_METHOD], "POST")
66+
self.assertEqual(span.attributes[HTTP_METHOD], "POST")
6467
self.assertEqual(span.attributes["aws.region"], "us-west-2")
6568

6669
# Create an instance
@@ -73,7 +76,7 @@ def test_ec2_client(self):
7376
assert_span_http_status_code(span, 200)
7477
self.assertEqual(span.attributes["endpoint"], "ec2")
7578
self.assertEqual(span.attributes["http_method"], "runinstances")
76-
self.assertEqual(span.attributes[SpanAttributes.HTTP_METHOD], "POST")
79+
self.assertEqual(span.attributes[HTTP_METHOD], "POST")
7780
self.assertEqual(span.attributes["aws.region"], "us-west-2")
7881
self.assertEqual(span.name, "ec2.command")
7982

@@ -120,7 +123,7 @@ def test_s3_client(self):
120123
self.assertEqual(len(spans), 1)
121124
span = spans[0]
122125
assert_span_http_status_code(span, 200)
123-
self.assertEqual(span.attributes[SpanAttributes.HTTP_METHOD], "GET")
126+
self.assertEqual(span.attributes[HTTP_METHOD], "GET")
124127
self.assertEqual(span.attributes["aws.operation"], "get_all_buckets")
125128

126129
# Create a bucket command
@@ -130,7 +133,7 @@ def test_s3_client(self):
130133
self.assertEqual(len(spans), 2)
131134
span = spans[1]
132135
assert_span_http_status_code(span, 200)
133-
self.assertEqual(span.attributes[SpanAttributes.HTTP_METHOD], "PUT")
136+
self.assertEqual(span.attributes[HTTP_METHOD], "PUT")
134137
self.assertEqual(span.attributes["path"], "/")
135138
self.assertEqual(span.attributes["aws.operation"], "create_bucket")
136139

@@ -143,7 +146,7 @@ def test_s3_client(self):
143146
assert_span_http_status_code(span, 200)
144147
self.assertEqual(span.attributes["endpoint"], "s3")
145148
self.assertEqual(span.attributes["http_method"], "head")
146-
self.assertEqual(span.attributes[SpanAttributes.HTTP_METHOD], "HEAD")
149+
self.assertEqual(span.attributes[HTTP_METHOD], "HEAD")
147150
self.assertEqual(span.attributes["aws.operation"], "head_bucket")
148151
self.assertEqual(span.name, "s3.command")
149152

@@ -240,7 +243,7 @@ def test_lambda_client(self):
240243
assert_span_http_status_code(span, 200)
241244
self.assertEqual(span.attributes["endpoint"], "lambda")
242245
self.assertEqual(span.attributes["http_method"], "get")
243-
self.assertEqual(span.attributes[SpanAttributes.HTTP_METHOD], "GET")
246+
self.assertEqual(span.attributes[HTTP_METHOD], "GET")
244247
self.assertEqual(span.attributes["aws.region"], "us-east-2")
245248
self.assertEqual(span.attributes["aws.operation"], "list_functions")
246249

‎instrumentation/opentelemetry-instrumentation-botocore/src/opentelemetry/instrumentation/botocore/extensions/bedrock.py‎

Lines changed: 38 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -499,18 +499,20 @@ def _converse_on_success(
499499
[stop_reason],
500500
)
501501

502-
event_logger = instrumentor_context.event_logger
503-
choice = _Choice.from_converse(result, capture_content)
504-
# this path is used by streaming apis, in that case we are already out of the span
505-
# context so need to add the span context manually
506-
span_ctx = span.get_span_context()
507-
event_logger.emit(
508-
choice.to_choice_event(
509-
trace_id=span_ctx.trace_id,
510-
span_id=span_ctx.span_id,
511-
trace_flags=span_ctx.trace_flags,
502+
# In case of an early stream closure, the result may not contain outputs
503+
if self._stream_has_output_content(result):
504+
event_logger = instrumentor_context.event_logger
505+
choice = _Choice.from_converse(result, capture_content)
506+
# this path is used by streaming apis, in that case we are already out of the span
507+
# context so need to add the span context manually
508+
span_ctx = span.get_span_context()
509+
event_logger.emit(
510+
choice.to_choice_event(
511+
trace_id=span_ctx.trace_id,
512+
span_id=span_ctx.span_id,
513+
trace_flags=span_ctx.trace_flags,
514+
)
512515
)
513-
)
514516

515517
metrics = instrumentor_context.metrics
516518
metrics_attributes = self._extract_metrics_attributes()
@@ -602,11 +604,14 @@ def _on_stream_error_callback(
602604
span: Span,
603605
exception,
604606
instrumentor_context: _BotocoreInstrumentorContext,
607+
span_ended: bool,
605608
):
606609
span.set_status(Status(StatusCode.ERROR, str(exception)))
607610
if span.is_recording():
608611
span.set_attribute(ERROR_TYPE, type(exception).__qualname__)
609-
span.end()
612+
613+
if not span_ended:
614+
span.end()
610615

611616
metrics = instrumentor_context.metrics
612617
metrics_attributes = {
@@ -638,15 +643,17 @@ def on_success(
638643
result["stream"], EventStream
639644
):
640645

641-
def stream_done_callback(response):
646+
def stream_done_callback(response, span_ended):
642647
self._converse_on_success(
643648
span, response, instrumentor_context, capture_content
644649
)
645-
span.end()
646650

647-
def stream_error_callback(exception):
651+
if not span_ended:
652+
span.end()
653+
654+
def stream_error_callback(exception, span_ended):
648655
self._on_stream_error_callback(
649-
span, exception, instrumentor_context
656+
span, exception, instrumentor_context, span_ended
650657
)
651658

652659
result["stream"] = ConverseStreamWrapper(
@@ -677,16 +684,17 @@ def stream_error_callback(exception):
677684
elif self._call_context.operation == "InvokeModelWithResponseStream":
678685
if "body" in result and isinstance(result["body"], EventStream):
679686

680-
def invoke_model_stream_done_callback(response):
687+
def invoke_model_stream_done_callback(response, span_ended):
681688
# the callback gets data formatted as the simpler converse API
682689
self._converse_on_success(
683690
span, response, instrumentor_context, capture_content
684691
)
685-
span.end()
692+
if not span_ended:
693+
span.end()
686694

687-
def invoke_model_stream_error_callback(exception):
695+
def invoke_model_stream_error_callback(exception, span_ended):
688696
self._on_stream_error_callback(
689-
span, exception, instrumentor_context
697+
span, exception, instrumentor_context, span_ended
690698
)
691699

692700
result["body"] = InvokeModelWithResponseStreamWrapper(
@@ -781,9 +789,11 @@ def _handle_amazon_nova_response(
781789
GEN_AI_RESPONSE_FINISH_REASONS, [response_body["stopReason"]]
782790
)
783791

784-
event_logger = instrumentor_context.event_logger
785-
choice = _Choice.from_converse(response_body, capture_content)
786-
event_logger.emit(choice.to_choice_event())
792+
# In case of an early stream closure, the result may not contain outputs
793+
if self._stream_has_output_content(response_body):
794+
event_logger = instrumentor_context.event_logger
795+
choice = _Choice.from_converse(response_body, capture_content)
796+
event_logger.emit(choice.to_choice_event())
787797

788798
metrics = instrumentor_context.metrics
789799
metrics_attributes = self._extract_metrics_attributes()
@@ -1004,3 +1014,8 @@ def on_error(
10041014
duration,
10051015
attributes=metrics_attributes,
10061016
)
1017+
1018+
def _stream_has_output_content(self, response_body: dict[str, Any]):
1019+
return (
1020+
"output" in response_body and "message" in response_body["output"]
1021+
)

‎instrumentation/opentelemetry-instrumentation-botocore/src/opentelemetry/instrumentation/botocore/extensions/bedrock_utils.py‎

Lines changed: 34 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -65,14 +65,15 @@ def __init__(
6565
self._message = None
6666
self._content_block = {}
6767
self._record_message = False
68+
self._ended = False
6869

6970
def __iter__(self):
7071
try:
7172
for event in self.__wrapped__:
7273
self._process_event(event)
7374
yield event
7475
except EventStreamError as exc:
75-
self._stream_error_callback(exc)
76+
self._handle_stream_error(exc)
7677
raise
7778

7879
def _process_event(self, event):
@@ -133,11 +134,23 @@ def _process_event(self, event):
133134

134135
if output_tokens := usage.get("outputTokens"):
135136
self._response["usage"]["outputTokens"] = output_tokens
136-
137-
self._stream_done_callback(self._response)
137+
self._complete_stream(self._response)
138138

139139
return
140140

141+
def close(self):
142+
self.__wrapped__.close()
143+
# Treat the stream as done to ensure the span end.
144+
self._complete_stream(self._response)
145+
146+
def _complete_stream(self, response):
147+
self._stream_done_callback(response, self._ended)
148+
self._ended = True
149+
150+
def _handle_stream_error(self, exc):
151+
self._stream_error_callback(exc, self._ended)
152+
self._ended = True
153+
141154

142155
# pylint: disable=abstract-method
143156
class InvokeModelWithResponseStreamWrapper(ObjectProxy):
@@ -163,14 +176,28 @@ def __init__(
163176
self._content_block = {}
164177
self._tool_json_input_buf = ""
165178
self._record_message = False
179+
self._ended = False
180+
181+
def close(self):
182+
self.__wrapped__.close()
183+
# Treat the stream as done to ensure the span end.
184+
self._stream_done_callback(self._response, self._ended)
185+
186+
def _complete_stream(self, response):
187+
self._stream_done_callback(response, self._ended)
188+
self._ended = True
189+
190+
def _handle_stream_error(self, exc):
191+
self._stream_error_callback(exc, self._ended)
192+
self._ended = True
166193

167194
def __iter__(self):
168195
try:
169196
for event in self.__wrapped__:
170197
self._process_event(event)
171198
yield event
172199
except EventStreamError as exc:
173-
self._stream_error_callback(exc)
200+
self._handle_stream_error(exc)
174201
raise
175202

176203
def _process_event(self, event):
@@ -213,7 +240,7 @@ def _process_amazon_titan_chunk(self, chunk):
213240
self._response["output"] = {
214241
"message": {"content": [{"text": chunk["outputText"]}]}
215242
}
216-
self._stream_done_callback(self._response)
243+
self._complete_stream(self._response)
217244

218245
def _process_amazon_nova_chunk(self, chunk):
219246
# pylint: disable=too-many-branches
@@ -283,7 +310,7 @@ def _process_amazon_nova_chunk(self, chunk):
283310
if output_tokens := usage.get("outputTokens"):
284311
self._response["usage"]["outputTokens"] = output_tokens
285312

286-
self._stream_done_callback(self._response)
313+
self._complete_stream(self._response)
287314
return
288315

289316
def _process_anthropic_claude_chunk(self, chunk):
@@ -355,7 +382,7 @@ def _process_anthropic_claude_chunk(self, chunk):
355382
self._record_message = False
356383
self._message = None
357384

358-
self._stream_done_callback(self._response)
385+
self._complete_stream(self._response)
359386
return
360387

361388

0 commit comments

Comments
 (0)