kafka_consumer: store broker_timestamps as marshal instead of json

[piochelepiotr]

What does this PR do?

Persists the kafka_consumer DSM broker_timestamps cache with marshal (a compact binary codec, base64-wrapped so it still rides the string-based persistent-cache API) instead of json.

No change to emitted metrics — kafka.estimated_consumer_lag and friends are computed from the same data.

Motivation

Cluster-check runners running kafka_consumer with enable_cluster_monitoring on large clusters have been OOMing (#incident-57455). Investigation (repro pod + live /proc + the Datadog continuous profiler) showed the growth is glibc malloc fragmentation fed by very high allocation churn — not a leak. The profiler's top Python allocator was _save_broker_timestamps -> json.dumps -> iterencode: every run re-serialized the entire (growing) timestamps dict, and JSON encodes every offset/timestamp as text.

Benchmarked at the incident's real cardinality (31,870 partitions), the broker_timestamps serialization on the save path dropped ~100x with marshal (and per-run load/save churn dropped several-fold overall). marshal keeps integer offsets and float timestamps native, avoiding JSON's per-number text encoding.

This is the source-side half of the mitigation; the allocator-side half (MALLOC_ARENA_MAX=2 on the runners) is a separate ops PR.

Why marshal (and not pickle)

pickle would be faster still, but it executes arbitrary opcodes on load — deserializing an on-disk cache file with pickle.loads turns a cache-tampering risk into remote code execution in the Agent process if anything else can write the run volume. marshal only (de)serializes basic built-in types and cannot construct/execute objects, so a tampered cache cannot run code (worst case is a crash — a large step down from RCE). marshal is also stdlib, so it adds no dependency and keeps the test-minimum-base-package job green (msgpack/orjson are not guaranteed in the minimum base).

Trade-offs, called out for review:

• marshal is not hardened against maliciously-crafted input — a tampered blob could crash the check (DoS), but not execute code.
• The marshal format is Python-version-specific, so the cache is discarded once (and rebuilt) after a Python upgrade or a format change. Handled by the try/except in _load_broker_timestamps.

Review checklist (to be filled by reviewers)

• Feature or bugfix MUST have appropriate tests — updated test_unit.py and test_integration.py to seed/read the cache in the marshal format with integer offset keys. Full unit suite (144) passes; lint clean.
• Add qa/required if this PR needs QA validation, or qa/skip-qa if it does not. → qa/required (agent-impacting runtime change; validate memory reduction + no estimated_consumer_lag regression on a live cluster).
• If you need to backport this PR to another branch, add the backport/<branch-name> label.

[datadog-prod-us1-5]

 

🎉 All green!

🧪 All tests passed
❄️ No new flaky tests detected

🎯 Code Coverage (details)
• Patch Coverage: 100.00%
• Overall Coverage: 92.11% (+3.89%)

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: cb18e07 | Docs | Datadog PR Page | Give us feedback!}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: aa959ece3d

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Avoid unpickling the persistent cache

When the Agent reads a cache file from its persistent cache directory, this now deserializes it with pickle.loads; if that cache entry is ever writable by another local process/container sharing the Agent run volume, the next kafka_consumer check run will execute attacker-controlled pickle opcodes under the Agent process. This cache previously used json.loads, so tampering could only corrupt the cache; use a non-executing binary format such as msgpack/struct-packed data, or authenticate the blob before unpickling.

Useful? React with 👍 / 👎.

[dd-octo-sts]

Validation Report

All 21 validations passed.

Show details

Validation	Description	Status
agent-reqs	Verify check versions match the Agent requirements file	✅
ci	Validate CI configuration and code coverage settings	✅
codeowners	Validate every integration has a CODEOWNERS entry	✅
config	Validate default configuration files against spec.yaml	✅
dep	Verify dependency pins are consistent and Agent-compatible	✅
http	Validate integrations use the HTTP wrapper correctly	✅
imports	Validate check imports do not use deprecated modules	✅
integration-style	Validate check code style conventions	✅
jmx-metrics	Validate JMX metrics definition files and config	✅
labeler	Validate PR labeler config matches integration directories	✅
legacy-signature	Validate no integration uses the legacy Agent check signature	✅
license-headers	Validate Python files have proper license headers	✅
licenses	Validate third-party license attribution list	✅
metadata	Validate metadata.csv metric definitions	✅
models	Validate configuration data models match spec.yaml	✅
openmetrics	Validate OpenMetrics integrations disable the metric limit	✅
package	Validate Python package metadata and naming	✅
qa-label	Validate the pull request declares whether it needs QA for the next Agent release	✅
readmes	Validate README files have required sections	✅
saved-views	Validate saved view JSON file structure and fields	✅
version	Validate version consistency between package and changelog	✅

View full run