Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.cb.get("dummy",ttl=remaining_time)
except:
pass
def tearDown(self):
super(ConnectionTestCaseBase, self).tearDown()
if hasattr(self, '_implDtorHook'):
self._implDtorHook()
else:
try:
self.checkCbRefcount()
finally:
del self.cb
class LogRecorder(SpanRecorder):
def record_span(self, span):
if os.environ.get("PYCBC_LOG_RECORDED_SPANS"):
logging.info("recording span: "+str(span.__dict__))
def basic_tracer():
return BasicTracer(LogRecorder())
try:
from opentracing_pyzipkin.tracer import Tracer
import requests
def http_transport(encoded_span):
# The collector expects a thrift-encoded list of spans.
import logging
import event
import itertools
from spanid import SpanID, Annotation
from basictracer import BasicTracer, SpanRecorder
def create_new_tracer(collector, sampler=None):
"""create_new_tracer creates a new appdash opentracing tracer using an Appdash collector.
"""
return BasicTracer(recorder=AppdashRecorder(collector), sampler=sampler)
class AppdashRecorder(SpanRecorder):
"""AppdashRecorder collects and records spans to a remote Appdash collector.
"""
def __init__(self, collector):
self._collector = collector
def record_span(self, span):
if not span.context.sampled:
return
span_id = SpanID()
span_id.trace = span.context.trace_id
span_id.span = span.context.span_id
if span.parent_id is not None:
span_id.parent = span.parent_id
self._collector.collect(span_id,
*event.MarshalEvent(event.SpanNameEvent(span.operation_name)))