diff --git a/.release-please-manifest.json b/.release-please-manifest.json index 94d5d8cc4..28ce5d0cc 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,4 +1,4 @@ { - ".": "2.23.1" + ".": "2.24.0" } \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 93db31182..447013720 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ [1]: https://pypi.org/project/google-cloud-pubsub/#history +## [2.24.0](https://github.com/googleapis/python-pubsub/compare/v2.23.1...v2.24.0) (2024-09-24) + + +### Features + +* Add OpenTelemetry support for Subscribe Side ([#1252](https://github.com/googleapis/python-pubsub/issues/1252)) ([1b6f3d2](https://github.com/googleapis/python-pubsub/commit/1b6f3d284095e138943576de8551df263f73a506)) +* Open Telemetry Publish Side Support ([#1241](https://github.com/googleapis/python-pubsub/issues/1241)) ([bb5f3d1](https://github.com/googleapis/python-pubsub/commit/bb5f3d1a7df2d661cccc336edc8eceb2161c6921)) + + +### Bug Fixes + +* Fix flaky test ([#1254](https://github.com/googleapis/python-pubsub/issues/1254)) ([1ae49de](https://github.com/googleapis/python-pubsub/commit/1ae49de09996a5cf19f592f996c46e0222d540fc)) + ## [2.23.1](https://github.com/googleapis/python-pubsub/compare/v2.23.0...v2.23.1) (2024-09-09) diff --git a/google/cloud/pubsub_v1/open_telemetry/context_propagation.py b/google/cloud/pubsub_v1/open_telemetry/context_propagation.py index 37fad3e20..bfa1aa638 100644 --- a/google/cloud/pubsub_v1/open_telemetry/context_propagation.py +++ b/google/cloud/pubsub_v1/open_telemetry/context_propagation.py @@ -12,7 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -from opentelemetry.propagators.textmap import Setter +from typing import Optional, List + +from opentelemetry.propagators.textmap import Setter, Getter from google.pubsub_v1 import PubsubMessage @@ -37,3 +39,17 @@ def set(self, carrier: PubsubMessage, key: str, value: str) -> None: None """ carrier.attributes["googclient_" + key] = value + + +class OpenTelemetryContextGetter(Getter): + """ + Used by Open Telemetry for context propagation. + """ + + def get(self, carrier: PubsubMessage, key: str) -> Optional[List[str]]: + if ("googclient_" + key) not in carrier.attributes: + return None + return [carrier.attributes["googclient_" + key]] + + def keys(self, carrier: PubsubMessage) -> List[str]: + return list(map(str, carrier.attributes.keys())) diff --git a/google/cloud/pubsub_v1/open_telemetry/subscribe_opentelemetry.py b/google/cloud/pubsub_v1/open_telemetry/subscribe_opentelemetry.py new file mode 100644 index 000000000..88870be60 --- /dev/null +++ b/google/cloud/pubsub_v1/open_telemetry/subscribe_opentelemetry.py @@ -0,0 +1,280 @@ +# Copyright 2024, Google LLC All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# 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 typing import Optional, List +from datetime import datetime + +from opentelemetry import trace, context +from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator +from opentelemetry.trace.propagation import set_span_in_context + +from google.cloud.pubsub_v1.open_telemetry.context_propagation import ( + OpenTelemetryContextGetter, +) +from google.pubsub_v1.types import PubsubMessage + +_OPEN_TELEMETRY_TRACER_NAME: str = "google.cloud.pubsub_v1" +_OPEN_TELEMETRY_MESSAGING_SYSTEM: str = "gcp_pubsub" + + +class SubscribeOpenTelemetry: + def __init__(self, message: PubsubMessage): + self._message: PubsubMessage = message + + # subscribe span will be initialized by the `start_subscribe_span` + # method. + self._subscribe_span: Optional[trace.Span] = None + + # subscriber concurrency control span will be initialized by the + # `start_subscribe_concurrency_control_span` method. + self._concurrency_control_span: Optional[trace.Span] = None + + # scheduler span will be initialized by the + # `start_subscribe_scheduler_span` method. + self._scheduler_span: Optional[trace.Span] = None + + # This will be set by `start_subscribe_span` method and will be used + # for other spans, such as process span. + self._subscription_id: Optional[str] = None + + # This will be set by `start_process_span` method. + self._process_span: Optional[trace.Span] = None + + # This will be set by `start_subscribe_span` method, if a publisher create span + # context was extracted from trace propagation. And will be used by spans like + # proces span to add links to the publisher create span. + self._publisher_create_span_context: Optional[context.Context] = None + + # This will be set by `start_subscribe_span` method and will be used + # for other spans, such as modack span. + self._project_id: Optional[str] = None + + @property + def subscription_id(self) -> Optional[str]: + return self._subscription_id + + @property + def project_id(self) -> Optional[str]: + return self._project_id + + @property + def subscribe_span(self) -> Optional[trace.Span]: + return self._subscribe_span + + def start_subscribe_span( + self, + subscription: str, + exactly_once_enabled: bool, + ack_id: str, + delivery_attempt: int, + ) -> None: + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + parent_span_context = TraceContextTextMapPropagator().extract( + carrier=self._message, + getter=OpenTelemetryContextGetter(), + ) + self._publisher_create_span_context = parent_span_context + split_subscription: List[str] = subscription.split("/") + assert len(split_subscription) == 4 + subscription_short_name = split_subscription[3] + self._project_id = split_subscription[1] + self._subscription_id = subscription_short_name + with tracer.start_as_current_span( + name=f"{subscription_short_name} subscribe", + context=parent_span_context if parent_span_context else None, + kind=trace.SpanKind.CONSUMER, + attributes={ + "messaging.system": _OPEN_TELEMETRY_MESSAGING_SYSTEM, + "messaging.destination.name": subscription_short_name, + "gcp.project_id": subscription.split("/")[1], + "messaging.message.id": self._message.message_id, + "messaging.message.body.size": len(self._message.data), + "messaging.gcp_pubsub.message.ack_id": ack_id, + "messaging.gcp_pubsub.message.ordering_key": self._message.ordering_key, + "messaging.gcp_pubsub.message.exactly_once_delivery": exactly_once_enabled, + "code.function": "_on_response", + "messaging.gcp_pubsub.message.delivery_attempt": delivery_attempt, + }, + end_on_exit=False, + ) as subscribe_span: + self._subscribe_span = subscribe_span + + def add_subscribe_span_event(self, event: str) -> None: + assert self._subscribe_span is not None + self._subscribe_span.add_event( + name=event, + attributes={ + "timestamp": str(datetime.now()), + }, + ) + + def end_subscribe_span(self) -> None: + assert self._subscribe_span is not None + self._subscribe_span.end() + + def set_subscribe_span_result(self, result: str) -> None: + assert self._subscribe_span is not None + self._subscribe_span.set_attribute( + key="messaging.gcp_pubsub.result", + value=result, + ) + + def start_subscribe_concurrency_control_span(self) -> None: + assert self._subscribe_span is not None + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + with tracer.start_as_current_span( + name="subscriber concurrency control", + kind=trace.SpanKind.INTERNAL, + context=set_span_in_context(self._subscribe_span), + end_on_exit=False, + ) as concurrency_control_span: + self._concurrency_control_span = concurrency_control_span + + def end_subscribe_concurrency_control_span(self) -> None: + assert self._concurrency_control_span is not None + self._concurrency_control_span.end() + + def start_subscribe_scheduler_span(self) -> None: + assert self._subscribe_span is not None + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + with tracer.start_as_current_span( + name="subscriber scheduler", + kind=trace.SpanKind.INTERNAL, + context=set_span_in_context(self._subscribe_span), + end_on_exit=False, + ) as scheduler_span: + self._scheduler_span = scheduler_span + + def end_subscribe_scheduler_span(self) -> None: + assert self._scheduler_span is not None + self._scheduler_span.end() + + def start_process_span(self) -> None: + assert self._subscribe_span is not None + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + publish_create_span_link: Optional[trace.Link] = None + if self._publisher_create_span_context: + publish_create_span: trace.Span = trace.get_current_span( + self._publisher_create_span_context + ) + span_context: Optional[ + trace.SpanContext + ] = publish_create_span.get_span_context() + publish_create_span_link = ( + trace.Link(span_context) if span_context else None + ) + + with tracer.start_as_current_span( + name=f"{self._subscription_id} process", + attributes={ + "messaging.system": _OPEN_TELEMETRY_MESSAGING_SYSTEM, + }, + kind=trace.SpanKind.INTERNAL, + context=set_span_in_context(self._subscribe_span), + links=[publish_create_span_link] if publish_create_span_link else None, + end_on_exit=False, + ) as process_span: + self._process_span = process_span + + def end_process_span(self) -> None: + assert self._process_span is not None + self._process_span.end() + + def add_process_span_event(self, event: str) -> None: + assert self._process_span is not None + self._process_span.add_event( + name=event, + attributes={ + "timestamp": str(datetime.now()), + }, + ) + + +def start_modack_span( + subscribe_span_links: List[trace.Link], + subscription_id: Optional[str], + message_count: int, + deadline: float, + project_id: Optional[str], + code_function: str, + receipt_modack: bool, +) -> trace.Span: + assert subscription_id is not None + assert project_id is not None + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + with tracer.start_as_current_span( + name=f"{subscription_id} modack", + attributes={ + "messaging.system": _OPEN_TELEMETRY_MESSAGING_SYSTEM, + "messaging.batch.message_count": message_count, + "messaging.gcp_pubsub.message.ack_deadline": deadline, + "messaging.destination.name": subscription_id, + "gcp.project_id": project_id, + "messaging.operation.name": "modack", + "code.function": code_function, + "messaging.gcp_pubsub.is_receipt_modack": receipt_modack, + }, + links=subscribe_span_links, + kind=trace.SpanKind.CLIENT, + end_on_exit=False, + ) as modack_span: + return modack_span + + +def start_ack_span( + subscription_id: str, + message_count: int, + project_id: str, + links: List[trace.Link], +) -> trace.Span: + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + with tracer.start_as_current_span( + name=f"{subscription_id} ack", + attributes={ + "messaging.system": _OPEN_TELEMETRY_MESSAGING_SYSTEM, + "messaging.batch.message_count": message_count, + "messaging.operation": "ack", + "gcp.project_id": project_id, + "messaging.destination.name": subscription_id, + "code.function": "ack", + }, + kind=trace.SpanKind.CLIENT, + links=links, + end_on_exit=False, + ) as ack_span: + return ack_span + + +def start_nack_span( + subscription_id: str, + message_count: int, + project_id: str, + links: List[trace.Link], +) -> trace.Span: + tracer = trace.get_tracer(_OPEN_TELEMETRY_TRACER_NAME) + with tracer.start_as_current_span( + name=f"{subscription_id} nack", + attributes={ + "messaging.system": _OPEN_TELEMETRY_MESSAGING_SYSTEM, + "messaging.batch.message_count": message_count, + "messaging.operation": "nack", + "gcp.project_id": project_id, + "messaging.destination.name": subscription_id, + "code.function": "modify_ack_deadline", + }, + kind=trace.SpanKind.CLIENT, + links=links, + end_on_exit=False, + ) as nack_span: + return nack_span diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py b/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py index 15ad4abb3..fe3771432 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/dispatcher.py @@ -26,11 +26,17 @@ import warnings from google.api_core.retry import exponential_sleep_generator +from opentelemetry import trace + from google.cloud.pubsub_v1.subscriber._protocol import helper_threads from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber.exceptions import ( AcknowledgeStatus, ) +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + start_ack_span, + start_nack_span, +) if typing.TYPE_CHECKING: # pragma: NO COVER import queue @@ -232,16 +238,69 @@ def ack(self, items: Sequence[requests.AckRequest]) -> None: items_gen = iter(items) ack_ids_gen = (item.ack_id for item in items) total_chunks = int(math.ceil(len(items) / _ACK_IDS_BATCH_SIZE)) + subscription_id: Optional[str] = None + project_id: Optional[str] = None + for item in items: + if item.opentelemetry_data: + item.opentelemetry_data.add_subscribe_span_event("ack start") + if subscription_id is None: + subscription_id = item.opentelemetry_data.subscription_id + if project_id is None: + project_id = item.opentelemetry_data.project_id for _ in range(total_chunks): ack_reqs_dict = { req.ack_id: req for req in itertools.islice(items_gen, _ACK_IDS_BATCH_SIZE) } + + subscribe_links: List[trace.Link] = [] + subscribe_spans: List[trace.Span] = [] + for ack_req in ack_reqs_dict.values(): + if ack_req.opentelemetry_data: + subscribe_span: Optional[ + trace.Span + ] = ack_req.opentelemetry_data.subscribe_span + if ( + subscribe_span + and subscribe_span.get_span_context().trace_flags.sampled + ): + subscribe_links.append( + trace.Link(subscribe_span.get_span_context()) + ) + subscribe_spans.append(subscribe_span) + ack_span: Optional[trace.Span] = None + if subscription_id and project_id: + ack_span = start_ack_span( + subscription_id, + len(ack_reqs_dict), + project_id, + subscribe_links, + ) + if ( + ack_span and ack_span.get_span_context().trace_flags.sampled + ): # pragma: NO COVER + ack_span_context: trace.SpanContext = ack_span.get_span_context() + for subscribe_span in subscribe_spans: + subscribe_span.add_link( + context=ack_span_context, + attributes={ + "messaging.operation.name": "ack", + }, + ) + requests_completed, requests_to_retry = self._manager.send_unary_ack( ack_ids=list(itertools.islice(ack_ids_gen, _ACK_IDS_BATCH_SIZE)), ack_reqs_dict=ack_reqs_dict, ) + if ack_span: + ack_span.end() + + for completed_ack in requests_completed: + if completed_ack.opentelemetry_data: + completed_ack.opentelemetry_data.add_subscribe_span_event("ack end") + completed_ack.opentelemetry_data.set_subscribe_span_result("acked") + completed_ack.opentelemetry_data.end_subscribe_span() # Remove the completed messages from lease management. self.drop(requests_completed) @@ -267,7 +326,7 @@ def _start_retry_thread(self, thread_name, thread_target): # a back-end timeout error or other permanent failure. retry_thread.start() - def _retry_acks(self, requests_to_retry): + def _retry_acks(self, requests_to_retry: List[requests.AckRequest]): retry_delay_gen = exponential_sleep_generator( initial=_MIN_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS, maximum=_MAX_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS, @@ -282,10 +341,62 @@ def _retry_acks(self, requests_to_retry): time.sleep(time_to_wait) ack_reqs_dict = {req.ack_id: req for req in requests_to_retry} + subscription_id: Optional[str] = None + project_id: Optional[str] = None + subscribe_links: List[trace.Link] = [] + subscribe_spans: List[trace.Span] = [] + for req in requests_to_retry: + if req.opentelemetry_data: + req.opentelemetry_data.add_subscribe_span_event("ack start") + if subscription_id is None: + subscription_id = req.opentelemetry_data.subscription_id + if project_id is None: + project_id = req.opentelemetry_data.project_id + subscribe_span: Optional[ + trace.Span + ] = req.opentelemetry_data.subscribe_span + if ( + subscribe_span + and subscribe_span.get_span_context().trace_flags.sampled + ): + subscribe_links.append( + trace.Link(subscribe_span.get_span_context()) + ) + subscribe_spans.append(subscribe_span) + ack_span: Optional[trace.Span] = None + if subscription_id and project_id: + ack_span = start_ack_span( + subscription_id, + len(ack_reqs_dict), + project_id, + subscribe_links, + ) + if ( + ack_span and ack_span.get_span_context().trace_flags.sampled + ): # pragma: NO COVER + ack_span_context: trace.SpanContext = ack_span.get_span_context() + for subscribe_span in subscribe_spans: + subscribe_span.add_link( + context=ack_span_context, + attributes={ + "messaging.operation.name": "ack", + }, + ) + requests_completed, requests_to_retry = self._manager.send_unary_ack( ack_ids=[req.ack_id for req in requests_to_retry], ack_reqs_dict=ack_reqs_dict, ) + + if ack_span: + ack_span.end() + + for completed_ack in requests_completed: + if completed_ack.opentelemetry_data: + completed_ack.opentelemetry_data.add_subscribe_span_event("ack end") + completed_ack.opentelemetry_data.set_subscribe_span_result("acked") + completed_ack.opentelemetry_data.end_subscribe_span() + assert ( len(requests_to_retry) <= _ACK_IDS_BATCH_SIZE ), "Too many requests to be retried." @@ -336,15 +447,63 @@ def modify_ack_deadline( deadline_seconds_gen = (item.seconds for item in items) total_chunks = int(math.ceil(len(items) / _ACK_IDS_BATCH_SIZE)) + subscription_id: Optional[str] = None + project_id: Optional[str] = None + + for item in items: + if item.opentelemetry_data: + if math.isclose(item.seconds, 0): + item.opentelemetry_data.add_subscribe_span_event("nack start") + if subscription_id is None: + subscription_id = item.opentelemetry_data.subscription_id + if project_id is None: + project_id = item.opentelemetry_data.project_id + else: + item.opentelemetry_data.add_subscribe_span_event("modack start") for _ in range(total_chunks): ack_reqs_dict = { req.ack_id: req for req in itertools.islice(items_gen, _ACK_IDS_BATCH_SIZE) } + subscribe_links: List[trace.Link] = [] + subscribe_spans: List[trace.Span] = [] + for ack_req in ack_reqs_dict.values(): + if ack_req.opentelemetry_data and math.isclose(ack_req.seconds, 0): + subscribe_span: Optional[ + trace.Span + ] = ack_req.opentelemetry_data.subscribe_span + if ( + subscribe_span + and subscribe_span.get_span_context().trace_flags.sampled + ): + subscribe_links.append( + trace.Link(subscribe_span.get_span_context()) + ) + subscribe_spans.append(subscribe_span) + nack_span: Optional[trace.Span] = None + if subscription_id and project_id: + nack_span = start_nack_span( + subscription_id, + len(ack_reqs_dict), + project_id, + subscribe_links, + ) + if ( + nack_span and nack_span.get_span_context().trace_flags.sampled + ): # pragma: NO COVER + nack_span_context: trace.SpanContext = nack_span.get_span_context() + for subscribe_span in subscribe_spans: + subscribe_span.add_link( + context=nack_span_context, + attributes={ + "messaging.operation.name": "nack", + }, + ) requests_to_retry: List[requests.ModAckRequest] + requests_completed: Optional[List[requests.ModAckRequest]] = None if default_deadline is None: # no further work needs to be done for `requests_to_retry` - _, requests_to_retry = self._manager.send_unary_modack( + requests_completed, requests_to_retry = self._manager.send_unary_modack( modify_deadline_ack_ids=list( itertools.islice(ack_ids_gen, _ACK_IDS_BATCH_SIZE) ), @@ -355,7 +514,7 @@ def modify_ack_deadline( default_deadline=None, ) else: - _, requests_to_retry = self._manager.send_unary_modack( + requests_completed, requests_to_retry = self._manager.send_unary_modack( modify_deadline_ack_ids=itertools.islice( ack_ids_gen, _ACK_IDS_BATCH_SIZE ), @@ -363,10 +522,28 @@ def modify_ack_deadline( ack_reqs_dict=ack_reqs_dict, default_deadline=default_deadline, ) + if nack_span: + nack_span.end() assert ( len(requests_to_retry) <= _ACK_IDS_BATCH_SIZE ), "Too many requests to be retried." + for completed_modack in requests_completed: + if completed_modack.opentelemetry_data: + # nack is a modack with 0 extension seconds. + if math.isclose(completed_modack.seconds, 0): + completed_modack.opentelemetry_data.set_subscribe_span_result( + "nacked" + ) + completed_modack.opentelemetry_data.add_subscribe_span_event( + "nack end" + ) + completed_modack.opentelemetry_data.end_subscribe_span() + else: + completed_modack.opentelemetry_data.add_subscribe_span_event( + "modack end" + ) + # Retry on a separate thread so the dispatcher thread isn't blocked # by sleeps. if requests_to_retry: @@ -390,11 +567,67 @@ def _retry_modacks(self, requests_to_retry): time.sleep(time_to_wait) ack_reqs_dict = {req.ack_id: req for req in requests_to_retry} + + subscription_id = None + project_id = None + subscribe_links = [] + subscribe_spans = [] + for ack_req in ack_reqs_dict.values(): + if ack_req.opentelemetry_data and math.isclose(ack_req.seconds, 0): + if subscription_id is None: + subscription_id = ack_req.opentelemetry_data.subscription_id + if project_id is None: + project_id = ack_req.opentelemetry_data.project_id + subscribe_span = ack_req.opentelemetry_data.subscribe_span + if ( + subscribe_span + and subscribe_span.get_span_context().trace_flags.sampled + ): + subscribe_links.append( + trace.Link(subscribe_span.get_span_context()) + ) + subscribe_spans.append(subscribe_span) + nack_span = None + if subscription_id and project_id: + nack_span = start_nack_span( + subscription_id, + len(ack_reqs_dict), + project_id, + subscribe_links, + ) + if ( + nack_span and nack_span.get_span_context().trace_flags.sampled + ): # pragma: NO COVER + nack_span_context: trace.SpanContext = nack_span.get_span_context() + for subscribe_span in subscribe_spans: + subscribe_span.add_link( + context=nack_span_context, + attributes={ + "messaging.operation.name": "nack", + }, + ) requests_completed, requests_to_retry = self._manager.send_unary_modack( modify_deadline_ack_ids=[req.ack_id for req in requests_to_retry], modify_deadline_seconds=[req.seconds for req in requests_to_retry], ack_reqs_dict=ack_reqs_dict, ) + if nack_span: + nack_span.end() + for completed_modack in requests_completed: + if completed_modack.opentelemetry_data: + # nack is a modack with 0 extension seconds. + if math.isclose(completed_modack.seconds, 0): + completed_modack.opentelemetry_data.set_subscribe_span_result( + "nacked" + ) + completed_modack.opentelemetry_data.add_subscribe_span_event( + "nack end" + ) + completed_modack.opentelemetry_data.end_subscribe_span() + else: + completed_modack.opentelemetry_data.add_subscribe_span_event( + "modack end" + ) def nack(self, items: Sequence[requests.NackRequest]) -> None: """Explicitly deny receipt of messages. @@ -405,7 +638,10 @@ def nack(self, items: Sequence[requests.NackRequest]) -> None: self.modify_ack_deadline( [ requests.ModAckRequest( - ack_id=item.ack_id, seconds=0, future=item.future + ack_id=item.ack_id, + seconds=0, + future=item.future, + opentelemetry_data=item.opentelemetry_data, ) for item in items ] diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py b/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py index 16018e384..5abdb7081 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py @@ -23,6 +23,9 @@ from typing import Dict, Iterable, Optional, Union from google.cloud.pubsub_v1.subscriber._protocol.dispatcher import _MAX_BATCH_LATENCY +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) try: from collections.abc import KeysView @@ -50,6 +53,7 @@ class _LeasedMessage(typing.NamedTuple): size: int ordering_key: Optional[str] + opentelemetry_data: Optional[SubscribeOpenTelemetry] class Leaser(object): @@ -98,6 +102,7 @@ def add(self, items: Iterable[requests.LeaseRequest]) -> None: sent_time=float("inf"), size=item.byte_size, ordering_key=item.ordering_key, + opentelemetry_data=item.opentelemetry_data, ) self._bytes += item.byte_size else: @@ -175,6 +180,17 @@ def maintain_leases(self) -> None: "Dropping %s items because they were leased too long.", len(to_drop) ) assert self._manager.dispatcher is not None + for drop_msg in to_drop: + leased_message = leased_messages.get(drop_msg.ack_id) + if leased_message and leased_message.opentelemetry_data: + leased_message.opentelemetry_data.add_process_span_event( + "expired" + ) + leased_message.opentelemetry_data.end_process_span() + leased_message.opentelemetry_data.set_subscribe_span_result( + "expired" + ) + leased_message.opentelemetry_data.end_subscribe_span() self._manager.dispatcher.drop(to_drop) # Remove dropped items from our copy of the leased messages (they @@ -198,14 +214,28 @@ def maintain_leases(self) -> None: # is inactive. assert self._manager.dispatcher is not None ack_id_gen = (ack_id for ack_id in ack_ids) + opentelemetry_data = [ + message.opentelemetry_data + for message in list(leased_messages.values()) + if message.opentelemetry_data + ] expired_ack_ids = self._manager._send_lease_modacks( - ack_id_gen, deadline + ack_id_gen, + deadline, + opentelemetry_data, ) start_time = time.time() # If exactly once delivery is enabled, we should drop all expired ack_ids from lease management. if self._manager._exactly_once_delivery_enabled() and len(expired_ack_ids): assert self._manager.dispatcher is not None + for ack_id in expired_ack_ids: + msg = leased_messages.get(ack_id) + if msg and msg.opentelemetry_data: + msg.opentelemetry_data.add_process_span_event("expired") + msg.opentelemetry_data.end_process_span() + msg.opentelemetry_data.set_subscribe_span_result("expired") + msg.opentelemetry_data.end_subscribe_span() self._manager.dispatcher.drop( [ requests.DropRequest( diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/messages_on_hold.py b/google/cloud/pubsub_v1/subscriber/_protocol/messages_on_hold.py index 63c2edbfa..3d4c2a392 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/messages_on_hold.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/messages_on_hold.py @@ -100,6 +100,8 @@ def put(self, message: "subscriber.message.Message") -> None: Args: message: The message to put on hold. """ + if message.opentelemetry_data: + message.opentelemetry_data.start_subscribe_scheduler_span() self._messages_on_hold.append(message) self._size = self._size + 1 diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/requests.py b/google/cloud/pubsub_v1/subscriber/_protocol/requests.py index 9cd387545..6fd35896b 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/requests.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/requests.py @@ -15,6 +15,10 @@ import typing from typing import NamedTuple, Optional +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) + if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1.subscriber import futures @@ -27,6 +31,7 @@ class AckRequest(NamedTuple): time_to_ack: float ordering_key: Optional[str] future: Optional["futures.Future"] + opentelemetry_data: Optional[SubscribeOpenTelemetry] = None class DropRequest(NamedTuple): @@ -39,12 +44,14 @@ class LeaseRequest(NamedTuple): ack_id: str byte_size: int ordering_key: Optional[str] + opentelemetry_data: Optional[SubscribeOpenTelemetry] = None class ModAckRequest(NamedTuple): ack_id: str seconds: float future: Optional["futures.Future"] + opentelemetry_data: Optional[SubscribeOpenTelemetry] = None class NackRequest(NamedTuple): @@ -52,3 +59,4 @@ class NackRequest(NamedTuple): byte_size: int ordering_key: Optional[str] future: Optional["futures.Future"] + opentelemetry_data: Optional[SubscribeOpenTelemetry] = None diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py index c01dd7f2e..4c9e1c20e 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py @@ -23,6 +23,7 @@ from typing import Any, Dict, Callable, Iterable, List, Optional, Set, Tuple import uuid +from opentelemetry import trace import grpc # type: ignore from google.api_core import bidi @@ -38,6 +39,9 @@ AcknowledgeError, AcknowledgeStatus, ) +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) import google.cloud.pubsub_v1.subscriber.message from google.cloud.pubsub_v1.subscriber import futures from google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler @@ -46,6 +50,9 @@ from google.rpc.error_details_pb2 import ErrorInfo # type: ignore from google.rpc import code_pb2 # type: ignore from google.rpc import status_pb2 +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + start_modack_span, +) if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1 import subscriber @@ -123,6 +130,9 @@ def _wrap_callback_errors( message: The Pub/Sub message. """ try: + if message.opentelemetry_data: + message.opentelemetry_data.end_subscribe_concurrency_control_span() + message.opentelemetry_data.start_process_span() callback(message) except BaseException as exc: # Note: the likelihood of this failing is extremely low. This just adds @@ -582,7 +592,8 @@ def _maybe_release_messages(self) -> None: msg = self._messages_on_hold.get() if not msg: break - + if msg.opentelemetry_data: + msg.opentelemetry_data.end_subscribe_scheduler_span() self._schedule_message_on_hold(msg) released_ack_ids.append(msg.ack_id) @@ -618,6 +629,8 @@ def _schedule_message_on_hold( ) assert self._scheduler is not None assert self._callback is not None + if msg.opentelemetry_data: + msg.opentelemetry_data.start_subscribe_concurrency_control_span() self._scheduler.schedule(self._callback, msg) def send_unary_ack( @@ -1007,22 +1020,85 @@ def _get_initial_request( return request def _send_lease_modacks( - self, ack_ids: Iterable[str], ack_deadline: float, warn_on_invalid=True + self, + ack_ids: Iterable[str], + ack_deadline: float, + opentelemetry_data: List[SubscribeOpenTelemetry], + warn_on_invalid=True, + receipt_modack: bool = False, ) -> Set[str]: exactly_once_enabled = False + + modack_span: Optional[trace.Span] = None + if self._client.open_telemetry_enabled: + subscribe_span_links: List[trace.Link] = [] + subscribe_spans: List[trace.Span] = [] + subscription_split: List[str] = self._subscription.split("/") + assert len(subscription_split) == 4 + subscription_id: str = subscription_split[3] + project_id: str = subscription_split[1] + for data in opentelemetry_data: + subscribe_span: Optional[trace.Span] = data.subscribe_span + if ( + subscribe_span + and subscribe_span.get_span_context().trace_flags.sampled + ): + subscribe_span_links.append( + trace.Link(subscribe_span.get_span_context()) + ) + subscribe_spans.append(subscribe_span) + modack_span = start_modack_span( + subscribe_span_links, + subscription_id, + len(opentelemetry_data), + ack_deadline, + project_id, + "_send_lease_modacks", + receipt_modack, + ) + if ( + modack_span and modack_span.get_span_context().trace_flags.sampled + ): # pragma: NO COVER + modack_span_context: trace.SpanContext = modack_span.get_span_context() + for subscribe_span in subscribe_spans: + subscribe_span.add_link( + context=modack_span_context, + attributes={ + "messaging.operation.name": "modack", + }, + ) + with self._exactly_once_enabled_lock: exactly_once_enabled = self._exactly_once_enabled if exactly_once_enabled: - items = [ - requests.ModAckRequest(ack_id, ack_deadline, futures.Future()) - for ack_id in ack_ids - ] + eod_items: List[requests.ModAckRequest] = [] + if self._client.open_telemetry_enabled: + for ack_id, data in zip( + ack_ids, opentelemetry_data + ): # pragma: NO COVER # Identical code covered in the same function below + assert data is not None + eod_items.append( + requests.ModAckRequest( + ack_id, + ack_deadline, + futures.Future(), + data, + ) + ) + else: + eod_items = [ + requests.ModAckRequest(ack_id, ack_deadline, futures.Future()) + for ack_id in ack_ids + ] assert self._dispatcher is not None - self._dispatcher.modify_ack_deadline(items, ack_deadline) - + self._dispatcher.modify_ack_deadline(eod_items, ack_deadline) + if ( + modack_span + ): # pragma: NO COVER # Identical code covered in the same function below + modack_span.end() expired_ack_ids = set() - for req in items: + for req in eod_items: try: assert req.future is not None req.future.result() @@ -1039,12 +1115,27 @@ def _send_lease_modacks( expired_ack_ids.add(req.ack_id) return expired_ack_ids else: - items = [ - requests.ModAckRequest(ack_id, self.ack_deadline, None) - for ack_id in ack_ids - ] + items: List[requests.ModAckRequest] = [] + if self._client.open_telemetry_enabled: + for ack_id, data in zip(ack_ids, opentelemetry_data): + assert data is not None + items.append( + requests.ModAckRequest( + ack_id, + self.ack_deadline, + None, + data, + ) + ) + else: + items = [ + requests.ModAckRequest(ack_id, self.ack_deadline, None) + for ack_id in ack_ids + ] assert self._dispatcher is not None self._dispatcher.modify_ack_deadline(items, ack_deadline) + if modack_span: + modack_span.end() return set() def _exactly_once_delivery_enabled(self) -> bool: @@ -1075,6 +1166,18 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None: # protobuf message to significantly gain on attribute access performance. received_messages = response._pb.received_messages + subscribe_opentelemetry: List[SubscribeOpenTelemetry] = [] + if self._client.open_telemetry_enabled: + for received_message in received_messages: + opentelemetry_data = SubscribeOpenTelemetry(received_message.message) + opentelemetry_data.start_subscribe_span( + self._subscription, + response.subscription_properties.exactly_once_delivery_enabled, + received_message.ack_id, + received_message.delivery_attempt, + ) + subscribe_opentelemetry.append(opentelemetry_data) + _LOGGER.debug( "Processing %s received message(s), currently on hold %s (bytes %s).", len(received_messages), @@ -1100,7 +1203,11 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None: # received them. ack_id_gen = (message.ack_id for message in received_messages) expired_ack_ids = self._send_lease_modacks( - ack_id_gen, self.ack_deadline, warn_on_invalid=False + ack_id_gen, + self.ack_deadline, + subscribe_opentelemetry, + warn_on_invalid=False, + receipt_modack=True, ) with self._pause_resume_lock: @@ -1110,6 +1217,7 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None: ) return + i: int = 0 for received_message in received_messages: if ( not self._exactly_once_delivery_enabled() @@ -1122,12 +1230,16 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None: self._scheduler.queue, self._exactly_once_delivery_enabled, ) + if self._client.open_telemetry_enabled: + message.opentelemetry_data = subscribe_opentelemetry[i] + i = i + 1 self._messages_on_hold.put(message) self._on_hold_bytes += message.size req = requests.LeaseRequest( ack_id=message.ack_id, byte_size=message.size, ordering_key=message.ordering_key, + opentelemetry_data=message.opentelemetry_data, ) self._leaser.add([req]) diff --git a/google/cloud/pubsub_v1/subscriber/client.py b/google/cloud/pubsub_v1/subscriber/client.py index 0d0d36a0c..175095077 100644 --- a/google/cloud/pubsub_v1/subscriber/client.py +++ b/google/cloud/pubsub_v1/subscriber/client.py @@ -14,6 +14,7 @@ from __future__ import absolute_import +import sys import os import typing from typing import cast, Any, Callable, Optional, Sequence, Union @@ -67,7 +68,16 @@ class Client(subscriber_client.SubscriberClient): ) """ - def __init__(self, **kwargs: Any): + def __init__( + self, + subscriber_options: Union[types.SubscriberOptions, Sequence] = (), + **kwargs: Any + ): + assert ( + isinstance(subscriber_options, types.SubscriberOptions) + or len(subscriber_options) == 0 + ), "subscriber_options must be of type SubscriberOptions or an empty sequence." + # Sanity check: Is our goal to use the emulator? # If so, create a grpc insecure channel with the emulator host # as the target. @@ -82,6 +92,32 @@ def __init__(self, **kwargs: Any): self._target = self._transport._host self._closed = False + self.subscriber_options = types.SubscriberOptions(*subscriber_options) + + # Set / override Open Telemetry option. + self._open_telemetry_enabled = ( + self.subscriber_options.enable_open_telemetry_tracing + ) + # OpenTelemetry features used by the library are not supported in Python versions <= 3.7. + # Refer https://github.com/open-telemetry/opentelemetry-python/issues/3993#issuecomment-2211976389 + if ( + self.subscriber_options.enable_open_telemetry_tracing + and sys.version_info.major == 3 + and sys.version_info.minor < 8 + ): + warnings.warn( + message="Open Telemetry for Python version 3.7 or lower is not supported. Disabling Open Telemetry tracing.", + category=RuntimeWarning, + ) + self._open_telemetry_enabled = False + + @property + def open_telemetry_enabled(self) -> bool: + """ + Returns True if Open Telemetry is enabled. False otherwise. + """ + return self._open_telemetry_enabled # pragma: NO COVER + @classmethod def from_service_account_file( # type: ignore[override] cls, filename: str, **kwargs: Any diff --git a/google/cloud/pubsub_v1/subscriber/message.py b/google/cloud/pubsub_v1/subscriber/message.py index f744966a2..61f60c4d9 100644 --- a/google/cloud/pubsub_v1/subscriber/message.py +++ b/google/cloud/pubsub_v1/subscriber/message.py @@ -24,6 +24,9 @@ from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber import futures from google.cloud.pubsub_v1.subscriber.exceptions import AcknowledgeStatus +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) if typing.TYPE_CHECKING: # pragma: NO COVER @@ -85,6 +88,8 @@ class Message(object): information on this type. publish_time (google.protobuf.timestamp_pb2.Timestamp): The time that this message was originally published. + opentelemetry_data (google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry.SubscribeOpenTelemetry) + Open Telemetry data associated with this message. None if Open Telemetry is not enabled. """ def __init__( @@ -144,6 +149,9 @@ def __init__( self._ordering_key = message.ordering_key self._size = message.ByteSize() + # None if Open Telemetry is disabled. Else contains OpenTelemetry data. + self._opentelemetry_data: Optional[SubscribeOpenTelemetry] = None + def __repr__(self): # Get an abbreviated version of the data. abbv_data = self._message.data @@ -158,6 +166,14 @@ def __repr__(self): pretty_attrs = pretty_attrs.lstrip() return _MESSAGE_REPR.format(abbv_data, str(self.ordering_key), pretty_attrs) + @property + def opentelemetry_data(self): + return self._opentelemetry_data # pragma: NO COVER + + @opentelemetry_data.setter + def opentelemetry_data(self, data): + self._opentelemetry_data = data # pragma: NO COVER + @property def attributes(self) -> "containers.ScalarMap": """Return the attributes of the underlying Pub/Sub Message. @@ -252,6 +268,9 @@ def ack(self) -> None: https://cloud.google.com/pubsub/docs/exactly-once-delivery." """ + if self.opentelemetry_data: + self.opentelemetry_data.add_process_span_event("ack called") + self.opentelemetry_data.end_process_span() time_to_ack = math.ceil(time.time() - self._received_timestamp) self._request_queue.put( requests.AckRequest( @@ -260,6 +279,7 @@ def ack(self) -> None: time_to_ack=time_to_ack, ordering_key=self.ordering_key, future=None, + opentelemetry_data=self.opentelemetry_data, ) ) @@ -302,6 +322,9 @@ def ack_with_response(self) -> "futures.Future": pubsub_v1.subscriber.exceptions.AcknowledgeError exception will be thrown. """ + if self.opentelemetry_data: + self.opentelemetry_data.add_process_span_event("ack called") + self.opentelemetry_data.end_process_span() req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): future = futures.Future() @@ -317,6 +340,7 @@ def ack_with_response(self) -> "futures.Future": time_to_ack=time_to_ack, ordering_key=self.ordering_key, future=req_future, + opentelemetry_data=self.opentelemetry_data, ) ) return future @@ -357,7 +381,12 @@ def modify_ack_deadline(self, seconds: int) -> None: against. """ self._request_queue.put( - requests.ModAckRequest(ack_id=self._ack_id, seconds=seconds, future=None) + requests.ModAckRequest( + ack_id=self._ack_id, + seconds=seconds, + future=None, + opentelemetry_data=self.opentelemetry_data, + ) ) def modify_ack_deadline_with_response(self, seconds: int) -> "futures.Future": @@ -416,7 +445,10 @@ def modify_ack_deadline_with_response(self, seconds: int) -> "futures.Future": self._request_queue.put( requests.ModAckRequest( - ack_id=self._ack_id, seconds=seconds, future=req_future + ack_id=self._ack_id, + seconds=seconds, + future=req_future, + opentelemetry_data=self.opentelemetry_data, ) ) @@ -429,12 +461,16 @@ def nack(self) -> None: may take place immediately or after a delay, and may arrive at this subscriber or another. """ + if self.opentelemetry_data: + self.opentelemetry_data.add_process_span_event("nack called") + self.opentelemetry_data.end_process_span() self._request_queue.put( requests.NackRequest( ack_id=self._ack_id, byte_size=self.size, ordering_key=self.ordering_key, future=None, + opentelemetry_data=self.opentelemetry_data, ) ) @@ -472,6 +508,9 @@ def nack_with_response(self) -> "futures.Future": will be thrown. """ + if self.opentelemetry_data: + self.opentelemetry_data.add_process_span_event("nack called") + self.opentelemetry_data.end_process_span() req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): future = futures.Future() @@ -486,6 +525,7 @@ def nack_with_response(self) -> "futures.Future": byte_size=self.size, ordering_key=self.ordering_key, future=req_future, + opentelemetry_data=self.opentelemetry_data, ) ) diff --git a/google/cloud/pubsub_v1/types.py b/google/cloud/pubsub_v1/types.py index c4282e685..7e94a7250 100644 --- a/google/cloud/pubsub_v1/types.py +++ b/google/cloud/pubsub_v1/types.py @@ -131,6 +131,29 @@ class PublishFlowControl(NamedTuple): """The action to take when publish flow control limits are exceeded.""" +# Define the default subscriber options. +# +# This class is used when creating a subscriber client to pass in options +# to enable/disable features. +class SubscriberOptions(NamedTuple): + """ + Options for the subscriber client. + Attributes: + enable_open_telemetry_tracing (bool): + Whether to enable OpenTelemetry tracing. Defaults to False. + """ + + enable_open_telemetry_tracing: bool = False + """ + Whether to enable OpenTelemetry tracing. + + Warning: traces are subject to change. The name and attributes of a span might + change without notice. Only use run traces interactively. Don't use in + automation. Running non-interactive traces can cause problems if the underlying + trace architecture changes without notice. + """ + + # Define the default publisher options. # # This class is used when creating a publisher client to pass in options @@ -175,7 +198,14 @@ class PublisherOptions(NamedTuple): ) enable_open_telemetry_tracing: bool = False # disabled by default - """Open Telemetry tracing is enabled if this is set to True.""" + """ + Open Telemetry tracing is enabled if this is set to True. + + Warning: traces are subject to change. The name and attributes of a span might + change without notice. Only use run traces interactively. Don't use in + automation. Running non-interactive traces can cause problems if the underlying + trace architecture changes without notice. + """ # Define the type class and default values for flow control settings. diff --git a/google/pubsub/gapic_version.py b/google/pubsub/gapic_version.py index 008f4dd36..07de09d56 100644 --- a/google/pubsub/gapic_version.py +++ b/google/pubsub/gapic_version.py @@ -13,4 +13,4 @@ # See the License for the specific language governing permissions and # limitations under the License. # -__version__ = "2.23.1" # {x-release-please-version} +__version__ = "2.24.0" # {x-release-please-version} diff --git a/google/pubsub_v1/gapic_version.py b/google/pubsub_v1/gapic_version.py index 008f4dd36..07de09d56 100644 --- a/google/pubsub_v1/gapic_version.py +++ b/google/pubsub_v1/gapic_version.py @@ -13,4 +13,4 @@ # See the License for the specific language governing permissions and # limitations under the License. # -__version__ = "2.23.1" # {x-release-please-version} +__version__ = "2.24.0" # {x-release-please-version} diff --git a/samples/generated_samples/snippet_metadata_google.pubsub.v1.json b/samples/generated_samples/snippet_metadata_google.pubsub.v1.json index c1602f5ba..fd163b590 100644 --- a/samples/generated_samples/snippet_metadata_google.pubsub.v1.json +++ b/samples/generated_samples/snippet_metadata_google.pubsub.v1.json @@ -8,7 +8,7 @@ ], "language": "PYTHON", "name": "google-cloud-pubsub", - "version": "2.23.1" + "version": "2.24.0" }, "snippets": [ { diff --git a/tests/unit/pubsub_v1/subscriber/test_dispatcher.py b/tests/unit/pubsub_v1/subscriber/test_dispatcher.py index 89d72c61d..5483c48c5 100644 --- a/tests/unit/pubsub_v1/subscriber/test_dispatcher.py +++ b/tests/unit/pubsub_v1/subscriber/test_dispatcher.py @@ -17,11 +17,17 @@ import sys import threading +from opentelemetry import trace + from google.cloud.pubsub_v1.subscriber._protocol import dispatcher from google.cloud.pubsub_v1.subscriber._protocol import helper_threads from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber._protocol import streaming_pull_manager from google.cloud.pubsub_v1.subscriber import futures +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) +from google.pubsub_v1.types import PubsubMessage # special case python < 3.8 if sys.version_info.major == 3 and sys.version_info.minor < 8: @@ -365,6 +371,125 @@ def test_unknown_request_type(): dispatcher_.dispatch_callback(items) +def test_opentelemetry_modify_ack_deadline(span_exporter): + manager = mock.create_autospec( + streaming_pull_manager.StreamingPullManager, instance=True + ) + dispatcher_ = dispatcher.Dispatcher(manager, mock.sentinel.queue) + opentelemetry_data = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + + items = [ + requests.ModAckRequest( + ack_id="ack_id_string", + seconds=60, + future=None, + opentelemetry_data=opentelemetry_data, + ) + ] + manager.send_unary_modack.return_value = (items, []) + dispatcher_.modify_ack_deadline(items) + + # Subscribe span would not have ended as part of a modack. So, end it + # in the test, so that we can export and assert its contents. + opentelemetry_data.end_subscribe_span() + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + subscribe_span = spans[0] + + assert len(subscribe_span.events) == 2 + assert subscribe_span.events[0].name == "modack start" + assert subscribe_span.events[1].name == "modack end" + + +@pytest.mark.skipif( + sys.version_info < (3, 8), + reason="Open Telemetry not supported below Python version 3.8", +) +def test_opentelemetry_ack(span_exporter): + manager = mock.create_autospec( + streaming_pull_manager.StreamingPullManager, instance=True + ) + dispatcher_ = dispatcher.Dispatcher(manager, mock.sentinel.queue) + + data1 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data1.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + data2 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data2.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + items = [ + requests.AckRequest( + ack_id="ack_id_string", + byte_size=0, + time_to_ack=20, + ordering_key="", + future=None, + opentelemetry_data=data1, + ), + requests.AckRequest( + ack_id="ack_id_string2", + byte_size=0, + time_to_ack=20, + ordering_key="", + future=None, + opentelemetry_data=data2, + ), + ] + manager.send_unary_ack.return_value = (items, []) + mock_span_context = mock.Mock(spec=trace.SpanContext) + mock_span_context.trace_flags.sampled = False + with mock.patch.object( + data2._subscribe_span, "get_span_context", return_value=mock_span_context + ): + dispatcher_.ack(items) + + spans = span_exporter.get_finished_spans() + + assert len(spans) == 3 + ack_span = spans[0] + + for subscribe_span in spans[1:]: + assert subscribe_span.attributes["messaging.gcp_pubsub.result"] == "acked" + assert len(subscribe_span.events) == 2 + assert subscribe_span.events[0].name == "ack start" + assert subscribe_span.events[1].name == "ack end" + + # This subscribe span is sampled, so we expect it to be linked to the ack + # span. + assert len(spans[1].links) == 1 + assert spans[1].links[0].context == ack_span.context + assert len(spans[1].links[0].attributes) == 1 + assert spans[1].links[0].attributes["messaging.operation.name"] == "ack" + # This subscribe span is not sampled, so we expect it to not be linked to + # the ack span + assert len(spans[2].links) == 0 + + assert ack_span.name == "subscriptionID ack" + assert ack_span.kind == trace.SpanKind.CLIENT + assert ack_span.parent is None + assert len(ack_span.links) == 1 + assert ack_span.attributes["messaging.system"] == "gcp_pubsub" + assert ack_span.attributes["messaging.batch.message_count"] == 2 + assert ack_span.attributes["messaging.operation"] == "ack" + assert ack_span.attributes["gcp.project_id"] == "projectID" + assert ack_span.attributes["messaging.destination.name"] == "subscriptionID" + assert ack_span.attributes["code.function"] == "ack" + + def test_ack(): manager = mock.create_autospec( streaming_pull_manager.StreamingPullManager, instance=True @@ -481,6 +606,92 @@ def test_retry_acks_in_new_thread(): assert ctor_call.kwargs["daemon"] +@pytest.mark.skipif( + sys.version_info < (3, 8), + reason="Open Telemetry not supported below Python version 3.8", +) +def test_opentelemetry_retry_acks(span_exporter): + manager = mock.create_autospec( + streaming_pull_manager.StreamingPullManager, instance=True + ) + dispatcher_ = dispatcher.Dispatcher(manager, mock.sentinel.queue) + data1 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data1.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + data2 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data2.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + + f = futures.Future() + items = [ + requests.AckRequest( + ack_id="ack_id_string", + byte_size=0, + time_to_ack=20, + ordering_key="", + future=f, + opentelemetry_data=data1, + ), + requests.AckRequest( + ack_id="ack_id_string2", + byte_size=0, + time_to_ack=20, + ordering_key="", + future=f, + opentelemetry_data=data2, + ), + ] + manager.send_unary_ack.side_effect = [(items, [])] + mock_span_context = mock.Mock(spec=trace.SpanContext) + mock_span_context.trace_flags.sampled = False + with mock.patch("time.sleep", return_value=None): + with mock.patch.object( + data2._subscribe_span, "get_span_context", return_value=mock_span_context + ): + dispatcher_._retry_acks(items) + + spans = span_exporter.get_finished_spans() + + assert len(spans) == 3 + ack_span = spans[0] + + for subscribe_span in spans[1:]: + assert "messaging.gcp_pubsub.result" in subscribe_span.attributes + assert subscribe_span.attributes["messaging.gcp_pubsub.result"] == "acked" + assert len(subscribe_span.events) == 2 + assert subscribe_span.events[0].name == "ack start" + assert subscribe_span.events[1].name == "ack end" + + # This subscribe span is sampled, so we expect it to be linked to the ack + # span. + assert len(spans[1].links) == 1 + assert spans[1].links[0].context == ack_span.context + assert len(spans[1].links[0].attributes) == 1 + assert spans[1].links[0].attributes["messaging.operation.name"] == "ack" + # This subscribe span is not sampled, so we expect it to not be linked to + # the ack span + assert len(spans[2].links) == 0 + + assert ack_span.name == "subscriptionID ack" + assert ack_span.kind == trace.SpanKind.CLIENT + assert ack_span.parent is None + assert len(ack_span.links) == 1 + assert ack_span.attributes["messaging.system"] == "gcp_pubsub" + assert ack_span.attributes["messaging.batch.message_count"] == 2 + assert ack_span.attributes["messaging.operation"] == "ack" + assert ack_span.attributes["gcp.project_id"] == "projectID" + assert ack_span.attributes["messaging.destination.name"] == "subscriptionID" + assert ack_span.attributes["code.function"] == "ack" + + def test_retry_acks(): manager = mock.create_autospec( streaming_pull_manager.StreamingPullManager, instance=True @@ -544,6 +755,125 @@ def test_retry_modacks_in_new_thread(): assert ctor_call.kwargs["daemon"] +def test_opentelemetry_retry_modacks(span_exporter): + manager = mock.create_autospec( + streaming_pull_manager.StreamingPullManager, instance=True + ) + dispatcher_ = dispatcher.Dispatcher(manager, mock.sentinel.queue) + + opentelemetry_data = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + + f = futures.Future() + items = [ + requests.ModAckRequest( + ack_id="ack_id_string", + seconds=20, + future=f, + opentelemetry_data=opentelemetry_data, + ) + ] + manager.send_unary_modack.side_effect = [(items, [])] + with mock.patch("time.sleep", return_value=None): + dispatcher_._retry_modacks(items) + + # Subscribe span wouldn't be ended for modacks. So, end it in the test, so + # that we can export and assert its contents. + opentelemetry_data.end_subscribe_span() + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + subscribe_span = spans[0] + + assert len(subscribe_span.events) == 1 + assert subscribe_span.events[0].name == "modack end" + + +@pytest.mark.skipif( + sys.version_info < (3, 8), + reason="Open Telemetry not supported below Python version 3.8", +) +def test_opentelemetry_retry_nacks(span_exporter): + manager = mock.create_autospec( + streaming_pull_manager.StreamingPullManager, instance=True + ) + dispatcher_ = dispatcher.Dispatcher(manager, mock.sentinel.queue) + + data1 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data1.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id1", + delivery_attempt=5, + ) + data2 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data2.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id2", + delivery_attempt=5, + ) + + f = futures.Future() + items = [ + requests.ModAckRequest( + ack_id="ack_id1", + seconds=0, + future=f, + opentelemetry_data=data1, + ), + requests.ModAckRequest( + ack_id="ack_id2", + seconds=0, + future=f, + opentelemetry_data=data2, + ), + ] + manager.send_unary_modack.side_effect = [(items, [])] + mock_span_context = mock.Mock(spec=trace.SpanContext) + mock_span_context.trace_flags.sampled = False + with mock.patch("time.sleep", return_value=None): + with mock.patch.object( + data2._subscribe_span, "get_span_context", return_value=mock_span_context + ): + dispatcher_._retry_modacks(items) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 3 + nack_span = spans[0] + + for subscribe_span in spans[1:]: + assert "messaging.gcp_pubsub.result" in subscribe_span.attributes + assert subscribe_span.attributes["messaging.gcp_pubsub.result"] == "nacked" + assert len(subscribe_span.events) == 1 + assert subscribe_span.events[0].name == "nack end" + + # This subscribe span is sampled, so we expect it to be linked to the nack + # span. + assert len(spans[1].links) == 1 + assert spans[1].links[0].context == nack_span.context + assert len(spans[1].links[0].attributes) == 1 + assert spans[1].links[0].attributes["messaging.operation.name"] == "nack" + # This subscribe span is not sampled, so we expect it to not be linked to + # the nack span + assert len(spans[2].links) == 0 + + assert nack_span.name == "subscriptionID nack" + assert nack_span.kind == trace.SpanKind.CLIENT + assert nack_span.parent is None + assert len(nack_span.links) == 1 + assert nack_span.attributes["messaging.system"] == "gcp_pubsub" + assert nack_span.attributes["messaging.batch.message_count"] == 2 + assert nack_span.attributes["messaging.operation"] == "nack" + assert nack_span.attributes["gcp.project_id"] == "projectID" + assert nack_span.attributes["messaging.destination.name"] == "subscriptionID" + assert nack_span.attributes["code.function"] == "modify_ack_deadline" + + def test_retry_modacks(): manager = mock.create_autospec( streaming_pull_manager.StreamingPullManager, instance=True @@ -633,6 +963,103 @@ def test_drop_ordered_messages(): manager.maybe_resume_consumer.assert_called_once() +@pytest.mark.skipif( + sys.version_info < (3, 8), + reason="Open Telemetry not supported below Python version 3.8", +) +def test_opentelemetry_nack(span_exporter): + manager = mock.create_autospec( + streaming_pull_manager.StreamingPullManager, instance=True + ) + dispatcher_ = dispatcher.Dispatcher(manager, mock.sentinel.queue) + + data1 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data1.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=5, + ) + data2 = SubscribeOpenTelemetry(message=PubsubMessage(data=b"foo")) + data2.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id2", + delivery_attempt=5, + ) + + items = [ + requests.NackRequest( + ack_id="ack_id", + byte_size=10, + ordering_key="", + future=None, + opentelemetry_data=data1, + ), + requests.NackRequest( + ack_id="ack_id2", + byte_size=10, + ordering_key="", + future=None, + opentelemetry_data=data2, + ), + ] + response_items = [ + requests.ModAckRequest( + ack_id="ack_id", + seconds=0, + future=None, + opentelemetry_data=data1, + ), + requests.ModAckRequest( + ack_id="ack_id2", + seconds=0, + future=None, + opentelemetry_data=data2, + ), + ] + manager.send_unary_modack.return_value = (response_items, []) + + mock_span_context = mock.Mock(spec=trace.SpanContext) + mock_span_context.trace_flags.sampled = False + with mock.patch.object( + data2._subscribe_span, "get_span_context", return_value=mock_span_context + ): + dispatcher_.nack(items) + + spans = span_exporter.get_finished_spans() + + assert len(spans) == 3 + nack_span = spans[0] + for subscribe_span in spans[1:]: + assert "messaging.gcp_pubsub.result" in subscribe_span.attributes + assert subscribe_span.attributes["messaging.gcp_pubsub.result"] == "nacked" + assert len(subscribe_span.events) == 2 + assert subscribe_span.events[0].name == "nack start" + assert subscribe_span.events[1].name == "nack end" + + # This subscribe span is sampled, so we expect it to be linked to the nack + # span. + assert len(spans[1].links) == 1 + assert spans[1].links[0].context == nack_span.context + assert len(spans[1].links[0].attributes) == 1 + assert spans[1].links[0].attributes["messaging.operation.name"] == "nack" + # This subscribe span is not sampled, so we expect it to not be linked to + # the nack span + assert len(spans[2].links) == 0 + + assert nack_span.name == "subscriptionID nack" + assert nack_span.kind == trace.SpanKind.CLIENT + assert nack_span.parent is None + assert len(nack_span.links) == 1 + assert nack_span.attributes["messaging.system"] == "gcp_pubsub" + assert nack_span.attributes["messaging.batch.message_count"] == 2 + assert nack_span.attributes["messaging.operation"] == "nack" + assert nack_span.attributes["gcp.project_id"] == "projectID" + assert nack_span.attributes["messaging.destination.name"] == "subscriptionID" + assert nack_span.attributes["code.function"] == "modify_ack_deadline" + + def test_nack(): manager = mock.create_autospec( streaming_pull_manager.StreamingPullManager, instance=True diff --git a/tests/unit/pubsub_v1/subscriber/test_leaser.py b/tests/unit/pubsub_v1/subscriber/test_leaser.py index f38717c6f..b5b5cac20 100644 --- a/tests/unit/pubsub_v1/subscriber/test_leaser.py +++ b/tests/unit/pubsub_v1/subscriber/test_leaser.py @@ -22,6 +22,10 @@ from google.cloud.pubsub_v1.subscriber._protocol import leaser from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber._protocol import streaming_pull_manager +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) +from google.cloud.pubsub_v1.subscriber import message # special case python < 3.8 if sys.version_info.major == 3 and sys.version_info.minor < 8: @@ -136,6 +140,101 @@ def trigger_done(timeout): leaser._stop_event.wait = trigger_done +def test_opentelemetry_dropped_message_process_span(span_exporter): + manager = create_manager() + leaser_ = leaser.Leaser(manager) + make_sleep_mark_event_as_done(leaser_) + msg = mock.create_autospec( + message.Message, instance=True, ack_id="ack_foo", size=10 + ) + msg.message_id = 3 + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + opentelemetry_data.start_process_span() + leaser_.add( + [ + requests.LeaseRequest( + ack_id="my ack id", + byte_size=50, + ordering_key="", + opentelemetry_data=opentelemetry_data, + ) + ] + ) + leased_messages_dict = leaser_._leased_messages + + # Setting the `sent_time`` to be less than `cutoff` in order to make the leased message expire. + # This will exercise the code path where the message would be dropped from the leaser + leased_messages_dict["my ack id"] = leased_messages_dict["my ack id"]._replace( + sent_time=0 + ) + + manager._send_lease_modacks.return_value = set() + leaser_.maintain_leases() + + opentelemetry_data.end_subscribe_span() + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + process_span, subscribe_span = spans + + assert process_span.name == "subscriptionID process" + assert subscribe_span.name == "subscriptionID subscribe" + + assert len(process_span.events) == 1 + assert process_span.events[0].name == "expired" + + assert process_span.parent == subscribe_span.context + + +def test_opentelemetry_expired_message_exactly_once_process_span(span_exporter): + manager = create_manager() + leaser_ = leaser.Leaser(manager) + make_sleep_mark_event_as_done(leaser_) + msg = mock.create_autospec( + message.Message, instance=True, ack_id="ack_foo", size=10 + ) + msg.message_id = 3 + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id", + delivery_attempt=4, + ) + opentelemetry_data.start_process_span() + leaser_.add( + [ + requests.LeaseRequest( + ack_id="my ack id", + byte_size=50, + ordering_key="", + opentelemetry_data=opentelemetry_data, + ) + ] + ) + + manager._send_lease_modacks.return_value = ["my ack id"] + leaser_.maintain_leases() + + opentelemetry_data.end_subscribe_span() + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + process_span, subscribe_span = spans + + assert process_span.name == "subscriptionID process" + assert subscribe_span.name == "subscriptionID subscribe" + + assert len(process_span.events) == 1 + assert process_span.events[0].name == "expired" + + assert process_span.parent == subscribe_span.context + + def test_maintain_leases_ack_ids(): manager = create_manager() leaser_ = leaser.Leaser(manager) diff --git a/tests/unit/pubsub_v1/subscriber/test_message.py b/tests/unit/pubsub_v1/subscriber/test_message.py index 49b07b7fd..8d9d2566e 100644 --- a/tests/unit/pubsub_v1/subscriber/test_message.py +++ b/tests/unit/pubsub_v1/subscriber/test_message.py @@ -29,6 +29,9 @@ from google.protobuf import timestamp_pb2 from google.pubsub_v1 import types as gapic_types from google.cloud.pubsub_v1.subscriber.exceptions import AcknowledgeStatus +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) RECEIVED = datetime.datetime(2012, 4, 21, 15, 0, tzinfo=datetime.timezone.utc) @@ -131,6 +134,155 @@ def check_call_types(mock, *args, **kwargs): assert isinstance(call_args[n], argtype) +def test_opentelemetry_ack(span_exporter): + SUBSCRIPTION = "projects/projectID/subscriptions/subscriptionID" + msg = create_message(b"data", ack_id="ack_id") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription=SUBSCRIPTION, + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=2, + ) + opentelemetry_data.start_process_span() + msg.opentelemetry_data = opentelemetry_data + msg.ack() + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + process_span, subscribe_span = spans + + assert subscribe_span.name == "subscriptionID subscribe" + assert len(subscribe_span.events) == 0 + + assert process_span.name == "subscriptionID process" + assert len(process_span.events) == 1 + assert process_span.events[0].name == "ack called" + + +def test_opentelemetry_ack_with_response(span_exporter): + SUBSCRIPTION = "projects/projectID/subscriptions/subscriptionID" + msg = create_message(b"data", ack_id="ack_id") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription=SUBSCRIPTION, + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=2, + ) + opentelemetry_data.start_process_span() + msg.opentelemetry_data = opentelemetry_data + msg.ack_with_response() + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + process_span, subscribe_span = spans + + assert subscribe_span.name == "subscriptionID subscribe" + assert len(subscribe_span.events) == 0 + + assert process_span.name == "subscriptionID process" + assert len(process_span.events) == 1 + assert process_span.events[0].name == "ack called" + + +def test_opentelemetry_nack(span_exporter): + SUBSCRIPTION = "projects/projectID/subscriptions/subscriptionID" + msg = create_message(b"data", ack_id="ack_id") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription=SUBSCRIPTION, + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=2, + ) + opentelemetry_data.start_process_span() + msg.opentelemetry_data = opentelemetry_data + msg.nack() + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + process_span, subscribe_span = spans + + assert subscribe_span.name == "subscriptionID subscribe" + assert len(subscribe_span.events) == 0 + + assert process_span.name == "subscriptionID process" + assert len(process_span.events) == 1 + assert process_span.events[0].name == "nack called" + + +def test_opentelemetry_nack_with_response(span_exporter): + SUBSCRIPTION = "projects/projectID/subscriptions/subscriptionID" + msg = create_message(b"data", ack_id="ack_id") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription=SUBSCRIPTION, + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=2, + ) + opentelemetry_data.start_process_span() + msg.opentelemetry_data = opentelemetry_data + msg.nack_with_response() + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + + process_span, subscribe_span = spans + + assert subscribe_span.name == "subscriptionID subscribe" + assert len(subscribe_span.events) == 0 + + assert process_span.name == "subscriptionID process" + assert len(process_span.events) == 1 + assert process_span.events[0].name == "nack called" + + +def test_opentelemetry_modack(span_exporter): + SUBSCRIPTION = "projects/projectID/subscriptions/subscriptionID" + msg = create_message(b"data", ack_id="ack_id") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription=SUBSCRIPTION, + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=2, + ) + msg.opentelemetry_data = opentelemetry_data + msg.modify_ack_deadline(3) + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + + assert len(spans[0].events) == 0 + + +def test_opentelemetry_modack_with_response(span_exporter): + SUBSCRIPTION = "projects/projectID/subscriptions/subscriptionID" + msg = create_message(b"data", ack_id="ack_id") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription=SUBSCRIPTION, + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=2, + ) + msg.opentelemetry_data = opentelemetry_data + msg.modify_ack_deadline_with_response(3) + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + + assert len(spans[0].events) == 0 + + def test_ack(): msg = create_message(b"foo", ack_id="bogus_ack_id") with mock.patch.object(msg._request_queue, "put") as put: diff --git a/tests/unit/pubsub_v1/subscriber/test_messages_on_hold.py b/tests/unit/pubsub_v1/subscriber/test_messages_on_hold.py index 5e1dcf91b..64963de48 100644 --- a/tests/unit/pubsub_v1/subscriber/test_messages_on_hold.py +++ b/tests/unit/pubsub_v1/subscriber/test_messages_on_hold.py @@ -14,9 +14,14 @@ import queue +from opentelemetry import trace + +from google.pubsub_v1 import types as gapic_types from google.cloud.pubsub_v1.subscriber import message from google.cloud.pubsub_v1.subscriber._protocol import messages_on_hold -from google.pubsub_v1 import types as gapic_types +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) def make_message(ack_id, ordering_key): @@ -37,6 +42,32 @@ def test_init(): assert moh.get() is None +def test_opentelemetry_subscriber_scheduler_span(span_exporter): + moh = messages_on_hold.MessagesOnHold() + msg = make_message(ack_id="ack1", ordering_key="") + opentelemetry_data = SubscribeOpenTelemetry(msg) + msg.opentelemetry_data = opentelemetry_data + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + moh.put(msg) + opentelemetry_data.end_subscribe_scheduler_span() + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + + assert len(spans) == 2 + + subscribe_scheduler_span, subscribe_span = spans + + assert subscribe_scheduler_span.name == "subscriber scheduler" + assert subscribe_scheduler_span.kind == trace.SpanKind.INTERNAL + assert subscribe_scheduler_span.parent == subscribe_span.context + + def test_put_and_get_unordered_messages(): moh = messages_on_hold.MessagesOnHold() diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index d4ce2cfdb..4d2d1b98e 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py @@ -18,6 +18,20 @@ import threading import time import types as stdlib_types +import datetime +import queue +import math + +from opentelemetry import trace +from google.protobuf import timestamp_pb2 +from google.api_core import datetime_helpers + +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) +from google.cloud.pubsub_v1.subscriber.message import Message +from google.cloud.pubsub_v1.types import PubsubMessage + # special case python < 3.8 if sys.version_info.major == 3 and sys.version_info.minor < 8: @@ -179,11 +193,16 @@ def test_constructor_with_max_duration_per_lease_extension_too_high(): assert manager._stream_ack_deadline == 600 -def make_manager(**kwargs): +def make_manager( + enable_open_telemetry: bool = False, + subscription_name: str = "subscription-name", + **kwargs, +): client_ = mock.create_autospec(client.Client, instance=True) + client_.open_telemetry_enabled = enable_open_telemetry scheduler_ = mock.create_autospec(scheduler.Scheduler, instance=True) return streaming_pull_manager.StreamingPullManager( - client_, "subscription-name", scheduler=scheduler_, **kwargs + client_, subscription_name, scheduler=scheduler_, **kwargs ) @@ -509,6 +528,45 @@ def test__maybe_release_messages_on_overload(): manager._scheduler.schedule.assert_not_called() +def test_opentelemetry__maybe_release_messages_subscribe_scheduler_span(span_exporter): + manager = make_manager( + flow_control=types.FlowControl(max_messages=10, max_bytes=1000) + ) + manager._callback = mock.sentinel.callback + + # Init leaser message count to 11, so that when subtracting the 3 messages + # that are on hold, there is still room for another 2 messages before the + # max load is hit. + _leaser = manager._leaser = mock.create_autospec(leaser.Leaser) + fake_leaser_add(_leaser, init_msg_count=8, assumed_msg_size=10) + msg = mock.create_autospec( + message.Message, instance=True, ack_id="ack_foo", size=10 + ) + msg.message_id = 3 + opentelemetry_data = SubscribeOpenTelemetry(msg) + msg.opentelemetry_data = opentelemetry_data + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + manager._messages_on_hold.put(msg) + manager._maybe_release_messages() + opentelemetry_data.end_subscribe_span() + spans = span_exporter.get_finished_spans() + + assert len(spans) == 2 + + subscriber_scheduler_span, subscribe_span = spans + + assert subscriber_scheduler_span.name == "subscriber scheduler" + assert subscribe_span.name == "subscriptionID subscribe" + + assert subscriber_scheduler_span.parent == subscribe_span.context + assert subscriber_scheduler_span.kind == trace.SpanKind.INTERNAL + + def test__maybe_release_messages_below_overload(): manager = make_manager( flow_control=types.FlowControl(max_messages=10, max_bytes=1000) @@ -574,6 +632,86 @@ def test__maybe_release_messages_negative_on_hold_bytes_warning(caplog): assert manager._on_hold_bytes == 0 # should be auto-corrected +@pytest.mark.skipif( + sys.version_info < (3, 8), + reason="Open Telemetry not supported below Python version 3.8", +) +@pytest.mark.parametrize( + "receipt_modack", + [ + True, + False, + ], +) +def test_opentelemetry__send_lease_modacks(span_exporter, receipt_modack): + manager, _, _, _, _, _ = make_running_manager( + enable_open_telemetry=True, + subscription_name="projects/projectID/subscriptions/subscriptionID", + ) + data1 = SubscribeOpenTelemetry( + message=gapic_types.PubsubMessage(data=b"foo", message_id="1") + ) + data2 = SubscribeOpenTelemetry( + message=gapic_types.PubsubMessage(data=b"bar", message_id="2") + ) + + data1.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id1", + delivery_attempt=2, + ) + data2.start_subscribe_span( + subscription="projects/projectID/subscriptions/subscriptionID", + exactly_once_enabled=True, + ack_id="ack_id1", + delivery_attempt=2, + ) + mock_span_context = mock.Mock(spec=trace.SpanContext) + mock_span_context.trace_flags.sampled = False + with mock.patch.object( + data1._subscribe_span, "get_span_context", return_value=mock_span_context + ): + manager._send_lease_modacks( + ack_ids=["ack_id1", "ack_id2"], + ack_deadline=20, + opentelemetry_data=[data1, data2], + receipt_modack=receipt_modack, + ) + data1.end_subscribe_span() + data2.end_subscribe_span() + spans = span_exporter.get_finished_spans() + assert len(spans) == 3 + modack_span, subscribe_span1, subscribe_span2 = spans + + assert len(subscribe_span1.events) == 0 + assert len(subscribe_span2.events) == 0 + + assert len(subscribe_span1.links) == 0 + assert len(subscribe_span2.links) == 1 + assert subscribe_span2.links[0].context == modack_span.context + assert subscribe_span2.links[0].attributes["messaging.operation.name"] == "modack" + + assert modack_span.name == "subscriptionID modack" + assert modack_span.parent is None + assert modack_span.kind == trace.SpanKind.CLIENT + assert len(modack_span.links) == 1 + modack_span_attributes = modack_span.attributes + assert modack_span_attributes["messaging.system"] == "gcp_pubsub" + assert modack_span_attributes["messaging.batch.message_count"] == 2 + assert math.isclose( + modack_span_attributes["messaging.gcp_pubsub.message.ack_deadline"], 20 + ) + assert modack_span_attributes["messaging.destination.name"] == "subscriptionID" + assert modack_span_attributes["gcp.project_id"] == "projectID" + assert modack_span_attributes["messaging.operation.name"] == "modack" + assert modack_span_attributes["code.function"] == "_send_lease_modacks" + assert ( + modack_span_attributes["messaging.gcp_pubsub.is_receipt_modack"] + == receipt_modack + ) + + def test_send_unary_ack(): manager = make_manager() @@ -1224,14 +1362,17 @@ def test_open_has_been_closed(): manager.open(mock.sentinel.callback, mock.sentinel.on_callback_error) -def make_running_manager(**kwargs): - manager = make_manager(**kwargs) +def make_running_manager( + enable_open_telemetry: bool = False, + subscription_name: str = "subscription-name", + **kwargs, +): + manager = make_manager(enable_open_telemetry, subscription_name, **kwargs) manager._consumer = mock.create_autospec(bidi.BackgroundConsumer, instance=True) manager._consumer.is_active = True manager._dispatcher = mock.create_autospec(dispatcher.Dispatcher, instance=True) manager._leaser = mock.create_autospec(leaser.Leaser, instance=True) manager._heartbeater = mock.create_autospec(heartbeater.Heartbeater, instance=True) - return ( manager, manager._consumer, @@ -2626,3 +2767,148 @@ def test_process_requests_mixed_success_and_failure_modacks(): # message with ack_id 'ackid3' succeeds assert requests_completed[1].ack_id == "ackid3" assert future3.result() == subscriber_exceptions.AcknowledgeStatus.SUCCESS + + +@pytest.mark.skipif( + sys.version_info < (3, 8), + reason="Open Telemetry not supported below Python version 3.8", +) +def test_opentelemetry__on_response_subscribe_span_create(span_exporter): + manager, _, _, leaser, _, _ = make_running_manager( + enable_open_telemetry=True, + subscription_name="projects/projectID/subscriptions/subscriptionID", + ) + + fake_leaser_add(leaser, init_msg_count=0, assumed_msg_size=42) + manager._callback = mock.sentinel.callback + + response = gapic_types.StreamingPullResponse( + received_messages=[ + gapic_types.ReceivedMessage( + ack_id="ack1", + message=gapic_types.PubsubMessage(data=b"foo", message_id="1"), + ), + gapic_types.ReceivedMessage( + ack_id="ack2", + message=gapic_types.PubsubMessage(data=b"bar", message_id="2"), + delivery_attempt=6, + ), + ] + ) + + manager._on_response(response) + + spans = span_exporter.get_finished_spans() + + # Subscribe span is still active, hence unexported. + # Subscriber scheduler spans corresponding to the two messages would be started in `messages_on_hold.put()`` + # and ended in `_maybe_release_messages` + assert len(spans) == 3 + modack_span = spans[0] + + for span in spans[1:]: + assert span.name == "subscriber scheduler" + assert span.kind == trace.SpanKind.INTERNAL + assert span.parent is not None + assert len(span.attributes) == 0 + + assert modack_span.name == "subscriptionID modack" + assert modack_span.kind == trace.SpanKind.CLIENT + assert modack_span.parent is None + assert len(modack_span.links) == 2 + + +RECEIVED = datetime.datetime(2012, 4, 21, 15, 0, tzinfo=datetime.timezone.utc) +RECEIVED_SECONDS = datetime_helpers.to_milliseconds(RECEIVED) // 1000 +PUBLISHED_MICROS = 123456 +PUBLISHED = RECEIVED + datetime.timedelta(days=1, microseconds=PUBLISHED_MICROS) +PUBLISHED_SECONDS = datetime_helpers.to_milliseconds(PUBLISHED) // 1000 + + +def create_message( + data, + ack_id="ACKID", + delivery_attempt=0, + ordering_key="", + exactly_once_delivery_enabled=False, + **attrs, +): # pragma: NO COVER + with mock.patch.object(time, "time") as time_: + time_.return_value = RECEIVED_SECONDS + gapic_pubsub_message = PubsubMessage( + attributes=attrs, + data=data, + message_id="message_id", + publish_time=timestamp_pb2.Timestamp( + seconds=PUBLISHED_SECONDS, nanos=PUBLISHED_MICROS * 1000 + ), + ordering_key=ordering_key, + ) + msg = Message( + # The code under test uses a raw protobuf PubsubMessage, i.e. w/o additional + # Python class wrappers, hence the "_pb" + message=gapic_pubsub_message._pb, + ack_id=ack_id, + delivery_attempt=delivery_attempt, + request_queue=queue.Queue(), + exactly_once_delivery_enabled_func=lambda: exactly_once_delivery_enabled, + ) + return msg + + +def test_opentelemetry_subscriber_concurrency_control_span(span_exporter): + manager, _, _, leaser, _, _ = make_running_manager( + enable_open_telemetry=True, + subscription_name="projects/projectID/subscriptions/subscriptionID", + ) + manager._callback = mock.Mock() + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + msg.opentelemetry_data = opentelemetry_data + manager._schedule_message_on_hold(msg) + opentelemetry_data.end_subscribe_concurrency_control_span() + opentelemetry_data.end_subscribe_span() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + + concurrency_control_span, subscribe_span = spans + assert concurrency_control_span.name == "subscriber concurrency control" + assert subscribe_span.name == "subscriptionID subscribe" + assert opentelemetry_data.subscription_id == "subscriptionID" + + assert concurrency_control_span.parent == subscribe_span.context + + +def test_opentelemetry_subscriber_concurrency_control_span_end(span_exporter): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + opentelemetry_data.start_subscribe_concurrency_control_span() + msg.opentelemetry_data = opentelemetry_data + streaming_pull_manager._wrap_callback_errors(mock.Mock(), mock.Mock(), msg) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + + concurrency_control_span = spans[0] + concurrency_control_span.name == "subscriber concurrency control" + + +def test_opentelemetry_wrap_callback_error(span_exporter): + msg = create_message(b"foo") + streaming_pull_manager._wrap_callback_errors(mock.Mock(), mock.Mock(), msg) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 0 diff --git a/tests/unit/pubsub_v1/subscriber/test_subscribe_opentelemetry.py b/tests/unit/pubsub_v1/subscriber/test_subscribe_opentelemetry.py new file mode 100644 index 000000000..2fb89aa7c --- /dev/null +++ b/tests/unit/pubsub_v1/subscriber/test_subscribe_opentelemetry.py @@ -0,0 +1,202 @@ +# Copyright 2024, Google LLC All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# 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. + +import datetime +import time +import sys +import queue +import pytest + +from google.protobuf import timestamp_pb2 +from google.api_core import datetime_helpers +from opentelemetry import trace +from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator +from google.cloud.pubsub_v1.open_telemetry.context_propagation import ( + OpenTelemetryContextSetter, +) + +from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( + SubscribeOpenTelemetry, +) +from google.cloud.pubsub_v1.subscriber.message import Message +from google.cloud.pubsub_v1.types import PubsubMessage + +# special case python < 3.8 +if sys.version_info.major == 3 and sys.version_info.minor < 8: + import mock +else: + from unittest import mock + +RECEIVED = datetime.datetime(2012, 4, 21, 15, 0, tzinfo=datetime.timezone.utc) +RECEIVED_SECONDS = datetime_helpers.to_milliseconds(RECEIVED) // 1000 +PUBLISHED_MICROS = 123456 +PUBLISHED = RECEIVED + datetime.timedelta(days=1, microseconds=PUBLISHED_MICROS) +PUBLISHED_SECONDS = datetime_helpers.to_milliseconds(PUBLISHED) // 1000 + + +def create_message( + data, + ack_id="ACKID", + delivery_attempt=0, + ordering_key="", + exactly_once_delivery_enabled=False, + **attrs +): # pragma: NO COVER + with mock.patch.object(time, "time") as time_: + time_.return_value = RECEIVED_SECONDS + gapic_pubsub_message = PubsubMessage( + attributes=attrs, + data=data, + message_id="message_id", + publish_time=timestamp_pb2.Timestamp( + seconds=PUBLISHED_SECONDS, nanos=PUBLISHED_MICROS * 1000 + ), + ordering_key=ordering_key, + ) + msg = Message( + # The code under test uses a raw protobuf PubsubMessage, i.e. w/o additional + # Python class wrappers, hence the "_pb" + message=gapic_pubsub_message._pb, + ack_id=ack_id, + delivery_attempt=delivery_attempt, + request_queue=queue.Queue(), + exactly_once_delivery_enabled_func=lambda: exactly_once_delivery_enabled, + ) + return msg + + +def test_opentelemetry_set_subscribe_span_result(span_exporter): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + msg.opentelemetry_data = opentelemetry_data + opentelemetry_data.set_subscribe_span_result("acked") + opentelemetry_data.end_subscribe_span() + spans = span_exporter.get_finished_spans() + + assert len(spans) == 1 + + assert "messaging.gcp_pubsub.result" in spans[0].attributes + assert spans[0].attributes["messaging.gcp_pubsub.result"] == "acked" + + +def test_opentelemetry_set_subscribe_span_result_assert_error(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.set_subscribe_span_result("hi") + + +def test_opentelemetry_start_subscribe_concurrency_control_span_no_subscribe_span(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.start_subscribe_concurrency_control_span() + + +def test_opentelemetry_end_subscribe_concurrency_control_span_assertion_error(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.end_subscribe_concurrency_control_span() + + +def test_opentelemetry_start_subscribe_scheduler_span_assertion_error(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.start_subscribe_scheduler_span() + + +def test_opentelemetry_end_subscribe_scheduler_span_assertion_error(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.end_subscribe_scheduler_span() + + +def test_opentelemetry_start_process_span_assertion_error(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.start_process_span() + + +def test_opentelemetry_end_process_span_assertion_error(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + with pytest.raises(AssertionError): + opentelemetry_data.end_process_span() + + +def test_opentelemetry_start_process_span_publisher_link(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + msg.opentelemetry_data = opentelemetry_data + tracer = trace.get_tracer("foo") + publisher_create_span = None + with tracer.start_as_current_span(name="name") as span: + publisher_create_span = span + TraceContextTextMapPropagator().inject( + carrier=msg._message, + setter=OpenTelemetryContextSetter(), + ) + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + opentelemetry_data.start_process_span() + assert len(opentelemetry_data._process_span.links) == 1 + assert ( + opentelemetry_data._process_span.links[0].context.span_id + == publisher_create_span.get_span_context().span_id + ) + + +def test_opentelemetry_start_process_span_no_publisher_span(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + msg.opentelemetry_data = opentelemetry_data + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + opentelemetry_data.start_process_span() + # Assert that when no context is propagated, the subscriber span has no parent. + assert opentelemetry_data._subscribe_span.parent is None + # Assert that when there is no publisher create span context propagated, + # There are no links created in the process span. + assert len(opentelemetry_data._process_span.links) == 0 + + +def test_opentelemetry_project_id_set_after_create_subscribe_span(): + msg = create_message(b"foo") + opentelemetry_data = SubscribeOpenTelemetry(msg) + msg.opentelemetry_data = opentelemetry_data + opentelemetry_data.start_subscribe_span( + subscription="projects/projectId/subscriptions/subscriptionID", + exactly_once_enabled=False, + ack_id="ack_id", + delivery_attempt=4, + ) + assert opentelemetry_data.project_id == "projectId" diff --git a/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py b/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py index 16a6150af..7c0ebfd83 100644 --- a/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py +++ b/tests/unit/pubsub_v1/subscriber/test_subscriber_client.py @@ -30,6 +30,10 @@ from google.cloud.pubsub_v1.subscriber import futures from google.pubsub_v1.services.subscriber import client as subscriber_client from google.pubsub_v1.services.subscriber.transports.grpc import SubscriberGrpcTransport +from google.cloud.pubsub_v1.open_telemetry.context_propagation import ( + OpenTelemetryContextGetter, +) +from google.pubsub_v1.types import PubsubMessage def test_init_default_client_info(creds): @@ -317,3 +321,48 @@ async def test_sync_pull_warning_if_return_immediately_async(creds): warning_msg = str(warned[0].message) assert "return_immediately" in warning_msg assert "deprecated" in warning_msg + + +@pytest.mark.parametrize( + "enable_open_telemetry", + [ + True, + False, + ], +) +def test_opentelemetry_subscriber_setting(creds, enable_open_telemetry): + options = types.SubscriberOptions( + enable_open_telemetry_tracing=enable_open_telemetry, + ) + if sys.version_info >= (3, 8) or enable_open_telemetry is False: + client = subscriber.Client(credentials=creds, subscriber_options=options) + assert client.subscriber_options == options + assert client._open_telemetry_enabled == enable_open_telemetry + else: + with pytest.warns( + RuntimeWarning, + match="Open Telemetry for Python version 3.7 or lower is not supported. Disabling Open Telemetry tracing.", + ): + client = subscriber.Client(credentials=creds, subscriber_options=options) + assert client._open_telemetry_enabled is False + + +def test_opentelemetry_propagator_get(): + message = PubsubMessage(data=b"foo") + message.attributes["key1"] = "value1" + message.attributes["googclient_key2"] = "value2" + + assert OpenTelemetryContextGetter().get(message, "key2") == ["value2"] + + assert OpenTelemetryContextGetter().get(message, "key1") is None + + +def test_opentelemetry_propagator_keys(): + message = PubsubMessage(data=b"foo") + message.attributes["key1"] = "value1" + message.attributes["googclient_key2"] = "value2" + + assert sorted(OpenTelemetryContextGetter().keys(message)) == [ + "googclient_key2", + "key1", + ]