diff --git a/.changelog/5870.fixed b/.changelog/5870.fixed new file mode 100644 index 00000000000..9cdf454ca44 --- /dev/null +++ b/.changelog/5870.fixed @@ -0,0 +1 @@ +`opentelemetry-propagator-b3`: preserve and re-propagate a received debug flag diff --git a/propagator/opentelemetry-propagator-b3/src/opentelemetry/propagators/b3/__init__.py b/propagator/opentelemetry-propagator-b3/src/opentelemetry/propagators/b3/__init__.py index 0a6cff5d072..1679a3b597f 100644 --- a/propagator/opentelemetry-propagator-b3/src/opentelemetry/propagators/b3/__init__.py +++ b/propagator/opentelemetry-propagator-b3/src/opentelemetry/propagators/b3/__init__.py @@ -7,7 +7,7 @@ from typing_extensions import deprecated from opentelemetry import trace -from opentelemetry.context import Context +from opentelemetry.context import Context, create_key, get_value, set_value from opentelemetry.propagators.textmap import ( CarrierT, Getter, @@ -18,6 +18,8 @@ ) from opentelemetry.trace import format_span_id, format_trace_id +_DEBUG_FLAG_KEY = create_key("b3-debug-flag") + class B3MultiFormat(TextMapPropagator): """Propagator for the B3 HTTP multi-header format. @@ -32,6 +34,8 @@ class B3MultiFormat(TextMapPropagator): SAMPLED_KEY = "x-b3-sampled" FLAGS_KEY = "x-b3-flags" _SAMPLE_PROPAGATE_VALUES = {"1", "True", "true", "d"} + _DEBUG_SAMPLED_VALUE = "d" + _DEBUG_FLAGS_VALUE = "1" _trace_id_regex = re_compile(r"[\da-fA-F]{16}|[\da-fA-F]{32}") _span_id_regex = re_compile(r"[\da-fA-F]{16}") @@ -64,11 +68,21 @@ def extract( trace_id, span_id, sampled = fields elif len(fields) == 4: trace_id, span_id, sampled, _ = fields + + debug = sampled == self._DEBUG_SAMPLED_VALUE + sampling_state_alone = len(fields) == 1 else: - trace_id = _extract_first_element(getter.get(carrier, self.TRACE_ID_KEY)) or trace_id - span_id = _extract_first_element(getter.get(carrier, self.SPAN_ID_KEY)) or span_id - sampled = _extract_first_element(getter.get(carrier, self.SAMPLED_KEY)) or sampled - flags = _extract_first_element(getter.get(carrier, self.FLAGS_KEY)) or flags + trace_id = _extract_first_element(getter.get(carrier, self.TRACE_ID_KEY)) + span_id = _extract_first_element(getter.get(carrier, self.SPAN_ID_KEY)) + sampled = _extract_first_element(getter.get(carrier, self.SAMPLED_KEY)) + flags = _extract_first_element(getter.get(carrier, self.FLAGS_KEY)) + + debug = flags == self._DEBUG_FLAGS_VALUE + sampling_state_alone = trace_id is None and span_id is None and bool(sampled or flags) + + trace_id = trace.INVALID_TRACE_ID if trace_id is None else trace_id + span_id = trace.INVALID_SPAN_ID if span_id is None else span_id + sampled = "0" if sampled is None else sampled if ( trace_id == trace.INVALID_TRACE_ID @@ -76,6 +90,8 @@ def extract( or self._trace_id_regex.fullmatch(trace_id) is None or self._span_id_regex.fullmatch(span_id) is None ): + if sampling_state_alone: + context = set_value(_DEBUG_FLAG_KEY, debug, context) return context trace_id = int(trace_id, 16) @@ -88,6 +104,8 @@ def extract( if sampled in self._SAMPLE_PROPAGATE_VALUES or flags == "1": options |= trace.TraceFlags.SAMPLED + context = set_value(_DEBUG_FLAG_KEY, debug, context) + return trace.set_span_in_context( trace.NonRecordingSpan( trace.SpanContext( @@ -121,7 +139,10 @@ def inject( format_trace_id(span_context.trace_id), ) setter.set(carrier, self.SPAN_ID_KEY, format_span_id(span_context.span_id)) - setter.set(carrier, self.SAMPLED_KEY, "1" if sampled else "0") + if get_value(_DEBUG_FLAG_KEY, context) is True: + setter.set(carrier, self.FLAGS_KEY, self._DEBUG_FLAGS_VALUE) + else: + setter.set(carrier, self.SAMPLED_KEY, "1" if sampled else "0") @property def fields(self) -> set[str]: @@ -129,6 +150,7 @@ def fields(self) -> set[str]: self.TRACE_ID_KEY, self.SPAN_ID_KEY, self.SAMPLED_KEY, + self.FLAGS_KEY, } @@ -153,10 +175,15 @@ def inject( sampled = (trace.TraceFlags.SAMPLED & span_context.trace_flags) != 0 + if get_value(_DEBUG_FLAG_KEY, context) is True: + sampling_state = self._DEBUG_SAMPLED_VALUE + else: + sampling_state = "1" if sampled else "0" + fields = [ format_trace_id(span_context.trace_id), format_span_id(span_context.span_id), - "1" if sampled else "0", + sampling_state, ] setter.set(carrier, self.SINGLE_HEADER_KEY, "-".join(fields)) diff --git a/propagator/opentelemetry-propagator-b3/tests/test_b3_format.py b/propagator/opentelemetry-propagator-b3/tests/test_b3_format.py index bad73518d44..432a57494c6 100644 --- a/propagator/opentelemetry-propagator-b3/tests/test_b3_format.py +++ b/propagator/opentelemetry-propagator-b3/tests/test_b3_format.py @@ -6,8 +6,9 @@ from unittest.mock import Mock import opentelemetry.trace as trace_api -from opentelemetry.context import Context, get_current +from opentelemetry.context import Context, attach, detach, get_current, get_value from opentelemetry.propagators.b3 import ( # pylint: disable=no-name-in-module,import-error + _DEBUG_FLAG_KEY, B3MultiFormat, B3SingleFormat, ) @@ -35,7 +36,7 @@ def get_child_parent_new_carrier(old_carrier, propagator): ) new_carrier = {} - ctx = trace_api.set_span_in_context(child) + ctx = trace_api.set_span_in_context(child, ctx) propagator.inject(new_carrier, context=ctx) return child, parent, new_carrier @@ -76,6 +77,12 @@ def assertSampled(self, carrier): def assertNotSampled(self, carrier): pass + def assertDebug(self, carrier): + pass + + def assertNotDebug(self, carrier): + pass + def test_extract_multi_header(self): """Test the extraction of B3 headers.""" propagator = self.get_propagator() @@ -177,9 +184,9 @@ def test_disabled_sampling(self): self.assertNotSampled(new_carrier) def test_flags(self): - """x-b3-flags set to "1" should result in propagation.""" + """x-b3-flags set to "1" should result in propagation of the debug flag.""" propagator = self.get_propagator() - _, _, new_carrier = self.get_child_parent_new_carrier( + _, parent, new_carrier = self.get_child_parent_new_carrier( { propagator.TRACE_ID_KEY: self.serialized_trace_id, propagator.SPAN_ID_KEY: self.serialized_span_id, @@ -187,21 +194,233 @@ def test_flags(self): } ) - self.assertSampled(new_carrier) + self.assertTrue(parent.context.trace_flags.sampled) + self.assertDebug(new_carrier) def test_flags_and_sampling(self): - """Propagate if b3 flags and sampling are set.""" + """x-b3-flags set to "1" wins over x-b3-sampled set to "0".""" propagator = self.get_propagator() - _, _, new_carrier = self.get_child_parent_new_carrier( + _, parent, new_carrier = self.get_child_parent_new_carrier( { propagator.TRACE_ID_KEY: self.serialized_trace_id, propagator.SPAN_ID_KEY: self.serialized_span_id, propagator.FLAGS_KEY: "1", + propagator.SAMPLED_KEY: "0", + } + ) + + self.assertTrue(parent.context.trace_flags.sampled) + self.assertDebug(new_carrier) + + def test_extract_debug_flag_from_multi_header(self): + """A received x-b3-flags of "1" is stored in the extracted context.""" + propagator = self.get_propagator() + new_ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.FLAGS_KEY: "1", + } + ) + + self.assertTrue(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_debug_flag_from_single_header(self): + """A received sampling state of "d" is stored in the extracted context.""" + propagator = self.get_propagator() + new_ctx = propagator.extract( + {propagator.SINGLE_HEADER_KEY: f"{self.serialized_trace_id}-{self.serialized_span_id}-d"} + ) + + self.assertTrue(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_without_debug_flag(self): + """A sampled trace carrying no debug flag leaves none in the extracted context.""" + propagator = self.get_propagator() + new_ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.SAMPLED_KEY: "1", + } + ) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_without_debug_flag_for_an_explicit_zero(self): + """An x-b3-flags of "0" is not a debug flag.""" + propagator = self.get_propagator() + new_ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.FLAGS_KEY: "0", + } + ) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_without_debug_flag_for_a_single_header_accept(self): + """A single header sampling state of "1" is not a debug flag.""" + propagator = self.get_propagator() + new_ctx = propagator.extract( + {propagator.SINGLE_HEADER_KEY: f"{self.serialized_trace_id}-{self.serialized_span_id}-1"} + ) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_multi_header_sampled_d_is_not_debug(self): + """In the multi header form only x-b3-flags carries the debug flag.""" + propagator = self.get_propagator() + new_ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.SAMPLED_KEY: "d", + } + ) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + self.assertTrue(trace_api.get_current_span(new_ctx).get_span_context().trace_flags.sampled) + + def test_inject_debug_flag_received_in_single_header(self): + """A debug flag received as the sampling state "d" is re-propagated on inject.""" + propagator = self.get_propagator() + _, _, new_carrier = self.get_child_parent_new_carrier( + {propagator.SINGLE_HEADER_KEY: f"{self.serialized_trace_id}-{self.serialized_span_id}-d"} + ) + + self.assertDebug(new_carrier) + + def test_inject_without_debug_flag(self): + """A sampled trace with no received debug flag is injected without one.""" + propagator = self.get_propagator() + _, _, new_carrier = self.get_child_parent_new_carrier( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.SAMPLED_KEY: "1", } ) + self.assertNotDebug(new_carrier) self.assertSampled(new_carrier) + def test_inject_debug_flag_for_unsampled_span(self): + """A received debug flag is injected in place of the span's unsampled decision.""" + propagator = self.get_propagator() + ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.FLAGS_KEY: "1", + } + ) + child = trace._Span( + "child", + trace_api.SpanContext( + int(self.serialized_trace_id, 16), + id_generator.RandomIdGenerator().generate_span_id(), + is_remote=False, + trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.DEFAULT), + ), + ) + + new_carrier = {} + propagator.inject(new_carrier, context=trace_api.set_span_in_context(child, ctx)) + + self.assertDebug(new_carrier) + + def test_extract_debug_flag_from_single_header_without_ids(self): + """A sampling state of "d" sent on its own still carries the debug flag.""" + propagator = self.get_propagator() + new_ctx = propagator.extract({propagator.SINGLE_HEADER_KEY: "d"}) + + self.assertTrue(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_debug_flag_from_multi_header_without_ids(self): + """An x-b3-flags of "1" sent on its own still carries the debug flag.""" + propagator = self.get_propagator() + new_ctx = propagator.extract({propagator.FLAGS_KEY: "1"}) + + self.assertTrue(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_clears_a_debug_flag_left_by_an_earlier_carrier(self): + """The debug flag in the returned context comes from the carrier being extracted.""" + propagator = self.get_propagator() + debug_ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.FLAGS_KEY: "1", + } + ) + new_ctx = propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.SAMPLED_KEY: "1", + }, + debug_ctx, + ) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + new_carrier = {} + propagator.inject(new_carrier, context=new_ctx) + + self.assertNotDebug(new_carrier) + + def test_extract_clears_a_debug_flag_for_a_single_header_sampling_state_sent_alone(self): + """A sampling state of "0" sent on its own clears a debug flag left by an earlier carrier.""" + propagator = self.get_propagator() + debug_ctx = propagator.extract({propagator.SINGLE_HEADER_KEY: "d"}) + new_ctx = propagator.extract({propagator.SINGLE_HEADER_KEY: "0"}, debug_ctx) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_extract_clears_a_debug_flag_for_a_multi_header_sampling_state_sent_alone(self): + """An x-b3-sampled sent on its own clears a debug flag left by an earlier carrier.""" + propagator = self.get_propagator() + debug_ctx = propagator.extract({propagator.FLAGS_KEY: "1"}) + new_ctx = propagator.extract({propagator.SAMPLED_KEY: "1"}, debug_ctx) + + self.assertFalse(get_value(_DEBUG_FLAG_KEY, new_ctx)) + + def test_inject_debug_flag_received_without_ids(self): + """A debug flag received with no ids is re-propagated onto the span that follows it.""" + propagator = self.get_propagator() + tracer = trace.TracerProvider().get_tracer("sdk_tracer_provider") + token = attach(propagator.extract({propagator.SINGLE_HEADER_KEY: "d"})) + try: + new_carrier = {} + with tracer.start_as_current_span("server"): + propagator.inject(new_carrier) + finally: + detach(token) + + self.assertDebug(new_carrier) + + def test_inject_debug_flag_from_attached_context(self): + """The debug flag survives the extracted context being attached rather than passed.""" + propagator = self.get_propagator() + token = attach( + propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.FLAGS_KEY: "1", + } + ) + ) + try: + new_carrier = {} + propagator.inject(new_carrier) + finally: + detach(token) + + self.assertDebug(new_carrier) + def test_derived_ctx_is_returned_for_success(self): """Ensure returned context is derived from the given context.""" old_ctx = Context({"k1": "v1"}) @@ -262,6 +481,23 @@ def test_extract_invalid_single_header_to_implicit_ctx(self): self.assertDictEqual(Context(), new_ctx) + def test_extract_invalid_single_header_ids_to_explicit_ctx(self): + """Given a single header whose IDs are unparsable, do not modify context""" + old_ctx = Context({"k1": "v1"}) + propagator = self.get_propagator() + + carrier = {propagator.SINGLE_HEADER_KEY: "abc123-def456"} + new_ctx = propagator.extract(carrier, old_ctx) + + self.assertDictEqual(new_ctx, old_ctx) + + def test_extract_invalid_single_header_ids_to_implicit_ctx(self): + propagator = self.get_propagator() + carrier = {propagator.SINGLE_HEADER_KEY: "abc123-def456"} + new_ctx = propagator.extract(carrier) + + self.assertDictEqual(Context(), new_ctx) + def test_extract_missing_trace_id_to_explicit_ctx(self): """Given no trace ID, do not modify context""" old_ctx = Context({"k1": "v1"}) @@ -358,6 +594,31 @@ def test_extract_missing_span_id_to_implicit_ctx(self): self.assertDictEqual(Context(), new_ctx) + def test_extract_empty_ids_to_explicit_ctx(self): + """Given empty trace and span IDs, do not modify context""" + old_ctx = Context({"k1": "v1"}) + propagator = self.get_propagator() + + carrier = { + propagator.TRACE_ID_KEY: "", + propagator.SPAN_ID_KEY: "", + propagator.FLAGS_KEY: "1", + } + new_ctx = propagator.extract(carrier, old_ctx) + + self.assertDictEqual(new_ctx, old_ctx) + + def test_extract_empty_ids_to_implicit_ctx(self): + propagator = self.get_propagator() + carrier = { + propagator.TRACE_ID_KEY: "", + propagator.SPAN_ID_KEY: "", + propagator.FLAGS_KEY: "1", + } + new_ctx = propagator.extract(carrier) + + self.assertDictEqual(Context(), new_ctx) + def test_extract_empty_carrier_to_explicit_ctx(self): """Given no headers at all, do not modify context""" old_ctx = Context({"k1": "v1"}) @@ -400,6 +661,18 @@ def test_fields(self): with tracer.start_as_current_span("child"): propagator.inject({}, setter=mock_setter) + propagator.inject( + {}, + context=propagator.extract( + { + propagator.TRACE_ID_KEY: self.serialized_trace_id, + propagator.SPAN_ID_KEY: self.serialized_span_id, + propagator.FLAGS_KEY: "1", + } + ), + setter=mock_setter, + ) + inject_fields = set() for call in mock_setter.mock_calls: @@ -448,6 +721,14 @@ def assertSampled(self, carrier): def assertNotSampled(self, carrier): self.assertEqual(carrier[self.get_propagator().SAMPLED_KEY], "0") + def assertDebug(self, carrier): + propagator = self.get_propagator() + self.assertEqual(carrier[propagator.FLAGS_KEY], "1") + self.assertNotIn(propagator.SAMPLED_KEY, carrier) + + def assertNotDebug(self, carrier): + self.assertNotIn(self.get_propagator().FLAGS_KEY, carrier) + class TestB3SingleFormat(AbstractB3FormatTestCase, unittest.TestCase): @classmethod @@ -463,3 +744,9 @@ def assertSampled(self, carrier): def assertNotSampled(self, carrier): self.assertEqual(carrier[self.get_propagator().SINGLE_HEADER_KEY].split("-")[2], "0") + + def assertDebug(self, carrier): + self.assertEqual(carrier[self.get_propagator().SINGLE_HEADER_KEY].split("-")[2], "d") + + def assertNotDebug(self, carrier): + self.assertNotEqual(carrier[self.get_propagator().SINGLE_HEADER_KEY].split("-")[2], "d")