fix(host_agent): an ingest writes the newest five samples, so a full backlog can land (#5149)
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 <noreply@anthropic.com>
This commit is contained in:
@@ -191,6 +191,11 @@ async def ingest():
|
|||||||
|
|
||||||
token_hash = _hash_token(raw)
|
token_hash = _hash_token(raw)
|
||||||
phases = _Phases()
|
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:
|
try:
|
||||||
async with current_app.db_sessionmaker() as session:
|
async with current_app.db_sessionmaker() as session:
|
||||||
reg = (await session.execute(
|
reg = (await session.execute(
|
||||||
@@ -326,13 +331,31 @@ async def ingest():
|
|||||||
phases.report(host.name, accepted, len(metric_rows),
|
phases.report(host.name, accepted, len(metric_rows),
|
||||||
sum(len(c) for _, c in docker_snapshots),
|
sum(len(c) for _, c in docker_snapshots),
|
||||||
sum(len(lines) for _, lines in docker_log_batches))
|
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
|
# 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).
|
# says where its time went, so a slow phase shows in the logs (#5149).
|
||||||
SLOW_INGEST_SECONDS = 2.0
|
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:
|
class _Phases:
|
||||||
"""Wall-clock per phase of one ingest request."""
|
"""Wall-clock per phase of one ingest request."""
|
||||||
|
|||||||
@@ -56,3 +56,20 @@ async def test_a_dropped_ingest_says_how_far_it_got(app, monkeypatch, caplog):
|
|||||||
with caplog.at_level(logging.WARNING):
|
with caplog.at_level(logging.WARNING):
|
||||||
phases.cancelled("abcd1234")
|
phases.cancelled("abcd1234")
|
||||||
assert "dropped after 10.1s for token=abcd1234… (completed: parse=0.2s) samples=21 metric_rows=840" in caplog.text
|
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)
|
||||||
|
|||||||
Reference in New Issue
Block a user