Skip to content

Commit 3a2706b

Browse files
xrmxxrmx
committed
Reduce use of shared state in the instrumentor class
1 parent 986f3db commit 3a2706b

File tree

1 file changed

+103
-58
lines changed
  • instrumentation/opentelemetry-instrumentation-httpx/src/opentelemetry/instrumentation/httpx

1 file changed

+103
-58
lines changed

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

Lines changed: 103 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -194,6 +194,7 @@ async def async_response_hook(span, request, response):
194194
import logging
195195
import typing
196196
from asyncio import iscoroutinefunction
197+
from functools import partial
197198
from types import TracebackType
198199

199200
import httpx
@@ -734,45 +735,53 @@ def _instrument(self, **kwargs):
734735
``async_response_hook``: Async``response_hook`` for ``httpx.AsyncClient``
735736
"""
736737
tracer_provider = kwargs.get("tracer_provider")
737-
_request_hook = kwargs.get("request_hook")
738-
self._request_hook = _request_hook if callable(_request_hook) else None
739-
_response_hook = kwargs.get("response_hook")
740-
self._response_hook = (
741-
_response_hook if callable(_response_hook) else None
742-
)
743-
_async_request_hook = kwargs.get("async_request_hook")
744-
self._async_request_hook = (
745-
_async_request_hook
746-
if iscoroutinefunction(_async_request_hook)
738+
request_hook = kwargs.get("request_hook")
739+
response_hook = kwargs.get("response_hook")
740+
async_request_hook = kwargs.get("async_request_hook")
741+
async_request_hook = (
742+
async_request_hook
743+
if iscoroutinefunction(async_request_hook)
747744
else None
748745
)
749-
_async_response_hook = kwargs.get("async_response_hook")
750-
self._async_response_hook = (
751-
_async_response_hook
752-
if iscoroutinefunction(_async_response_hook)
746+
async_response_hook = kwargs.get("async_response_hook")
747+
async_response_hook = (
748+
async_response_hook
749+
if iscoroutinefunction(async_response_hook)
753750
else None
754751
)
755752

756753
_OpenTelemetrySemanticConventionStability._initialize()
757-
self._sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
754+
sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
758755
_OpenTelemetryStabilitySignalType.HTTP,
759756
)
760-
self._tracer = get_tracer(
757+
tracer = get_tracer(
761758
__name__,
762759
instrumenting_library_version=__version__,
763760
tracer_provider=tracer_provider,
764-
schema_url=_get_schema_url(self._sem_conv_opt_in_mode),
761+
schema_url=_get_schema_url(sem_conv_opt_in_mode),
765762
)
766763

767764
wrap_function_wrapper(
768765
"httpx",
769766
"HTTPTransport.handle_request",
770-
self._handle_request_wrapper,
767+
partial(
768+
self._handle_request_wrapper,
769+
tracer=tracer,
770+
sem_conv_opt_in_mode=sem_conv_opt_in_mode,
771+
request_hook=request_hook,
772+
response_hook=response_hook,
773+
),
771774
)
772775
wrap_function_wrapper(
773776
"httpx",
774777
"AsyncHTTPTransport.handle_async_request",
775-
self._handle_async_request_wrapper,
778+
partial(
779+
self._handle_async_request_wrapper,
780+
tracer=tracer,
781+
sem_conv_opt_in_mode=sem_conv_opt_in_mode,
782+
async_request_hook=async_request_hook,
783+
async_response_hook=async_response_hook,
784+
),
776785
)
777786

778787
def _uninstrument(self, **kwargs):
@@ -781,7 +790,17 @@ def _uninstrument(self, **kwargs):
781790
unwrap(httpx.HTTPTransport, "handle_request")
782791
unwrap(httpx.AsyncHTTPTransport, "handle_async_request")
783792

784-
def _handle_request_wrapper(self, wrapped, instance, args, kwargs):
793+
def _handle_request_wrapper(
794+
self,
795+
wrapped,
796+
instance,
797+
args,
798+
kwargs,
799+
tracer,
800+
sem_conv_opt_in_mode,
801+
request_hook,
802+
response_hook,
803+
):
785804
if not is_http_instrumentation_enabled():
786805
return wrapped(*args, **kwargs)
787806

@@ -796,17 +815,17 @@ def _handle_request_wrapper(self, wrapped, instance, args, kwargs):
796815
span_attributes,
797816
url,
798817
method_original,
799-
self._sem_conv_opt_in_mode,
818+
sem_conv_opt_in_mode,
800819
)
801820

802821
request_info = RequestInfo(method, url, headers, stream, extensions)
803822

804-
with self._tracer.start_as_current_span(
823+
with tracer.start_as_current_span(
805824
span_name, kind=SpanKind.CLIENT, attributes=span_attributes
806825
) as span:
807826
exception = None
808-
if callable(self._request_hook):
809-
self._request_hook(span, request_info)
827+
if callable(request_hook):
828+
request_hook(span, request_info)
810829

811830
_inject_propagation_headers(headers, args, kwargs)
812831

@@ -827,19 +846,17 @@ def _handle_request_wrapper(self, wrapped, instance, args, kwargs):
827846
span,
828847
status_code,
829848
http_version,
830-
self._sem_conv_opt_in_mode,
849+
sem_conv_opt_in_mode,
831850
)
832-
if callable(self._response_hook):
833-
self._response_hook(
851+
if callable(response_hook):
852+
response_hook(
834853
span,
835854
request_info,
836855
ResponseInfo(status_code, headers, stream, extensions),
837856
)
838857

839858
if exception:
840-
if span.is_recording() and _report_new(
841-
self._sem_conv_opt_in_mode
842-
):
859+
if span.is_recording() and _report_new(sem_conv_opt_in_mode):
843860
span.set_attribute(
844861
ERROR_TYPE, type(exception).__qualname__
845862
)
@@ -848,7 +865,15 @@ def _handle_request_wrapper(self, wrapped, instance, args, kwargs):
848865
return response
849866

850867
async def _handle_async_request_wrapper(
851-
self, wrapped, instance, args, kwargs
868+
self,
869+
wrapped,
870+
instance,
871+
args,
872+
kwargs,
873+
tracer,
874+
sem_conv_opt_in_mode,
875+
async_request_hook,
876+
async_response_hook,
852877
):
853878
if not is_http_instrumentation_enabled():
854879
return await wrapped(*args, **kwargs)
@@ -864,17 +889,17 @@ async def _handle_async_request_wrapper(
864889
span_attributes,
865890
url,
866891
method_original,
867-
self._sem_conv_opt_in_mode,
892+
sem_conv_opt_in_mode,
868893
)
869894

870895
request_info = RequestInfo(method, url, headers, stream, extensions)
871896

872-
with self._tracer.start_as_current_span(
897+
with tracer.start_as_current_span(
873898
span_name, kind=SpanKind.CLIENT, attributes=span_attributes
874899
) as span:
875900
exception = None
876-
if callable(self._async_request_hook):
877-
await self._async_request_hook(span, request_info)
901+
if callable(async_request_hook):
902+
await async_request_hook(span, request_info)
878903

879904
_inject_propagation_headers(headers, args, kwargs)
880905

@@ -895,20 +920,18 @@ async def _handle_async_request_wrapper(
895920
span,
896921
status_code,
897922
http_version,
898-
self._sem_conv_opt_in_mode,
923+
sem_conv_opt_in_mode,
899924
)
900925

901-
if callable(self._async_response_hook):
902-
await self._async_response_hook(
926+
if callable(async_response_hook):
927+
await async_response_hook(
903928
span,
904929
request_info,
905930
ResponseInfo(status_code, headers, stream, extensions),
906931
)
907932

908933
if exception:
909-
if span.is_recording() and _report_new(
910-
self._sem_conv_opt_in_mode
911-
):
934+
if span.is_recording() and _report_new(sem_conv_opt_in_mode):
912935
span.set_attribute(
913936
ERROR_TYPE, type(exception).__qualname__
914937
)
@@ -944,59 +967,81 @@ def instrument_client(
944967
)
945968
return
946969

947-
# FIXME: sharing state in the instrumentor instance maybe it's not that great, need to pass tracer and semconv to each
948-
# instance separately
949970
_OpenTelemetrySemanticConventionStability._initialize()
950-
self._sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
971+
sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
951972
_OpenTelemetryStabilitySignalType.HTTP,
952973
)
953-
self._tracer = get_tracer(
974+
tracer = get_tracer(
954975
__name__,
955976
instrumenting_library_version=__version__,
956977
tracer_provider=tracer_provider,
957-
schema_url=_get_schema_url(self._sem_conv_opt_in_mode),
978+
schema_url=_get_schema_url(sem_conv_opt_in_mode),
958979
)
959980

960981
if iscoroutinefunction(request_hook):
961-
self._async_request_hook = request_hook
962-
self._request_hook = None
982+
async_request_hook = request_hook
983+
request_hook = None
963984
else:
964-
self._request_hook = request_hook
965-
self._async_request_hook = None
985+
request_hook = request_hook
986+
async_request_hook = None
966987

967988
if iscoroutinefunction(response_hook):
968-
self._async_response_hook = response_hook
969-
self._response_hook = None
989+
async_response_hook = response_hook
990+
response_hook = None
970991
else:
971-
self._response_hook = response_hook
972-
self._async_response_hook = None
992+
response_hook = response_hook
993+
async_response_hook = None
973994

974995
if hasattr(client._transport, "handle_request"):
975996
wrap_function_wrapper(
976997
client._transport,
977998
"handle_request",
978-
self._handle_request_wrapper,
999+
partial(
1000+
self._handle_request_wrapper,
1001+
tracer=tracer,
1002+
sem_conv_opt_in_mode=sem_conv_opt_in_mode,
1003+
request_hook=request_hook,
1004+
response_hook=response_hook,
1005+
),
9791006
)
9801007
for transport in client._mounts.values():
9811008
# FIXME: check it's not wrapped already?
9821009
wrap_function_wrapper(
9831010
transport,
9841011
"handle_request",
985-
self._handle_request_wrapper,
1012+
partial(
1013+
self._handle_request_wrapper,
1014+
tracer=tracer,
1015+
sem_conv_opt_in_mode=sem_conv_opt_in_mode,
1016+
request_hook=request_hook,
1017+
response_hook=response_hook,
1018+
),
9861019
)
9871020
client._is_instrumented_by_opentelemetry = True
9881021
if hasattr(client._transport, "handle_async_request"):
9891022
wrap_function_wrapper(
9901023
client._transport,
9911024
"handle_async_request",
992-
self._handle_async_request_wrapper,
1025+
partial(
1026+
self._handle_async_request_wrapper,
1027+
tracer=tracer,
1028+
sem_conv_opt_in_mode=sem_conv_opt_in_mode,
1029+
async_request_hook=async_request_hook,
1030+
async_response_hook=async_response_hook,
1031+
),
9931032
)
9941033
for transport in client._mounts.values():
9951034
# FIXME: check it's not wrapped already?
9961035
wrap_function_wrapper(
9971036
transport,
9981037
"handle_async_request",
999-
self._handle_async_request_wrapper,
1038+
partial(
1039+
self._handle_async_request_wrapper,
1040+
tracer=tracer,
1041+
sem_conv_opt_in_mode=sem_conv_opt_in_mode,
1042+
async_request_hook=async_request_hook,
1043+
async_response_hook=async_response_hook,
1044+
),
10001045
)
10011046
client._is_instrumented_by_opentelemetry = True
10021047

0 commit comments

Comments
 (0)