22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.8

/localstack-core/localstack/utils/xray/trace_header.py

# This file is part of LocalStack.
# It is adapted from aws-xray-sdk-python licensed under the Apache License 2.0.
# You may obtain a copy of the Apache License 2.0 at http://www.apache.org/licenses/LICENSE-2.0
# Original source: https://github.com/aws/aws-xray-sdk-python/blob/master/aws_xray_sdk/core/models/trace_header.py
# Modifications:
# * Add optional lineage field for https://docs.aws.amazon.com/lambda/latest/dg/invocation-recursion.html
# * Add ensure_root_exists(), ensure_parent_exists(), and ensure_sampled_exists()
# * Add generate_random_id() from https://github.com/aws/aws-xray-sdk-python/blob/d3a202719e659968fe6dcc04fe14c7f3045b53e8/aws_xray_sdk/core/models/entity.py#L308

import binascii
import logging
import os

from localstack.utils.xray.traceid import TraceId

log = logging.getLogger(__name__)

ROOT = "Root"
PARENT = "Parent"
SAMPLE = "Sampled"
SELF = "Self"
LINEAGE = "Lineage"

HEADER_DELIMITER = ";"


def generate_random_id():
    """
    Generate a random 16-digit hex str.
    This is used for generating segment/subsegment id.
    """
    return binascii.b2a_hex(os.urandom(8)).decode("utf-8")


class TraceHeader:
    """
    The sampling decision and trace ID are added to HTTP requests in
    tracing headers named ``X-Amzn-Trace-Id``. The first X-Ray-integrated
    service that the request hits adds a tracing header, which is read
    by the X-Ray SDK and included in the response. Learn more about
    `Tracing Header <http://docs.aws.amazon.com/xray/latest/devguide/xray-concepts.html#xray-concepts-tracingheader>`_.
    """

    def __init__(self, root=None, parent=None, sampled=None, data=None, lineage=None):
        """
        :param str root: trace id
        :param str parent: parent id
        :param int sampled: 0 means not sampled, 1 means sampled
        :param dict data: arbitrary data fields
        :param str lineage: lineage
        """
        self._root = root
        self._parent = parent
        self._sampled = None
        self._lineage = lineage
        self._data = data

        if sampled is not None:
            if sampled == "?":
                self._sampled = sampled
            if sampled is True or sampled == "1" or sampled == 1:
                self._sampled = 1
            if sampled is False or sampled == "0" or sampled == 0:
                self._sampled = 0

    @classmethod
    def from_header_str(cls, header):
        """
        Create a TraceHeader object from a tracing header string
        extracted from a http request headers.
        """
        if not header:
            return cls()

        try:
            params = header.strip().split(HEADER_DELIMITER)
            header_dict = {}
            data = {}

            for param in params:
                entry = param.split("=")
                key = entry[0]
                if key in (ROOT, PARENT, SAMPLE, LINEAGE):
                    header_dict[key] = entry[1]
                # Ignore any "Self=" trace ids injected from ALB.
                elif key != SELF:
                    data[key] = entry[1]

            return cls(
                root=header_dict.get(ROOT, None),
                parent=header_dict.get(PARENT, None),
                sampled=header_dict.get(SAMPLE, None),
                lineage=header_dict.get(LINEAGE, None),
                data=data,
            )

        except Exception:
            log.warning("malformed tracing header %s, ignore.", header)
            return cls()

    def to_header_str(self):
        """
        Convert to a tracing header string that can be injected to
        outgoing http request headers.
        """
        h_parts = []
        if self.root:
            h_parts.append(ROOT + "=" + self.root)
        if self.parent:
            h_parts.append(PARENT + "=" + self.parent)
        if self.sampled is not None:
            h_parts.append(SAMPLE + "=" + str(self.sampled))
        if self.lineage is not None:
            h_parts.append(LINEAGE + "=" + str(self.lineage))
        if self.data:
            for key in self.data:
                h_parts.append(key + "=" + self.data[key])

        return HEADER_DELIMITER.join(h_parts)

    def ensure_root_exists(self):
        """
        Ensures that a root trace id exists by generating one if None.
        Return self to allow for chaining.
        """
        if self._root is None:
            self._root = TraceId().to_id()
        return self

    # TODO: remove this hack once LocalStack supports X-Ray integration.
    #  This hack is only needed because we do not create segment ids in many places, but then expect downstream
    #  segments to have a valid parent link (e.g., Lambda invocations).
    def ensure_parent_exists(self):
        """
        Ensures that a parent segment link exists by generating a random one.
        Return self to allow for chaining.
        """
        if self._parent is None:
            self._parent = generate_random_id()
        return self

    def ensure_sampled_exists(self, sampled=None):
        """
        Ensures that the sampled flag is set.
        Return self to allow for chaining.
        """
        if sampled is None:
            self._sampled = 1
        else:
            if sampled == "?":
                self._sampled = sampled
            if sampled is True or sampled == "1" or sampled == 1:
                self._sampled = 1
            if sampled is False or sampled == "0" or sampled == 0:
                self._sampled = 0
        return self

    @property
    def root(self):
        """
        Return trace id of the header
        """
        return self._root

    @property
    def parent(self):
        """
        Return the parent segment id in the header
        """
        return self._parent

    @property
    def sampled(self):
        """
        Return the sampling decision in the header.
        It's 0 or 1 or '?'.
        """
        return self._sampled

    @property
    def lineage(self):
        """
        Return the lineage in the header
        """
        return self._lineage

    @property
    def data(self):
        """
        Return the arbitrary fields in the trace header.
        """
        return self._data

