Rename and refactor the code to remove redundency and additional inline comments

Author: mxiamxia

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/metrics/otlp_aws_emf_exporter.py renamed to aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/metrics/aws_cloudwatch_emf_exporter.py

@@ -35,34 +35,28 @@
 logger = logging.getLogger(__name__)
 
 
-class CloudWatchEMFExporter(MetricExporter):
+class AwsCloudWatchEMFExporter(MetricExporter):
     """
     OpenTelemetry metrics exporter for CloudWatch EMF format.
 
     This exporter converts OTel metrics into CloudWatch EMF logs which are then
     sent to CloudWatch Logs. CloudWatch Logs automatically extracts the metrics
     from the EMF logs.
+
+    https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html
+
     """
 
     # OTel to CloudWatch unit mapping
+    # Ref: opentelemetry-collector-contrib/blob/main/exporter/awsemfexporter/grouped_metric.go#L188
     UNIT_MAPPING = {
+        "1": "",
+        "ns": "",
         "ms": "Milliseconds",
         "s": "Seconds",
         "us": "Microseconds",
-        "ns": "Nanoseconds",
         "By": "Bytes",
-        "KiBy": "Kilobytes",
-        "MiBy": "Megabytes",
-        "GiBy": "Gigabytes",
-        "TiBy": "Terabytes",
-        "Bi": "Bits",
-        "KiBi": "Kilobits",
-        "MiBi": "Megabits",
-        "GiBi": "Gigabits",
-        "TiBi": "Terabits",
-        "%": "Percent",
-        "1": "Count",
-        "{count}": "Count",
+        "bit": "Bits",
     }
 
     def __init__(
@@ -102,6 +96,9 @@ def __init__(
         # Ensure log group exists
         self._ensure_log_group_exists()
 
+        # Ensure log stream exists
+        self._ensure_log_stream_exists()
+
     def _generate_log_stream_name(self) -> str:
         """Generate a unique log stream name."""
 
@@ -120,6 +117,17 @@ def _ensure_log_group_exists(self):
                 logger.error("Failed to create log group %s : %s", self.log_group_name, error)
                 raise
 
+    def _ensure_log_stream_exists(self):
+        try:
+            self.logs_client.create_log_stream(logGroupName=self.log_group_name, logStreamName=self.log_stream_name)
+            logger.info("Created log stream: %s", self.log_stream_name)
+        except ClientError as error:
+            if error.response.get("Error", {}).get("Code") == "ResourceAlreadyExistsException":
+                logger.debug("Log stream %s already exists", self.log_stream_name)
+            else:
+                logger.error("Failed to create log stream %s : %s", self.log_group_name, error)
+                raise
+
     def _get_metric_name(self, record: Any) -> Optional[str]:
         """Get the metric name from the metric record or data point."""
         # For compatibility with older record format
@@ -250,20 +258,28 @@ def _create_emf_log(self, metric_records: List[Any], resource: Resource, timesta
         emf_log["Version"] = "1"
 
         # Add resource attributes to EMF log but not as dimensions
+        # OTel collector EMF Exporter has a resource_to_telemetry_conversion flag that will convert resource attributes
+        # as regular metric attributes(potential dimensions). However, for this SDK EMF implementation,
+        # we align with the OpenTelemetry concept that all metric attributes are treated as dimensions.
+        # And have resource attributes as just additional metadata in EMF, added otel.resource as prefix to distinguish.
         if resource and resource.attributes:
             for key, value in resource.attributes.items():
-                emf_log[f"resource.{key}"] = str(value)
+                emf_log[f"otel.resource.{key}"] = str(value)
 
         # Initialize collections for dimensions and metrics
-        all_attributes = {}
+
         metric_definitions = []
 
+        # Collect attributes from all records (they should be the same for all records in the group)
+        # Only collect once from the first record and apply to all records
+        all_attributes = (
+            metric_records[0].attributes
+            if metric_records and hasattr(metric_records[0], "attributes") and metric_records[0].attributes
+            else {}
+        )
+
         # Process each metric record
         for record in metric_records:
-            # Collect attributes from all records (they should be the same for all records in the group)
-            if hasattr(record, "attributes") and record.attributes:
-                for key, value in record.attributes.items():
-                    all_attributes[key] = value
 
             metric_name = self._get_metric_name(record)
 
@@ -279,9 +295,11 @@ def _create_emf_log(self, metric_records: List[Any], resource: Resource, timesta
                 metric_data["Unit"] = unit
 
             # Process gauge metrics (only type supported in PR 1)
-            if hasattr(record, "value"):
+            if not hasattr(record, "value"):
                 # Store value directly in emf_log
-                emf_log[metric_name] = record.value
+                logger.debug("Skipping metric %s as it does not have valid metric value", metric_name)
+
+            emf_log[metric_name] = record.value
 
             # Add to metric definitions list
             metric_definitions.append({"Name": metric_name, **metric_data})
@@ -307,6 +325,8 @@ def _send_log_event(self, log_event: Dict[str, Any]):
         Send a log event to CloudWatch Logs.
 
         Basic implementation for PR 1 - sends individual events directly.
+
+        TODO: Batching event and follow CloudWatch Logs quato constraints - number of events & size limit per payload
         """
         try:
             # Send the log event
@@ -354,15 +374,16 @@ def export(
                         if not (hasattr(metric, "data") and hasattr(metric.data, "data_points")):
                             continue
 
-                        # Process only Gauge metrics in PR 1
-                        if isinstance(metric.data, Gauge):
+                        # Process metrics based on type
+                        metric_type = type(metric.data)
+                        if metric_type == Gauge:
                             for dp in metric.data.data_points:
                                 record, timestamp_ms = self._convert_gauge(metric, dp)
                                 grouped_metrics[self._group_by_attributes_and_timestamp(record, timestamp_ms)].append(
                                     record
                                 )
                         else:
-                            logger.warning("Unsupported Metric Type: %s", type(metric.data))
+                            logger.debug("Unsupported Metric Type: %s", metric_type)
 
                     # Now process each group separately to create one EMF log per group
                     for (_, timestamp_ms), metric_records in grouped_metrics.items():
@@ -390,27 +411,31 @@ def force_flush(self, timeout_millis: int = 10000) -> bool:
         """
         Force flush any pending metrics.
 
+        TODO: will add logic to handle gracefule shutdown
+
         Args:
             timeout_millis: Timeout in milliseconds
 
         Returns:
             True if successful, False otherwise
         """
-        logger.debug("CloudWatchEMFExporter force flushes the buffered metrics")
+        logger.debug("AWsCloudWatchEMFExporter force flushes the buffered metrics")
         return True
 
     def shutdown(self, timeout_millis: Optional[int] = None, **kwargs: Any) -> bool:
         """
         Shutdown the exporter.
         Override to handle timeout and other keyword arguments, but do nothing.
 
+        TODO: will add logic to handle gracefule shutdown
+
         Args:
             timeout_millis: Ignored timeout in milliseconds
             **kwargs: Ignored additional keyword arguments
         """
         # Intentionally do nothing
         self.force_flush(timeout_millis)
-        logger.debug("CloudWatchEMFExporter shutdown called with timeout_millis=%s", timeout_millis)
+        logger.debug("AwsCloudWatchEMFExporter shutdown called with timeout_millis=%s", timeout_millis)
         return True
 
 
@@ -420,7 +445,7 @@ def create_emf_exporter(
     log_stream_name: Optional[str] = None,
     aws_region: Optional[str] = None,
     **kwargs,
-) -> CloudWatchEMFExporter:
+) -> AwsCloudWatchEMFExporter:
     """
     Convenience function to create a CloudWatch EMF exporter with DELTA temporality.
 
@@ -430,10 +455,10 @@ def create_emf_exporter(
         log_stream_name: CloudWatch log stream name (auto-generated if None)
         aws_region: AWS region (auto-detected if None)
         debug: Whether to enable debug printing of EMF logs
-        **kwargs: Additional arguments passed to the CloudWatchEMFExporter
+        **kwargs: Additional arguments passed to the AwsCloudWatchEMFExporter
 
     Returns:
-        Configured CloudWatchEMFExporter instance
+        Configured AwsCloudWatchEMFExporter instance
     """
 
     # Set up temporality preference - always use DELTA for CloudWatch
@@ -447,7 +472,7 @@ def create_emf_exporter(
     }
 
     # Create and return the exporter
-    return CloudWatchEMFExporter(
+    return AwsCloudWatchEMFExporter(
         namespace=namespace,
         log_group_name=log_group_name,
         log_stream_name=log_stream_name,

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/metrics/test_otlp_aws_emf_exporter.py renamed to aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/metrics/test_aws_cloudwatch_emf_exporter.py

@@ -8,8 +8,8 @@
 
 from botocore.exceptions import ClientError
 
-from amazon.opentelemetry.distro.exporter.otlp.aws.metrics.otlp_aws_emf_exporter import (
-    CloudWatchEMFExporter,
+from amazon.opentelemetry.distro.exporter.otlp.aws.metrics.aws_cloudwatch_emf_exporter import (
+    AwsCloudWatchEMFExporter,
     create_emf_exporter,
 )
 from opentelemetry.sdk.metrics.export import Gauge, MetricExportResult
@@ -83,7 +83,7 @@ def test_create_emf_exporter_default_args(self, mock_session):
 
         exporter = create_emf_exporter()
 
-        self.assertIsInstance(exporter, CloudWatchEMFExporter)
+        self.assertIsInstance(exporter, AwsCloudWatchEMFExporter)
         self.assertEqual(exporter.namespace, "OTelPython")
 
     @patch("botocore.session.Session")
@@ -101,14 +101,14 @@ def test_create_emf_exporter_custom_args(self, mock_session):
             namespace="CustomNamespace", log_group_name="/custom/log/group", aws_region="us-west-2"
         )
 
-        self.assertIsInstance(exporter, CloudWatchEMFExporter)
+        self.assertIsInstance(exporter, AwsCloudWatchEMFExporter)
         self.assertEqual(exporter.namespace, "CustomNamespace")
         self.assertEqual(exporter.log_group_name, "/custom/log/group")
 
 
 # pylint: disable=too-many-public-methods
-class TestCloudWatchEMFExporter(unittest.TestCase):
-    """Test CloudWatchEMFExporter class."""
+class TestAwsCloudWatchEMFExporter(unittest.TestCase):
+    """Test AwsCloudWatchEMFExporter class."""
 
     def setUp(self):
         """Set up test fixtures."""
@@ -121,7 +121,7 @@ def setUp(self):
             mock_client.describe_log_groups.return_value = {"logGroups": []}
             mock_client.create_log_group.return_value = {}
 
-            self.exporter = CloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
+            self.exporter = AwsCloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
 
     def test_initialization(self):
         """Test exporter initialization."""
@@ -140,7 +140,7 @@ def test_initialization_with_custom_params(self, mock_session):
         mock_client.describe_log_groups.return_value = {"logGroups": []}
         mock_client.create_log_group.return_value = {}
 
-        exporter = CloudWatchEMFExporter(
+        exporter = AwsCloudWatchEMFExporter(
             namespace="CustomNamespace",
             log_group_name="custom-log-group",
             log_stream_name="custom-stream",
@@ -152,13 +152,18 @@ def test_initialization_with_custom_params(self, mock_session):
 
     def test_get_unit_mapping(self):
         """Test unit mapping functionality."""
-        # Test known units
+        # Test known units from UNIT_MAPPING
         self.assertEqual(self.exporter._get_unit(Mock(unit="ms")), "Milliseconds")
         self.assertEqual(self.exporter._get_unit(Mock(unit="s")), "Seconds")
+        self.assertEqual(self.exporter._get_unit(Mock(unit="us")), "Microseconds")
         self.assertEqual(self.exporter._get_unit(Mock(unit="By")), "Bytes")
-        self.assertEqual(self.exporter._get_unit(Mock(unit="%")), "Percent")
+        self.assertEqual(self.exporter._get_unit(Mock(unit="bit")), "Bits")
 
-        # Test unknown unit
+        # Test units that map to empty string
+        self.assertEqual(self.exporter._get_unit(Mock(unit="1")), "")
+        self.assertEqual(self.exporter._get_unit(Mock(unit="ns")), "")
+
+        # Test unknown unit (returns as-is)
         self.assertEqual(self.exporter._get_unit(Mock(unit="unknown")), "unknown")
 
         # Test empty unit (should return None due to falsy check)
@@ -326,7 +331,7 @@ def test_force_flush_no_pending_events(self):
 
         self.assertTrue(result)
 
-    @patch.object(CloudWatchEMFExporter, "force_flush")
+    @patch.object(AwsCloudWatchEMFExporter, "force_flush")
     def test_shutdown(self, mock_force_flush):
         """Test shutdown functionality."""
         mock_force_flush.return_value = True
@@ -375,8 +380,8 @@ def test_create_emf_log_with_resource(self):
         self.assertEqual(result["Version"], "1")
 
         # Check resource attributes are prefixed
-        self.assertEqual(result["resource.service.name"], "test-service")
-        self.assertEqual(result["resource.service.version"], "1.0.0")
+        self.assertEqual(result["otel.resource.service.name"], "test-service")
+        self.assertEqual(result["otel.resource.service.version"], "1.0.0")
 
         # Check metric attributes
         self.assertEqual(result["env"], "test")
@@ -489,7 +494,7 @@ def test_initialization_with_env_region(self, mock_session, mock_env_get):
         mock_client.describe_log_groups.return_value = {"logGroups": []}
         mock_client.create_log_group.return_value = {}
 
-        exporter = CloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
+        exporter = AwsCloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
 
         # Just verify the exporter was created successfully with region handling
         self.assertIsNotNone(exporter)
@@ -511,7 +516,7 @@ def test_ensure_log_group_exists_create_failure(self, mock_session):
         mock_client.create_log_group.side_effect = ClientError({"Error": {"Code": "AccessDenied"}}, "CreateLogGroup")
 
         with self.assertRaises(ClientError):
-            CloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
+            AwsCloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
 
     @patch("botocore.session.Session")
     def test_ensure_log_group_exists_success(self, mock_session):
@@ -526,7 +531,7 @@ def test_ensure_log_group_exists_success(self, mock_session):
         mock_client.describe_log_groups.return_value = {"logGroups": [{"logGroupName": "test-log-group"}]}
 
         # This should not raise an exception
-        exporter = CloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
+        exporter = AwsCloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
         self.assertIsNotNone(exporter)
         # Verify describe was called but create was not
         mock_client.describe_log_groups.assert_called_once_with(logGroupNamePrefix="test-log-group", limit=1)