From 385a579afed1204e47bfd964dae92ea0f3f03118 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 6 Oct 2026 18:46:25 -0400 Subject: [PATCH] fix(host_agent): an ingest writes the newest five samples, so a full backlog can land (#5149) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measured on dev (f802c9f): every dropped ingest carried samples=21, metric_rows≈4,200 and never finished the INSERT flush inside the agent's 10 s ("completed: parse=0.1s"). The rollback left the agent's ring buffer full, so each later post was the same size and failed the same way: no host's metrics have landed since. An ingest now keeps the newest MAX_SAMPLES_PER_INGEST (5) samples by ts, logs how many it left out, and returns samples_dropped. The older samples are the ones the ring buffer would have overwritten anyway. Why the database writes so few rows a second is not yet known; the slow-ingest line now measures it on a post that finishes. Co-Authored-By: Claude Opus 5.5 --- plugins/host_agent/routes.py | 25 ++++++++++++++++++- .../plugins/host_agent/test_ingest_timing.py | 17 +++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index e70a138..bb27b68 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -191,6 +191,11 @@ async def ingest(): token_hash = _hash_token(raw) phases = _Phases() + samples, dropped = newest_samples(samples, MAX_SAMPLES_PER_INGEST) + if dropped: + current_app.logger.warning( + "host_agent ingest: kept the newest %d of %d samples for token=%s…", + len(samples), len(samples) + dropped, token_hash[:8]) try: async with current_app.db_sessionmaker() as session: reg = (await session.execute( @@ -326,13 +331,31 @@ async def ingest(): phases.report(host.name, accepted, len(metric_rows), sum(len(c) for _, c in docker_snapshots), sum(len(lines) for _, lines in docker_log_batches)) - return jsonify({"ok": True, "samples_accepted": accepted}), 200 + return jsonify({"ok": True, "samples_accepted": accepted, "samples_dropped": dropped}), 200 # The agent gives up on a post after 10 s (agent.py); an ingest past this # says where its time went, so a slow phase shows in the logs (#5149). SLOW_INGEST_SECONDS = 2.0 +# How many samples one ingest writes (#5149). An agent that could not post +# sends its whole buffer (up to 21 samples, ~4,000 metric rows); on the live +# instance that did not finish inside the agent's 10 s, the rollback left the +# buffer full, and every later post failed the same way. The newest samples +# are kept: the older ones would be lost to the agent's ring buffer anyway. +MAX_SAMPLES_PER_INGEST = 5 + + +def newest_samples(samples: list, limit: int) -> tuple[list, int]: + """(the newest `limit` samples, oldest first; how many were left out). + + Ordered by their ISO-8601 `ts` (the agent writes them all in one form); + one without a ts sorts oldest and is rejected later as malformed.""" + if len(samples) <= limit: + return samples, 0 + ordered = sorted(samples, key=lambda s: str(s.get("ts") or "") if isinstance(s, dict) else "") + return ordered[-limit:], len(samples) - limit + class _Phases: """Wall-clock per phase of one ingest request.""" diff --git a/tests/plugins/host_agent/test_ingest_timing.py b/tests/plugins/host_agent/test_ingest_timing.py index 9d31844..a3b5f3e 100644 --- a/tests/plugins/host_agent/test_ingest_timing.py +++ b/tests/plugins/host_agent/test_ingest_timing.py @@ -56,3 +56,20 @@ async def test_a_dropped_ingest_says_how_far_it_got(app, monkeypatch, caplog): with caplog.at_level(logging.WARNING): phases.cancelled("abcd1234") assert "dropped after 10.1s for token=abcd1234… (completed: parse=0.2s) samples=21 metric_rows=840" in caplog.text + + +def _s(minute): + return {"ts": f"2026-10-06T12:{minute:02d}:00+00:00", "cpu_pct": 1.0} + + +def test_a_full_backlog_is_cut_to_its_newest_samples(): + # Sent oldest first (the agent's buffer, then the new sample) or not. + backlog = [_s(m) for m in range(21)] + kept, dropped = routes.newest_samples(list(reversed(backlog)), 5) + assert [s["ts"] for s in kept] == [_s(m)["ts"] for m in range(16, 21)] + assert dropped == 16 + + +def test_a_small_post_is_left_alone(): + posted = [_s(3), _s(1)] + assert routes.newest_samples(posted, 5) == (posted, 0)