1	# This file is part of LocalStack.
2	# It is adapted from aws-xray-sdk-python licensed under the Apache License 2.0.
3	# You may obtain a copy of the Apache License 2.0 at http://www.apache.org/licenses/LICENSE-2.0
4	# Original source: https://github.com/aws/aws-xray-sdk-python/blob/master/aws_xray_sdk/core/models/trace_header.py
5	# Modifications:
6	# * Add optional lineage field for https://docs.aws.amazon.com/lambda/latest/dg/invocation-recursion.html
7	# * Add ensure_root_exists(), ensure_parent_exists(), and ensure_sampled_exists()
8	# * Add generate_random_id() from https://github.com/aws/aws-xray-sdk-python/blob/d3a202719e659968fe6dcc04fe14c7f3045b53e8/aws_xray_sdk/core/models/entity.py#L308
9
10	import binascii	1✔
11	import logging	1✔
12	import os	1✔
13
14	from localstack.utils.xray.traceid import TraceId	1✔
15
16	log = logging.getLogger(__name__)	1✔
17
18	ROOT = "Root"	1✔
19	PARENT = "Parent"	1✔
20	SAMPLE = "Sampled"	1✔
21	SELF = "Self"	1✔
22	LINEAGE = "Lineage"	1✔
23
24	HEADER_DELIMITER = ";"	1✔
25
26
27	def generate_random_id():	1✔
28	"""
29	Generate a random 16-digit hex str.
30	This is used for generating segment/subsegment id.
31	"""
32	return binascii.b2a_hex(os.urandom(8)).decode("utf-8")	1✔
33
34
35	class TraceHeader:	1✔
36	"""
37	The sampling decision and trace ID are added to HTTP requests in
38	tracing headers named ``X-Amzn-Trace-Id``. The first X-Ray-integrated
39	service that the request hits adds a tracing header, which is read
40	by the X-Ray SDK and included in the response. Learn more about
41	`Tracing Header <http://docs.aws.amazon.com/xray/latest/devguide/xray-concepts.html#xray-concepts-tracingheader>`_.
42	"""
43
44	def __init__(self, root=None, parent=None, sampled=None, data=None, lineage=None):	1✔
45	"""
46	:param str root: trace id
47	:param str parent: parent id
48	:param int sampled: 0 means not sampled, 1 means sampled
49	:param dict data: arbitrary data fields
50	:param str lineage: lineage
51	"""
52	self._root = root	1✔
53	self._parent = parent	1✔
54	self._sampled = None	1✔
55	self._lineage = lineage	1✔
56	self._data = data	1✔
57
58	if sampled is not None:	1✔
59	if sampled == "?":	1✔
60	self._sampled = sampled	×
61	if sampled is True or sampled == "1" or sampled == 1:	1✔
62	self._sampled = 1	1✔
63	if sampled is False or sampled == "0" or sampled == 0:	1✔
UNCOV 64	self._sampled = 0	×
65
66	@classmethod	1✔
67	def from_header_str(cls, header):	1✔
68	"""
69	Create a TraceHeader object from a tracing header string
70	extracted from a http request headers.
71	"""
72	if not header:	1✔
73	return cls()	1✔
74
75	try:	1✔
76	params = header.strip().split(HEADER_DELIMITER)	1✔
77	header_dict = {}	1✔
78	data = {}	1✔
79
80	for param in params:	1✔
81	entry = param.split("=")	1✔
82	key = entry[0]	1✔
83	if key in (ROOT, PARENT, SAMPLE, LINEAGE):	1✔
84	header_dict[key] = entry[1]	1✔
85	# Ignore any "Self=" trace ids injected from ALB.
86	elif key != SELF:	×
87	data[key] = entry[1]	×
88
89	return cls(	1✔
90	root=header_dict.get(ROOT, None),
91	parent=header_dict.get(PARENT, None),
92	sampled=header_dict.get(SAMPLE, None),
93	lineage=header_dict.get(LINEAGE, None),
94	data=data,
95	)
96
97	except Exception:	×
98	log.warning("malformed tracing header %s, ignore.", header)	×
99	return cls()	×
100
101	def to_header_str(self):	1✔
102	"""
103	Convert to a tracing header string that can be injected to
104	outgoing http request headers.
105	"""
106	h_parts = []	1✔
107	if self.root:	1✔
108	h_parts.append(ROOT + "=" + self.root)	1✔
109	if self.parent:	1✔
110	h_parts.append(PARENT + "=" + self.parent)	1✔
111	if self.sampled is not None:	1✔
112	h_parts.append(SAMPLE + "=" + str(self.sampled))	1✔
113	if self.lineage is not None:	1✔
114	h_parts.append(LINEAGE + "=" + str(self.lineage))	×
115	if self.data:	1✔
116	for key in self.data:	×
117	h_parts.append(key + "=" + self.data[key])	×
118
119	return HEADER_DELIMITER.join(h_parts)	1✔
120
121	def ensure_root_exists(self):	1✔
122	"""
123	Ensures that a root trace id exists by generating one if None.
124	Return self to allow for chaining.
125	"""
126	if self._root is None:	1✔
127	self._root = TraceId().to_id()	1✔
128	return self	1✔
129
130	# TODO: remove this hack once LocalStack supports X-Ray integration.
131	# This hack is only needed because we do not create segment ids in many places, but then expect downstream
132	# segments to have a valid parent link (e.g., Lambda invocations).
133	def ensure_parent_exists(self):	1✔
134	"""
135	Ensures that a parent segment link exists by generating a random one.
136	Return self to allow for chaining.
137	"""
138	if self._parent is None:	1✔
139	self._parent = generate_random_id()	1✔
140	return self	1✔
141
142	def ensure_sampled_exists(self, sampled=None):	1✔
143	"""
144	Ensures that the sampled flag is set.
145	Return self to allow for chaining.
146	"""
147	if sampled is None:	1✔
148	self._sampled = 1	1✔
149	else:
150	if sampled == "?":	×
151	self._sampled = sampled	×
152	if sampled is True or sampled == "1" or sampled == 1:	×
153	self._sampled = 1	×
154	if sampled is False or sampled == "0" or sampled == 0:	×
155	self._sampled = 0	×
156	return self	1✔
157
158	@property	1✔
159	def root(self):	1✔
160	"""
161	Return trace id of the header
162	"""
163	return self._root	1✔
164
165	@property	1✔
166	def parent(self):	1✔
167	"""
168	Return the parent segment id in the header
169	"""
170	return self._parent	1✔
171
172	@property	1✔
173	def sampled(self):	1✔
174	"""
175	Return the sampling decision in the header.
176	It's 0 or 1 or '?'.
177	"""
178	return self._sampled	1✔
179
180	@property	1✔
181	def lineage(self):	1✔
182	"""
183	Return the lineage in the header
184	"""
185	return self._lineage	1✔
186
187	@property	1✔
188	def data(self):	1✔
189	"""
190	Return the arbitrary fields in the trace header.
191	"""
192	return self._data	1✔

localstack / localstack / 22709357475

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous