codekingpro/portable-devtools

1# Copyright The OpenTelemetry Authors2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14#15import re16import typing17 18from opentelemetry import trace19from opentelemetry.context.context import Context20from opentelemetry.propagators import textmap21from opentelemetry.trace import format_span_id, format_trace_id22from opentelemetry.trace.span import TraceState23 24 25class TraceContextTextMapPropagator(textmap.TextMapPropagator):26    """Extracts and injects using w3c TraceContext's headers."""27 28    _TRACEPARENT_HEADER_NAME = "traceparent"29    _TRACESTATE_HEADER_NAME = "tracestate"30    _TRACEPARENT_HEADER_FORMAT = (31        "^[ \t]*([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})"32        + "(-.*)?[ \t]*$"33    )34    _TRACEPARENT_HEADER_FORMAT_RE = re.compile(_TRACEPARENT_HEADER_FORMAT)35 36    def extract(37        self,38        carrier: textmap.CarrierT,39        context: typing.Optional[Context] = None,40        getter: textmap.Getter[textmap.CarrierT] = textmap.default_getter,41    ) -> Context:42        """Extracts SpanContext from the carrier.43 44        See `opentelemetry.propagators.textmap.TextMapPropagator.extract`45        """46        if context is None:47            context = Context()48 49        header = getter.get(carrier, self._TRACEPARENT_HEADER_NAME)50 51        if not header:52            return context53 54        match = re.search(self._TRACEPARENT_HEADER_FORMAT_RE, header[0])55        if not match:56            return context57 58        version: str = match.group(1)59        trace_id: str = match.group(2)60        span_id: str = match.group(3)61        trace_flags: str = match.group(4)62 63        if trace_id == "0" * 32 or span_id == "0" * 16:64            return context65 66        if version == "00":67            if match.group(5):  # type: ignore68                return context69        if version == "ff":70            return context71 72        tracestate_headers = getter.get(carrier, self._TRACESTATE_HEADER_NAME)73        if tracestate_headers is None:74            tracestate = None75        else:76            tracestate = TraceState.from_header(tracestate_headers)77 78        span_context = trace.SpanContext(79            trace_id=int(trace_id, 16),80            span_id=int(span_id, 16),81            is_remote=True,82            trace_flags=trace.TraceFlags(int(trace_flags, 16)),83            trace_state=tracestate,84        )85        return trace.set_span_in_context(86            trace.NonRecordingSpan(span_context), context87        )88 89    def inject(90        self,91        carrier: textmap.CarrierT,92        context: typing.Optional[Context] = None,93        setter: textmap.Setter[textmap.CarrierT] = textmap.default_setter,94    ) -> None:95        """Injects SpanContext into the carrier.96 97        See `opentelemetry.propagators.textmap.TextMapPropagator.inject`98        """99        span = trace.get_current_span(context)100        span_context = span.get_span_context()101        if span_context == trace.INVALID_SPAN_CONTEXT:102            return103        traceparent_string = f"00-{format_trace_id(span_context.trace_id)}-{format_span_id(span_context.span_id)}-{span_context.trace_flags:02x}"104        setter.set(carrier, self._TRACEPARENT_HEADER_NAME, traceparent_string)105        if span_context.trace_state:106            tracestate_string = span_context.trace_state.to_header()107            setter.set(108                carrier, self._TRACESTATE_HEADER_NAME, tracestate_string109            )110 111    @property112    def fields(self) -> typing.Set[str]:113        """Returns a set with the fields set in `inject`.114 115        See116        `opentelemetry.propagators.textmap.TextMapPropagator.fields`117        """118        return {self._TRACEPARENT_HEADER_NAME, self._TRACESTATE_HEADER_NAME}119