Closes datacontract#106 - add gRPC support to opentelemetry integration

CHANGELOG.md

@@ -7,7 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 - Added export format **great-expectations**: `datacontract export --format great-expectations`  
-
+- Added gRPC support to opentelemetry integration for publishing test results
 
 
 ## [0.9.7] - 2024-03-15

README.md

@@ -159,14 +159,13 @@ The metric name is "datacontract.cli.test.result" and it uses the following enco
 # Fetch current data contract, execute tests on production, and publish result to open telemetry
 $ EXPORT OTEL_SERVICE_NAME=datacontract-cli
 $ EXPORT OTEL_EXPORTER_OTLP_ENDPOINT=https://YOUR_ID.apm.westeurope.azure.elastic-cloud.com:443
-$ EXPORT OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer%20secret (Optional, when using SaaS Products)
-$ EXPORT OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf (Optional, because it is the default value)
+$ EXPORT OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer%20secret # Optional, when using SaaS Products
+$ EXPORT OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf # Optional, default is http/protobuf - use value grpc to use the gRPC protocol instead
 # Send to OpenTelemetry
 $ datacontract test https://demo.datamesh-manager.com/demo279750347121/datacontracts/4df9d6ee-e55d-4088-9598-b635b2fdcbbc/datacontract.yaml --server production --publish-to-opentelemetry
 ```
 
 Current limitations:
-- no gRPC support
 - currently, only ConsoleExporter and OTLP Exporter
 - Metrics only, no logs yet (but loosely planned)
 

datacontract/integration/publish_opentelemetry.py

@@ -1,10 +1,11 @@
 import logging
 import os
 from importlib import metadata
-from uuid import uuid4
 import math
 
 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as OTLPgRPCMetricExporter
+
 from opentelemetry.metrics import Observation
 
 from datacontract.model.run import \
@@ -13,7 +14,8 @@
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader
 
-# Publishes metrics of a test run. 
+
+# Publishes metrics of a test run.
 # Metric contains the values: 
 # 0 == test run passed, 
 # 1 == test run has warnings
@@ -26,7 +28,7 @@
 # OTEL_SERVICE_NAME=datacontract-cli
 # OTEL_EXPORTER_OTLP_ENDPOINT=https://YOUR_ID.apm.westeurope.azure.elastic-cloud.com:443
 # OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer%20secret (Optional, when using SaaS Products)
-# OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf (Optional, because it is the default value)
+# OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf and OTEL_EXPORTER_OTLP_PROTOCOL=grpc
 #
 # Current limitations:
 # - no gRPC support
@@ -80,10 +82,20 @@ def _to_observation(run):
 
 class Telemetry:
     def __init__(self):
-        self.exporter = ConsoleMetricExporter()
-        self.remote_exporter = OTLPMetricExporter()
+
+        protocol = os.getenv("OTEL_EXPORTER_OTLP_PROTOCOL")
+
+        # lower to allow grpc, GRPC and alike values.
+        if protocol and protocol.lower() == "grpc":
+            self.remote_exporter = OTLPgRPCMetricExporter()
+        else:
+            # Fallback to default OTEL http/protobuf which is used when the variable is not set.
+            # This Exporter also works for http/json.
+            self.remote_exporter = OTLPMetricExporter()
+
+        self.console_exporter = ConsoleMetricExporter()
         # using math.inf so it does not collect periodically. we do this in collect ourselves, one-time.
-        self.reader = PeriodicExportingMetricReader(self.exporter, export_interval_millis=math.inf)
+        self.reader = PeriodicExportingMetricReader(self.console_exporter, export_interval_millis=math.inf)
         self.remote_reader = PeriodicExportingMetricReader(self.remote_exporter, export_interval_millis=math.inf)
         provider = MeterProvider(metric_readers=[self.reader, self.remote_reader])
         metrics.set_meter_provider(provider)

pyproject.toml

@@ -36,6 +36,8 @@ dependencies = [
   "s3fs==2024.2.0",
   "rdflib==7.0.0",
   "avro==1.11.3",
+  "opentelemetry-exporter-otlp-proto-grpc~=1.16.0",
+  "opentelemetry-exporter-otlp-proto-http~=1.16.0"
 ]
 
 [project.optional-dependencies]

tests/test_examples_s3_csv.py

@@ -23,9 +23,9 @@ def minio_container():
         yield minio_container
 
 
-def test_examples_s3_csv(minio_container):
-    os.environ['DATACONTRACT_S3_ACCESS_KEY_ID'] = s3_access_key
-    os.environ['DATACONTRACT_S3_SECRET_ACCESS_KEY'] = s3_secret_access_key
+def test_examples_s3_csv(minio_container, monkeypatch):
+    monkeypatch.setenv("DATACONTRACT_S3_ACCESS_KEY_ID", s3_access_key)
+    monkeypatch.setenv("DATACONTRACT_S3_SECRET_ACCESS_KEY", s3_secret_access_key)
     data_contract_str = _prepare_s3_files(minio_container)
     data_contract = DataContract(data_contract_str=data_contract_str)
 

tests/test_examples_s3_json.py

@@ -20,9 +20,9 @@ def minio_container():
         yield minio_container
 
 
-def test_examples_s3_json(minio_container):
-    os.environ['DATACONTRACT_S3_ACCESS_KEY_ID'] = "test-access"
-    os.environ['DATACONTRACT_S3_SECRET_ACCESS_KEY'] = "test-secret"
+def test_examples_s3_json(minio_container, monkeypatch):
+    monkeypatch.setenv("DATACONTRACT_S3_ACCESS_KEY_ID","test-access")
+    monkeypatch.setenv("DATACONTRACT_S3_SECRET_ACCESS_KEY","test-secret")
     data_contract_str = _prepare_s3_files(minio_container)
     data_contract = DataContract(data_contract_str=data_contract_str)
 

tests/test_integration_opentelemetry.py

@@ -1,10 +1,12 @@
 import logging
 from uuid import uuid4
 
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as OTLPgRPCMetricExporter
+
 from opentelemetry.metrics import Observation
-from typer.testing import CliRunner
 
-from datacontract.integration.publish_opentelemetry import _to_observation
+from datacontract.integration.publish_opentelemetry import _to_observation, Telemetry
 from datacontract.model.run import Run
 
 logging.basicConfig(level=logging.DEBUG, force=True)
@@ -29,3 +31,41 @@ def test_convert_to_observation():
     actual = _to_observation(run)
 
     assert expected == actual
+
+
+def test_http_exporter_chosen_without_env():
+    telemetry = Telemetry()
+
+    assert isinstance(telemetry.remote_exporter,
+                      OTLPMetricExporter), ("Should use OTLPMetricExporter when OTEL_EXPORTER_OTLP_PROTOCOL is "
+                                            "not set")
+
+
+def test_http_exporter_chosen_with_env_httpprotobuf(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf")
+
+    telemetry = Telemetry()
+
+    assert isinstance(telemetry.remote_exporter,
+                      OTLPMetricExporter), ("Should use OTLPMetricExporter when OTEL_EXPORTER_OTLP_PROTOCOL is "
+                                            "http/protobuf")
+
+
+def test_http_exporter_chosen_with_env_httpjson(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/json")
+
+    telemetry = Telemetry()
+
+    assert isinstance(telemetry.remote_exporter,
+                      OTLPMetricExporter), ("Should use OTLPMetricExporter when OTEL_EXPORTER_OTLP_PROTOCOL is "
+                                            "http/json")
+
+
+def test_grpc_exporter_chosen_with_env_GRPC(monkeypatch):
+    monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "GRPC")
+
+    telemetry = Telemetry()
+
+    assert isinstance(telemetry.remote_exporter,
+                      OTLPgRPCMetricExporter), ("Should use OTLPMetricExporter from gRPC package "
+                                                "when OTEL_EXPORTER_OTLP_PROTOCOL is GRPC")