refactor(memory-monitor): humanize error/log byte values and log configured thresholds on init

- Add _format_bytes helper that renders byte counts as GB/MB/KB with 2 decimal places (decimal units).
- Replace raw byte values in the fail-fast internal_message and the two critical-but-not-raising log paths with _format_bytes(...).
- Emit a one-shot INFO log on MemoryMonitor instantiation reporting the configured thresholds and check intervals.
- Clarify docstrings that the tightened high-pressure check interval overrides the configurable check_interval (default 5000), addressing Copilot review feedback.
- Tests: parametrized _format_bytes coverage, instantiation-log assertion, and assertions that the internal_message uses the humanized format.

Co-Authored-By: patrick.nilan@airbyte.io <patrick.nilan@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/utils/memory_monitor.py

@@ -25,6 +25,24 @@
 # Process-level anonymous RSS from /proc/self/status (Linux only, no extra dependency)
 _PROC_SELF_STATUS = Path("/proc/self/status")
 
+
+def _format_bytes(num_bytes: int) -> str:
+    """Render a byte count as a short human-readable string with 2 decimals.
+
+    Uses decimal units (GB = 10^9, MB = 10^6, KB = 10^3) so that raw cgroup
+    byte values render close to the way operators describe container limits
+    (e.g. a 2_147_483_648-byte limit renders as ``2.15 GB`` rather than the
+    binary ``2.00 GiB``).  Values below 1 KB are rendered as plain bytes.
+    """
+    if num_bytes >= 1_000_000_000:
+        return f"{num_bytes / 1_000_000_000:.2f} GB"
+    if num_bytes >= 1_000_000:
+        return f"{num_bytes / 1_000_000:.2f} MB"
+    if num_bytes >= 1_000:
+        return f"{num_bytes / 1_000:.2f} KB"
+    return f"{num_bytes} B"
+
+
 # Raise AirbyteTracedException when BOTH conditions are met:
 #   1. cgroup usage >= critical threshold
 #   2. anonymous memory >= anon-share threshold of *current cgroup usage*
@@ -104,8 +122,8 @@ class MemoryMonitor:
       transitions, not periodic sampling
 
     **High-pressure polling:** Once cgroup usage first crosses 90%, the check
-    interval permanently tightens from 5000 to 100 messages to narrow the race
-    window near OOM.
+    interval permanently tightens from the configured ``check_interval``
+    (default 5000) to 100 messages to narrow the race window near OOM.
 
     **Fail-fast:** Raises ``AirbyteTracedException`` with
     ``FailureType.system_error`` when *both*:
