add cycle detection

Author: liustve

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/exporter/otlp/aws/logs/aws_batch_log_record_processor.py

@@ -103,7 +103,8 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:
         Estimates the size in bytes of a log by calculating the size of its body and its attributes
         and adding a buffer amount to account for other log metadata information.
         Will process complex log structures up to the specified depth limit.
-        If the depth limit of the log structure is exceeded, returns the truncated calculation
+        Includes cycle detection to prevent processing the same complex log content (Maps, Arrays)
+        more than once. If the depth limit of the log structure is exceeded, returns the truncated calculation
         to everything up to that point.
 
         Args:
@@ -114,9 +115,13 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:
             int: The estimated size of the log object in bytes
         """
 
-        # Use a queue to prevent excessive recursive calls.
-        # We calculate based on the size of the log record body and attributes for the log.
-        queue: List[tuple[AnyValue, int]] = [(log.log_record.body, 0), (log.log_record.attributes, -1)]
+        # Queue is a list of (log_content, depth) where:
+        # log_content is the current piece of log data being processed
+        # depth tracks how many levels deep we've traversed to reach this data
+        queue = [(log.log_record.body, 0), (log.log_record.attributes, -1)]
+
+        # Track visited objects to avoid calculating the same complex log content more than once
+        visited = set()
 
         size: int = self._BASE_LOG_BUFFER_BYTE_SIZE
 
@@ -130,6 +135,9 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:
 
                 next_val, current_depth = data
 
+                if not next_val:
+                    continue
+
                 if isinstance(next_val, (str, bytes)):
                     size += len(next_val)
                     continue
@@ -142,7 +150,14 @@ def _estimate_log_size(self, log: LogData, depth: int = 3) -> int:
                     size += len(str(next_val))
                     continue
 
+                # next_val must be Sequence["AnyValue"] or Mapping[str, "AnyValue"],
                 if current_depth <= depth:
+                    # Guaranteed to be unique, see: https://www.w3schools.com/python/ref_func_id.asp
+                    obj_id = id(next_val)
+                    if obj_id in visited:
+                        continue
+                    visited.add(obj_id)
+
                     if isinstance(next_val, Sequence):
                         for content in next_val:
                             new_queue.append((cast(AnyValue, content), current_depth + 1))

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/exporter/otlp/aws/logs/test_aws_batch_log_record_processor.py

@@ -114,6 +114,23 @@ def test_process_log_data_primitive(self):
 
             self.assertEqual(actual_size, expected_size)
 
+    def test_process_log_data_with_cycle(self):
+        """Test that processor handles processing logs with circular references only once"""
+        cyclic_dict: dict = {"data": "test"}
+        cyclic_dict["self_ref"] = cyclic_dict
+
+        log = self.generate_test_log_data(
+            log_body=cyclic_dict,
+            attr_key="",
+            attr_val="",
+            log_body_depth=-1,
+            attr_depth=-1,
+            count=1,
+        )
+        expected_size = self.base_log_size + len("data") + len("self_ref") + len("test")
+        actual_size = self.processor._estimate_log_size(log[0])
+        self.assertEqual(actual_size, expected_size)
+
     @patch(
         "amazon.opentelemetry.distro.exporter.otlp.aws.logs.aws_batch_log_record_processor.attach",
         return_value=MagicMock(),
@@ -236,7 +253,6 @@ def generate_test_log_data(
         attr_depth=3,
         count=5,
         create_map=True,
-        instrumentation_scope=InstrumentationScope("test-scope", "1.0.0"),
     ) -> List[LogData]:
 
         def generate_nested_value(depth, value, create_map=True) -> AnyValue: