Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if format in self._propagators:
self._propagators[format].inject(span_context, carrier)
else:
raise UnsupportedFormatException()
def extract(self, format, carrier):
if format in self._propagators:
return self._propagators[format].extract(carrier)
else:
raise UnsupportedFormatException()
def record(self, span):
self.recorder.record_span(span)
class NoopRecorder(SpanRecorder):
def record_span(self, span):
pass
import time
import traceback
import warnings
from basictracer.recorder import SpanRecorder
from opentracing.logs import ERROR_KIND, STACK
from lightstep.http_converter import HttpConverter
from lightstep.thrift_converter import ThriftConverter
from . import constants
from . import util
from lightstep.thrift_connection import _ThriftConnection
from lightstep.http_connection import _HTTPConnection
class Recorder(SpanRecorder):
"""Recorder translates, buffers, and reports basictracer.BasicSpans.
These reports are sent to a LightStep collector at the provided address.
For parameter semantics, see Tracer() documentation; Recorder() respects
component_name, access_token, collector_host, collector_port,
collector_encryption, tags, max_span_records, periodic_flush_seconds,
verbosity, and certificate_verification,
"""
def __init__(self,
component_name=None,
access_token='',
collector_host='collector.lightstep.com',
collector_port=443,
collector_encryption='tls',
tags=None,
class SpanRecorder(six.with_metaclass(ABCMeta, object)):
"""SpanRecorder is a simple abstract interface built around record_span.
"""
@abstractmethod
def record_span(self, span):
"""After the call to finish(), each BasicSpan is passed as `span` to
SpanRecorder.record_span.
:param BasicSpan span: the finish()'d BasicSpan object.
"""
pass
class InMemoryRecorder(SpanRecorder):
"""InMemoryRecorder stores all received spans in an internal list.
This recorder is not suitable for production use, only for testing.
"""
def __init__(self):
self.spans = []
self.mux = threading.Lock()
def record_span(self, span):
with self.mux:
self.spans.append(span)
def get_spans(self):
with self.mux:
return self.spans[:]