|
| 1 | +# Copyright The OpenTelemetry Authors |
| 2 | +# |
| 3 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +# you may not use this file except in compliance with the License. |
| 5 | +# You may obtain a copy of the License at |
| 6 | +# |
| 7 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +# |
| 9 | +# Unless required by applicable law or agreed to in writing, software |
| 10 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +# See the License for the specific language governing permissions and |
| 13 | +# limitations under the License. |
| 14 | + |
| 15 | +"""OTLP Exporter""" |
| 16 | + |
| 17 | +import logging |
| 18 | +from abc import ABC, abstractmethod |
| 19 | +from collections.abc import Mapping, Sequence |
| 20 | +from time import sleep |
| 21 | + |
| 22 | +from backoff import expo |
| 23 | +from google.rpc.error_details_pb2 import RetryInfo |
| 24 | +from grpc import ( |
| 25 | + ChannelCredentials, |
| 26 | + RpcError, |
| 27 | + StatusCode, |
| 28 | + insecure_channel, |
| 29 | + secure_channel, |
| 30 | +) |
| 31 | + |
| 32 | +from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue |
| 33 | +from opentelemetry.proto.resource.v1.resource_pb2 import Resource |
| 34 | + |
| 35 | +logger = logging.getLogger(__name__) |
| 36 | + |
| 37 | + |
| 38 | +def _translate_key_values(key, value): |
| 39 | + |
| 40 | + if isinstance(value, bool): |
| 41 | + any_value = AnyValue(bool_value=value) |
| 42 | + |
| 43 | + elif isinstance(value, str): |
| 44 | + any_value = AnyValue(string_value=value) |
| 45 | + |
| 46 | + elif isinstance(value, int): |
| 47 | + any_value = AnyValue(int_value=value) |
| 48 | + |
| 49 | + elif isinstance(value, float): |
| 50 | + any_value = AnyValue(double_value=value) |
| 51 | + |
| 52 | + elif isinstance(value, Sequence): |
| 53 | + any_value = AnyValue(array_value=value) |
| 54 | + |
| 55 | + elif isinstance(value, Mapping): |
| 56 | + any_value = AnyValue(kvlist_value=value) |
| 57 | + |
| 58 | + else: |
| 59 | + raise Exception( |
| 60 | + "Invalid type {} of value {}".format(type(value), value) |
| 61 | + ) |
| 62 | + |
| 63 | + return KeyValue(key=key, value=any_value) |
| 64 | + |
| 65 | + |
| 66 | +def _get_resource_data( |
| 67 | + sdk_resource_instrumentation_library_data, resource_class, name |
| 68 | +): |
| 69 | + |
| 70 | + resource_data = [] |
| 71 | + |
| 72 | + for ( |
| 73 | + sdk_resource, |
| 74 | + instrumentation_library_data, |
| 75 | + ) in sdk_resource_instrumentation_library_data.items(): |
| 76 | + |
| 77 | + collector_resource = Resource() |
| 78 | + |
| 79 | + for key, value in sdk_resource.labels.items(): |
| 80 | + |
| 81 | + try: |
| 82 | + # pylint: disable=no-member |
| 83 | + collector_resource.attributes.append( |
| 84 | + _translate_key_values(key, value) |
| 85 | + ) |
| 86 | + except Exception as error: # pylint: disable=broad-except |
| 87 | + logger.exception(error) |
| 88 | + |
| 89 | + resource_data.append( |
| 90 | + resource_class( |
| 91 | + **{ |
| 92 | + "resource": collector_resource, |
| 93 | + "instrumentation_library_{}".format(name): [ |
| 94 | + instrumentation_library_data |
| 95 | + ], |
| 96 | + } |
| 97 | + ) |
| 98 | + ) |
| 99 | + |
| 100 | + return resource_data |
| 101 | + |
| 102 | + |
| 103 | +# pylint: disable=no-member |
| 104 | +class OTLPExporterMixin(ABC): |
| 105 | + """OTLP span/metric exporter |
| 106 | +
|
| 107 | + Args: |
| 108 | + endpoint: OpenTelemetry Collector receiver endpoint |
| 109 | + credentials: ChannelCredentials object for server authentication |
| 110 | + metadata: Metadata to send when exporting |
| 111 | + """ |
| 112 | + |
| 113 | + def __init__( |
| 114 | + self, |
| 115 | + endpoint: str = "localhost:55680", |
| 116 | + credentials: ChannelCredentials = None, |
| 117 | + metadata: tuple = None, |
| 118 | + ): |
| 119 | + super().__init__() |
| 120 | + |
| 121 | + self._metadata = metadata |
| 122 | + self._collector_span_kwargs = None |
| 123 | + |
| 124 | + if credentials is None: |
| 125 | + self._client = self._stub(insecure_channel(endpoint)) |
| 126 | + else: |
| 127 | + self._client = self._stub(secure_channel(endpoint, credentials)) |
| 128 | + |
| 129 | + @abstractmethod |
| 130 | + def _translate_data(self, data): |
| 131 | + pass |
| 132 | + |
| 133 | + def _export(self, data): |
| 134 | + # expo returns a generator that yields delay values which grow |
| 135 | + # exponentially. Once delay is greater than max_value, the yielded |
| 136 | + # value will remain constant. |
| 137 | + # max_value is set to 900 (900 seconds is 15 minutes) to use the same |
| 138 | + # value as used in the Go implementation. |
| 139 | + |
| 140 | + max_value = 900 |
| 141 | + |
| 142 | + for delay in expo(max_value=max_value): |
| 143 | + |
| 144 | + if delay == max_value: |
| 145 | + return self._result.FAILURE |
| 146 | + |
| 147 | + try: |
| 148 | + self._client.Export( |
| 149 | + request=self._translate_data(data), |
| 150 | + metadata=self._metadata, |
| 151 | + ) |
| 152 | + |
| 153 | + return self._result.SUCCESS |
| 154 | + |
| 155 | + except RpcError as error: |
| 156 | + |
| 157 | + if error.code() in [ |
| 158 | + StatusCode.CANCELLED, |
| 159 | + StatusCode.DEADLINE_EXCEEDED, |
| 160 | + StatusCode.PERMISSION_DENIED, |
| 161 | + StatusCode.UNAUTHENTICATED, |
| 162 | + StatusCode.RESOURCE_EXHAUSTED, |
| 163 | + StatusCode.ABORTED, |
| 164 | + StatusCode.OUT_OF_RANGE, |
| 165 | + StatusCode.UNAVAILABLE, |
| 166 | + StatusCode.DATA_LOSS, |
| 167 | + ]: |
| 168 | + |
| 169 | + retry_info_bin = dict(error.trailing_metadata()).get( |
| 170 | + "google.rpc.retryinfo-bin" |
| 171 | + ) |
| 172 | + if retry_info_bin is not None: |
| 173 | + retry_info = RetryInfo() |
| 174 | + retry_info.ParseFromString(retry_info_bin) |
| 175 | + delay = ( |
| 176 | + retry_info.retry_delay.seconds |
| 177 | + + retry_info.retry_delay.nanos / 1.0e9 |
| 178 | + ) |
| 179 | + |
| 180 | + logger.debug( |
| 181 | + "Waiting %ss before retrying export of span", delay |
| 182 | + ) |
| 183 | + sleep(delay) |
| 184 | + continue |
| 185 | + |
| 186 | + if error.code() == StatusCode.OK: |
| 187 | + return self._result.SUCCESS |
| 188 | + |
| 189 | + return self.result.FAILURE |
| 190 | + |
| 191 | + return self._result.FAILURE |
| 192 | + |
| 193 | + def shutdown(self): |
| 194 | + pass |
0 commit comments