From 33d98b0cc3215f983219595bf9f60150180bd690 Mon Sep 17 00:00:00 2001 From: Phillip Verheyden Date: Wed, 22 Dec 2021 12:28:14 -0600 Subject: [PATCH] Refactor nested inner functions out for Pylint failure for too many local variables --- CHANGELOG.md | 10 +- .../instrumentation/asgi/__init__.py | 99 +++++++++++-------- 2 files changed, 62 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3806b1b2db..c543bf5764 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased](https://github.com/open-telemetry/opentelemetry-python/compare/v1.8.0-0.27b0...HEAD) +### Added + +- `opentelemetry-instrumentation-asgi` now returns a `traceresponse` response header. + ([#817](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/817)) + ### Fixed - `opentelemetry-instrumentation-flask` Flask: Conditionally create SERVER spans @@ -18,11 +23,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `opentelemetry-instrumentation-aws-lambda` Adds support for configurable flush timeout via `OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT` property. ([#825](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/825)) -### Added - -`opentelemetry-instrumenation-asgi` now returns a `traceresponse` response header. - ([#817](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/817)) - ### Fixed - `opentelemetry-exporter-richconsole` Fixed attribute error on parentless spans. diff --git a/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py b/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py index df16268f51..5fb4914a04 100644 --- a/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py @@ -338,49 +338,64 @@ async def __call__(self, scope, receive, send): if callable(self.server_request_hook): self.server_request_hook(span, scope) + otel_receive = self._get_otel_receive( + span_name, scope, receive + ) + server_span_context = trace.context_api.get_current() + otel_send = self._get_otel_send( + span, span_name, server_span_context, scope, send + ) - @wraps(receive) - async def wrapped_receive(): - with self.tracer.start_as_current_span( - " ".join((span_name, scope["type"], "receive")) - ) as receive_span: - if callable(self.client_request_hook): - self.client_request_hook(receive_span, scope) - message = await receive() - if receive_span.is_recording(): - if message["type"] == "websocket.receive": - set_status_code(receive_span, 200) - receive_span.set_attribute("type", message["type"]) - return message - - @wraps(send) - async def wrapped_send(message): - with self.tracer.start_as_current_span( - " ".join((span_name, scope["type"], "send")) - ) as send_span: - if callable(self.client_response_hook): - self.client_response_hook(send_span, message) - if send_span.is_recording(): - if message["type"] == "http.response.start": - status_code = message["status"] - set_status_code(span, status_code) - set_status_code(send_span, status_code) - elif message["type"] == "websocket.send": - set_status_code(span, 200) - set_status_code(send_span, 200) - send_span.set_attribute("type", message["type"]) - - propagator = get_global_response_propagator() - if propagator: - propagator.inject( - message, - context=server_span_context, - setter=asgi_setter, - ) - - await send(message) - - await self.app(scope, wrapped_receive, wrapped_send) + await self.app(scope, otel_receive, otel_send) finally: context.detach(token) + + def _get_otel_receive(self, span_name, scope, receive): + @wraps(receive) + async def wrapped_receive(): + with self.tracer.start_as_current_span( + " ".join((span_name, scope["type"], "receive")) + ) as receive_span: + if callable(self.client_request_hook): + self.client_request_hook(receive_span, scope) + message = await receive() + if receive_span.is_recording(): + if message["type"] == "websocket.receive": + set_status_code(receive_span, 200) + receive_span.set_attribute("type", message["type"]) + return message + + return wrapped_receive + + async def _get_otel_send( + self, span, span_name, server_span_context, scope, send + ): + @wraps(send) + async def wrapped_send(message): + with self.tracer.start_as_current_span( + " ".join((span_name, scope["type"], "send")) + ) as send_span: + if callable(self.client_response_hook): + self.client_response_hook(send_span, message) + if send_span.is_recording(): + if message["type"] == "http.response.start": + status_code = message["status"] + set_status_code(span, status_code) + set_status_code(send_span, status_code) + elif message["type"] == "websocket.send": + set_status_code(span, 200) + set_status_code(send_span, 200) + send_span.set_attribute("type", message["type"]) + + propagator = get_global_response_propagator() + if propagator: + propagator.inject( + message, + context=server_span_context, + setter=asgi_setter, + ) + + await send(message) + + return wrapped_send