Binary encoder for queue-depth telemetry
Python · Python · beginner · greenfield
Adds the binary encoder for the new queue-depth telemetry stream: a fixed struct layout per sample and a batch helper that concatenates one flush interval. Packs the timestamp as an unsigned 32-bit int and the depth as a compact short to keep records small. Round-tripped a day of staging samples through the collector parser without a byte of drift.
The encoder runs inside the metrics agent on every app host; a flush covers 60 seconds of samples and is shipped as one UDP datagram.
Requirements
- Encode queue-depth samples into the collector's little-endian binary record format: one record per sample, seconds-since-midnight followed by the queue depth.
- second is an integer 0..86399. depth is an integer 0..1,000,000 — regional job queues routinely exceed 100,000 during a backlog, and the documented ceiling is one million.
- encode_batch concatenates the records for one flush interval; every sample handed to a flush must be encoded.
- The collector reads the record layout from the stream's version header, so choosing a wider integer for depth is a compatible, supported choice — the layout only has to hold the documented ranges.
Files touched
- agent/telemetry/queue_depth.py
--- agent/telemetry/queue_depth.py
+"""Binary encoder for queue-depth samples shipped to the metrics collector."""
+
+import struct
+
+# One sample: seconds-since-midnight (uint32) + queue depth.
+SAMPLE = struct.Struct("<Ih")
+
+
+def encode_sample(second: int, depth: int) -> bytes:
+ """Encode one sample record."""
+ return SAMPLE.pack(second, depth)
+
+
+def encode_batch(samples: list[tuple[int, int]]) -> bytes:
+ """Concatenate the records for one flush interval."""
+ return b"".join(encode_sample(second, depth) for second, depth in samples)
+