Move all OTLP exporter tests that are testing the

underlying behavior in the mixin to the mixin unit
tests, instead of having them specified multiple
times in the Metric/Log/Trace exporters.

Fix the shutdown tests which were flaky, so that
they just test whether a pending export call completes
or not.  Update shutdown so it doesn't release the
lock -- in cases where an export call is pending,
export then also releases the lock causing a Runtime
Error: https://docs.python.org/3/library/threading.html#threading.Lock.release.

Author: DylanRussell

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

@@ -289,11 +289,11 @@ def _export(
         # expo returns a generator that yields delay values which grow
         # exponentially. Once delay is greater than max_value, the yielded
         # value will remain constant.
-        for delay in _create_exp_backoff_generator(max_value=max_value):
-            if delay == max_value or self._shutdown:
-                return self._result.FAILURE
+        with self._export_lock:
+            for delay in _create_exp_backoff_generator(max_value=max_value):
+                if delay == max_value or self._shutdown:
+                    return self._result.FAILURE
 
-            with self._export_lock:
                 try:
                     self._client.Export(
                         request=self._translate_data(data),
@@ -352,15 +352,14 @@ def _export(
 
         return self._result.FAILURE
 
-    def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
+    def shutdown(self, timeout_millis: float = 30_000) -> None:
         if self._shutdown:
             logger.warning("Exporter already shutdown, ignoring call")
             return
         # wait for the last export if any
         self._export_lock.acquire(timeout=timeout_millis / 1e3)
         self._shutdown = True
         self._channel.close()
-        self._export_lock.release()
 
     @property
     @abstractmethod

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/trace_exporter/__init__.py

@@ -142,8 +142,8 @@ def _translate_data(
     def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         return self._export(spans)
 
-    def shutdown(self) -> None:
-        OTLPExporterMixin.shutdown(self)
+    def shutdown(self, timeout_millis: float = 30_000) -> None:
+        OTLPExporterMixin.shutdown(self, timeout_millis)
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         """Nothing is buffered in this exporter, so this method does nothing."""

exporter/opentelemetry-exporter-otlp-proto-grpc/tests/logs/test_otlp_logs_exporter.py

@@ -15,33 +15,20 @@
 # pylint: disable=too-many-lines
 
 import time
-from concurrent.futures import ThreadPoolExecutor
 from os.path import dirname
 from unittest import TestCase
 from unittest.mock import patch
 
-from google.protobuf.duration_pb2 import (  # pylint: disable=no-name-in-module
-    Duration,
-)
 from google.protobuf.json_format import MessageToDict
-from google.rpc.error_details_pb2 import (  # pylint: disable=no-name-in-module
-    RetryInfo,
-)
-from grpc import ChannelCredentials, Compression, StatusCode, server
+from grpc import ChannelCredentials, Compression
 
 from opentelemetry._logs import SeverityNumber
 from opentelemetry.exporter.otlp.proto.common._internal import _encode_value
 from opentelemetry.exporter.otlp.proto.grpc._log_exporter import (
     OTLPLogExporter,
 )
-from opentelemetry.exporter.otlp.proto.grpc.version import __version__
 from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import (
     ExportLogsServiceRequest,
-    ExportLogsServiceResponse,
-)
-from opentelemetry.proto.collector.logs.v1.logs_service_pb2_grpc import (
-    LogsServiceServicer,
-    add_LogsServiceServicer_to_server,
 )
 from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue
 from opentelemetry.proto.common.v1.common_pb2 import (
@@ -53,7 +40,6 @@
     Resource as OTLPResource,
 )
 from opentelemetry.sdk._logs import LogData, LogRecord
-from opentelemetry.sdk._logs.export import LogExportResult
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_LOGS_CERTIFICATE,
     OTEL_EXPORTER_OTLP_LOGS_CLIENT_CERTIFICATE,
@@ -70,62 +56,9 @@
 THIS_DIR = dirname(__file__)
 
 
-class LogsServiceServicerUNAVAILABLEDelay(LogsServiceServicer):
-    # pylint: disable=invalid-name,unused-argument,no-self-use
-    def Export(self, request, context):
-        context.set_code(StatusCode.UNAVAILABLE)
-
-        context.send_initial_metadata(
-            (("google.rpc.retryinfo-bin", RetryInfo().SerializeToString()),)
-        )
-        context.set_trailing_metadata(
-            (
-                (
-                    "google.rpc.retryinfo-bin",
-                    RetryInfo(
-                        retry_delay=Duration(nanos=int(1e7))
-                    ).SerializeToString(),
-                ),
-            )
-        )
-
-        return ExportLogsServiceResponse()
-
-
-class LogsServiceServicerUNAVAILABLE(LogsServiceServicer):
-    # pylint: disable=invalid-name,unused-argument,no-self-use
-    def Export(self, request, context):
-        context.set_code(StatusCode.UNAVAILABLE)
-
-        return ExportLogsServiceResponse()
-
-
-class LogsServiceServicerSUCCESS(LogsServiceServicer):
-    # pylint: disable=invalid-name,unused-argument,no-self-use
-    def Export(self, request, context):
-        context.set_code(StatusCode.OK)
-
-        return ExportLogsServiceResponse()
-
-
-class LogsServiceServicerALREADY_EXISTS(LogsServiceServicer):
-    # pylint: disable=invalid-name,unused-argument,no-self-use
-    def Export(self, request, context):
-        context.set_code(StatusCode.ALREADY_EXISTS)
-
-        return ExportLogsServiceResponse()
-
-
 class TestOTLPLogExporter(TestCase):
     def setUp(self):
         self.exporter = OTLPLogExporter()
-
-        self.server = server(ThreadPoolExecutor(max_workers=10))
-
-        self.server.add_insecure_port("127.0.0.1:4317")
-
-        self.server.start()
-
         self.log_data_1 = LogData(
             log_record=LogRecord(
                 timestamp=int(time.time() * 1e9),
@@ -204,9 +137,6 @@ def setUp(self):
             ),
         )
 
-    def tearDown(self):
-        self.server.stop(None)
-
     def test_exporting(self):
         # pylint: disable=protected-access
         self.assertEqual(self.exporter._exporting, "logs")
@@ -296,145 +226,6 @@ def test_env_variables_with_only_certificate(
 
         mock_logger_error.assert_not_called()
 
-    @patch(
-        "opentelemetry.exporter.otlp.proto.grpc.exporter.ssl_channel_credentials"
-    )
-    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.secure_channel")
-    @patch(
-        "opentelemetry.exporter.otlp.proto.grpc._log_exporter.OTLPLogExporter._stub"
-    )
-    # pylint: disable=unused-argument
-    def test_no_credentials_error(
-        self, mock_ssl_channel, mock_secure, mock_stub
-    ):
-        OTLPLogExporter(insecure=False)
-        self.assertTrue(mock_ssl_channel.called)
-
-    # pylint: disable=no-self-use
-    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.insecure_channel")
-    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.secure_channel")
-    def test_otlp_exporter_endpoint(self, mock_secure, mock_insecure):
-        expected_endpoint = "localhost:4317"
-        endpoints = [
-            (
-                "http://localhost:4317",
-                None,
-                mock_insecure,
-            ),
-            (
-                "localhost:4317",
-                None,
-                mock_secure,
-            ),
-            (
-                "http://localhost:4317",
-                True,
-                mock_insecure,
-            ),
-            (
-                "localhost:4317",
-                True,
-                mock_insecure,
-            ),
-            (
-                "http://localhost:4317",
-                False,
-                mock_secure,
-            ),
-            (
-                "localhost:4317",
-                False,
-                mock_secure,
-            ),
-            (
-                "https://localhost:4317",
-                False,
-                mock_secure,
-            ),
-            (
-                "https://localhost:4317",
-                None,
-                mock_secure,
-            ),
-            (
-                "https://localhost:4317",
-                True,
-                mock_secure,
-            ),
-        ]
-
-        # pylint: disable=C0209
-        for endpoint, insecure, mock_method in endpoints:
-            OTLPLogExporter(endpoint=endpoint, insecure=insecure)
-            self.assertEqual(
-                1,
-                mock_method.call_count,
-                "expected {} to be called for {} {}".format(
-                    mock_method, endpoint, insecure
-                ),
-            )
-            self.assertEqual(
-                expected_endpoint,
-                mock_method.call_args[0][0],
-                "expected {} got {} {}".format(
-                    expected_endpoint, mock_method.call_args[0][0], endpoint
-                ),
-            )
-            mock_method.reset_mock()
-
-    def test_otlp_headers_from_env(self):
-        # pylint: disable=protected-access
-        self.assertEqual(
-            self.exporter._headers,
-            (("user-agent", "OTel-OTLP-Exporter-Python/" + __version__),),
-        )
-
-    @patch(
-        "opentelemetry.exporter.otlp.proto.grpc.exporter._create_exp_backoff_generator"
-    )
-    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.sleep")
-    def test_unavailable(self, mock_sleep, mock_expo):
-        mock_expo.configure_mock(**{"return_value": [0.01]})
-
-        add_LogsServiceServicer_to_server(
-            LogsServiceServicerUNAVAILABLE(), self.server
-        )
-        self.assertEqual(
-            self.exporter.export([self.log_data_1]), LogExportResult.FAILURE
-        )
-        mock_sleep.assert_called_with(0.01)
-
-    @patch(
-        "opentelemetry.exporter.otlp.proto.grpc.exporter._create_exp_backoff_generator"
-    )
-    @patch("opentelemetry.exporter.otlp.proto.grpc.exporter.sleep")
-    def test_unavailable_delay(self, mock_sleep, mock_expo):
-        mock_expo.configure_mock(**{"return_value": [1]})
-
-        add_LogsServiceServicer_to_server(
-            LogsServiceServicerUNAVAILABLEDelay(), self.server
-        )
-        self.assertEqual(
-            self.exporter.export([self.log_data_1]), LogExportResult.FAILURE
-        )
-        mock_sleep.assert_called_with(0.01)
-
-    def test_success(self):
-        add_LogsServiceServicer_to_server(
-            LogsServiceServicerSUCCESS(), self.server
-        )
-        self.assertEqual(
-            self.exporter.export([self.log_data_1]), LogExportResult.SUCCESS
-        )
-
-    def test_failure(self):
-        add_LogsServiceServicer_to_server(
-            LogsServiceServicerALREADY_EXISTS(), self.server
-        )
-        self.assertEqual(
-            self.exporter.export([self.log_data_1]), LogExportResult.FAILURE
-        )
-
     def export_log_and_deserialize(self, log_data):
         # pylint: disable=protected-access
         translated_data = self.exporter._translate_data([log_data])