/usr/lib/python2.7/dist-packages/aws_xray_sdk/ext/util.py is in python-aws-xray-sdk 0.95-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | import re
from aws_xray_sdk.core.models.trace_header import TraceHeader
from aws_xray_sdk.core.models import http
first_cap_re = re.compile('(.)([A-Z][a-z]+)')
all_cap_re = re.compile('([a-z0-9])([A-Z])')
def inject_trace_header(headers, entity):
"""
Extract trace id, entity id and sampling decision
from the input entity and inject these information
to headers.
:param dict headers: http headers to inject
:param Entity entity: trace entity that the trace header
value generated from.
"""
if not entity:
return
to_insert = TraceHeader(
root=entity.trace_id,
parent=entity.id,
sampled=entity.sampled,
)
value = to_insert.to_header_str()
headers[http.XRAY_HEADER] = value
def calculate_sampling_decision(trace_header, recorder,
service_name, method, path):
"""
Return 1 if should sample and 0 if should not.
The sampling decision coming from ``trace_header`` always has
the highest precedence. If the ``trace_header`` doesn't contain
sampling decision then it checks if sampling is enabled or not
in the recorder. If not enbaled it returns 1. Otherwise it uses
sampling rule to decide.
"""
if trace_header.sampled is not None:
return trace_header.sampled
elif not recorder.sampling:
return 1
elif recorder.sampler.should_trace(
service_name=service_name,
method=method,
path=path,
):
return 1
else:
return 0
def construct_xray_header(headers):
"""
Construct a ``TraceHeader`` object from dictionary headers
of the incoming request. This method should always return
a ``TraceHeader`` object regardless of tracing header's presence
in the incoming request.
"""
header_str = headers.get(http.XRAY_HEADER) or headers.get(http.ALT_XRAY_HEADER)
if header_str:
return TraceHeader.from_header_str(header_str)
else:
return TraceHeader()
def calculate_segment_name(host_name, recorder):
"""
Returns the segment name based on recorder configuration and
input host name. This is a helper generally used in web framework
middleware where a host name is available from incoming request's headers.
"""
if recorder.dynamic_naming:
return recorder.dynamic_naming.get_name(host_name)
else:
return recorder.service
def to_snake_case(name):
"""
Convert the input string to snake-cased string.
"""
s1 = first_cap_re.sub(r'\1_\2', name)
# handle acronym words
return all_cap_re.sub(r'\1_\2', s1).lower()
|