refactor(logging): Optimize truncation limits for better 256KB compliance

Co-authored-by: ntindle <8845353+ntindle@users.noreply.github.com>

autogpt_platform/backend/backend/util/logging.py

@@ -35,27 +35,43 @@ class TruncatedLogger:
 
     def info(self, msg: str, **extra):
         msg = self._wrap(msg, **extra)
-        self.logger.info(msg, extra=self._get_metadata(**extra))
+        truncated_extra = self._truncate_large_data(extra)
+        truncated_metadata = self._truncate_large_data(self.metadata)
+        self.logger.info(
+            msg, extra={"json_fields": {**truncated_metadata, **truncated_extra}}
+        )
 
     def warning(self, msg: str, **extra):
         msg = self._wrap(msg, **extra)
-        self.logger.warning(msg, extra=self._get_metadata(**extra))
+        truncated_extra = self._truncate_large_data(extra)
+        truncated_metadata = self._truncate_large_data(self.metadata)
+        self.logger.warning(
+            msg, extra={"json_fields": {**truncated_metadata, **truncated_extra}}
+        )
 
     def error(self, msg: str, **extra):
         msg = self._wrap(msg, **extra)
-        self.logger.error(msg, extra=self._get_metadata(**extra))
+        truncated_extra = self._truncate_large_data(extra)
+        truncated_metadata = self._truncate_large_data(self.metadata)
+        self.logger.error(
+            msg, extra={"json_fields": {**truncated_metadata, **truncated_extra}}
+        )
 
     def debug(self, msg: str, **extra):
         msg = self._wrap(msg, **extra)
-        self.logger.debug(msg, extra=self._get_metadata(**extra))
+        truncated_extra = self._truncate_large_data(extra)
+        truncated_metadata = self._truncate_large_data(self.metadata)
+        self.logger.debug(
+            msg, extra={"json_fields": {**truncated_metadata, **truncated_extra}}
+        )
 
     def exception(self, msg: str, **extra):
         msg = self._wrap(msg, **extra)
-        self.logger.exception(msg, extra=self._get_metadata(**extra))
-
-    def _get_metadata(self, **extra):
-        metadata = {**self.metadata, **extra}
-        return {"json_fields": metadata} if metadata else {}
+        truncated_extra = self._truncate_large_data(extra)
+        truncated_metadata = self._truncate_large_data(self.metadata)
+        self.logger.exception(
+            msg, extra={"json_fields": {**truncated_metadata, **truncated_extra}}
+        )
 
     def _wrap(self, msg: str, **extra):
         extra_msg = str(extra or "")
@@ -64,6 +80,26 @@ class TruncatedLogger:
             text = text[: self.max_length] + "..."
         return text
 
+    def _truncate_large_data(self, data, max_size=4000):
+        if isinstance(data, dict):
+            # For large dictionaries, limit the number of keys and truncate values
+            if len(data) > 20:
+                # Keep only first 20 keys for very large dictionaries
+                truncated_dict = {}
+                for i, (k, v) in enumerate(data.items()):
+                    if i >= 20:
+                        truncated_dict["... [additional_keys_truncated]"] = f"{len(data) - 20} more keys"
+                        break
+                    truncated_dict[k] = self._truncate_large_data(v, max_size)
+                return truncated_dict
+            else:
+                return {k: self._truncate_large_data(v, max_size) for k, v in data.items()}
+        elif isinstance(data, list):
+            return [self._truncate_large_data(v, max_size) for v in data[:50]]
+        elif isinstance(data, str) and len(data) > max_size:
+            return data[:max_size] + "... [truncated]"
+        return data
+
 
 class PrefixFilter(logging.Filter):
     def __init__(self, prefix: str):