feat: readings kept through a storage stall; an agent's backlog arrives in order (#5995)
CI / lint (push) Successful in 2s
CI / unit (push) Successful in 1m39s
CI / integration (push) Successful in 1m49s
CI / publish (push) Successful in 1m49s

On 2026-10-17 the NFS under Postgres stalled for ~15 minutes. Two things
went wrong for the readings that would have explained it:

- The SNMP poll stored before it polled again, so while the database hung
  it stopped polling. It now collects on schedule and hands each round to a
  Holding (steward/core/holding.py): one writer stores what is held, oldest
  first, each reading at the time it was collected. Bounded at two hours of
  polls, with a 30-minute deadline on a store; the device list read falls
  back to the last one read after 10 s.
- "Mount not answering" (10 checks in a row) went pending on Docker02 and
  Docker03 and never fired: the agent sent each post's backlog with its
  newest, healthy sample, and that sample ended every post's run. Agent
  1.22.0 queues the new sample behind the backlog, so Steward receives a
  host's samples in the order they were taken.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-17 09:36:47 -04:00
co-authored by Claude Opus 5.5
parent bd1c32d2a4
commit 41fc8073db
7 changed files with 387 additions and 77 deletions
+2 -2
View File
@@ -327,7 +327,7 @@ The agent's self-reported hostname in the payload is advisory. The real identity
### Failure behavior ### Failure behavior
- POST fails → push sample to ring buffer (max 20), exponential backoff `30s → 60s → 120s → 300s cap`, resets on first success. - POST fails → push sample to ring buffer (max 20), exponential backoff `30s → 60s → 120s → 300s cap`, resets on first success.
- On next success → flush all buffered samples in a single batched POST. - On next success → flush the buffer oldest first, five samples a POST (`MAX_SAMPLES_PER_POST`). The new sample queues behind the backlog, so Steward always receives a host's samples in the order they were taken (1.22.0, #5995): a rule counting checks in a row then sees a stall's run of breaching checks unbroken. Before, each post carried the newest sample too, and after the 2026-10-17 NFS stall its healthy reading ended every post's run, so "Mount not answering" (10 in a row) never fired.
- Ring buffer full → drop oldest, keep newest. Logged at DEBUG (expected during extended outages). - Ring buffer full → drop oldest, keep newest. Logged at DEBUG (expected during extended outages).
- Malformed payload rejection (400) → drop the sample (don't re-buffer, it'll just be rejected again). Indicates agent/server version skew. - Malformed payload rejection (400) → drop the sample (don't re-buffer, it'll just be rejected again). Indicates agent/server version skew.
- Token rejection (401) → log ERROR with remediation hint, continue retry loop at backoff cadence. Admin fixes via UI + conf file edit; no restart needed. - Token rejection (401) → log ERROR with remediation hint, continue retry loop at backoff cadence. Admin fixes via UI + conf file edit; no restart needed.
@@ -461,7 +461,7 @@ Content-Type: application/json
### Design points ### Design points
- **`samples` is always a list**, even for a single sample. This is how the ring buffer drains after a failure — agent POSTs `[current, ...buffered]` in a single request. One-path server code. - **`samples` is always a list**, even for a single sample. This is how the ring buffer drains after a failure — agent POSTs the oldest buffered samples, with the current one only once the backlog is through. One-path server code.
- **`ts` is agent-reported, ISO-8601 UTC.** Server trusts it for `PluginMetric.recorded_at`. Host clocks drift but NTP is reliable enough for our purposes. - **`ts` is agent-reported, ISO-8601 UTC.** Server trusts it for `PluginMetric.recorded_at`. Host clocks drift but NTP is reliable enough for our purposes.
- **If `|ts - now| > 5 minutes`**, server logs a WARN but still accepts. Don't reject, don't lose data over clock skew. - **If `|ts - now| > 5 minutes`**, server logs a WARN but still accepts. Don't reject, don't lose data over clock skew.
- **`metadata` is sent on every POST**, not just on change. Server-side diff detects actual changes and only writes on change. Cost per POST is one dict — negligible. Benefit: server can cleanly detect agent restarts. - **`metadata` is sent on every POST**, not just on change. Server-side diff detects actual changes and only writes on change. Cost per POST is one dict — negligible. Benefit: server can cleanly detect agent restarts.
+14 -8
View File
@@ -22,7 +22,7 @@ from collections import deque
from concurrent.futures import ThreadPoolExecutor, wait as wait_futures from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
from datetime import datetime, timezone from datetime import datetime, timezone
AGENT_VERSION = "1.21.0" AGENT_VERSION = "1.22.0"
# Default path to the local Docker Engine socket. Overridable via the # Default path to the local Docker Engine socket. Overridable via the
# `docker_socket` config key; collection is silently skipped if it's absent or # `docker_socket` config key; collection is silently skipped if it's absent or
@@ -1956,10 +1956,16 @@ def _drop_logs(sample: dict) -> dict:
return sample return sample
def split_backlog(buffered: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple: def split_backlog(queued: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple:
"""(the buffered samples to send with this one, the ones held for later): """(the samples to send in this post, the ones held for later): the oldest
the oldest limit-1, leaving room for the new sample.""" `limit` of the backlog with the new sample queued behind it.
return buffered[:limit - 1], buffered[limit - 1:]
The new sample waits its turn. Sent ahead of the backlog it reached Steward
before older samples did, and a rule counting checks in a row saw each
post's run of stalled checks ended by the newest, healthy one: after the
2026-10-17 NFS stall "Mount not answering" (10 in a row) never fired
(#5995)."""
return queued[:limit], queued[limit:]
def next_backoff(current: int) -> int: def next_backoff(current: int) -> int:
@@ -2074,10 +2080,10 @@ def main_loop(conf_path: str) -> int:
docker_logs_enabled=docker_logs_enabled, docker_logs_enabled=docker_logs_enabled,
docker_log_exclude=docker_log_exclude) docker_log_exclude=docker_log_exclude)
buffered = buffer.drain() buffered = buffer.drain()
send, held = split_backlog(buffered) send, held = split_backlog(buffered + [sample])
maint_events, maint_keys = maintenance_results(maint_seen) maint_events, maint_keys = maintenance_results(maint_seen)
payload = build_payload( payload = build_payload(
samples=send + [sample], samples=send,
hostname=hostname, hostname=hostname,
metadata=metadata, metadata=metadata,
events=events + ([lost_contact_event(lost_since)] if lost_since else []) events=events + ([lost_contact_event(lost_since)] if lost_since else [])
@@ -2098,7 +2104,7 @@ def main_loop(conf_path: str) -> int:
rate_state["updates_due"] = True rate_state["updates_due"] = True
for s in held: for s in held:
buffer.push(s) buffer.push(s)
if send: if buffered:
_log("INFO", f"flushed {len(send)} buffered samples" _log("INFO", f"flushed {len(send)} buffered samples"
+ (f", {len(held)} still held" if held else "")) + (f", {len(held)} still held" if held else ""))
backoff = 0 backoff = 0
+146 -60
View File
@@ -3,7 +3,9 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from sqlalchemy import select from sqlalchemy import select
@@ -47,15 +49,75 @@ def make_poll_task(app: "Quart") -> ScheduledTask:
) )
async def _do_poll(app: "Quart", poll=None, walk=None) -> None: # Polls collected but not yet stored, written one store at a time (#5995): a
"""Poll every device at once, then write each one's outcome to its row # stall of the storage under Postgres delays them instead of losing them. Two
and its answers to the metric tables. A device whose vendor tables are # hours at the default one-minute poll.
due is walked first, so a new disk is read in the same poll. `poll` and HELD_POLLS = 120
`walk` stand in for poller.poll_device and poller.walk_tables in tests.""" # How long the device list may take to read before the last one read stands
global RATES # in, so a stalled database doesn't stop the polling it is about to explain.
from steward.core.alerts import MetricBatch DEVICES_READ_S = 10.0
# How long a poll waits for its own store before collecting on.
STORE_WAIT_S = 20.0
# Each device as last read: plain copies, so a poll can go on without the
# database, and a walk's rows found meanwhile can be kept on them.
DEVICES: list | None = None
@dataclass
class Polled:
"""One device's poll, collected and waiting to be stored."""
at: datetime
device_id: str
name: str
failed: str | None = None # the poll raised: why, in words
error: str | None = None
scan: object = None # a TableScan when the tables were walked
answered: bool = False
missing: list = field(default_factory=list)
texts: dict = field(default_factory=dict)
values: list = field(default_factory=list) # (metric, value)
def _snapshot(device) -> SimpleNamespace:
from sqlalchemy import inspect as sa_inspect
return SimpleNamespace(**{a.key: getattr(device, a.key)
for a in sa_inspect(type(device)).column_attrs})
async def _devices(app: "Quart") -> list:
"""The devices to poll: read now, or as last read when the database
doesn't answer within DEVICES_READ_S."""
global DEVICES
from .models import SnmpDevice from .models import SnmpDevice
async def read() -> list:
async with app.db_sessionmaker() as session:
return [_snapshot(d) for d in (await session.execute(select(SnmpDevice))).scalars()]
try:
DEVICES = await asyncio.wait_for(read(), DEVICES_READ_S)
except Exception as exc:
if DEVICES is None:
logger.warning("SNMP devices not read (%s); nothing to poll yet", exc or type(exc).__name__)
return []
logger.warning("SNMP devices not read within %.0f s (%s); polling the %d last read",
DEVICES_READ_S, exc or type(exc).__name__, len(DEVICES))
return DEVICES
async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
"""Poll every device at once, hold what each answered, and store it: the
device's outcome on its row, its answers in the metric tables, each at
the time it was collected. A device whose vendor tables are due is walked
first, so a new disk is read in the same poll. When the database doesn't
take the store within STORE_WAIT_S the poll returns and the next one
collects on; what is held is stored, oldest first, once it does (#5995).
`poll` and `walk` stand in for poller.poll_device and poller.walk_tables
in tests."""
global RATES
from steward.core.holding import holding
from .poller import poll_device, walk_tables from .poller import poll_device, walk_tables
from .readings import state_of, tables_for from .readings import state_of, tables_for
from .service import RateKeeper, credentials, readings_for from .service import RateKeeper, credentials, readings_for
@@ -64,67 +126,91 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None:
walk = walk or walk_tables walk = walk or walk_tables
if RATES is None: if RATES is None:
RATES = RateKeeper() RATES = RateKeeper()
held = holding(app, "SNMP poll", store=lambda rounds: _store(app, rounds), limit=HELD_POLLS)
async with app.db_sessionmaker() as session: devices = await _devices(app)
devices = (await session.execute(select(SnmpDevice))).scalars().all() if devices:
if not devices: started = datetime.now(timezone.utc)
return scans, asked = {}, {}
started = datetime.now(timezone.utc) async def one(d):
scans, asked = {}, {} creds = credentials(d)
if tables_due(d, started):
scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or []))
found = scans.get(d.id)
asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None)
return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]])
async def one(d): results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True)
creds = credentials(d)
if tables_due(d, started):
scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or []))
found = scans.get(d.id)
asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None)
return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]])
results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True) now = datetime.now(timezone.utc)
polls = []
for device, result in zip(devices, results):
if isinstance(result, BaseException):
logger.warning("SNMP poll of %s failed: %s", device.name, result)
polls.append(Polled(now, device.id, device.name, failed=f"SNMP poll failed: {result}"))
device.last_error = polls[-1].failed
continue
p = Polled(now, device.id, device.name, error=result.error, scan=scans.get(device.id),
answered=bool(result.readings or result.texts), missing=list(result.missing))
if p.scan is not None and p.scan.error is None:
# Kept on the copy too, so a stall doesn't walk it again every poll.
device.tables, device.tables_at = p.scan.rows, now
device.last_error = result.error
missing = set(result.missing)
for reading in asked[device.id]:
value = result.readings.get(reading.key)
if reading.unit == "state":
# Its word is shown as answered; its level is stored,
# unless Steward doesn't know the word (level None).
raw = value if value is not None else result.texts.get(reading.key)
if raw is None:
continue
p.texts[reading.key], value = state_of(reading, raw)
elif value is None:
if reading.key in result.texts: # a word where a number was asked for
missing.add(reading.key)
continue
elif reading.counter:
value = RATES.rate(device.id, reading, value, now)
else:
value = value * reading.scale + reading.offset
if value is not None:
p.values.append((reading.key, value))
p.missing = sorted(missing)
polls.append(p)
held.hold(polls)
await held.flush(STORE_WAIT_S)
async def _store(app: "Quart", rounds: list[list[Polled]]) -> None:
"""Write held polls, oldest first, in one transaction."""
from steward.core.alerts import MetricBatch
from .models import SnmpDevice
now = datetime.now(timezone.utc)
async with app.db_sessionmaker() as session: async with app.db_sessionmaker() as session:
async with session.begin(): async with session.begin():
batch = MetricBatch() batch = MetricBatch()
for device, result in zip(devices, results): for polls in rounds:
row = await session.get(SnmpDevice, device.id) for p in polls:
if row is None: # deleted while it was being polled row = await session.get(SnmpDevice, p.device_id)
continue if row is None: # deleted since it was polled
row.last_polled_at = now
if isinstance(result, BaseException):
logger.warning("SNMP poll of %s failed: %s", device.name, result)
row.last_error = f"SNMP poll failed: {result}"
continue
scan = scans.get(device.id)
if scan is not None and scan.error is None:
row.tables, row.tables_at = scan.rows, now
row.last_error = result.error
missing = set(result.missing)
if result.readings or result.texts:
row.last_ok_at = now
texts = dict(row.texts or {}) if result.error else {}
for reading in asked[device.id]:
value = result.readings.get(reading.key)
if reading.unit == "state":
# Its word is shown as answered; its level is stored,
# unless Steward doesn't know the word (level None).
raw = value if value is not None else result.texts.get(reading.key)
if raw is None:
continue
texts[reading.key], value = state_of(reading, raw)
elif value is None:
if reading.key in result.texts: # a word where a number was asked for
missing.add(reading.key)
continue continue
elif reading.counter: row.last_polled_at = p.at
value = RATES.rate(device.id, reading, value, now) if p.failed:
else: row.last_error = p.failed
value = value * reading.scale + reading.offset
if value is None:
continue continue
batch.add(source_module="snmp", resource_name=device.name, if p.scan is not None and p.scan.error is None:
metric_name=reading.key, value=value, recorded_at=now) row.tables, row.tables_at = p.scan.rows, p.at
row.missing = sorted(missing) row.last_error = p.error
row.texts = texts if p.answered:
row.last_ok_at = p.at
# A failed poll keeps the words it had; one that answered says them afresh.
texts = dict(row.texts or {}) if p.error else {}
texts.update(p.texts)
row.missing, row.texts = p.missing, texts
for key, value in p.values:
batch.add(source_module="snmp", resource_name=p.name,
metric_name=key, value=value, recorded_at=p.at)
await batch.record(session) await batch.record(session)
+99
View File
@@ -0,0 +1,99 @@
"""Collected readings held in memory until the database takes them (#5995).
Steward's Postgres sits on the NFS storage it watches. When that storage
stalls, every write waits, and a collector that stores what it found before
collecting again stops collecting for as long as the stall lasts: on
2026-10-17 the NAS readings went blind for the fifteen minutes they mattered.
A collector built on a Holding collects on schedule whatever the database is
doing, and hands each round to the holding. One writer at a time stores what
is held, oldest first, each item at the time it was collected, so a stall
delays the readings instead of losing them, and alert rules count them in the
order they happened.
held = holding(app, "SNMP poll", store=_store, limit=120)
held.hold(round) # one collected round
await held.flush(wait_s=20) # store it, or leave it to the writer
The holding is bounded: past `limit` rounds the oldest is dropped, and that
is logged with how many have gone. A store has `deadline_s`: generous, past
any stall seen so far, so a connection that will never answer is given up
and the rounds stay held for a fresh one rather than waiting forever.
"""
from __future__ import annotations
import asyncio
import logging
from collections import deque
from typing import Awaitable, Callable
logger = logging.getLogger(__name__)
class Holding:
def __init__(self, name: str, store: Callable[[list], Awaitable[None]], limit: int,
deadline_s: float = 1800.0) -> None:
self.name = name
self.store = store
self.limit = limit
self.deadline_s = deadline_s
self._held: deque[tuple[int, object]] = deque()
self._seq = 0
self._writer: asyncio.Task | None = None
self.dropped = 0
def __len__(self) -> int:
return len(self._held)
def hold(self, item) -> None:
"""Hold one collected round until it is stored."""
self._seq += 1
self._held.append((self._seq, item))
while len(self._held) > self.limit:
self._held.popleft()
self.dropped += 1
logger.warning("%s: the database hasn't taken a round in %d; dropped the oldest "
"(%d dropped so far)", self.name, self.limit, self.dropped)
async def flush(self, wait_s: float) -> bool:
"""Start the writer if it isn't running, and wait up to `wait_s` for
it. True when nothing is left held. A writer still waiting on the
database is left to finish: the next round joins what it stores next."""
writer = self._writer
if writer is None or writer.done() or writer.get_loop() is not asyncio.get_running_loop():
if not self._held:
return True
writer = self._writer = asyncio.create_task(self._write())
done, _ = await asyncio.wait({writer}, timeout=wait_s)
if not done:
logger.warning("%s: the database hasn't taken %d held round(s) in %.0f s; "
"collecting on", self.name, len(self._held), wait_s)
return not self._held
async def _write(self) -> None:
while self._held:
batch = list(self._held)
try:
await asyncio.wait_for(self.store([item for _, item in batch]), self.deadline_s)
except TimeoutError:
logger.error("%s: storing %d held round(s) took over %.0f s; given up, they stay "
"held (the database may still finish it)", self.name, len(batch),
self.deadline_s)
return
except Exception:
logger.exception("%s: storing %d held round(s) failed; they stay held",
self.name, len(batch))
return
last = batch[-1][0]
while self._held and self._held[0][0] <= last:
self._held.popleft()
def holding(app, name: str, store: Callable[[list], Awaitable[None]], limit: int) -> Holding:
"""The app's Holding for one collector, made on first use."""
holdings = getattr(app, "_holdings", None)
if holdings is None:
holdings = app._holdings = {}
if name not in holdings:
holdings[name] = Holding(name, store, limit)
return holdings[name]
+64
View File
@@ -0,0 +1,64 @@
"""Collected rounds held in memory until the database takes them (#5995):
stored oldest first, one writer at a time, bounded, and kept through a
store that fails or never answers."""
import asyncio
from types import SimpleNamespace as NS
from steward.core.holding import Holding, holding
def test_rounds_held_while_the_store_hangs_are_stored_in_order_once_it_answers():
async def main():
gate, stored = asyncio.Event(), []
async def store(rounds):
await gate.wait()
stored.append(list(rounds))
held = Holding("t", store, limit=10)
held.hold("a")
assert await held.flush(0.05) is False # the store hangs: the caller goes on
held.hold("b")
assert await held.flush(0.05) is False and len(held) == 2
gate.set()
assert await held.flush(1.0) is True and len(held) == 0
return stored
assert asyncio.run(main()) == [["a"], ["b"]] # "b" waited for the store in flight
def test_past_its_limit_the_oldest_round_goes_and_is_counted():
held = Holding("t", store=None, limit=2)
for r in ("a", "b", "c"):
held.hold(r)
assert len(held) == 2 and held.dropped == 1 and [i for _, i in held._held] == ["b", "c"]
def test_a_store_that_fails_or_overruns_its_deadline_keeps_the_rounds_held():
async def main():
calls = []
async def failing(rounds):
calls.append(list(rounds))
raise RuntimeError("database refused")
held = Holding("t", failing, limit=10)
held.hold("a")
assert await held.flush(1.0) is False and len(held) == 1
async def never(rounds):
await asyncio.sleep(3600)
stuck = Holding("t", never, limit=10, deadline_s=0.05)
stuck.hold("a")
assert await stuck.flush(1.0) is False and len(stuck) == 1 # given up, not lost
return calls
assert asyncio.run(main()) == [["a"]]
def test_an_app_has_one_holding_per_collector():
app = NS()
first = holding(app, "SNMP poll", store=None, limit=5)
assert holding(app, "SNMP poll", store=None, limit=5) is first
assert holding(app, "UniFi poll", store=None, limit=5) is not first
@@ -120,11 +120,11 @@ def test_backoff_schedule_stays_inside_the_stale_window():
assert next_backoff(60) == 60 # capped below Steward's 180 s stale window (#5154) assert next_backoff(60) == 60 # capped below Steward's 180 s stale window (#5154)
def test_a_backlog_drains_a_few_samples_a_post_oldest_first(): def test_a_backlog_drains_a_few_samples_a_post_oldest_first_and_the_new_one_waits_its_turn():
from plugins.host_agent.agent import MAX_SAMPLES_PER_POST, split_backlog from plugins.host_agent.agent import MAX_SAMPLES_PER_POST, split_backlog
buffered = list(range(20)) queued = list(range(20)) + ["new"]
send, held = split_backlog(buffered) send, held = split_backlog(queued)
assert send == [0, 1, 2, 3] and held == list(range(4, 20)) assert send == [0, 1, 2, 3, 4] and len(send) == MAX_SAMPLES_PER_POST
assert len(send) + 1 == MAX_SAMPLES_PER_POST # room for the new sample assert held == list(range(5, 20)) + ["new"] # never ahead of an older sample
assert split_backlog([7]) == ([7], []) assert split_backlog(["new"]) == (["new"], [])
assert split_backlog([]) == ([], []) assert split_backlog([7, "new"]) == ([7, "new"], [])
+55
View File
@@ -0,0 +1,55 @@
"""An SNMP poll collects on schedule while the database hangs, and what it
held is stored oldest first, each at its own time, once it answers (#5995)."""
import asyncio
from contextlib import asynccontextmanager
from types import SimpleNamespace as NS
import plugins.snmp.scheduler as scheduler
import plugins.snmp.service as service
from plugins.snmp.poller import Credentials, PollResult
def _device():
return NS(id="d1", name="nas", address="10.0.0.5", port=161, version="2c", presets=["system"],
interfaces="", custom=None, tables={}, tables_at=None, last_error=None, texts={})
def test_polls_go_on_while_the_database_hangs_and_are_stored_in_order_after(monkeypatch):
async def main():
stalled = asyncio.Event()
@asynccontextmanager
async def hung_session():
await stalled.wait() # the storage under Postgres has stalled
yield None
app = NS(db_sessionmaker=hung_session)
gate, stored = asyncio.Event(), []
async def store(app_, rounds):
await gate.wait()
stored.extend(rounds)
uptimes = iter([100.0, 200.0])
async def poll(address, port, creds, oids):
return PollResult(readings={"uptime": next(uptimes)})
monkeypatch.setattr(scheduler, "DEVICES", [_device()])
monkeypatch.setattr(scheduler, "DEVICES_READ_S", 0.05)
monkeypatch.setattr(scheduler, "STORE_WAIT_S", 0.05)
monkeypatch.setattr(scheduler, "_store", store)
monkeypatch.setattr(service, "credentials", lambda d: Credentials(community="public"))
scheduler.RATES = None
await scheduler._do_poll(app, poll=poll) # the device list is the last one read
await asyncio.sleep(0.01)
await scheduler._do_poll(app, poll=poll) # the first store hasn't returned: polled anyway
gate.set()
await app._holdings["SNMP poll"].flush(1.0)
return stored
rounds = asyncio.run(main())
assert [[(p.name, p.values) for p in r] for r in rounds] == [
[("nas", [("uptime", 100.0)])], [("nas", [("uptime", 200.0)])]]
assert rounds[0][0].at < rounds[1][0].at # each stored at the time it was collected