@@ -138,6 +156,16 @@ def __init__(
         self._probed = False
         self._high_pressure_mode = False
         self._critical_logged = False
+        logger.info(
+            "MemoryMonitor instantiated with critical threshold: %d%%, "
+            "anon share of usage threshold: %d%%, high-pressure threshold: %d%%, "
+            "check interval: %d messages (tightens to %d under high pressure).",
+            int(_CRITICAL_THRESHOLD * 100),
+            int(_ANON_SHARE_OF_USAGE_THRESHOLD * 100),
+            int(_HIGH_PRESSURE_THRESHOLD * 100),
+            self._check_interval,
+            _HIGH_PRESSURE_CHECK_INTERVAL,
+        )
 
     def _probe_cgroup(self) -> None:
         """Detect which cgroup version (if any) is available.
@@ -216,7 +244,8 @@ def check_memory_usage(self) -> None:
         Intended to be called on every message. The monitor internally tracks
         a message counter and only reads cgroup files every ``check_interval``
         messages (default 5000). Once usage crosses 90%, the interval tightens
-        to 100 messages for the remainder of the sync.
+        to 100 messages for the remainder of the sync regardless of the
+        configured ``check_interval``.
 
         Logging is event-based (one-shot on state transitions), not periodic.
 
@@ -240,8 +269,6 @@ def check_memory_usage(self) -> None:
         usage_bytes, limit_bytes = memory_info
         usage_ratio = usage_bytes / limit_bytes
         usage_percent = int(usage_ratio * 100)
-        usage_gb = usage_bytes / (1024**3)
-        limit_gb = limit_bytes / (1024**3)
 
         if usage_ratio >= _HIGH_PRESSURE_THRESHOLD and not self._high_pressure_mode:
             self._high_pressure_mode = True
@@ -262,8 +289,9 @@ def check_memory_usage(self) -> None:
                     raise AirbyteTracedException(
                         message=f"Source memory usage exceeded critical threshold ({usage_percent}% of container limit).",
                         internal_message=(
-                            f"Cgroup memory: {usage_bytes} / {limit_bytes} bytes ({usage_percent}%). "
-                            f"Anonymous memory ({anon_source}): {anon_bytes} bytes "
+                            f"Cgroup memory: {_format_bytes(usage_bytes)} / "
+                            f"{_format_bytes(limit_bytes)} ({usage_percent}%). "
+                            f"Anonymous memory ({anon_source}): {_format_bytes(anon_bytes)} "
                             f"({int(anon_share * 100)}% of current cgroup usage). "
                             f"Thresholds: cgroup >= {int(_CRITICAL_THRESHOLD * 100)}%, "
                             f"anon share of usage >= {int(_ANON_SHARE_OF_USAGE_THRESHOLD * 100)}%."
@@ -273,13 +301,17 @@ def check_memory_usage(self) -> None:
                 elif not self._critical_logged:
                     self._critical_logged = True
                     logger.info(
-                        "Cgroup usage crossed %d%% but anonymous memory is only %d%% of current cgroup usage; not raising.",
+                        "Cgroup usage crossed %d%% (%s of %s) but anonymous memory is only %d%% of current cgroup usage; not raising.",
                         int(_CRITICAL_THRESHOLD * 100),
+                        _format_bytes(usage_bytes),
+                        _format_bytes(limit_bytes),
                         int(anon_share * 100),
                     )
             elif not self._critical_logged:
                 self._critical_logged = True
                 logger.warning(
-                    "Cgroup usage crossed %d%% but anonymous memory signal unavailable; skipping fail-fast.",
+                    "Cgroup usage crossed %d%% (%s of %s) but anonymous memory signal unavailable; skipping fail-fast.",
                     int(_CRITICAL_THRESHOLD * 100),
+                    _format_bytes(usage_bytes),
+                    _format_bytes(limit_bytes),
                 )

unit_tests/utils/test_memory_monitor.py

@@ -17,6 +17,7 @@
     _CGROUP_V2_STAT,
     _PROC_SELF_STATUS,
     MemoryMonitor,
+    _format_bytes,
     _read_cgroup_v2_anon_bytes,
     _read_process_anon_rss_bytes,
 )
@@ -88,6 +89,46 @@ def test_check_interval_negative_raises() -> None:
         MemoryMonitor(check_interval=-1)
 
 
+def test_init_logs_configured_thresholds(caplog: pytest.LogCaptureFixture) -> None:
+    """Instantiation should emit one INFO line with the configured thresholds and intervals."""
+    with caplog.at_level(logging.INFO, logger="airbyte"):
+        MemoryMonitor(check_interval=1234)
+    init_logs = [r for r in caplog.records if "MemoryMonitor instantiated" in r.message]
+    assert len(init_logs) == 1
+    msg = init_logs[0].message
+    assert "critical threshold: 95%" in msg
+    assert "anon share of usage threshold: 85%" in msg
+    assert "high-pressure threshold: 90%" in msg
+    assert "check interval: 1234 messages" in msg
+    assert "tightens to 100 under high pressure" in msg
+
+
+# ---------------------------------------------------------------------------
+# _format_bytes — unit tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("num_bytes", "expected"),
+    [
+        (0, "0 B"),
+        (999, "999 B"),
+        (1_000, "1.00 KB"),
+        (1_500, "1.50 KB"),
+        (999_999, "1000.00 KB"),
+        (1_000_000, "1.00 MB"),
+        (960_000_000, "960.00 MB"),
+        (999_999_999, "1000.00 MB"),
+        (1_000_000_000, "1.00 GB"),
+        (2_109_915_136, "2.11 GB"),
+        (2_147_483_648, "2.15 GB"),
+    ],
+)
+def test_format_bytes(num_bytes: int, expected: str) -> None:
+    """`_format_bytes` renders byte counts with 2 decimals using decimal units."""
+    assert _format_bytes(num_bytes) == expected
+
+
 # ---------------------------------------------------------------------------
 # check_memory_usage — no-op paths
 # ---------------------------------------------------------------------------
@@ -96,6 +137,7 @@ def test_check_interval_negative_raises() -> None:
 def test_noop_when_no_cgroup(caplog: pytest.LogCaptureFixture) -> None:
     """check_memory_usage should be a no-op when cgroup is unavailable."""
     monitor = MemoryMonitor()
+    caplog.clear()  # discard the one-shot instantiation log
     with (
         caplog.at_level(logging.WARNING, logger="airbyte"),
         patch.object(Path, "exists", return_value=False),
@@ -107,6 +149,7 @@ def test_noop_when_no_cgroup(caplog: pytest.LogCaptureFixture) -> None:
 def test_noop_when_limit_is_max(caplog: pytest.LogCaptureFixture) -> None:
     """When cgroup v2 memory.max is 'max' (unlimited), should be a no-op."""
     monitor = MemoryMonitor(check_interval=1)
+    caplog.clear()
     with (
         caplog.at_level(logging.WARNING, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -119,6 +162,7 @@ def test_noop_when_limit_is_max(caplog: pytest.LogCaptureFixture) -> None:
 def test_noop_when_limit_is_zero(caplog: pytest.LogCaptureFixture) -> None:
     """When cgroup limit file contains '0', should be a no-op."""
     monitor = MemoryMonitor(check_interval=1)
+    caplog.clear()
     with (
         caplog.at_level(logging.WARNING, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -136,6 +180,7 @@ def test_noop_when_limit_is_zero(caplog: pytest.LogCaptureFixture) -> None:
 def test_no_log_below_threshold(caplog: pytest.LogCaptureFixture) -> None:
     """No log should be emitted when usage is below the 90% high-pressure threshold."""
     monitor = MemoryMonitor(check_interval=1)
+    caplog.clear()
     with (
         caplog.at_level(logging.DEBUG, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -182,6 +227,7 @@ def mock_read_text(self: Path) -> str:
 def test_check_interval_skips_intermediate_calls(caplog: pytest.LogCaptureFixture) -> None:
     """Monitor should only check cgroup files every check_interval messages."""
     monitor = MemoryMonitor(check_interval=5000)
+    caplog.clear()
     with (
         caplog.at_level(logging.INFO, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -205,6 +251,7 @@ def test_check_interval_skips_intermediate_calls(caplog: pytest.LogCaptureFixtur
 def test_malformed_cgroup_file_degrades_gracefully(caplog: pytest.LogCaptureFixture) -> None:
     """Malformed cgroup files should not crash the sync."""
     monitor = MemoryMonitor(check_interval=1)
+    caplog.clear()
     with (
         caplog.at_level(logging.WARNING, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -217,6 +264,7 @@ def test_malformed_cgroup_file_degrades_gracefully(caplog: pytest.LogCaptureFixt
 def test_empty_cgroup_file_degrades_gracefully(caplog: pytest.LogCaptureFixture) -> None:
     """Empty cgroup file content should not crash the sync."""
     monitor = MemoryMonitor(check_interval=1)
+    caplog.clear()
     with (
         caplog.at_level(logging.WARNING, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -233,6 +281,7 @@ def mock_read_text(self: Path) -> str:
         raise OSError("Permission denied")
 
     monitor = MemoryMonitor(check_interval=1)
+    caplog.clear()
     with (
         caplog.at_level(logging.WARNING, logger="airbyte"),
         patch.object(Path, "exists", _v2_exists),
@@ -342,6 +391,14 @@ def test_raises_when_cgroup_critical_and_anon_share_of_usage_above_threshold() -
     assert "critical threshold" in (exc_info.value.message or "")
     assert "96%" in (exc_info.value.message or "")
     assert "anon share of usage" in (exc_info.value.internal_message or "")
+    # Human-readable byte formatting: 960 MB usage, 1.00 GB limit, 840 MB anon.
+    internal = exc_info.value.internal_message or ""
+    assert "960.00 MB" in internal
+    assert "1.00 GB" in internal
+    assert "840.00 MB" in internal
+    # Raw byte counts should no longer appear in the message.
+    assert "960000000" not in internal
+    assert "1000000000" not in internal
 
 
 def test_no_raise_when_cgroup_critical_but_anon_share_of_usage_below_threshold(