Count dropped log lines per point (#126)

* test: benchmark the forward writer per sink

Discard, loopback tcp and loopback udp, so the serial forward path has a measured per-line ceiling to compare future changes against.

* feat: count dropped log lines per point

log_lines_dropped_total{point=forward|subscriber|console} on /metrics. Every drop on the agent side was already counted and warned about; none of it was observable without reading logs. The shim stays out: it is a separate process and reports its drops once when it exits.

* feat: count what journald rate-limits

journald drops a service's excess lines inside itself and tells the sender nothing, so the shim and console counters never see it; the reader ORs journald's SD_MESSAGE_JOURNAL_DROPPED notices into its match and books N_DROPPED for eru units and the agent's own senders as point=journald. On a stock node a 60000-line burst keeps 37500. The counter now registers where it is declared, so a node without an api addr counts the same way.

* review: apply the simplify round

forward keeps its one-line error shape, the test helper carries a broadcaster, the counter asserts compare exactly, and the writer benchmark swaps the no-op arm for an encode arm so the agent's own per-line cost is visible next to the socket floor.

* perf: encode forward lines into a reused buffer

json.Marshal returns cap==len, so appending the newline reallocated and copied every line. A buffer and json.Encoder kept on the StreamEncoder cut the encode from 480 B/5 allocs to 64 B/4 allocs per line; the tcp arm of BenchmarkWriterWrite went from 6.5-7.7 us to 4.3-4.6 us.

* fix: stop inferring journald log drops

Journald suppression notices are service-wide, delayed until another allowed message, and cannot close a per-workload counter. Keep the metric limited to drops the agent observes directly.

* docs: keep the journald rate limit and the no-blocking property

The revert of the journald point also removed the operator facts that stand on their own: journald drops a fast service's lines silently at its own per-service limit, a unit's LogRateLimitBurst= is the override, and nothing on the log path blocks a workload.

* docs: narrow the log drop guarantee

Journald writes have no deadline, so the README must not promise that every logging failure is non-blocking. Keep the claim limited to observable drops.

* review: say what the shim guards against

The comment promised the container never blocks on the journal; the code only drops a line the journal refuses, and a journal that stops draining its socket still back-pressures the container (journal.Send has no deadline).

Author: CMGS

README.md

@@ -11,9 +11,9 @@ Eru's per-node agent. It watches the workloads a node runs — containerd contai
 
 - **Node heartbeat** — reports the node alive to core on an interval, with a ttl three times the interval so one lost report does not evict the node; a clean shutdown removes the status, a `SIGUSR1` restart keeps it.
 - **Workload health checks** — TCP and HTTP probes declared per workload, run on every runtime event and on a periodic sweep, with the result published through core.
-- **Log forwarding** — ships each line as a JSON record to `tcp://`, `udp://` or `journal://` targets, sharded over several targets by workload id, read from the node's journal with a persisted cursor, or read straight off a VM's serial console. Container output reaches the journal through `eru-agent log-shim`, containerd's binary logger; console output is written there by the agent itself, so every workload kind has the same history.
+- **Log forwarding** — ships each line as a JSON record to `tcp://`, `udp://` or `journal://` targets, sharded over several targets by workload id, read from the node's journal with a persisted cursor, or read straight off a VM's serial console. Container output reaches the journal through `eru-agent log-shim`, containerd's binary logger; console output is written there by the agent itself, so every workload kind has the same history. A down forward target, a tailing client that falls behind, or journald's own per-service rate limit can drop lines; the agent counts only the drops it can observe.
 - **Live log tailing** — `GET /log/?app=<name>` streams the logs of one application straight off the node.
-- **Prometheus metrics** — per-workload cpu, memory, per-nic network and per-device block io gauges on `/metrics`, optionally pushed to statsd as well. Sampled straight from cgroup v2 files, so a tick makes no call to any daemon.
+- **Prometheus metrics** — per-workload cpu, memory, per-nic network and per-device block io gauges on `/metrics`, optionally pushed to statsd as well, plus a node-level `log_lines_dropped_total{point}` counter for the log path. Sampled straight from cgroup v2 files, so a tick makes no call to any daemon.
 - **Runtimes per node** — a node declares the runtimes it hosts under a required `runtimes:` section: containerd containers, systemd process pods, cocoon VMs, or several at once; the heartbeat needs every one of them alive. The old `runtime:` and `docker:` keys are gone, with no compatibility shim. A mock runtime and store cover development.
 - **Node-side helper modes** — the same binary is containerd's task logger (`eru-agent log-shim`) and its CNI hook (`eru-agent oci-hook`), so a container node needs no eru daemon beyond the agent.
 

collector/console.go

@@ -22,6 +22,8 @@ const (
 	consoleRetryMax = 5 * time.Second
 )
 
+var droppedByConsole = LogLinesDropped.WithLabelValues(DropPointConsole)
+
 type journalFunc func(message string, priority journal.Priority, vars map[string]string) error
 
 // pathFunc answers where the vm's console is now: a restart moves it, and core rewrites the meta file.
@@ -129,5 +131,6 @@ func (c *Console) emit(line string, handle EntryHandler) {
 	// journald holds the history core reads back over ssh, the same way it does for a container
 	if err := c.send(line, journal.PriInfo, c.vars); err != nil {
 		c.dropped++
+		droppedByConsole.Inc()
 	}
 }

collector/console_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/coreos/go-systemd/v22/journal"
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -154,13 +155,15 @@ func TestEmitStampsTheWorkloadAndTheConsoleStream(t *testing.T) {
 func TestEmitKeepsForwardingWhenTheJournalRefuses(t *testing.T) {
 	console := NewConsole(consoleWorkload, consoleAppname, at("unused"))
 	console.send = func(string, journal.Priority, map[string]string) error { return errJournalRefused }
+	counted := testutil.ToFloat64(droppedByConsole)
 
 	var entries []*Entry
 	console.emit("boot", func(e *Entry) { entries = append(entries, e) })
 	console.emit("login:", func(e *Entry) { entries = append(entries, e) })
 
 	assert.Len(t, entries, 2)
 	assert.Equal(t, 2, console.dropped)
+	assert.Equal(t, counted+2, testutil.ToFloat64(droppedByConsole))
 }
 
 func at(path string) pathFunc {

collector/metrics.go

@@ -11,18 +11,29 @@ import (
 	"github.com/projecteru2/core/log"
 	coreutils "github.com/projecteru2/core/utils"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/projecteru2/agent/source"
 )
 
 const (
-	labelNIC = "nic"
-	labelDev = "dev"
+	DropPointSubscriber = "subscriber"
+	DropPointConsole    = "console"
+	DropPointForward    = "forward"
+
+	labelNIC   = "nic"
+	labelDev   = "dev"
+	labelPoint = "point"
 
 	metricMemMaxUsage = "mem_max_usage"
 )
 
 var (
+	LogLinesDropped = promauto.With(prometheus.DefaultRegisterer).NewCounterVec(prometheus.CounterOpts{
+		Name: "log_lines_dropped_total",
+		Help: "workload log lines that never reached their destination, by the point that lost them.",
+	}, []string{labelPoint})
+
 	clientsMutex sync.Mutex
 	clients      = map[string]*MetricsClient{}
 

docs/architecture.md

@@ -80,7 +80,7 @@ Either way, each line becomes the same JSON record:
 
 The record goes to two places: the configured forwarder for that workload, and the in-process broadcaster that serves `/log/`. Non-utf8 bytes are escaped as `\xNN` so a binary blob on stdout cannot corrupt the stream.
 
-**Log broadcaster.** `GET /log/?app=<name>` hijacks the connection and subscribes to every record whose `name` matches. Each reader broadcasts its lines in order, and one workload's lines come from one reader, so a subscriber sees a workload's lines in order. A subscriber whose connection breaks — detected by a read on the hijacked socket — is cancelled and dropped.
+**Log broadcaster.** `GET /log/?app=<name>` hijacks the connection and subscribes to every record whose `name` matches. Each reader broadcasts its lines in order, and one workload's lines come from one reader, so a subscriber sees a workload's lines in order. A subscriber whose connection breaks — detected by a read on the hijacked socket — is cancelled and dropped; one that falls more than 256 lines behind loses the excess rather than stalling the node's forwarding, counted in `log_lines_dropped_total{point="subscriber"}`.
 
 ## Status reporting
 

docs/configuration.md

@@ -90,10 +90,12 @@ log:
   stdout: false
 ```
 
-`forwards` is a list of targets. Supported schemes are `tcp://`, `udp://` and `journal://`; a target with any other scheme is accepted and silently discards, with a warning at startup. Each workload is pinned to one target by hashing its id, so several targets share the load without duplicating lines. A target that is down is retried every 30 seconds in the background while its lines are dropped; a line too large for a udp datagram loses only itself.
+`forwards` is a list of targets. Supported schemes are `tcp://`, `udp://` and `journal://`; a target with any other scheme is accepted and silently discards, with a warning at startup. Each workload is pinned to one target by hashing its id, so several targets share the load without duplicating lines. A target that is down is retried every 30 seconds in the background while its lines are dropped; a line too large for a udp datagram loses only itself. Both count in `log_lines_dropped_total{point="forward"}` on `/metrics`.
 
 A VM's output is its serial console, so the agent reads the console its meta file names, one goroutine per VM, forwarding each line and writing it to journald so the history is there too. Every other runtime logs to journald natively, and the agent runs one `journalctl --follow --output=json SYSLOG_IDENTIFIER=eru` for the whole node, resuming from the cursor it saved under `state_dir`. That path needs `journalctl` on the node, and needs every eru workload to log under the `eru` syslog identifier: process units get it from core's `systemd-run`, containers from `eru-agent log-shim`.
 
+journald rate-limits every service on its own — by default `RateLimitBurst=10000` per `RateLimitIntervalSec=30s`, per priority class, scaled up with the journal's free space — and drops the excess without telling the sender, so a workload that logs faster than that loses lines before the agent ever sees them. A unit's own `LogRateLimitBurst=` overrides the node default; journald's "Suppressed N messages" notices are the only record of what was lost.
+
 `stdout: true` additionally writes every forwarded line to the agent's own log.
 
 ## `metrics`

docs/metrics.md

@@ -16,6 +16,24 @@ Sampling happens every `metrics.step` seconds. Rates are computed over that wind
 
 The node-wide cpu split that the two `cpu_host_*_usage` ratios divide by is read from `/proc/stat` at most once a second and shared by every workload, instead of once per workload per tick.
 
+## Counters
+
+One node-level counter, with no workload labels:
+
+| Counter | Meaning |
+|---|---|
+| `log_lines_dropped_total{point}` | workload log lines the agent observed losing, by where they were lost |
+
+| `point` | when |
+|---|---|
+| `forward` | the workload's forward target was down, timed out on a write, or refused an oversized datagram; the line is not replayed after the reconnect |
+| `subscriber` | a `GET /log` client fell more than 256 lines behind |
+| `console` | journald refused a VM console line |
+
+Lines a `log-shim` could not journal are not counted here: the shim is a separate process and reports its drops once, when it exits.
+
+Journald's internal rate-limit drops are not counted either: its notices cover a whole service and priority bucket, appear only after a later message passes the limit, and cannot provide a complete per-workload total.
+
 ## Gauges
 
 Every gauge carries the same constant labels: `containerID`, `hostname`, `appname`, `entrypoint`, `orchestrator` and `labels` — the last being the workload's own labels, minus eru's internal ones, flattened to `k=v,k=v`.

logs/enc.go

@@ -1,6 +1,7 @@
 package logs
 
 import (
+	"bytes"
 	"encoding/json"
 	"io"
 	"sync"
@@ -17,19 +18,24 @@ type Encoder interface {
 }
 
 type StreamEncoder struct {
-	wt io.WriteCloser
+	wt  io.WriteCloser
+	buf bytes.Buffer
+	enc *json.Encoder
 }
 
 func NewStreamEncoder(wt io.WriteCloser) *StreamEncoder {
-	return &StreamEncoder{wt: wt}
+	e := &StreamEncoder{wt: wt}
+	e.enc = json.NewEncoder(&e.buf)
+	return e
 }
 
+// Encode runs under Writer's lock, which is what lets one buffer serve every line.
 func (e *StreamEncoder) Encode(logline *types.Log) error {
-	data, err := json.Marshal(logline)
-	if err != nil {
+	e.buf.Reset()
+	if err := e.enc.Encode(logline); err != nil {
 		return err
 	}
-	_, err = e.wt.Write(append(data, '\n'))
+	_, err := e.wt.Write(e.buf.Bytes())
 	return err
 }
 

logs/writer_test.go

@@ -2,6 +2,7 @@ package logs
 
 import (
 	"errors"
+	"io"
 	"net"
 	"strings"
 	"testing"
@@ -127,3 +128,70 @@ func TestReconnect(t *testing.T) {
 	writer.reconnect(ctx)
 	assert.NoError(t, writer.Write(ctx, &types.Log{}))
 }
+
+func BenchmarkWriterWrite(b *testing.B) {
+	line := &types.Log{
+		ID:         "0123456789abcdef0123456789abcdef",
+		Name:       "app",
+		Type:       common.StreamStdout,
+		EntryPoint: "web",
+		Ident:      "ident",
+		Data:       strings.Repeat("x", 200),
+		Datetime:   "2026-08-31T00:00:00.000000000Z",
+		Extra:      map[string]string{"zone": "z", "node": "n"},
+	}
+	ctx := b.Context()
+	writers := map[string]*Writer{"encode": {addr: "encode", scheme: "tcp", enc: NewStreamEncoder(nopSink{})}}
+	for name, addr := range map[string]string{"tcp": drainingTCP(b), "udp": drainingUDP(b)} {
+		w, err := NewWriter(ctx, addr, false)
+		require.NoError(b, err)
+		writers[name] = w
+	}
+	for name, w := range writers {
+		b.Run(name, func(b *testing.B) {
+			b.ReportAllocs()
+			for b.Loop() {
+				if err := w.Write(ctx, line); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}
+
+type nopSink struct{}
+
+func (nopSink) Write(p []byte) (int, error) { return len(p), nil }
+
+func (nopSink) Close() error { return nil }
+
+func drainingTCP(b *testing.B) string {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(b, err)
+	b.Cleanup(func() { _ = l.Close() })
+	go func() {
+		for {
+			conn, err := l.Accept()
+			if err != nil {
+				return
+			}
+			go func() { _, _ = io.Copy(io.Discard, conn) }()
+		}
+	}()
+	return "tcp://" + l.Addr().String()
+}
+
+func drainingUDP(b *testing.B) string {
+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
+	require.NoError(b, err)
+	b.Cleanup(func() { _ = conn.Close() })
+	go func() {
+		buf := make([]byte, 64<<10)
+		for {
+			if _, _, err := conn.ReadFrom(buf); err != nil {
+				return
+			}
+		}
+	}()
+	return "udp://" + conn.LocalAddr().String()
+}

logshim/logshim.go

@@ -105,7 +105,7 @@ func (s *stream) pump(reader io.Reader) {
 		if err != nil {
 			return
 		}
-		// a journal that cannot keep up must not block the container, so a refused line is lost
+		// a journal that refuses a line must not fail the container, so the line is lost
 		if sendErr := s.send(string(line), s.priority, vars); sendErr != nil {
 			s.dropped++
 			s.err = sendErr