Writes through to span: attributes, events, links and the end,
with the exit the last set_status chose (an error status ends it
with a defect). Times default to clock's now.
Functions#
__init__#
__init__(span, clock) -> None| Parameter | Type | Description |
|---|---|---|
span | Span | |
clock | ClockService |
def __init__(self, span: Span, clock: ClockService) -> None:
self._span = span
self._clock = clock
self._exit: Exit[object, object] = Success(None)
self._context = SpanContext(
hex_id(span.trace_id),
hex_id(span.span_id),
False,
TraceFlags(TraceFlags.SAMPLED if span.sampled else TraceFlags.DEFAULT),
)get_span_context#
get_span_context() -> SpanContextReturns SpanContext
def get_span_context(self) -> SpanContext:
return self._contextset_attribute#
set_attribute(key, value) -> None| Parameter | Type | Description |
|---|---|---|
key | str | |
value | types.AttributeValue |
def set_attribute(self, key: str, value: types.AttributeValue) -> None:
self._span.attribute(key, value)set_attributes#
set_attributes(attributes) -> None| Parameter | Type | Description |
|---|---|---|
attributes | Mapping[str, types.AttributeValue] |
def set_attributes(self, attributes: Mapping[str, types.AttributeValue]) -> None:
for key, value in attributes.items():
self._span.attribute(key, value)add_event#
add_event(name, attributes=None, timestamp=None) -> None| Parameter | Type | Description |
|---|---|---|
name | str | |
attributes | types.Attributes = None | |
timestamp | int | None = None |
def add_event(
self, name: str, attributes: types.Attributes = None, timestamp: int | None = None
) -> None:
self._span.event(name, self._time(timestamp), attributes)add_link#
add_link(context, attributes=None) -> None| Parameter | Type | Description |
|---|---|---|
context | SpanContext | |
attributes | types.Attributes = None |
def add_link(self, context: SpanContext, attributes: types.Attributes = None) -> None:
self._span.add_links([_link(context, attributes)])update_name#
update_name(name) -> None| Parameter | Type | Description |
|---|---|---|
name | str |
def update_name(self, name: str) -> None:
passis_recording#
is_recording() -> boolReturns bool
def is_recording(self) -> bool:
return Trueset_status#
set_status(status, description=None) -> None| Parameter | Type | Description |
|---|---|---|
status | Status | StatusCode | |
description | str | None = None |
def set_status(self, status: Status | StatusCode, description: str | None = None) -> None:
code = status.status_code if isinstance(status, Status) else status
if code is StatusCode.ERROR:
message = status.description if isinstance(status, Status) else description
self._exit = die(Exception(message or "Unknown error"))
else:
self._exit = Success(None)end#
end(end_time=None) -> None| Parameter | Type | Description |
|---|---|---|
end_time | int | None = None |
def end(self, end_time: int | None = None) -> None:
self._span.end(self._time(end_time), self._exit)record_exception#
record_exception(exception, attributes=None, timestamp=None, escaped=False) -> None| Parameter | Type | Description |
|---|---|---|
exception | BaseException | |
attributes | types.Attributes = None | |
timestamp | int | None = None | |
escaped | bool = False |
def record_exception(
self,
exception: BaseException,
attributes: types.Attributes = None,
timestamp: int | None = None,
escaped: bool = False,
) -> None:
self._span.event(
str(exception),
self._time(timestamp),
{
"exception.type": type(exception).__name__,
"exception.message": str(exception),
"exception.stacktrace": "".join(traceback.format_exception(exception)),
},
)