Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.kind = kind
self._set_status_on_exception = set_status_on_exception
self.span_processor = span_processor
self.status = None
self._lock = threading.Lock()
if attributes is None:
self.attributes = Span.empty_attributes
else:
self.attributes = BoundedDict.from_map(
MAX_NUM_ATTRIBUTES, attributes
)
if events is None:
self.events = Span.empty_events
else:
self.events = BoundedList.from_seq(MAX_NUM_EVENTS, events)
if links is None:
self.links = Span.empty_links
else:
self.links = BoundedList.from_seq(MAX_NUM_LINKS, links)
self.end_time = None # type: Optional[int]
self.start_time = None # type: Optional[int]
self.instrumentation_info = instrumentation_info
def add_lazy_event(self, event: trace_api.Event) -> None:
with self._lock:
if not self.is_recording_events():
return
has_ended = self.end_time is not None
if not has_ended:
if self.events is Span.empty_events:
self.events = BoundedList(MAX_NUM_EVENTS)
if has_ended:
logger.warning("Calling add_event() on an ended span.")
return
self.events.append(event)