Set gRPC user-agent when calling google APIs (#216)

aabmass · web-flow · commit 4d581e955257 · 2022-12-05T11:57:17.000-05:00
diff --git a/opentelemetry-exporter-gcp-monitoring/src/opentelemetry/exporter/cloud_monitoring/__init__.py b/opentelemetry-exporter-gcp-monitoring/src/opentelemetry/exporter/cloud_monitoring/__init__.py
@@ -19,11 +19,11 @@
 from typing import Dict, List, NoReturn, Optional, Set, Union
 
 import google.auth
+import pkg_resources
 from google.api.distribution_pb2 import Distribution
 from google.api.label_pb2 import LabelDescriptor
 from google.api.metric_pb2 import Metric as GMetric
 from google.api.metric_pb2 import MetricDescriptor
-from google.api.monitored_resource_pb2 import MonitoredResource
 from google.cloud.monitoring_v3 import (
     CreateMetricDescriptorRequest,
     CreateTimeSeriesRequest,
@@ -33,12 +33,16 @@
     TimeSeries,
     TypedValue,
 )
+from google.cloud.monitoring_v3.services.metric_service.transports.grpc import (
+    MetricServiceGrpcTransport,
+)
 
 # pylint: disable=no-name-in-module
 from google.protobuf.timestamp_pb2 import Timestamp
 from opentelemetry.exporter.cloud_monitoring._resource import (
     get_monitored_resource,
 )
+from opentelemetry.exporter.cloud_monitoring.version import __version__
 from opentelemetry.sdk.metrics.export import (
     Gauge,
     Histogram,
@@ -50,14 +54,25 @@
     NumberDataPoint,
     Sum,
 )
-from opentelemetry.sdk.resources import Resource
 
 logger = logging.getLogger(__name__)
 MAX_BATCH_WRITE = 200
 WRITE_INTERVAL = 10
 UNIQUE_IDENTIFIER_KEY = "opentelemetry_id"
 NANOS_PER_SECOND = 10**9
 
+_OTEL_SDK_VERSION = pkg_resources.get_distribution("opentelemetry-sdk").version
+_USER_AGENT = f"opentelemetry-python {_OTEL_SDK_VERSION}; google-cloud-metric-exporter {__version__}"
+
+# Set user-agent metadata, see https://github.com/grpc/grpc/issues/23644 and default options
+# from
+# https://github.com/googleapis/python-monitoring/blob/v2.11.3/google/cloud/monitoring_v3/services/metric_service/transports/grpc.py#L175-L178
+_OPTIONS = [
+    ("grpc.max_send_message_length", -1),
+    ("grpc.max_receive_message_length", -1),
+    ("grpc.primary_user_agent", _USER_AGENT),
+]
+
 
 # pylint is unable to resolve members of protobuf objects
 # pylint: disable=no-member
@@ -88,7 +103,13 @@ def __init__(
         # Default preferred_temporality is all CUMULATIVE so need to customize
         super().__init__()
 
-        self.client = client or MetricServiceClient()
+        self.client = client or MetricServiceClient(
+            transport=MetricServiceGrpcTransport(
+                channel=MetricServiceGrpcTransport.create_channel(
+                    options=_OPTIONS,
+                )
+            )
+        )
         self.project_id: str
         if not project_id:
             _, default_project_id = google.auth.default()
@@ -122,7 +143,7 @@ def _batch_write(self, series: List[TimeSeries]) -> None:
                     time_series=series[
                         write_ind : write_ind + MAX_BATCH_WRITE
                     ],
-                )
+                ),
             )
             write_ind += MAX_BATCH_WRITE
 
diff --git a/opentelemetry-exporter-gcp-monitoring/tests/fixtures/gcmfake.py b/opentelemetry-exporter-gcp-monitoring/tests/fixtures/gcmfake.py
@@ -15,15 +15,16 @@
 from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
-from typing import Callable, Iterable, List, Mapping, Tuple, Type
+from functools import partial
+from typing import Callable, Iterable, List, Mapping, Tuple, Type, cast
+from unittest.mock import patch
 
 import grpc
 import pytest
 from google.api.metric_pb2 import MetricDescriptor
 from google.cloud.monitoring_v3 import (
     CreateMetricDescriptorRequest,
     CreateTimeSeriesRequest,
-    MetricServiceClient,
 )
 from google.cloud.monitoring_v3.services.metric_service.transports import (
     MetricServiceGrpcTransport,
@@ -34,8 +35,8 @@
 from google.protobuf.message import Message
 from grpc import (
     GenericRpcHandler,
-    RpcContext,
     RpcMethodHandler,
+    ServicerContext,
     insecure_channel,
     method_handlers_generic_handler,
     unary_unary_rpc_method_handler,
@@ -46,8 +47,14 @@
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
 
-# Mapping of fully qualified GCM API method names to list of requests received
-GcmCalls = Mapping[str, List[Message]]
+
+@dataclass
+class GcmCall:
+    message: Message
+    user_agent: str
+
+
+GcmCalls = Mapping[str, List[GcmCall]]
 
 PROJECT_ID = "fakeproject"
 
@@ -89,12 +96,16 @@ def __init__(self):
     def _make_impl(
         self,
         method: str,
-        behavior: Callable[[Message, RpcContext], Message],
+        behavior: Callable[[Message, ServicerContext], Message],
         deserializer,
         serializer,
     ) -> Tuple[str, RpcMethodHandler]:
-        def impl(req: Message, context: RpcContext) -> Message:
-            self._calls[f"/{self._service}/{method}"].append(req)
+        def impl(req: Message, context: ServicerContext) -> Message:
+            metadata_dict = dict(context.invocation_metadata())
+            user_agent = cast(str, metadata_dict["user-agent"])
+            self._calls[f"/{self._service}/{method}"].append(
+                GcmCall(message=req, user_agent=user_agent)
+            )
             return behavior(req, context)
 
         return method, unary_unary_rpc_method_handler(
@@ -114,7 +125,7 @@ def get_calls(self) -> GcmCalls:
 
 @dataclass
 class GcmFake:
-    client: MetricServiceClient
+    exporter: CloudMonitoringMetricsExporter
     get_calls: Callable[[], GcmCalls]
 
 
@@ -131,10 +142,17 @@ def fixture_gcmfake() -> Iterable[GcmFake]:
             server = grpc.server(executor, handlers=[handler])
             port = server.add_insecure_port("localhost:0")
             server.start()
-            with insecure_channel(f"localhost:{port}") as channel:
+
+            # patch MetricServiceGrpcTransport.create_channel staticmethod to return an insecure
+            # channel but otherwise respect any parameters passed to it
+            with patch.object(
+                MetricServiceGrpcTransport,
+                "create_channel",
+                partial(insecure_channel, target=f"localhost:{port}"),
+            ):
                 yield GcmFake(
-                    client=MetricServiceClient(
-                        transport=MetricServiceGrpcTransport(channel=channel),
+                    exporter=CloudMonitoringMetricsExporter(
+                        project_id=PROJECT_ID
                     ),
                     get_calls=handler.get_calls,
                 )
@@ -161,11 +179,7 @@ def make_meter_provider(**kwargs) -> MeterProvider:
         mp = MeterProvider(
             **{
                 "metric_readers": [
-                    PeriodicExportingMetricReader(
-                        CloudMonitoringMetricsExporter(
-                            project_id=PROJECT_ID, client=gcmfake.client
-                        )
-                    )
+                    PeriodicExportingMetricReader(gcmfake.exporter)
                 ],
                 "shutdown_on_exit": False,
                 **kwargs,
diff --git a/opentelemetry-exporter-gcp-monitoring/tests/fixtures/snapshot_gcmcalls.py b/opentelemetry-exporter-gcp-monitoring/tests/fixtures/snapshot_gcmcalls.py
@@ -43,20 +43,20 @@ def serialize(
         exclude: Optional[PropertyFilter] = None,
         matcher: Optional[PropertyMatcher] = None,
     ) -> SerializedData:
-        calls = cast(GcmCalls, data)
+        gcmcalls = cast(GcmCalls, data)
         json = {}
-        for method, requests in calls.items():
+        for method, calls in gcmcalls.items():
             dict_requests = []
-            for request in requests:
-                if isinstance(request, proto.message.Message):
-                    request = type(request).pb(request)
-                elif isinstance(request, google.protobuf.message.Message):
+            for call in calls:
+                if isinstance(call.message, proto.message.Message):
+                    call.message = type(call.message).pb(call.message)
+                elif isinstance(call.message, google.protobuf.message.Message):
                     pass
                 else:
                     raise ValueError(
-                        f"Excepted a proto-plus or protobuf message, got {type(request)}"
+                        f"Excepted a proto-plus or protobuf message, got {type(call)}"
                     )
-                dict_requests.append(json_format.MessageToDict(request))
+                dict_requests.append(json_format.MessageToDict(call.message))
             json[method] = dict_requests
 
         return super().serialize(json, exclude=exclude, matcher=matcher)
diff --git a/opentelemetry-exporter-gcp-monitoring/tests/test_user_agent.py b/opentelemetry-exporter-gcp-monitoring/tests/test_user_agent.py
@@ -0,0 +1,39 @@
+# Copyright 2022 Google LLC
+#
+# 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
+#
+#      https://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 re
+
+from fixtures.gcmfake import GcmFake, GcmFakeMeterProvider
+
+
+def test_with_resource(
+    gcmfake_meter_provider: GcmFakeMeterProvider,
+    gcmfake: GcmFake,
+) -> None:
+    meter_provider = gcmfake_meter_provider()
+    counter = meter_provider.get_meter(__name__).create_counter(
+        "mycounter", description="foo", unit="{myunit}"
+    )
+    counter.add(12)
+    meter_provider.force_flush()
+
+    for calls in gcmfake.get_calls().values():
+        for call in calls:
+            assert (
+                re.match(
+                    r"^opentelemetry-python \S+; google-cloud-metric-exporter \S+ grpc-python/\S+",
+                    call.user_agent,
+                )
+                is not None
+            )
diff --git a/opentelemetry-exporter-gcp-trace/src/opentelemetry/exporter/cloud_trace/__init__.py b/opentelemetry-exporter-gcp-trace/src/opentelemetry/exporter/cloud_trace/__init__.py
@@ -90,6 +90,9 @@
 import pkg_resources
 from google.cloud.trace_v2 import BatchWriteSpansRequest, TraceServiceClient
 from google.cloud.trace_v2 import types as trace_types
+from google.cloud.trace_v2.services.trace_service.transports import (
+    TraceServiceGrpcTransport,
+)
 from google.protobuf.timestamp_pb2 import (  # pylint: disable=no-name-in-module
     Timestamp,
 )
@@ -113,6 +116,18 @@
 
 logger = logging.getLogger(__name__)
 
+_OTEL_SDK_VERSION = pkg_resources.get_distribution("opentelemetry-sdk").version
+_USER_AGENT = f"opentelemetry-python {_OTEL_SDK_VERSION}; google-cloud-trace-exporter {__version__}"
+
+# Set user-agent metadata, see https://github.com/grpc/grpc/issues/23644 and default options
+# from
+# https://github.com/googleapis/python-trace/blob/v1.7.3/google/cloud/trace_v1/services/trace_service/transports/grpc.py#L177-L180
+_OPTIONS = [
+    ("grpc.max_send_message_length", -1),
+    ("grpc.max_receive_message_length", -1),
+    ("grpc.primary_user_agent", _USER_AGENT),
+]
+
 MAX_NUM_LINKS = 128
 MAX_NUM_EVENTS = 32
 MAX_EVENT_ATTRS = 4
@@ -141,7 +156,14 @@ def __init__(
         client=None,
         resource_regex=None,
     ):
-        self.client: TraceServiceClient = client or TraceServiceClient()
+        self.client: TraceServiceClient = client or TraceServiceClient(
+            transport=TraceServiceGrpcTransport(
+                channel=TraceServiceGrpcTransport.create_channel(
+                    options=_OPTIONS
+                )
+            )
+        )
+
         if not project_id:
             project_id = environ.get(OTEL_EXPORTER_GCP_TRACE_PROJECT_ID)
         if not project_id:
diff --git a/opentelemetry-exporter-gcp-trace/tests/test_cloud_trace_exporter.py b/opentelemetry-exporter-gcp-trace/tests/test_cloud_trace_exporter.py
@@ -17,6 +17,9 @@
 from unittest import mock
 
 import pkg_resources
+from google.cloud.trace_v2.services.trace_service.transports import (
+    TraceServiceGrpcTransport,
+)
 from google.cloud.trace_v2.types import AttributeValue, BatchWriteSpansRequest
 from google.cloud.trace_v2.types import Span as ProtoSpan
 from google.cloud.trace_v2.types import TruncatableString
@@ -51,13 +54,18 @@
 # pylint: disable=too-many-public-methods
 class TestCloudTraceSpanExporter(unittest.TestCase):
     def setUp(self):
-        self.client_patcher = mock.patch(
-            "opentelemetry.exporter.cloud_trace.TraceServiceClient"
-        )
-        self.client_patcher.start()
+        self.patchers = [
+            mock.patch(
+                "opentelemetry.exporter.cloud_trace.TraceServiceClient"
+            ),
+            mock.patch.object(TraceServiceGrpcTransport, "create_channel"),
+        ]
+        for patcher in self.patchers:
+            patcher.start()
 
     def tearDown(self):
-        self.client_patcher.stop()
+        for patcher in self.patchers:
+            patcher.stop()
 
     @classmethod
     def setUpClass(cls):
