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_from_headers_tracestate_limit(self):
propagator = trace_context_http_header_format.\
TraceContextPropagator()
span_context = propagator.from_headers({
'traceparent':
'00-12345678901234567890123456789012-1234567890123456-00',
'tracestate':
','.join([
'a00=0,a01=1,a02=2,a03=3,a04=4,a05=5,a06=6,a07=7,a08=8,a09=9',
'b00=0,b01=1,b02=2,b03=3,b04=4,b05=5,b06=6,b07=7,b08=8,b09=9',
'c00=0,c01=1,c02=2,c03=3,c04=4,c05=5,c06=6,c07=7,c08=8,c09=9',
'd00=0,d01=1,d02=2',
]),
})
self.assertFalse(span_context.tracestate)
def test_constructor_default(self):
from opencensus.trace import print_exporter
from opencensus.trace.propagation \
import trace_context_http_header_format
from opencensus.trace.samplers import ProbabilitySampler
from opencensus.trace.span_context import SpanContext
from opencensus.trace.tracers import noop_tracer
tracer = tracer_module.Tracer()
assert isinstance(tracer.span_context, SpanContext)
assert isinstance(tracer.sampler, ProbabilitySampler)
assert isinstance(tracer.exporter, print_exporter.PrintExporter)
assert isinstance(
tracer.propagator,
trace_context_http_header_format.TraceContextPropagator)
assert isinstance(tracer.tracer, noop_tracer.NoopTracer)
def test_header_match(self):
propagator = trace_context_http_header_format.\
TraceContextPropagator()
trace_id = '12345678901234567890123456789012'
span_id = '1234567890123456'
# Trace option is not enabled.
span_context = propagator.from_headers({
'traceparent':
'00-12345678901234567890123456789012-1234567890123456-00',
})
self.assertEqual(span_context.trace_id, trace_id)
self.assertEqual(span_context.span_id, span_id)
self.assertFalse(span_context.trace_options.enabled)
# Trace option is enabled.
def test_from_headers_empty(self):
from opencensus.trace.span_context import SpanContext
propagator = trace_context_http_header_format.\
TraceContextPropagator()
span_context = propagator.from_headers({})
self.assertTrue(isinstance(span_context, SpanContext))
def test_to_headers_with_empty_tracestate(self):
from opencensus.trace import span_context
from opencensus.trace import trace_options
from opencensus.trace.tracestate import Tracestate
propagator = trace_context_http_header_format.\
TraceContextPropagator()
trace_id = '12345678901234567890123456789012'
span_id_hex = '1234567890123456'
span_context = span_context.SpanContext(
trace_id=trace_id,
span_id=span_id_hex,
tracestate=Tracestate(),
trace_options=trace_options.TraceOptions('1'))
headers = propagator.to_headers(span_context)
self.assertTrue('traceparent' in headers)
self.assertEqual(headers['traceparent'], '00-{}-{}-01'.format(
trace_id, span_id_hex))
def get_tracer_propagator():
global _tracer_propagator
if _tracer_propagator is None:
settings_ = getattr(settings, 'OPENCENSUS', {})
settings_ = settings_.get('TRACE', {})
_tracer_propagator = settings_.get('PROPAGATOR', None) or \
trace_context_http_header_format.TraceContextPropagator()
if isinstance(_tracer_propagator, str):
_tracer_propagator = configuration.load(_tracer_propagator)
return _tracer_propagator
def link(cls, headers):
# type: (Dict[str, str]) -> None
"""
Given a dictionary, extracts the context and links the context to the current tracer.
:param headers: A key value pair dictionary
:type headers: dict
"""
ctx = trace_context_http_header_format.TraceContextPropagator().from_headers(headers)
current_span = cls.get_current_span()
current_span.add_link(Link(ctx.trace_id, ctx.span_id))
self.get_response = get_response
settings = getattr(django.conf.settings, 'OPENCENSUS', {})
settings = settings.get('TRACE', {})
self.sampler = (settings.get('SAMPLER', None)
or samplers.ProbabilitySampler())
if isinstance(self.sampler, six.string_types):
self.sampler = configuration.load(self.sampler)
self.exporter = settings.get('EXPORTER', None) or \
print_exporter.PrintExporter()
if isinstance(self.exporter, six.string_types):
self.exporter = configuration.load(self.exporter)
self.propagator = settings.get('PROPAGATOR', None) or \
trace_context_http_header_format.TraceContextPropagator()
if isinstance(self.propagator, six.string_types):
self.propagator = configuration.load(self.propagator)
self.blacklist_paths = settings.get(BLACKLIST_PATHS, None)
self.blacklist_hostnames = settings.get(BLACKLIST_HOSTNAMES, None)
if django.VERSION >= (2,): # pragma: NO COVER
connection.execute_wrappers.append(_trace_db_call)
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from opencensus.trace import print_exporter, samplers
from opencensus.trace.propagation import trace_context_http_header_format
DEFAULT_PYRAMID_TRACER_CONFIG = {
'SAMPLER': samplers.AlwaysOnSampler(),
'EXPORTER': print_exporter.PrintExporter(),
'PROPAGATOR': trace_context_http_header_format.TraceContextPropagator(),
# https://cloud.google.com/appengine/docs/flexible/python/
# how-instances-are-managed#health_checking
'BLACKLIST_PATHS': ['_ah/health'],
}
class PyramidTraceSettings(object):
def __init__(self, registry):
self.settings = registry.settings.get('OPENCENSUS', {})
self.settings = self.settings.get(
'TRACE', DEFAULT_PYRAMID_TRACER_CONFIG)
_set_default_configs(self.settings, DEFAULT_PYRAMID_TRACER_CONFIG)
def __getattr__(self, attr):
# If not in defaults, it is something we cannot parse.