Enhancements by removing redundent code

Author: mxiamxia

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

@@ -120,18 +120,15 @@ def _ensure_log_group_exists(self):
                 logger.error("Failed to create log group %s : %s", self.log_group_name, error)
                 raise
 
-    def _get_metric_name(self, record) -> Optional[str]:
+    def _get_metric_name(self, record: Any) -> Optional[str]:
         """Get the metric name from the metric record or data point."""
-        # For metrics in MetricsData format
-        if hasattr(record, "name") and record.name.strip():
-            return record.name
         # For compatibility with older record format
-        if hasattr(record, "instrument") and hasattr(record.instrument, "name") and record.instrument.name.strip():
+        if hasattr(record, "instrument") and hasattr(record.instrument, "name") and record.instrument.name:
             return record.instrument.name
         # Return None if no valid metric name found
         return None
 
-    def _get_unit(self, instrument_or_metric) -> Optional[str]:
+    def _get_unit(self, instrument_or_metric: Any) -> Optional[str]:
         """Get CloudWatch unit from OTel instrument or metric unit."""
         # Check if we have an Instrument object or a metric with unit attribute
         if isinstance(instrument_or_metric, Instrument):
@@ -198,7 +195,7 @@ def _create_metric_record(self, metric_name: str, metric_unit: str, metric_descr
 
         return record
 
-    def _convert_gauge(self, metric, dp) -> Tuple[Any, int]:
+    def _convert_gauge(self, metric: Any, dp: Any) -> Tuple[Any, int]:
         """Convert a Gauge metric datapoint to a metric record.
 
         Args:
@@ -225,7 +222,7 @@ def _convert_gauge(self, metric, dp) -> Tuple[Any, int]:
 
         return record, timestamp_ms
 
-    def _group_by_attributes_and_timestamp(self, record, timestamp_ms) -> Tuple[str, int]:
+    def _group_by_attributes_and_timestamp(self, record: Any, timestamp_ms: int) -> Tuple[str, int]:
         """Group metric record by attributes and timestamp.
 
         Args:
@@ -239,7 +236,7 @@ def _group_by_attributes_and_timestamp(self, record, timestamp_ms) -> Tuple[str,
         attrs_key = self._get_attributes_key(record.attributes)
         return (attrs_key, timestamp_ms)
 
-    def _create_emf_log(self, metric_records, resource: Resource, timestamp: Optional[int] = None) -> Dict:
+    def _create_emf_log(self, metric_records: List[Any], resource: Resource, timestamp: Optional[int] = None) -> Dict:
         """
         Create EMF log dictionary from metric records.
 
@@ -305,31 +302,13 @@ def _create_emf_log(self, metric_records, resource: Resource, timestamp: Optiona
         return emf_log
 
     # pylint: disable=no-member
-    def _send_log_event(self, log_event: Dict):
+    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.
         """
         try:
-            # Create log group and stream if they don't exist
-            try:
-                self.logs_client.create_log_group(logGroupName=self.log_group_name)
-                logger.debug("Created log group: %s", self.log_group_name)
-            except ClientError as error:
-                # Check if it's a ResourceAlreadyExistsException (botocore exception handling)
-                if error.response.get("Error", {}).get("Code") == "ResourceAlreadyExistsException":
-                    logger.debug("Log group %s already exists", self.log_group_name)
-
-            # Create log stream if it doesn't exist
-            try:
-                self.logs_client.create_log_stream(logGroupName=self.log_group_name, logStreamName=self.log_stream_name)
-                logger.debug("Created log stream: %s", self.log_stream_name)
-            except ClientError as error:
-                # Check if it's a ResourceAlreadyExistsException (botocore exception handling)
-                if error.response.get("Error", {}).get("Code") == "ResourceAlreadyExistsException":
-                    logger.debug("Log stream %s already exists", self.log_stream_name)
-
             # Send the log event
             response = self.logs_client.put_log_events(
                 logGroupName=self.log_group_name, logStreamName=self.log_stream_name, logEvents=[log_event]
@@ -343,7 +322,9 @@ def _send_log_event(self, log_event: Dict):
             raise
 
     # pylint: disable=too-many-nested-blocks
-    def export(self, metrics_data: MetricsData, timeout_millis: Optional[int] = None, **kwargs) -> MetricExportResult:
+    def export(
+        self, metrics_data: MetricsData, timeout_millis: Optional[int] = None, **kwargs: Any
+    ) -> MetricExportResult:
         """
         Export metrics as EMF logs to CloudWatch.
 
@@ -418,7 +399,7 @@ def force_flush(self, timeout_millis: int = 10000) -> bool:
         logger.debug("CloudWatchEMFExporter force flushes the buffered metrics")
         return True
 
-    def shutdown(self, timeout_millis=None, **kwargs):
+    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.
@@ -438,7 +419,6 @@ def create_emf_exporter(
     log_group_name: str = "/aws/otel/python",
     log_stream_name: Optional[str] = None,
     aws_region: Optional[str] = None,
-    debug: bool = False,
     **kwargs,
 ) -> CloudWatchEMFExporter:
     """
@@ -466,11 +446,6 @@ def create_emf_exporter(
         UpDownCounter: AggregationTemporality.DELTA,
     }
 
-    # Configure logging if debug is enabled
-    if debug:
-        logging.basicConfig(level=logging.DEBUG)
-        logger.setLevel(logging.DEBUG)
-
     # Create and return the exporter
     return CloudWatchEMFExporter(
         namespace=namespace,

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

@@ -105,23 +105,6 @@ def test_create_emf_exporter_custom_args(self, mock_session):
         self.assertEqual(exporter.namespace, "CustomNamespace")
         self.assertEqual(exporter.log_group_name, "/custom/log/group")
 
-    @patch("botocore.session.Session")
-    @patch("logging.basicConfig")
-    def test_create_emf_exporter_debug_mode(self, mock_logging_config, mock_session):
-        """Test creating exporter with debug mode enabled."""
-        # Mock the botocore session to avoid AWS calls
-        mock_client = Mock()
-        mock_session_instance = Mock()
-        mock_session.return_value = mock_session_instance
-        mock_session_instance.create_client.return_value = mock_client
-        mock_client.describe_log_groups.return_value = {"logGroups": []}
-        mock_client.create_log_group.return_value = {}
-
-        exporter = create_emf_exporter(debug=True)
-
-        self.assertIsInstance(exporter, CloudWatchEMFExporter)
-        mock_logging_config.assert_called_once()
-
 
 # pylint: disable=too-many-public-methods
 class TestCloudWatchEMFExporter(unittest.TestCase):
@@ -190,17 +173,17 @@ def test_get_metric_name(self):
         record = Mock()
         record.instrument = Mock()
         record.instrument.name = "test_metric"
-        del record.name  # Ensure record.name doesn't exist
 
         result = self.exporter._get_metric_name(record)
         self.assertEqual(result, "test_metric")
 
-        # Test with record that has direct name attribute
-        record_with_name = Mock()
-        record_with_name.name = "direct_metric"
+        # Test with record that has empty instrument name (should return None)
+        record_empty = Mock()
+        record_empty.instrument = Mock()
+        record_empty.instrument.name = ""
 
-        result2 = self.exporter._get_metric_name(record_with_name)
-        self.assertEqual(result2, "direct_metric")
+        result_empty = self.exporter._get_metric_name(record_empty)
+        self.assertIsNone(result_empty)
 
     def test_get_dimension_names(self):
         """Test dimension names extraction."""
@@ -445,30 +428,18 @@ def test_export_with_gauge_metrics(self, mock_session):
 
     def test_get_metric_name_fallback(self):
         """Test metric name extraction fallback."""
-        # Test with record that has no name or instrument
+        # Test with record that has no instrument attribute
         record = Mock(spec=[])
 
         result = self.exporter._get_metric_name(record)
-        # Note: This test may fail against old installed version which returns "unknown_metric"
-        # The new implementation correctly returns None
-        self.assertTrue(result is None or result == "unknown_metric")
+        self.assertIsNone(result)
 
     def test_get_metric_name_empty_name(self):
-        """Test metric name extraction with empty name."""
-        # Test with record that has empty name
-        record = Mock()
-        record.name = ""
-
-        self.exporter._get_metric_name(record)
-        # Just verify the method can handle empty names without crashing
-        # The method should return some value and not crash
-        self.assertIsNotNone(self.exporter._get_metric_name)  # Method exists
-
+        """Test metric name extraction with empty instrument name."""
         # Test with record that has empty instrument name
         record = Mock()
         record.instrument = Mock()
         record.instrument.name = ""
-        del record.name  # Ensure record.name doesn't exist
 
         result = self.exporter._get_metric_name(record)
         self.assertIsNone(result)
@@ -496,10 +467,9 @@ def test_create_emf_log_skips_empty_metric_names(self):
         self.assertIn("valid_metric", result)
         self.assertEqual(result["valid_metric"], 20.0)
 
-        # Check that the valid metric is in the definitions
+        # Check that only the valid metric is in the definitions (empty names are skipped)
         cw_metrics = result["_aws"]["CloudWatchMetrics"][0]
-        # Note: Old version may include both metrics (1 or 2), new version skips empty names (only 1)
-        self.assertTrue(len(cw_metrics["Metrics"]) >= 1)
+        self.assertEqual(len(cw_metrics["Metrics"]), 1)
         # Ensure our valid metric is present
         metric_names = [m["Name"] for m in cw_metrics["Metrics"]]
         self.assertIn("valid_metric", metric_names)
@@ -543,6 +513,25 @@ def test_ensure_log_group_exists_create_failure(self, mock_session):
         with self.assertRaises(ClientError):
             CloudWatchEMFExporter(namespace="TestNamespace", log_group_name="test-log-group")
 
+    @patch("botocore.session.Session")
+    def test_ensure_log_group_exists_success(self, mock_session):
+        """Test log group existence check when log group already exists."""
+        # Mock the botocore session
+        mock_client = Mock()
+        mock_session_instance = Mock()
+        mock_session.return_value = mock_session_instance
+        mock_session_instance.create_client.return_value = mock_client
+
+        # Make describe succeed (log group exists)
+        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")
+        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)
+        mock_client.create_log_group.assert_not_called()
+
     def test_export_with_unsupported_metric_type(self):
         """Test export with unsupported metric types."""
         # Create mock metrics data with unsupported metric type