[python O11Y] Implement OpenTelemetry (#35292)
<!-- If you know who should review your pull request, please assign it to that person, otherwise the pull request would get assigned randomly. If your pull request is for a specific language, please add the appropriate lang label. --> Closes #35292 PiperOrigin-RevId: 595188404pull/35428/head
Xuan Wang
1 year ago
committed by
Copybara-Service
parent
853e8c62f0
commit
d2da19e99e
32 changed files with 1273 additions and 295 deletions
@ -0,0 +1,38 @@ |
|||||||
|
# Copyright 2023 gRPC authors. |
||||||
|
# |
||||||
|
# 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 Iterable, List |
||||||
|
|
||||||
|
from grpc_observability import _observability # pytype: disable=pyi-error |
||||||
|
from grpc_observability._open_telemetry_plugin import _OpenTelemetryPlugin |
||||||
|
|
||||||
|
|
||||||
|
class _OpenTelemetryExporterDelegator(_observability.Exporter): |
||||||
|
_plugins: Iterable[_OpenTelemetryPlugin] |
||||||
|
|
||||||
|
def __init__(self, plugins: Iterable[_OpenTelemetryPlugin]): |
||||||
|
self._plugins = plugins |
||||||
|
|
||||||
|
def export_stats_data( |
||||||
|
self, stats_data: List[_observability.StatsData] |
||||||
|
) -> None: |
||||||
|
# Records stats data to MeterProvider. |
||||||
|
for data in stats_data: |
||||||
|
for plugin in self._plugins: |
||||||
|
plugin.maybe_record_stats_data(data) |
||||||
|
|
||||||
|
def export_tracing_data( |
||||||
|
self, tracing_data: List[_observability.TracingData] |
||||||
|
) -> None: |
||||||
|
pass |
||||||
@ -0,0 +1,97 @@ |
|||||||
|
# Copyright 2023 gRPC authors. |
||||||
|
# |
||||||
|
# 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 collections |
||||||
|
from typing import List |
||||||
|
|
||||||
|
from grpc_observability._cyobservability import MetricsName |
||||||
|
|
||||||
|
|
||||||
|
class Metric( |
||||||
|
collections.namedtuple( |
||||||
|
"Metric", |
||||||
|
["name", "cyname", "unit", "description"], |
||||||
|
) |
||||||
|
): |
||||||
|
pass |
||||||
|
|
||||||
|
|
||||||
|
CLIENT_ATTEMPT_STARTED = Metric( |
||||||
|
"grpc.client.attempt.started", |
||||||
|
MetricsName.CLIENT_STARTED_RPCS, |
||||||
|
"{attempt}", |
||||||
|
"Number of client call attempts started", |
||||||
|
) |
||||||
|
CLIENT_ATTEMPT_DURATION = Metric( |
||||||
|
"grpc.client.attempt.duration", |
||||||
|
MetricsName.CLIENT_ROUNDTRIP_LATENCY, |
||||||
|
"s", |
||||||
|
"End-to-end time taken to complete a client call attempt", |
||||||
|
) |
||||||
|
CLIENT_RPC_DURATION = Metric( |
||||||
|
"grpc.client.call.duration", |
||||||
|
MetricsName.CLIENT_API_LATENCY, |
||||||
|
"s", |
||||||
|
"End-to-end time taken to complete a call from client's perspective", |
||||||
|
) |
||||||
|
CLIENT_ATTEMPT_SEND_BYTES = Metric( |
||||||
|
"grpc.client.attempt.sent_total_compressed_message_size", |
||||||
|
MetricsName.CLIENT_SEND_BYTES_PER_RPC, |
||||||
|
"By", |
||||||
|
"Compressed message bytes sent per client call attempt", |
||||||
|
) |
||||||
|
CLIENT_ATTEMPT_RECEIVED_BYTES = Metric( |
||||||
|
"grpc.client.attempt.rcvd_total_compressed_message_size", |
||||||
|
MetricsName.CLIENT_RECEIVED_BYTES_PER_RPC, |
||||||
|
"By", |
||||||
|
"Compressed message bytes received per call attempt", |
||||||
|
) |
||||||
|
SERVER_STARTED_RPCS = Metric( |
||||||
|
"grpc.server.call.started", |
||||||
|
MetricsName.SERVER_STARTED_RPCS, |
||||||
|
"{call}", |
||||||
|
"Number of server calls started", |
||||||
|
) |
||||||
|
SERVER_RPC_DURATION = Metric( |
||||||
|
"grpc.server.call.duration", |
||||||
|
MetricsName.SERVER_SERVER_LATENCY, |
||||||
|
"s", |
||||||
|
"End-to-end time taken to complete a call from server transport's perspective", |
||||||
|
) |
||||||
|
SERVER_RPC_SEND_BYTES = Metric( |
||||||
|
"grpc.server.call.sent_total_compressed_message_size", |
||||||
|
MetricsName.SERVER_SENT_BYTES_PER_RPC, |
||||||
|
"By", |
||||||
|
"Compressed message bytes sent per server call", |
||||||
|
) |
||||||
|
SERVER_RPC_RECEIVED_BYTES = Metric( |
||||||
|
"grpc.server.call.rcvd_total_compressed_message_size", |
||||||
|
MetricsName.SERVER_RECEIVED_BYTES_PER_RPC, |
||||||
|
"By", |
||||||
|
"Compressed message bytes received per server call", |
||||||
|
) |
||||||
|
|
||||||
|
|
||||||
|
def base_metrics() -> List[Metric]: |
||||||
|
return [ |
||||||
|
CLIENT_ATTEMPT_STARTED, |
||||||
|
CLIENT_ATTEMPT_DURATION, |
||||||
|
CLIENT_RPC_DURATION, |
||||||
|
CLIENT_ATTEMPT_SEND_BYTES, |
||||||
|
CLIENT_ATTEMPT_RECEIVED_BYTES, |
||||||
|
SERVER_STARTED_RPCS, |
||||||
|
SERVER_RPC_DURATION, |
||||||
|
SERVER_RPC_SEND_BYTES, |
||||||
|
SERVER_RPC_RECEIVED_BYTES, |
||||||
|
] |
||||||
@ -0,0 +1,155 @@ |
|||||||
|
# Copyright 2023 gRPC authors. |
||||||
|
# |
||||||
|
# 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 logging |
||||||
|
import time |
||||||
|
from typing import Any, Iterable, Optional |
||||||
|
|
||||||
|
import grpc |
||||||
|
|
||||||
|
# pytype: disable=pyi-error |
||||||
|
from grpc_observability import _cyobservability |
||||||
|
from grpc_observability._open_telemetry_exporter import ( |
||||||
|
_OpenTelemetryExporterDelegator, |
||||||
|
) |
||||||
|
from grpc_observability._open_telemetry_plugin import OpenTelemetryPlugin |
||||||
|
from grpc_observability._open_telemetry_plugin import _OpenTelemetryPlugin |
||||||
|
|
||||||
|
_LOGGER = logging.getLogger(__name__) |
||||||
|
|
||||||
|
ClientCallTracerCapsule = Any # it appears only once in the function signature |
||||||
|
ServerCallTracerFactoryCapsule = ( |
||||||
|
Any # it appears only once in the function signature |
||||||
|
) |
||||||
|
grpc_observability = Any # grpc_observability.py imports this module. |
||||||
|
|
||||||
|
GRPC_STATUS_CODE_TO_STRING = { |
||||||
|
grpc.StatusCode.OK: "OK", |
||||||
|
grpc.StatusCode.CANCELLED: "CANCELLED", |
||||||
|
grpc.StatusCode.UNKNOWN: "UNKNOWN", |
||||||
|
grpc.StatusCode.INVALID_ARGUMENT: "INVALID_ARGUMENT", |
||||||
|
grpc.StatusCode.DEADLINE_EXCEEDED: "DEADLINE_EXCEEDED", |
||||||
|
grpc.StatusCode.NOT_FOUND: "NOT_FOUND", |
||||||
|
grpc.StatusCode.ALREADY_EXISTS: "ALREADY_EXISTS", |
||||||
|
grpc.StatusCode.PERMISSION_DENIED: "PERMISSION_DENIED", |
||||||
|
grpc.StatusCode.UNAUTHENTICATED: "UNAUTHENTICATED", |
||||||
|
grpc.StatusCode.RESOURCE_EXHAUSTED: "RESOURCE_EXHAUSTED", |
||||||
|
grpc.StatusCode.FAILED_PRECONDITION: "FAILED_PRECONDITION", |
||||||
|
grpc.StatusCode.ABORTED: "ABORTED", |
||||||
|
grpc.StatusCode.OUT_OF_RANGE: "OUT_OF_RANGE", |
||||||
|
grpc.StatusCode.UNIMPLEMENTED: "UNIMPLEMENTED", |
||||||
|
grpc.StatusCode.INTERNAL: "INTERNAL", |
||||||
|
grpc.StatusCode.UNAVAILABLE: "UNAVAILABLE", |
||||||
|
grpc.StatusCode.DATA_LOSS: "DATA_LOSS", |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
# pylint: disable=no-self-use |
||||||
|
class OpenTelemetryObservability(grpc._observability.ObservabilityPlugin): |
||||||
|
"""OpenTelemetry based plugin implementation. |
||||||
|
|
||||||
|
Args: |
||||||
|
exporter: Exporter used to export data. |
||||||
|
plugin: OpenTelemetryPlugin to enable. |
||||||
|
""" |
||||||
|
|
||||||
|
exporter: "grpc_observability.Exporter" |
||||||
|
plugins: Iterable[OpenTelemetryPlugin] |
||||||
|
|
||||||
|
def __init__( |
||||||
|
self, |
||||||
|
*, |
||||||
|
plugins: Optional[Iterable[OpenTelemetryPlugin]] = None, |
||||||
|
exporter: "grpc_observability.Exporter" = None, |
||||||
|
): |
||||||
|
_plugins = [] |
||||||
|
if plugins: |
||||||
|
for plugin in plugins: |
||||||
|
_plugins.append(_OpenTelemetryPlugin(plugin)) |
||||||
|
|
||||||
|
if exporter: |
||||||
|
self.exporter = exporter |
||||||
|
else: |
||||||
|
self.exporter = _OpenTelemetryExporterDelegator(_plugins) |
||||||
|
|
||||||
|
try: |
||||||
|
_cyobservability.activate_stats() |
||||||
|
self.set_stats(True) |
||||||
|
except Exception as e: # pylint: disable=broad-except |
||||||
|
raise ValueError(f"Activate observability metrics failed with: {e}") |
||||||
|
|
||||||
|
def __enter__(self): |
||||||
|
try: |
||||||
|
_cyobservability.cyobservability_init(self.exporter) |
||||||
|
# TODO(xuanwn): Use specific exceptons |
||||||
|
except Exception as e: # pylint: disable=broad-except |
||||||
|
_LOGGER.exception("Initiate observability failed with: %s", e) |
||||||
|
|
||||||
|
grpc._observability.observability_init(self) |
||||||
|
return self |
||||||
|
|
||||||
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None: |
||||||
|
self.exit() |
||||||
|
|
||||||
|
def exit(self) -> None: |
||||||
|
# Sleep so we don't loss any data. If we shutdown export thread |
||||||
|
# immediately after exit, it's possible that core didn't call RecordEnd |
||||||
|
# in callTracer, and all data recorded by calling RecordEnd will be |
||||||
|
# lost. |
||||||
|
# CENSUS_EXPORT_BATCH_INTERVAL_SECS: The time equals to the time in |
||||||
|
# AwaitNextBatchLocked. |
||||||
|
# TODO(xuanwn): explicit synchronization |
||||||
|
# https://github.com/grpc/grpc/issues/33262 |
||||||
|
time.sleep(_cyobservability.CENSUS_EXPORT_BATCH_INTERVAL_SECS) |
||||||
|
self.set_tracing(False) |
||||||
|
self.set_stats(False) |
||||||
|
_cyobservability.observability_deinit() |
||||||
|
grpc._observability.observability_deinit() |
||||||
|
|
||||||
|
def create_client_call_tracer( |
||||||
|
self, method_name: bytes, target: bytes |
||||||
|
) -> ClientCallTracerCapsule: |
||||||
|
trace_id = b"TRACE_ID" |
||||||
|
capsule = _cyobservability.create_client_call_tracer( |
||||||
|
method_name, target, trace_id |
||||||
|
) |
||||||
|
return capsule |
||||||
|
|
||||||
|
def create_server_call_tracer_factory( |
||||||
|
self, |
||||||
|
) -> ServerCallTracerFactoryCapsule: |
||||||
|
capsule = _cyobservability.create_server_call_tracer_factory_capsule() |
||||||
|
return capsule |
||||||
|
|
||||||
|
def delete_client_call_tracer( |
||||||
|
self, client_call_tracer: ClientCallTracerCapsule |
||||||
|
) -> None: |
||||||
|
_cyobservability.delete_client_call_tracer(client_call_tracer) |
||||||
|
|
||||||
|
def save_trace_context( |
||||||
|
self, trace_id: str, span_id: str, is_sampled: bool |
||||||
|
) -> None: |
||||||
|
pass |
||||||
|
|
||||||
|
def record_rpc_latency( |
||||||
|
self, |
||||||
|
method: str, |
||||||
|
target: str, |
||||||
|
rpc_latency: float, |
||||||
|
status_code: grpc.StatusCode, |
||||||
|
) -> None: |
||||||
|
status_code = GRPC_STATUS_CODE_TO_STRING.get(status_code, "UNKNOWN") |
||||||
|
_cyobservability._record_rpc_latency( |
||||||
|
self.exporter, method, target, rpc_latency, status_code |
||||||
|
) |
||||||
@ -0,0 +1,254 @@ |
|||||||
|
# Copyright 2023 gRPC authors. |
||||||
|
# |
||||||
|
# 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 abc |
||||||
|
from typing import Dict, Iterable, List, Optional, Union |
||||||
|
|
||||||
|
# pytype: disable=pyi-error |
||||||
|
import grpc |
||||||
|
from grpc_observability import _open_telemetry_measures |
||||||
|
from grpc_observability._cyobservability import MetricsName |
||||||
|
from grpc_observability._observability import StatsData |
||||||
|
from opentelemetry.sdk.metrics import Counter |
||||||
|
from opentelemetry.sdk.metrics import Histogram |
||||||
|
from opentelemetry.sdk.metrics import Meter |
||||||
|
from opentelemetry.sdk.metrics import MeterProvider |
||||||
|
|
||||||
|
GRPC_METHOD_LABEL = "grpc.method" |
||||||
|
GRPC_TARGET_LABEL = "grpc.target" |
||||||
|
GRPC_OTHER_LABEL_VALUE = "other" |
||||||
|
|
||||||
|
|
||||||
|
class OpenTelemetryLabelInjector(abc.ABC): |
||||||
|
""" |
||||||
|
An interface that allows you to add additional labels on the calls traced. |
||||||
|
|
||||||
|
Please note that this class is still work in progress and NOT READY to be used. |
||||||
|
""" |
||||||
|
|
||||||
|
_labels: List[Dict[str, str]] |
||||||
|
|
||||||
|
def __init__(self): |
||||||
|
# Calls Python OTel API to detect resource and get labels, save |
||||||
|
# those lables to OpenTelemetryLabelInjector.labels. |
||||||
|
pass |
||||||
|
|
||||||
|
@abc.abstractmethod |
||||||
|
def get_labels(self): |
||||||
|
# Get additional labels for this OpenTelemetryLabelInjector. |
||||||
|
raise NotImplementedError() |
||||||
|
|
||||||
|
|
||||||
|
class OpenTelemetryPluginOption(abc.ABC): |
||||||
|
""" |
||||||
|
An interface that allows you to add additional function to OpenTelemetryPlugin. |
||||||
|
|
||||||
|
Please note that this class is still work in progress and NOT READY to be used. |
||||||
|
""" |
||||||
|
|
||||||
|
@abc.abstractmethod |
||||||
|
def is_active_on_method(self, method: str) -> bool: |
||||||
|
"""Determines whether this plugin option is active on a given method. |
||||||
|
|
||||||
|
Args: |
||||||
|
method: Required. The RPC method, for example: `/helloworld.Greeter/SayHello`. |
||||||
|
|
||||||
|
Returns: |
||||||
|
True if this this plugin option is active on the giving method, false otherwise. |
||||||
|
""" |
||||||
|
raise NotImplementedError() |
||||||
|
|
||||||
|
@abc.abstractmethod |
||||||
|
def is_active_on_server(self, channel_args: List[str]) -> bool: |
||||||
|
"""Determines whether this plugin option is active on a given server. |
||||||
|
|
||||||
|
Args: |
||||||
|
channel_args: Required. The channel args used for server. |
||||||
|
TODO(xuanwn): detail on what channel_args will contain. |
||||||
|
|
||||||
|
Returns: |
||||||
|
True if this this plugin option is active on the server, false otherwise. |
||||||
|
""" |
||||||
|
raise NotImplementedError() |
||||||
|
|
||||||
|
@abc.abstractmethod |
||||||
|
def get_label_injector(self) -> Optional[OpenTelemetryLabelInjector]: |
||||||
|
# Returns the LabelsInjector used by this plugin option, or None. |
||||||
|
raise NotImplementedError() |
||||||
|
|
||||||
|
|
||||||
|
# pylint: disable=no-self-use |
||||||
|
class OpenTelemetryPlugin: |
||||||
|
"""Describes a Plugin for OpenTelemetry observability.""" |
||||||
|
|
||||||
|
def get_plugin_options( |
||||||
|
self, |
||||||
|
) -> Iterable[OpenTelemetryPluginOption]: |
||||||
|
return [] |
||||||
|
|
||||||
|
def get_meter_provider(self) -> Optional[MeterProvider]: |
||||||
|
return None |
||||||
|
|
||||||
|
def target_attribute_filter( |
||||||
|
self, target: str # pylint: disable=unused-argument |
||||||
|
) -> bool: |
||||||
|
""" |
||||||
|
If set, this will be called per channel to decide whether to record the |
||||||
|
target attribute on client or to replace it with "other". |
||||||
|
This helps reduce the cardinality on metrics in cases where many channels |
||||||
|
are created with different targets in the same binary (which might happen |
||||||
|
for example, if the channel target string uses IP addresses directly). |
||||||
|
|
||||||
|
Args: |
||||||
|
target: The target for the RPC. |
||||||
|
|
||||||
|
Returns: |
||||||
|
bool: True means the original target string will be used, False means target string |
||||||
|
will be replaced with "other". |
||||||
|
""" |
||||||
|
return True |
||||||
|
|
||||||
|
def generic_method_attribute_filter( |
||||||
|
self, method: str # pylint: disable=unused-argument |
||||||
|
) -> bool: |
||||||
|
""" |
||||||
|
If set, this will be called with a generic method type to decide whether to |
||||||
|
record the method name or to replace it with "other". |
||||||
|
|
||||||
|
Note that pre-registered methods will always be recorded no matter what this |
||||||
|
function returns. |
||||||
|
|
||||||
|
Args: |
||||||
|
method: The method name for the RPC. |
||||||
|
|
||||||
|
Returns: |
||||||
|
bool: True means the original method name will be used, False means method name |
||||||
|
will be replaced with "other". |
||||||
|
""" |
||||||
|
return False |
||||||
|
|
||||||
|
|
||||||
|
class _OpenTelemetryPlugin: |
||||||
|
_plugin: OpenTelemetryPlugin |
||||||
|
_metric_to_recorder: Dict[MetricsName, Union[Counter, Histogram]] |
||||||
|
|
||||||
|
def __init__(self, plugin: OpenTelemetryPlugin): |
||||||
|
self._plugin = plugin |
||||||
|
self._metric_to_recorder = dict() |
||||||
|
|
||||||
|
meter_provider = self._plugin.get_meter_provider() |
||||||
|
if meter_provider: |
||||||
|
meter = meter_provider.get_meter("grpc-python", grpc.__version__) |
||||||
|
enabled_metrics = _open_telemetry_measures.base_metrics() |
||||||
|
self._metric_to_recorder = self._register_metrics( |
||||||
|
meter, enabled_metrics |
||||||
|
) |
||||||
|
|
||||||
|
def _should_record(self, stats_data: StatsData) -> bool: |
||||||
|
# Decide if this plugin should record the stats_data. |
||||||
|
return stats_data.name in self._metric_to_recorder.keys() |
||||||
|
|
||||||
|
def _record_stats_data(self, stats_data: StatsData) -> None: |
||||||
|
recorder = self._metric_to_recorder[stats_data.name] |
||||||
|
|
||||||
|
target = stats_data.labels.get(GRPC_TARGET_LABEL, "") |
||||||
|
if not self._plugin.target_attribute_filter(target): |
||||||
|
# Filter target name. |
||||||
|
stats_data.labels[GRPC_TARGET_LABEL] = GRPC_OTHER_LABEL_VALUE |
||||||
|
|
||||||
|
method = stats_data.labels.get(GRPC_METHOD_LABEL, "") |
||||||
|
if not self._plugin.generic_method_attribute_filter(method): |
||||||
|
# Filter method name. |
||||||
|
stats_data.labels[GRPC_METHOD_LABEL] = GRPC_OTHER_LABEL_VALUE |
||||||
|
|
||||||
|
value = 0 |
||||||
|
if stats_data.measure_double: |
||||||
|
value = stats_data.value_float |
||||||
|
else: |
||||||
|
value = stats_data.value_int |
||||||
|
if isinstance(recorder, Counter): |
||||||
|
recorder.add(value, attributes=stats_data.labels) |
||||||
|
elif isinstance(recorder, Histogram): |
||||||
|
recorder.record(value, attributes=stats_data.labels) |
||||||
|
|
||||||
|
# pylint: disable=no-self-use |
||||||
|
def maybe_record_stats_data(self, stats_data: List[StatsData]) -> None: |
||||||
|
# Records stats data to MeterProvider. |
||||||
|
if self._should_record(stats_data): |
||||||
|
self._record_stats_data(stats_data) |
||||||
|
|
||||||
|
def _register_metrics( |
||||||
|
self, meter: Meter, metrics: List[_open_telemetry_measures.Metric] |
||||||
|
) -> Dict[MetricsName, Union[Counter, Histogram]]: |
||||||
|
metric_to_recorder_map = {} |
||||||
|
recorder = None |
||||||
|
for metric in metrics: |
||||||
|
if metric == _open_telemetry_measures.CLIENT_ATTEMPT_STARTED: |
||||||
|
recorder = meter.create_counter( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.CLIENT_ATTEMPT_DURATION: |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.CLIENT_RPC_DURATION: |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.CLIENT_ATTEMPT_SEND_BYTES: |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif ( |
||||||
|
metric == _open_telemetry_measures.CLIENT_ATTEMPT_RECEIVED_BYTES |
||||||
|
): |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.SERVER_STARTED_RPCS: |
||||||
|
recorder = meter.create_counter( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.SERVER_RPC_DURATION: |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.SERVER_RPC_SEND_BYTES: |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
elif metric == _open_telemetry_measures.SERVER_RPC_RECEIVED_BYTES: |
||||||
|
recorder = meter.create_histogram( |
||||||
|
name=metric.name, |
||||||
|
unit=metric.unit, |
||||||
|
description=metric.description, |
||||||
|
) |
||||||
|
metric_to_recorder_map[metric.cyname] = recorder |
||||||
|
return metric_to_recorder_map |
||||||
@ -0,0 +1,299 @@ |
|||||||
|
# Copyright 2023 gRPC authors. |
||||||
|
# |
||||||
|
# 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 collections import defaultdict |
||||||
|
import datetime |
||||||
|
import logging |
||||||
|
import os |
||||||
|
import sys |
||||||
|
import time |
||||||
|
from typing import Any, Callable, Dict, List, Optional, Set |
||||||
|
import unittest |
||||||
|
|
||||||
|
import grpc_observability |
||||||
|
from grpc_observability import _open_telemetry_measures |
||||||
|
from grpc_observability._open_telemetry_plugin import GRPC_METHOD_LABEL |
||||||
|
from grpc_observability._open_telemetry_plugin import GRPC_OTHER_LABEL_VALUE |
||||||
|
from grpc_observability._open_telemetry_plugin import GRPC_TARGET_LABEL |
||||||
|
from opentelemetry.sdk.metrics import MeterProvider |
||||||
|
from opentelemetry.sdk.metrics.export import AggregationTemporality |
||||||
|
from opentelemetry.sdk.metrics.export import MetricExportResult |
||||||
|
from opentelemetry.sdk.metrics.export import MetricExporter |
||||||
|
from opentelemetry.sdk.metrics.export import MetricsData |
||||||
|
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader |
||||||
|
|
||||||
|
from tests.observability import _test_server |
||||||
|
|
||||||
|
logger = logging.getLogger(__name__) |
||||||
|
|
||||||
|
STREAM_LENGTH = 5 |
||||||
|
OTEL_EXPORT_INTERVAL_S = 0.5 |
||||||
|
|
||||||
|
|
||||||
|
class OTelMetricExporter(MetricExporter): |
||||||
|
"""Implementation of :class:`MetricExporter` that export metrics to the |
||||||
|
provided metric_list. |
||||||
|
|
||||||
|
all_metrics: A dict which key is grpc_observability._opentelemetry_measures.Metric.name, |
||||||
|
value is a list of labels recorded for that metric. |
||||||
|
An example item of this dict: |
||||||
|
{"grpc.client.attempt.started": |
||||||
|
[{'grpc.method': 'test/UnaryUnary', 'grpc.target': 'localhost:42517'}, |
||||||
|
{'grpc.method': 'other', 'grpc.target': 'localhost:42517'}]} |
||||||
|
""" |
||||||
|
|
||||||
|
def __init__( |
||||||
|
self, |
||||||
|
all_metrics: Dict[str, List], |
||||||
|
preferred_temporality: Dict[type, AggregationTemporality] = None, |
||||||
|
preferred_aggregation: Dict[ |
||||||
|
type, "opentelemetry.sdk.metrics.view.Aggregation" |
||||||
|
] = None, |
||||||
|
): |
||||||
|
super().__init__( |
||||||
|
preferred_temporality=preferred_temporality, |
||||||
|
preferred_aggregation=preferred_aggregation, |
||||||
|
) |
||||||
|
self.all_metrics = all_metrics |
||||||
|
|
||||||
|
def export( |
||||||
|
self, |
||||||
|
metrics_data: MetricsData, |
||||||
|
timeout_millis: float = 10_000, |
||||||
|
**kwargs, |
||||||
|
) -> MetricExportResult: |
||||||
|
self.record_metric(metrics_data) |
||||||
|
return MetricExportResult.SUCCESS |
||||||
|
|
||||||
|
def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None: |
||||||
|
pass |
||||||
|
|
||||||
|
def force_flush(self, timeout_millis: float = 10_000) -> bool: |
||||||
|
return True |
||||||
|
|
||||||
|
def record_metric(self, metrics_data: MetricsData) -> None: |
||||||
|
for resource_metric in metrics_data.resource_metrics: |
||||||
|
for scope_metric in resource_metric.scope_metrics: |
||||||
|
for metric in scope_metric.metrics: |
||||||
|
for data_point in metric.data.data_points: |
||||||
|
self.all_metrics[metric.name].append( |
||||||
|
data_point.attributes |
||||||
|
) |
||||||
|
|
||||||
|
|
||||||
|
class BaseTestOpenTelemetryPlugin(grpc_observability.OpenTelemetryPlugin): |
||||||
|
def __init__(self, provider: MeterProvider): |
||||||
|
self.provider = provider |
||||||
|
|
||||||
|
def get_meter_provider(self) -> Optional[MeterProvider]: |
||||||
|
return self.provider |
||||||
|
|
||||||
|
|
||||||
|
@unittest.skipIf( |
||||||
|
os.name == "nt" or "darwin" in sys.platform, |
||||||
|
"Observability is not supported in Windows and MacOS", |
||||||
|
) |
||||||
|
class OpenTelemetryObservabilityTest(unittest.TestCase): |
||||||
|
def setUp(self): |
||||||
|
self.all_metrics = defaultdict(list) |
||||||
|
otel_exporter = OTelMetricExporter(self.all_metrics) |
||||||
|
reader = PeriodicExportingMetricReader( |
||||||
|
exporter=otel_exporter, |
||||||
|
export_interval_millis=OTEL_EXPORT_INTERVAL_S * 1000, |
||||||
|
) |
||||||
|
self._provider = MeterProvider(metric_readers=[reader]) |
||||||
|
self._server = None |
||||||
|
self._port = None |
||||||
|
|
||||||
|
def tearDown(self): |
||||||
|
if self._server: |
||||||
|
self._server.stop(0) |
||||||
|
|
||||||
|
def testRecordUnaryUnary(self): |
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
server, port = _test_server.start_server() |
||||||
|
self._server = server |
||||||
|
_test_server.unary_unary_call(port=port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_all_metrics_names(self.all_metrics) |
||||||
|
|
||||||
|
def testRecordUnaryUnaryClientOnly(self): |
||||||
|
server, port = _test_server.start_server() |
||||||
|
self._server = server |
||||||
|
|
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
_test_server.unary_unary_call(port=port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_client_metrics_names(self.all_metrics) |
||||||
|
|
||||||
|
def testRecordUnaryStream(self): |
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
|
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
server, port = _test_server.start_server() |
||||||
|
self._server = server |
||||||
|
_test_server.unary_stream_call(port=port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_all_metrics_names(self.all_metrics) |
||||||
|
|
||||||
|
def testRecordStreamUnary(self): |
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
|
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
server, port = _test_server.start_server() |
||||||
|
self._server = server |
||||||
|
_test_server.stream_unary_call(port=port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_all_metrics_names(self.all_metrics) |
||||||
|
|
||||||
|
def testRecordStreamStream(self): |
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
|
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
server, port = _test_server.start_server() |
||||||
|
self._server = server |
||||||
|
_test_server.stream_stream_call(port=port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_all_metrics_names(self.all_metrics) |
||||||
|
|
||||||
|
def testTargetAttributeFilter(self): |
||||||
|
main_server, main_port = _test_server.start_server() |
||||||
|
backup_server, backup_port = _test_server.start_server() |
||||||
|
main_target = f"localhost:{main_port}" |
||||||
|
backup_target = f"localhost:{backup_port}" |
||||||
|
|
||||||
|
# Replace target label with 'other' for main_server. |
||||||
|
def target_filter(target: str) -> bool: |
||||||
|
if main_target in target: |
||||||
|
return False |
||||||
|
return True |
||||||
|
|
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
otel_plugin.target_attribute_filter = target_filter |
||||||
|
|
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
_test_server.unary_unary_call(port=main_port) |
||||||
|
_test_server.unary_unary_call(port=backup_port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_client_metrics_names(self.all_metrics) |
||||||
|
|
||||||
|
target_values = set() |
||||||
|
for label_list in self.all_metrics.values(): |
||||||
|
for labels in label_list: |
||||||
|
if GRPC_TARGET_LABEL in labels: |
||||||
|
target_values.add(labels[GRPC_TARGET_LABEL]) |
||||||
|
self.assertTrue(GRPC_OTHER_LABEL_VALUE in target_values) |
||||||
|
self.assertTrue(backup_target in target_values) |
||||||
|
|
||||||
|
main_server.stop(0) |
||||||
|
backup_server.stop(0) |
||||||
|
|
||||||
|
def testMethodAttributeFilter(self): |
||||||
|
# If method name is 'test/UnaryUnaryFiltered', is should be replaced with 'other'. |
||||||
|
FILTERED_METHOD_NAME = "test/UnaryUnaryFiltered" |
||||||
|
|
||||||
|
def method_filter(method: str) -> bool: |
||||||
|
if FILTERED_METHOD_NAME in method: |
||||||
|
return False |
||||||
|
return True |
||||||
|
|
||||||
|
otel_plugin = BaseTestOpenTelemetryPlugin(self._provider) |
||||||
|
otel_plugin.generic_method_attribute_filter = method_filter |
||||||
|
|
||||||
|
with grpc_observability.OpenTelemetryObservability( |
||||||
|
plugins=[otel_plugin] |
||||||
|
): |
||||||
|
server, port = _test_server.start_server() |
||||||
|
self._server = server |
||||||
|
_test_server.unary_unary_call(port=port) |
||||||
|
_test_server.unary_unary_filtered_call(port=port) |
||||||
|
|
||||||
|
self._validate_metrics_exist(self.all_metrics) |
||||||
|
self._validate_all_metrics_names(self.all_metrics) |
||||||
|
method_values = set() |
||||||
|
for label_list in self.all_metrics.values(): |
||||||
|
for labels in label_list: |
||||||
|
if GRPC_METHOD_LABEL in labels: |
||||||
|
method_values.add(labels[GRPC_METHOD_LABEL]) |
||||||
|
self.assertTrue(GRPC_OTHER_LABEL_VALUE in method_values) |
||||||
|
self.assertTrue(FILTERED_METHOD_NAME not in method_values) |
||||||
|
|
||||||
|
def assert_eventually( |
||||||
|
self, |
||||||
|
predicate: Callable[[], bool], |
||||||
|
*, |
||||||
|
timeout: Optional[datetime.timedelta] = None, |
||||||
|
message: Optional[Callable[[], str]] = None, |
||||||
|
) -> None: |
||||||
|
message = message or (lambda: "Proposition did not evaluate to true") |
||||||
|
timeout = timeout or datetime.timedelta(seconds=10) |
||||||
|
end = datetime.datetime.now() + timeout |
||||||
|
while datetime.datetime.now() < end: |
||||||
|
if predicate(): |
||||||
|
break |
||||||
|
time.sleep(0.5) |
||||||
|
else: |
||||||
|
self.fail(message() + " after " + str(timeout)) |
||||||
|
|
||||||
|
def _validate_metrics_exist(self, all_metrics: Dict[str, Any]) -> None: |
||||||
|
# Sleep here to make sure we have at least one export from OTel MetricExporter. |
||||||
|
self.assert_eventually( |
||||||
|
lambda: len(all_metrics.keys()) > 1, |
||||||
|
message=lambda: f"No metrics was exported", |
||||||
|
) |
||||||
|
|
||||||
|
def _validate_all_metrics_names(self, metric_names: Set[str]) -> None: |
||||||
|
self._validate_server_metrics_names(metric_names) |
||||||
|
self._validate_client_metrics_names(metric_names) |
||||||
|
|
||||||
|
def _validate_server_metrics_names(self, metric_names: Set[str]) -> None: |
||||||
|
for base_metric in _open_telemetry_measures.base_metrics(): |
||||||
|
if "grpc.server" in base_metric.name: |
||||||
|
self.assertTrue( |
||||||
|
base_metric.name in metric_names, |
||||||
|
msg=f"metric {base_metric.name} not found in exported metrics: {metric_names}!", |
||||||
|
) |
||||||
|
|
||||||
|
def _validate_client_metrics_names(self, metric_names: Set[str]) -> None: |
||||||
|
for base_metric in _open_telemetry_measures.base_metrics(): |
||||||
|
if "grpc.client" in base_metric.name: |
||||||
|
self.assertTrue( |
||||||
|
base_metric.name in metric_names, |
||||||
|
msg=f"metric {base_metric.name} not found in exported metrics: {metric_names}!", |
||||||
|
) |
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__": |
||||||
|
logging.basicConfig() |
||||||
|
unittest.main(verbosity=2) |
||||||
@ -0,0 +1,169 @@ |
|||||||
|
# Copyright 2023 gRPC authors. |
||||||
|
# |
||||||
|
# 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 concurrent import futures |
||||||
|
from typing import Tuple |
||||||
|
|
||||||
|
import grpc |
||||||
|
|
||||||
|
_REQUEST = b"\x00\x00\x00" |
||||||
|
_RESPONSE = b"\x00\x00\x00" |
||||||
|
|
||||||
|
_UNARY_UNARY = "/test/UnaryUnary" |
||||||
|
_UNARY_UNARY_FILTERED = "/test/UnaryUnaryFiltered" |
||||||
|
_UNARY_STREAM = "/test/UnaryStream" |
||||||
|
_STREAM_UNARY = "/test/StreamUnary" |
||||||
|
_STREAM_STREAM = "/test/StreamStream" |
||||||
|
STREAM_LENGTH = 5 |
||||||
|
TRIGGER_RPC_METADATA = ("control", "trigger_rpc") |
||||||
|
TRIGGER_RPC_TO_NEW_SERVER_METADATA = ("to_new_server", "") |
||||||
|
|
||||||
|
|
||||||
|
def handle_unary_unary(request, servicer_context): |
||||||
|
if TRIGGER_RPC_METADATA in servicer_context.invocation_metadata(): |
||||||
|
for k, v in servicer_context.invocation_metadata(): |
||||||
|
if "port" in k: |
||||||
|
unary_unary_call(port=int(v)) |
||||||
|
if "to_new_server" in k: |
||||||
|
second_server = grpc.server( |
||||||
|
futures.ThreadPoolExecutor(max_workers=10) |
||||||
|
) |
||||||
|
second_server.add_generic_rpc_handlers((_GenericHandler(),)) |
||||||
|
second_server_port = second_server.add_insecure_port("[::]:0") |
||||||
|
second_server.start() |
||||||
|
unary_unary_call(port=second_server_port) |
||||||
|
second_server.stop(0) |
||||||
|
return _RESPONSE |
||||||
|
|
||||||
|
|
||||||
|
def handle_unary_stream(request, servicer_context): |
||||||
|
for _ in range(STREAM_LENGTH): |
||||||
|
yield _RESPONSE |
||||||
|
|
||||||
|
|
||||||
|
def handle_stream_unary(request_iterator, servicer_context): |
||||||
|
return _RESPONSE |
||||||
|
|
||||||
|
|
||||||
|
def handle_stream_stream(request_iterator, servicer_context): |
||||||
|
for request in request_iterator: |
||||||
|
yield _RESPONSE |
||||||
|
|
||||||
|
|
||||||
|
class _MethodHandler(grpc.RpcMethodHandler): |
||||||
|
def __init__(self, request_streaming, response_streaming): |
||||||
|
self.request_streaming = request_streaming |
||||||
|
self.response_streaming = response_streaming |
||||||
|
self.request_deserializer = None |
||||||
|
self.response_serializer = None |
||||||
|
self.unary_unary = None |
||||||
|
self.unary_stream = None |
||||||
|
self.stream_unary = None |
||||||
|
self.stream_stream = None |
||||||
|
if self.request_streaming and self.response_streaming: |
||||||
|
self.stream_stream = handle_stream_stream |
||||||
|
elif self.request_streaming: |
||||||
|
self.stream_unary = handle_stream_unary |
||||||
|
elif self.response_streaming: |
||||||
|
self.unary_stream = handle_unary_stream |
||||||
|
else: |
||||||
|
self.unary_unary = handle_unary_unary |
||||||
|
|
||||||
|
|
||||||
|
class _GenericHandler(grpc.GenericRpcHandler): |
||||||
|
def service(self, handler_call_details): |
||||||
|
if handler_call_details.method == _UNARY_UNARY: |
||||||
|
return _MethodHandler(False, False) |
||||||
|
if handler_call_details.method == _UNARY_UNARY_FILTERED: |
||||||
|
return _MethodHandler(False, False) |
||||||
|
elif handler_call_details.method == _UNARY_STREAM: |
||||||
|
return _MethodHandler(False, True) |
||||||
|
elif handler_call_details.method == _STREAM_UNARY: |
||||||
|
return _MethodHandler(True, False) |
||||||
|
elif handler_call_details.method == _STREAM_STREAM: |
||||||
|
return _MethodHandler(True, True) |
||||||
|
else: |
||||||
|
return None |
||||||
|
|
||||||
|
|
||||||
|
def start_server(interceptors=None) -> Tuple[grpc.Server, int]: |
||||||
|
if interceptors: |
||||||
|
server = grpc.server( |
||||||
|
futures.ThreadPoolExecutor(max_workers=10), |
||||||
|
interceptors=interceptors, |
||||||
|
) |
||||||
|
else: |
||||||
|
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) |
||||||
|
server.add_generic_rpc_handlers((_GenericHandler(),)) |
||||||
|
port = server.add_insecure_port("[::]:0") |
||||||
|
server.start() |
||||||
|
return server, port |
||||||
|
|
||||||
|
|
||||||
|
def unary_unary_call(port, metadata=None): |
||||||
|
with grpc.insecure_channel(f"localhost:{port}") as channel: |
||||||
|
multi_callable = channel.unary_unary(_UNARY_UNARY) |
||||||
|
if metadata: |
||||||
|
unused_response, call = multi_callable.with_call( |
||||||
|
_REQUEST, metadata=metadata |
||||||
|
) |
||||||
|
else: |
||||||
|
unused_response, call = multi_callable.with_call(_REQUEST) |
||||||
|
|
||||||
|
|
||||||
|
def intercepted_unary_unary_call(port, interceptors, metadata=None): |
||||||
|
with grpc.insecure_channel(f"localhost:{port}") as channel: |
||||||
|
intercept_channel = grpc.intercept_channel(channel, interceptors) |
||||||
|
multi_callable = intercept_channel.unary_unary(_UNARY_UNARY) |
||||||
|
if metadata: |
||||||
|
unused_response, call = multi_callable.with_call( |
||||||
|
_REQUEST, metadata=metadata |
||||||
|
) |
||||||
|
else: |
||||||
|
unused_response, call = multi_callable.with_call(_REQUEST) |
||||||
|
|
||||||
|
|
||||||
|
def unary_unary_filtered_call(port, metadata=None): |
||||||
|
with grpc.insecure_channel(f"localhost:{port}") as channel: |
||||||
|
multi_callable = channel.unary_unary(_UNARY_UNARY_FILTERED) |
||||||
|
if metadata: |
||||||
|
unused_response, call = multi_callable.with_call( |
||||||
|
_REQUEST, metadata=metadata |
||||||
|
) |
||||||
|
else: |
||||||
|
unused_response, call = multi_callable.with_call(_REQUEST) |
||||||
|
|
||||||
|
|
||||||
|
def unary_stream_call(port): |
||||||
|
with grpc.insecure_channel(f"localhost:{port}") as channel: |
||||||
|
multi_callable = channel.unary_stream(_UNARY_STREAM) |
||||||
|
call = multi_callable(_REQUEST) |
||||||
|
for _ in call: |
||||||
|
pass |
||||||
|
|
||||||
|
|
||||||
|
def stream_unary_call(port): |
||||||
|
with grpc.insecure_channel(f"localhost:{port}") as channel: |
||||||
|
multi_callable = channel.stream_unary(_STREAM_UNARY) |
||||||
|
unused_response, call = multi_callable.with_call( |
||||||
|
iter([_REQUEST] * STREAM_LENGTH) |
||||||
|
) |
||||||
|
|
||||||
|
|
||||||
|
def stream_stream_call(port): |
||||||
|
with grpc.insecure_channel(f"localhost:{port}") as channel: |
||||||
|
multi_callable = channel.stream_stream(_STREAM_STREAM) |
||||||
|
call = multi_callable(iter([_REQUEST] * STREAM_LENGTH)) |
||||||
|
for _ in call: |
||||||
|
pass |
||||||
@ -0,0 +1,23 @@ |
|||||||
|
%YAML 1.2 |
||||||
|
--- | |
||||||
|
# Copyright 2023 The gRPC Authors |
||||||
|
# |
||||||
|
# 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. |
||||||
|
# |
||||||
|
# This file has been automatically generated from a template file. |
||||||
|
# Please make modifications to |
||||||
|
# `$REPO_ROOT/templates/src/python/grpcio/_parallel_compile_patch.py.template` |
||||||
|
# instead. This file can be regenerated from the template by running |
||||||
|
# `tools/buildgen/generate_projects.sh`. |
||||||
|
|
||||||
|
<%include file="../_parallel_compile_patch.py.include" /> |
||||||
Loading…
Reference in new issue