Sync

.circleci/config.yml

@@ -141,7 +141,7 @@ workflows:
       - build:
           matrix:
             parameters:
-              version: ["py38", "py37", "py36", "py35", "pypy3"]
+              version: ["py37", "py36", "py35", "pypy3"]
               package: ["core", "exporter", "instrumentation"]
       - build-py34:
           matrix:

.pylintrc

@@ -65,6 +65,7 @@ disable=missing-docstring,
       too-few-public-methods, # Might be good to re-enable this later.
       too-many-instance-attributes,
       too-many-arguments,
+      duplicate-code,
       ungrouped-imports, # Leave this up to isort
       wrong-import-order, # Leave this up to isort
       bad-continuation, # Leave this up to black

docs/getting-started.rst

@@ -312,10 +312,10 @@ Install the OpenTelemetry Collector exporter:
 
 .. code-block:: sh
 
-    pip install opentelemetry-instrumentation-otcollector
+    pip install opentelemetry-exporter-otlp
 
 And execute the following script:
 
-.. literalinclude:: getting_started/otcollector_example.py
+.. literalinclude:: getting_started/otlpcollector_example.py
     :language: python
     :lines: 15-

docs/getting_started/otcollector_example.py → docs/getting_started/otlpcollector_example.py

@@ -16,33 +16,29 @@
 import time
 
 from opentelemetry import metrics, trace
-from opentelemetry.ext.otcollector.metrics_exporter import (
-    CollectorMetricsExporter,
-)
-from opentelemetry.ext.otcollector.trace_exporter import CollectorSpanExporter
+from opentelemetry.exporter.otlp.metrics_exporter import OTLPMetricsExporter
+from opentelemetry.exporter.otlp.trace_exporter import OTLPSpanExporter
 from opentelemetry.sdk.metrics import Counter, MeterProvider
 from opentelemetry.sdk.metrics.export.controller import PushController
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchExportSpanProcessor
 
