Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create_worker(self, filters=None, api_class=DummyAPI, enable_stats=False):
with self.override_global_config(dict(health_metrics_enabled=enable_stats)):
self.dogstatsd = mock.Mock()
worker = AgentWriter(dogstatsd=self.dogstatsd, filters=filters)
worker._STATS_EVERY_INTERVAL = 1
self.api = api_class()
worker.api = self.api
for i in range(self.N_TRACES):
worker.write([
Span(tracer=None, name='name', trace_id=i, span_id=j, parent_id=j - 1 or None)
for j in range(7)
])
worker.stop()
worker.join()
return worker
def test_log_unfinished_spans(log, tracer_with_debug_logging):
# when the root parent is finished, notify if there are spans still pending
tracer = tracer_with_debug_logging
ctx = Context()
# manually create a root-child trace
root = Span(tracer=tracer, name='root')
child_1 = Span(tracer=tracer, name='child_1', trace_id=root.trace_id, parent_id=root.span_id)
child_2 = Span(tracer=tracer, name='child_2', trace_id=root.trace_id, parent_id=root.span_id)
child_1._parent = root
child_2._parent = root
ctx.add_span(root)
ctx.add_span(child_1)
ctx.add_span(child_2)
# close only the parent
root.finish()
unfinished_spans_log = log.call_args_list[-3][0][2]
child_1_log = log.call_args_list[-2][0][1]
child_2_log = log.call_args_list[-1][0][1]
assert 2 == unfinished_spans_log
assert 'name child_1' in child_1_log
assert 'name child_2' in child_2_log
assert 'duration 0.000000s' in child_1_log
assert 'duration 0.000000s' in child_2_log
def test_log_unfinished_spans_disabled(self, log):
# the trace finished status logging is disabled
tracer = get_dummy_tracer()
ctx = Context()
# manually create a root-child trace
root = Span(tracer=tracer, name='root')
child_1 = Span(tracer=tracer, name='child_1', trace_id=root.trace_id, parent_id=root.span_id)
child_2 = Span(tracer=tracer, name='child_2', trace_id=root.trace_id, parent_id=root.span_id)
child_1._parent = root
child_2._parent = root
ctx.add_span(root)
ctx.add_span(child_1)
ctx.add_span(child_2)
# close only the parent
root.finish()
# the logger has never been invoked to print unfinished spans
for call, _ in log.call_args_list:
msg = call[0]
assert 'the trace has %d unfinished spans' not in msg
def test_traceback_with_error(self):
s = Span(None, 'test.span')
try:
1 / 0
except ZeroDivisionError:
s.set_traceback()
else:
assert 0, 'should have failed'
assert s.error
assert 'by zero' in s.get_tag(errors.ERROR_MSG)
assert 'ZeroDivisionError' in s.get_tag(errors.ERROR_TYPE)
def test_ctx_mgr(self):
s = Span(self.tracer, 'bar')
assert not s.duration
assert not s.error
e = Exception('boo')
try:
with s:
time.sleep(0.01)
raise e
except Exception as out:
assert out == e
assert s.duration > 0, s.duration
assert s.error
assert s.get_tag(errors.ERROR_MSG) == 'boo'
assert 'Exception' in s.get_tag(errors.ERROR_TYPE)
assert s.get_tag(errors.ERROR_STACK)
def test_finish(self):
# ensure finish will record a span
ctx = Context()
s = Span(self.tracer, 'test.span', context=ctx)
ctx.add_span(s)
assert s.duration is None
sleep = 0.05
with s as s1:
assert s is s1
time.sleep(sleep)
assert s.duration >= sleep, '%s < %s' % (s.duration, sleep)
self.assert_span_count(1)
def test_set_tag_none(self):
s = Span(tracer=None, name='root.span', service='s', resource='r')
assert s.meta == dict()
s.set_tag('custom.key', 100)
assert s.meta == {'custom.key': '100'}
s.set_tag('custom.key', None)
assert s.meta == {'custom.key': 'None'}
if parent:
trace_id = parent.trace_id
parent_span_id = parent.span_id
else:
trace_id = context.trace_id
parent_span_id = context.span_id
if trace_id:
# child_of a non-empty context, so either a local child span or from a remote context
# when not provided, inherit from parent's service
if parent:
service = service or parent.service
span = Span(
self,
name,
trace_id=trace_id,
parent_id=parent_span_id,
service=service,
resource=resource,
span_type=span_type,
)
# Extra attributes when from a local parent
if parent:
span.sampled = parent.sampled
span._parent = parent
else:
# this is the root span of a new trace
name,
trace_id=trace_id,
parent_id=parent_span_id,
service=service,
resource=resource,
span_type=span_type,
)
# Extra attributes when from a local parent
if parent:
span.sampled = parent.sampled
span._parent = parent
else:
# this is the root span of a new trace
span = Span(
self,
name,
service=service,
resource=resource,
span_type=span_type,
)
span.sampled = self.sampler.sample(span)
# Old behavior
# DEV: The new sampler sets metrics and priority sampling on the span for us
if not isinstance(self.sampler, DatadogSampler):
if span.sampled:
# When doing client sampling in the client, keep the sample rate so that we can
# scale up statistics in the next steps of the pipeline.
if isinstance(self.sampler, RateSampler):
span.set_metric(SAMPLE_RATE_METRIC_KEY, self.sampler.sample_rate)