Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def gen_opentracing():
with tracer.start_active_span('asteroid') as pscope:
pscope.span.set_tag(ext.COMPONENT, "Python simple example app")
pscope.span.set_tag(ext.SPAN_KIND, ext.SPAN_KIND_RPC_SERVER)
pscope.span.set_tag(ext.PEER_HOSTNAME, "localhost")
pscope.span.set_tag(ext.HTTP_URL, "/python/simple/one")
pscope.span.set_tag(ext.HTTP_METHOD, "GET")
pscope.span.set_tag(ext.HTTP_STATUS_CODE, 200)
pscope.span.log_kv({"foo": "bar"})
with tracer.start_active_span('spacedust', child_of=pscope.span) as cscope:
cscope.span.set_tag(ext.SPAN_KIND, ext.SPAN_KIND_RPC_CLIENT)
cscope.span.set_tag(ext.PEER_HOSTNAME, "localhost")
cscope.span.set_tag(ext.HTTP_URL, "/python/simple/two")
cscope.span.set_tag(ext.HTTP_METHOD, "POST")
cscope.span.set_tag(ext.HTTP_STATUS_CODE, 204)
cscope.span.set_baggage_item("someBaggage", "someValue")
return "<center><h1>🐍 Generated some OT spans... 🦄</h1></center>"
def setUp(self):
""" Clear all spans before a test run """
self.http = urllib3.PoolManager()
self.recorder = tracer.recorder
self.recorder.clear_spans()
def execute_command_with_instana(wrapped, instance, args, kwargs):
parent_span = tracer.active_span
# If we're not tracing, just return
if parent_span is None or parent_span.operation_name in EXCLUDED_PARENT_SPANS:
return wrapped(*args, **kwargs)
with tracer.start_active_span("redis", child_of=parent_span) as scope:
try:
collect_tags(scope.span, instance, args, kwargs)
if (len(args) > 0):
scope.span.set_tag("command", args[0])
rv = wrapped(*args, **kwargs)
except Exception as e:
scope.span.log_exception(e)
raise
else:
return rv
def task_prerun(*args, **kwargs):
try:
ctx = None
task = kwargs.get('sender', None)
task_id = kwargs.get('task_id', None)
task = registry.tasks.get(task.name)
headers = task.request.get('headers', {})
if headers is not None:
ctx = tracer.extract(opentracing.Format.HTTP_HEADERS, headers)
scope = tracer.start_active_span("celery-worker", child_of=ctx)
scope.span.set_tag("task", task.name)
scope.span.set_tag("task_id", task_id)
add_broker_tags(scope.span, task.app.conf['broker_url'])
# Store the scope on the task to eventually close it out on the "after" signal
task_catalog_push(task, task_id, scope, True)
except:
logger.debug("task_prerun: ", exc_info=True)