-# create a CollectorSpanExporter
-span_exporter = CollectorSpanExporter(
-    # optional:
-    # endpoint="myCollectorUrl:55678",
-    # service_name="test_service",
-    # host_name="machine/container name",
+span_exporter = OTLPSpanExporter(
+    # optional
+    # endpoint:="myCollectorURL:55678",
+    # credentials=ChannelCredentials(credentials),
+    # metadata=(("metadata", "metadata")),
 )
 tracer_provider = TracerProvider()
 trace.set_tracer_provider(tracer_provider)
 span_processor = BatchExportSpanProcessor(span_exporter)
 tracer_provider.add_span_processor(span_processor)
 
-# create a CollectorMetricsExporter
-metric_exporter = CollectorMetricsExporter(
-    # optional:
-    # endpoint="myCollectorUrl:55678",
-    # service_name="test_service",
-    # host_name="machine/container name",
+metric_exporter = OTLPMetricsExporter(
+    # optional
+    # endpoint:="myCollectorURL:55678",
+    # credentials=ChannelCredentials(credentials),
+    # metadata=(("metadata", "metadata")),
 )
 
 # Meter is responsible for creating and recording metrics

exporter/opentelemetry-exporter-otlp/CHANGELOG.md

@@ -2,9 +2,10 @@
 
 ## Unreleased
 
-## Version 0.12b0
+- Add metric OTLP exporter
+  ([#835](https://github.com/open-telemetry/opentelemetry-python/pull/835))
 
-Released 2020-08-14
+## Version 0.12b0
 
 - Change package name to opentelemetry-exporter-otlp
   ([#953](https://github.com/open-telemetry/opentelemetry-python/pull/953))

exporter/opentelemetry-exporter-otlp/src/opentelemetry/exporter/otlp/exporter.py

@@ -0,0 +1,194 @@
+# Copyright The OpenTelemetry 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.
+
+"""OTLP Exporter"""
+
+import logging
+from abc import ABC, abstractmethod
+from collections.abc import Mapping, Sequence
+from time import sleep
+
+from backoff import expo
+from google.rpc.error_details_pb2 import RetryInfo
+from grpc import (
+    ChannelCredentials,
+    RpcError,
+    StatusCode,
+    insecure_channel,
+    secure_channel,
+)
+
+from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue
+from opentelemetry.proto.resource.v1.resource_pb2 import Resource
+
+logger = logging.getLogger(__name__)
+
+
+def _translate_key_values(key, value):
+
+    if isinstance(value, bool):
+        any_value = AnyValue(bool_value=value)
+
+    elif isinstance(value, str):
+        any_value = AnyValue(string_value=value)
+
+    elif isinstance(value, int):
+        any_value = AnyValue(int_value=value)
+
+    elif isinstance(value, float):
+        any_value = AnyValue(double_value=value)
+
+    elif isinstance(value, Sequence):
+        any_value = AnyValue(array_value=value)
+
+    elif isinstance(value, Mapping):
+        any_value = AnyValue(kvlist_value=value)
+
+    else:
+        raise Exception(
+            "Invalid type {} of value {}".format(type(value), value)
+        )
+
+    return KeyValue(key=key, value=any_value)
+
+
+def _get_resource_data(
+    sdk_resource_instrumentation_library_data, resource_class, name
+):
+
+    resource_data = []
+
+    for (
+        sdk_resource,
+        instrumentation_library_data,
+    ) in sdk_resource_instrumentation_library_data.items():
+
+        collector_resource = Resource()
+
+        for key, value in sdk_resource.labels.items():
+
+            try:
+                # pylint: disable=no-member
+                collector_resource.attributes.append(
+                    _translate_key_values(key, value)
+                )
+            except Exception as error:  # pylint: disable=broad-except
+                logger.exception(error)
+
+        resource_data.append(
+            resource_class(
+                **{
+                    "resource": collector_resource,
+                    "instrumentation_library_{}".format(name): [
+                        instrumentation_library_data
+                    ],
+                }
+            )
+        )
+
+    return resource_data
+
+
+# pylint: disable=no-member
+class OTLPExporterMixin(ABC):
+    """OTLP span/metric exporter
+
+    Args:
+        endpoint: OpenTelemetry Collector receiver endpoint
+        credentials: ChannelCredentials object for server authentication
+        metadata: Metadata to send when exporting
+    """
+
+    def __init__(
+        self,
+        endpoint: str = "localhost:55680",
+        credentials: ChannelCredentials = None,
+        metadata: tuple = None,
+    ):
+        super().__init__()
+
+        self._metadata = metadata
+        self._collector_span_kwargs = None
+
+        if credentials is None:
+            self._client = self._stub(insecure_channel(endpoint))
+        else:
+            self._client = self._stub(secure_channel(endpoint, credentials))
+
+    @abstractmethod
+    def _translate_data(self, data):
+        pass
+
+    def _export(self, data):
+        # expo returns a generator that yields delay values which grow
+        # exponentially. Once delay is greater than max_value, the yielded
+        # value will remain constant.
+        # max_value is set to 900 (900 seconds is 15 minutes) to use the same
+        # value as used in the Go implementation.
+
+        max_value = 900
+
+        for delay in expo(max_value=max_value):
+
+            if delay == max_value:
+                return self._result.FAILURE
+
+            try:
+                self._client.Export(
+                    request=self._translate_data(data),
+                    metadata=self._metadata,
+                )
+
+                return self._result.SUCCESS
+
+            except RpcError as error:
+
+                if error.code() in [
+                    StatusCode.CANCELLED,
+                    StatusCode.DEADLINE_EXCEEDED,
+                    StatusCode.PERMISSION_DENIED,
+                    StatusCode.UNAUTHENTICATED,
+                    StatusCode.RESOURCE_EXHAUSTED,
+                    StatusCode.ABORTED,
+                    StatusCode.OUT_OF_RANGE,
+                    StatusCode.UNAVAILABLE,
+                    StatusCode.DATA_LOSS,
+                ]:
+
+                    retry_info_bin = dict(error.trailing_metadata()).get(
+                        "google.rpc.retryinfo-bin"
+                    )
+                    if retry_info_bin is not None:
+                        retry_info = RetryInfo()
+                        retry_info.ParseFromString(retry_info_bin)
+                        delay = (
+                            retry_info.retry_delay.seconds
+                            + retry_info.retry_delay.nanos / 1.0e9
+                        )
+
+                    logger.debug(
+                        "Waiting %ss before retrying export of span", delay
+                    )
+                    sleep(delay)
+                    continue
+
+                if error.code() == StatusCode.OK:
+                    return self._result.SUCCESS
+
+                return self.result.FAILURE
+
+        return self._result.FAILURE
+
+    def shutdown(self):
+        pass