1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
# Copyright 2024 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Tracer, Provider and span context."""
import contextlib
from typing import Any, Iterator, Optional, Sequence
import logging
import pathlib
import sys
from opentelemetry import context as otel_context_api
from opentelemetry import trace as otel_trace_api
from opentelemetry.sdk import trace as otel_trace_sdk
from opentelemetry.util import types as otel_types
from . import config
from . import clearcut_span_exporter
from . import detector
@contextlib.contextmanager
def use_span(
span: otel_trace_api.Span,
end_on_exit: bool = False,
record_exception: bool = True,
set_status_on_exception: bool = True,
) -> Iterator[otel_trace_api.Span]:
"""Takes a non-active span and activates it in the current context."""
try:
token = otel_context_api.attach(
# This is needed since the key needs to be the same as
# used in the rest of opentelemetry code.
otel_context_api.set_value(otel_trace_api._SPAN_KEY, span))
try:
yield span
finally:
otel_context_api.detach(token)
except KeyboardInterrupt as exc:
if span.is_recording():
if record_exception:
span.record_exception(exc)
if set_status_on_exception:
span.set_status(otel_trace_api.StatusCode.OK)
raise
except BaseException as exc:
if span.is_recording():
# Record the exception as an event
if record_exception:
span.record_exception(exc)
# Set status in case exception was raised
if set_status_on_exception:
span.set_status(
otel_trace_api.Status(
status_code=otel_trace_api.StatusCode.ERROR,
description=f"{type(exc).__name__}: {exc}",
))
raise
finally:
if end_on_exit:
span.end()
class Tracer(otel_trace_api.Tracer):
"""Specific otel tracer."""
def __init__(self, inner: otel_trace_sdk.Tracer) -> None:
self._inner = inner
def start_span(
self,
name: str,
context: Optional[otel_context_api.Context] = None,
kind: otel_trace_api.SpanKind = otel_trace_api.SpanKind.INTERNAL,
attributes: otel_types.Attributes = None,
links: Optional[Sequence[otel_trace_api.Link]] = None,
start_time: Optional[int] = None,
record_exception: bool = True,
set_status_on_exception: bool = True,
) -> otel_trace_api.Span:
return self._inner.start_span(
name,
context=context,
kind=kind,
attributes=attributes,
links=links,
start_time=start_time,
record_exception=record_exception,
set_status_on_exception=set_status_on_exception,
)
@contextlib.contextmanager
def start_as_current_span(
self,
name: str,
context: Optional[otel_context_api.Context] = None,
kind: otel_trace_api.SpanKind = otel_trace_api.SpanKind.INTERNAL,
attributes: otel_types.Attributes = None,
links: Optional[Sequence[otel_trace_api.Link]] = None,
start_time: Optional[int] = None,
record_exception: bool = True,
set_status_on_exception: bool = True,
end_on_exit: bool = True,
) -> Iterator[otel_trace_api.Span]:
span = self.start_span(
name=name,
context=context,
kind=kind,
attributes=attributes,
links=links,
start_time=start_time,
record_exception=record_exception,
set_status_on_exception=set_status_on_exception,
)
with use_span(
span,
end_on_exit=end_on_exit,
record_exception=record_exception,
set_status_on_exception=set_status_on_exception,
) as span_context:
yield span_context
class TracerProvider(otel_trace_api.TracerProvider):
"""Specific otel tracer provider."""
def __init__(self, inner: otel_trace_sdk.TracerProvider) -> None:
self._inner = inner
def get_tracer(
self,
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
schema_url: Optional[str] = None,
) -> otel_trace_api.Tracer:
tracer = self._inner.get_tracer(
instrumenting_module_name=instrumenting_module_name,
instrumenting_library_version=instrumenting_library_version,
schema_url=schema_url,
)
return Tracer(tracer)
def __getattr__(self, name: str) -> Any:
"""Method allows to delegate method calls."""
return getattr(self._inner, name)
|
