update udp message schema

Author: srprash

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/aws_opentelemetry_configurator.py

@@ -16,7 +16,7 @@
     AwsMetricAttributesSpanExporterBuilder,
 )
 from amazon.opentelemetry.distro.aws_span_metrics_processor_builder import AwsSpanMetricsProcessorBuilder
-from amazon.opentelemetry.distro.otlp_udp_exporter import OtlpUdpMetricExporter, OtlpUdpSpanExporter
+from amazon.opentelemetry.distro.otlp_udp_exporter import OTLPUdpMetricExporter, OTLPUdpSpanExporter
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPHttpOTLPMetricExporter
 from opentelemetry.sdk._configuration import (
@@ -40,6 +40,7 @@
 from opentelemetry.sdk.extension.aws.resource.ecs import AwsEcsResourceDetector
 from opentelemetry.sdk.extension.aws.resource.eks import AwsEksResourceDetector
 from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics._internal.export import ConsoleMetricExporter
 from opentelemetry.sdk.metrics._internal.instrument import (
     Counter,
     Histogram,
@@ -223,7 +224,7 @@ def _customize_exporter(span_exporter: SpanExporter, resource: Resource) -> Span
     if not _is_application_signals_enabled():
         return span_exporter
     if _is_lambda_environment():
-        return AwsMetricAttributesSpanExporterBuilder(OtlpUdpSpanExporter(), resource).build()
+        return AwsMetricAttributesSpanExporterBuilder(OTLPUdpSpanExporter(), resource).build()
     return AwsMetricAttributesSpanExporterBuilder(span_exporter, resource).build()
 
 
@@ -302,7 +303,7 @@ def create_exporter(self):
         if _is_lambda_environment():
             # When running in Lambda, export Application Signals metrics over UDP
             application_signals_endpoint = os.environ.get(AWS_XRAY_DAEMON_ADDRESS_CONFIG, "127.0.0.1:2000")
-            return OtlpUdpMetricExporter(endpoint=application_signals_endpoint, preferred_temporality=temporality_dict)
+            return OTLPUdpMetricExporter(endpoint=application_signals_endpoint, preferred_temporality=temporality_dict)
 
         if protocol == "http/protobuf":
             application_signals_endpoint = os.environ.get(

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/otlp_udp_exporter.py

@@ -1,5 +1,6 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+import base64
 import socket
 from logging import Logger, getLogger
 from typing import Dict, Optional, Sequence, Tuple
@@ -18,27 +19,30 @@
 
 DEFAULT_ENDPOINT = "127.0.0.1:2000"
 PROTOCOL_HEADER = '{"format":"json","version":1}\n'
-PROTOCOL_DELIMITER = "\n"
+FORMAT_OTEL_METRICS_BINARY_PREFIX = "M1"
+FORMAT_OTEL_TRACES_BINARY_PREFIX = "T1"
 
 _logger: Logger = getLogger(__name__)
 
 
 class UdpExporter:
     def __init__(self, endpoint: Optional[str] = None):
-        self._endpoint = endpoint or DEFAULT_ENDPOINT  # TODO: read from some env var??
+        self._endpoint = endpoint or DEFAULT_ENDPOINT
         self._host, self._port = self._parse_endpoint(self._endpoint)
         self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self._socket.setblocking(False)
 
-    def send_data(self, data: str, signal_format: str):
-        udp_data = f'{{"format":"{signal_format}","data":{data}}}'
-        message = PROTOCOL_HEADER + udp_data
+    def send_data(self, data: bytes, signal_format_prefix: str):
+        # base64 encoding and then converting to string with utf-8
+        base64_encoded_string: str = base64.b64encode(data).decode("utf-8")
+        message = f"{PROTOCOL_HEADER}{signal_format_prefix}{base64_encoded_string}"
 
         try:
-            print("Sending UDP data: ", message)  # TODO: remove
+            _logger.debug("Sending UDP data: ", message)
             self._socket.sendto(message.encode("utf-8"), (self._host, int(self._port)))
         except Exception as exc:  # pylint: disable=broad-except
             _logger.error("Error sending UDP data: %s", exc)
+            raise
 
     def shutdown(self):
         self._socket.close()
@@ -55,7 +59,7 @@ def _parse_endpoint(self, endpoint: str) -> Tuple[str, int]:
         return host, port
 
 
-class OtlpUdpMetricExporter(MetricExporter):
+class OTLPUdpMetricExporter(MetricExporter):
     def __init__(
         self,
         endpoint: Optional[str] = None,
@@ -76,8 +80,13 @@ def export(
         **kwargs,
     ) -> MetricExportResult:
         serialized_data = encode_metrics(metrics_data).SerializeToString()
-        self._udp_exporter.send_data(data=serialized_data, signal_format="OTEL_V1_METRICS")  # TODO: Convert to constant
-        return MetricExportResult.SUCCESS  # TODO: send appropriate status back. Need to??
+
+        try:
+            self._udp_exporter.send_data(data=serialized_data, signal_format_prefix=FORMAT_OTEL_METRICS_BINARY_PREFIX)
+            return MetricExportResult.SUCCESS
+        except Exception as exc:  # pylint: disable=broad-except
+            _logger.error("Error exporting metrics: %s", exc)
+            return MetricExportResult.FAILURE
 
     # pylint: disable=no-self-use
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
@@ -88,15 +97,20 @@ def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         self._udp_exporter.shutdown()
 
 
-class OtlpUdpSpanExporter(SpanExporter):
+class OTLPUdpSpanExporter(SpanExporter):
     def __init__(self, endpoint: Optional[str] = None):
         self._udp_exporter = UdpExporter(endpoint=endpoint)
 
     @override
     def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         serialized_data = encode_spans(spans).SerializeToString()
-        self._udp_exporter.send_data(data=serialized_data, signal_format="OTEL_V1_TRACES")  # TODO: Convert to constant
-        return SpanExportResult.SUCCESS  # TODO: send appropriate status back. Need to??
+
+        try:
+            self._udp_exporter.send_data(data=serialized_data, signal_format_prefix=FORMAT_OTEL_TRACES_BINARY_PREFIX)
+            return SpanExportResult.SUCCESS
+        except Exception as exc:
+            _logger.error("Error exporting spans: %s", exc)
+            return SpanExportResult.FAILURE
 
     # pylint: disable=no-self-use
     @override

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_aws_opentelementry_configurator.py

@@ -19,7 +19,7 @@
 )
 from amazon.opentelemetry.distro.aws_opentelemetry_distro import AwsOpenTelemetryDistro
 from amazon.opentelemetry.distro.aws_span_metrics_processor import AwsSpanMetricsProcessor
-from amazon.opentelemetry.distro.otlp_udp_exporter import OtlpUdpMetricExporter, OtlpUdpSpanExporter
+from amazon.opentelemetry.distro.otlp_udp_exporter import OTLPUdpMetricExporter, OTLPUdpSpanExporter
 from amazon.opentelemetry.distro.sampler._aws_xray_sampling_client import _AwsXRaySamplingClient
 from amazon.opentelemetry.distro.sampler.aws_xray_remote_sampler import AwsXRayRemoteSampler
 from opentelemetry.environment_variables import OTEL_LOGS_EXPORTER, OTEL_METRICS_EXPORTER, OTEL_TRACES_EXPORTER
@@ -262,7 +262,7 @@ def test_customize_exporter(self):
         customized_exporter = _customize_exporter(mock_exporter, Resource.get_empty())
         self.assertNotEqual(mock_exporter, customized_exporter)
         self.assertIsInstance(customized_exporter, AwsMetricAttributesSpanExporter)
-        self.assertIsInstance(customized_exporter._delegate, OtlpUdpSpanExporter)
+        self.assertIsInstance(customized_exporter._delegate, OTLPUdpSpanExporter)
         os.environ.pop("OTEL_AWS_APPLICATION_SIGNALS_ENABLED", None)
         os.environ.pop("AWS_LAMBDA_FUNCTION_NAME", None)
 
@@ -301,7 +301,7 @@ def test_application_signals_exporter_provider(self):
         # When in Lambda, exporter should be UDP.
         os.environ.setdefault("AWS_LAMBDA_FUNCTION_NAME", "myLambdaFunc")
         exporter: MetricExporter = ApplicationSignalsExporterProvider().create_exporter()
-        self.assertIsInstance(exporter, OtlpUdpMetricExporter)
+        self.assertIsInstance(exporter, OTLPUdpMetricExporter)
         self.assertEqual("127.0.0.1:2000", exporter._udp_exporter._endpoint)
         os.environ.pop("AWS_LAMBDA_FUNCTION_NAME", None)
 

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_otlp_udp_exporter.py

@@ -1,5 +1,6 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
+import base64
 import socket
 import unittest
 from unittest import TestCase
@@ -8,8 +9,8 @@
 from amazon.opentelemetry.distro.otlp_udp_exporter import (
     DEFAULT_ENDPOINT,
     PROTOCOL_HEADER,
-    OtlpUdpMetricExporter,
-    OtlpUdpSpanExporter,
+    OTLPUdpMetricExporter,
+    OTLPUdpSpanExporter,
     UdpExporter,
 )
 from opentelemetry.sdk.metrics._internal.export import MetricExportResult
@@ -46,8 +47,10 @@ def test_udp_exporter_init_invalid_endpoint(self, mock_socket):
     def test_send_data(self, mock_socket):
         mock_socket_instance = mock_socket.return_value
         exporter = UdpExporter()
-        exporter.send_data("encoded_data", "signal")
-        expected_message = PROTOCOL_HEADER + '{"format":"signal","data":encoded_data}'
+        input_bytes: bytes = b"hello"
+        encoded_bytes: bytes = base64.b64encode(input_bytes)
+        exporter.send_data(input_bytes, "signal_prefix")
+        expected_message = PROTOCOL_HEADER + "signal_prefix" + encoded_bytes.decode("utf-8")
         mock_socket_instance.sendto.assert_called_once_with(expected_message.encode("utf-8"), ("127.0.0.1", 2000))
 
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.socket.socket")
@@ -58,54 +61,45 @@ def test_shutdown(self, mock_socket):
         mock_socket_instance.close.assert_called_once()
 
 
-class TestOtlpUdpMetricExporter(unittest.TestCase):
+class TestOTLPUdpMetricExporter(unittest.TestCase):
 
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.encode_metrics")
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.UdpExporter")
     def test_export(self, mock_udp_exporter, mock_encode_metrics):
         mock_udp_exporter_instance = mock_udp_exporter.return_value
         mock_encoded_data = MagicMock()
         mock_encode_metrics.return_value.SerializeToString.return_value = mock_encoded_data
-        exporter = OtlpUdpMetricExporter()
+        exporter = OTLPUdpMetricExporter()
         result = exporter.export(MagicMock())
-        mock_udp_exporter_instance.send_data.assert_called_once_with(
-            data=mock_encoded_data, signal_format="OTEL_V1_METRICS"
-        )
+        mock_udp_exporter_instance.send_data.assert_called_once_with(data=mock_encoded_data, signal_format_prefix="M1")
         self.assertEqual(result, MetricExportResult.SUCCESS)
 
     # pylint: disable=no-self-use
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.UdpExporter")
     def test_shutdown(self, mock_udp_exporter):
         mock_udp_exporter_instance = mock_udp_exporter.return_value
-        exporter = OtlpUdpMetricExporter()
+        exporter = OTLPUdpMetricExporter()
         exporter.shutdown()
         mock_udp_exporter_instance.shutdown.assert_called_once()
 
 
-class TestOtlpUdpSpanExporter(unittest.TestCase):
+class TestOTLPUdpSpanExporter(unittest.TestCase):
 
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.encode_spans")
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.UdpExporter")
     def test_export(self, mock_udp_exporter, mock_encode_spans):
         mock_udp_exporter_instance = mock_udp_exporter.return_value
         mock_encoded_data = MagicMock()
         mock_encode_spans.return_value.SerializeToString.return_value = mock_encoded_data
-        exporter = OtlpUdpSpanExporter()
+        exporter = OTLPUdpSpanExporter()
         result = exporter.export(MagicMock())
-        mock_udp_exporter_instance.send_data.assert_called_once_with(
-            data=mock_encoded_data, signal_format="OTEL_V1_TRACES"
-        )
+        mock_udp_exporter_instance.send_data.assert_called_once_with(data=mock_encoded_data, signal_format_prefix="T1")
         self.assertEqual(result, SpanExportResult.SUCCESS)
 
     # pylint: disable=no-self-use
     @patch("amazon.opentelemetry.distro.otlp_udp_exporter.UdpExporter")
     def test_shutdown(self, mock_udp_exporter):
         mock_udp_exporter_instance = mock_udp_exporter.return_value
-        exporter = OtlpUdpSpanExporter()
+        exporter = OTLPUdpSpanExporter()
         exporter.shutdown()
         mock_udp_exporter_instance.shutdown.assert_called_once()
-
-
-# TODO: remove this line for final PR
-if __name__ == "__main__":
-    unittest.main()