Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_should_sample_force_not_trace(self):
span_context = mock.Mock()
span_context.trace_options.enabled = False
sampler = mock.Mock()
sampler.should_sample.return_value = False
tracer = tracer_module.Tracer(
span_context=span_context, sampler=sampler)
sampled = tracer.should_sample()
self.assertFalse(sampled)
def test_get_tracer_context_tracer(self):
from opencensus.trace.tracers import context_tracer
sampler = mock.Mock()
sampler.should_sample.return_value = True
tracer = tracer_module.Tracer(sampler=sampler)
result = tracer.get_tracer()
assert isinstance(result, context_tracer.ContextTracer)
self.assertTrue(tracer.span_context.trace_options.enabled)
def test_finish_not_sampled(self):
from opencensus.trace.tracers import noop_tracer
sampler = mock.Mock()
sampler.should_sample.return_value = False
span_context = mock.Mock()
span_context.trace_options.enabled = False
tracer = tracer_module.Tracer(
span_context=span_context, sampler=sampler)
assert isinstance(tracer.tracer, noop_tracer.NoopTracer)
mock_tracer = mock.Mock()
tracer.tracer = mock_tracer
tracer.finish()
self.assertTrue(mock_tracer.finish.called)
def create_tracer(span_context=None):
return tracer_module.Tracer(
span_context=span_context,
sampler=get_tracer_sampler(),
exporter=get_tracer_exporter(),
propagator=get_tracer_propagator())
def _before_request(self):
"""A function to be run before each request.
See: http://flask.pocoo.org/docs/0.12/api/#flask.Flask.before_request
"""
# Do not trace if the url is blacklisted
if utils.disable_tracing_url(flask.request.url, self.blacklist_paths):
return
try:
span_context = self.propagator.from_headers(flask.request.headers)
tracer = tracer_module.Tracer(
span_context=span_context,
sampler=self.sampler,
exporter=self.exporter,
propagator=self.propagator)
span = tracer.start_span()
span.span_kind = span_module.SpanKind.SERVER
# Set the span name as the name of the current module name
span.name = '[{}]{}'.format(
flask.request.method,
flask.request.url)
tracer.add_attribute_to_current_span(
HTTP_HOST, flask.request.host
)
tracer.add_attribute_to_current_span(
HTTP_METHOD, flask.request.method
def __call__(self, *args, **kwargs):
kwds = kwargs.pop("kwds")
span_context_binary = kwargs.pop("span_context_binary")
propagator = binary_format.BinaryFormatPropagator()
kwargs["span_context"] = propagator.from_header(span_context_binary)
_tracer = tracer.Tracer(**kwargs)
execution_context.set_opencensus_tracer(_tracer)
with _tracer.span(name=threading.current_thread().name):
result = self.func(*args, **kwds)
execution_context.clean()
return result
import grpc
import demo_pb2
import demo_pb2_grpc
from logger import getJSONLogger
logger = getJSONLogger('emailservice-client')
from opencensus.trace.tracer import Tracer
from opencensus.trace.exporters import stackdriver_exporter
from opencensus.trace.ext.grpc import client_interceptor
try:
exporter = stackdriver_exporter.StackdriverExporter()
tracer = Tracer(exporter=exporter)
tracer_interceptor = client_interceptor.OpenCensusClientInterceptor(tracer, host_port='0.0.0.0:8080')
except:
tracer_interceptor = client_interceptor.OpenCensusClientInterceptor()
def send_confirmation_email(email, order):
channel = grpc.insecure_channel('0.0.0.0:8080')
channel = grpc.intercept_channel(channel, tracer_interceptor)
stub = demo_pb2_grpc.EmailServiceStub(channel)
try:
response = stub.SendOrderConfirmation(demo_pb2.SendOrderConfirmationRequest(
email = email,
order = order
))
logger.info('Request sent.')
except grpc.RpcError as err:
logger.error(err.details())
def _before_request(self, request):
if utils.disable_tracing_url(request.path, self._blacklist_paths):
return
try:
span_context = self.propagator.from_headers(request.headers)
tracer = tracer_module.Tracer(
span_context=span_context,
sampler=self.sampler,
exporter=self.exporter,
propagator=self.propagator)
span = tracer.start_span()
# Set the span name as the name of the current module name
span.name = '[{}]{}'.format(
request.method,
request.path)
span.span_kind = span_module.SpanKind.SERVER
tracer.add_attribute_to_current_span(
attribute_key=HTTP_HOST,
attribute_value=request.host_url)
def initialize_tracer(project_id):
exporter = stackdriver_exporter.StackdriverExporter(
project_id=project_id
)
tracer = opencensus.trace.tracer.Tracer(exporter=exporter)
return tracer
# [END trace_setup_python_configure]