Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def fake_pooled_func():
_tracer = execution_context.get_opencensus_tracer()
return _tracer.span_context.trace_id
mock_registry.settings = {}
middleware = pyramid_middleware.OpenCensusTweenFactory(
dummy_handler,
mock_registry,
)
request = DummyRequest(
registry=mock_registry,
path='/_ah/health',
headers={pyramid_trace_header: pyramid_trace_id},
)
middleware._before_request(request)
tracer = execution_context.get_opencensus_tracer()
assert isinstance(tracer, noop_tracer.NoopTracer)
span = tracer.current_span()
assert isinstance(span, NoopSpan)
def get_current_tracer(cls):
# type: () -> tracer_module.Tracer
"""
Get the current tracer from the execution context. Return None otherwise.
"""
return execution_context.get_opencensus_tracer()
# Check if request was sent from an exporter. If so, do not wrap.
if execution_context.is_exporter():
return requests_func(url, *args, **kwargs)
blacklist_hostnames = execution_context.get_opencensus_attr(
'blacklist_hostnames')
parsed_url = urlparse(url)
if parsed_url.port is None:
dest_url = parsed_url.hostname
else:
dest_url = '{}:{}'.format(parsed_url.hostname, parsed_url.port)
if utils.disable_tracing_hostname(dest_url, blacklist_hostnames):
return requests_func(url, *args, **kwargs)
path = parsed_url.path if parsed_url.path else '/'
_tracer = execution_context.get_opencensus_tracer()
_span = _tracer.start_span()
_span.name = '{}'.format(path)
_span.span_kind = span_module.SpanKind.CLIENT
# Add the requests host to attributes
_tracer.add_attribute_to_current_span(
HTTP_HOST, dest_url)
# Add the requests method to attributes
_tracer.add_attribute_to_current_span(
HTTP_METHOD, requests_func.__name__.upper())
# Add the requests path to attributes
_tracer.add_attribute_to_current_span(
HTTP_PATH, path)
# Get tracer from attribute `attr` passed.
if attr is not None:
tracer = rgetattr(obj, attr, None)
method = 'from attribute (%s)' % attr
# Get tracer from default attributes.
if tracer is None:
for _ in default_attributes:
tracer = rgetattr(obj, _, None)
if tracer is not None:
method = 'from default_attribute (%s)' % _
break
# Get tracer from OpenCensus context.
if tracer is None:
tracer = execution_context.get_opencensus_tracer()
if obj is not None:
for _ in default_attributes:
try:
rsetattr(obj, _, tracer)
method += ' + set to attribute %s' % _
break
except AttributeError:
pass
# If working with an instance of a class, set tracer to the proper
# instance attribute (either the passed `attr` or one of the default
# attributes).
if obj is not None:
attributes = [attr] if attr is not None else default_attributes
for _ in attributes:
try:
def _after_request(self, request, response):
if utils.disable_tracing_url(request.path, self._blacklist_paths):
return
try:
tracer = execution_context.get_opencensus_tracer()
tracer.add_attribute_to_current_span(
HTTP_STATUS_CODE,
response.status_code)
tracer.end_span()
tracer.finish()
except Exception: # pragma: NO COVER
log.error('Failed to trace request', exc_info=True)
def decorated_function(*args, **kwargs):
tracer = execution_context.get_opencensus_tracer()
with tracer.span(name=function_to_wrap.__name__):
return function_to_wrap(*args, **kwargs)
def main():
sampler = samplers.AlwaysOnSampler()
exporter = print_exporter.PrintExporter()
tracer = Tracer(sampler=sampler, exporter=exporter)
with tracer.span(name='root'):
tracer.add_attribute_to_current_span(
attribute_key='example key', attribute_value='example value')
function_to_trace()
with tracer.span(name='child'):
function_to_trace()
# Get the current tracer
tracer = execution_context.get_opencensus_tracer()
# Explicitly create spans
tracer.start_span()
# Get current span
execution_context.get_current_span()
# Explicitly end span
tracer.end_span()