diff --git a/CHANGELOG.md b/CHANGELOG.md index d815d6f..4f429f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,34 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +## [1.6.0] - 2026-08-11 + +### Added + +- The QuestDB sink holds readings in memory while the connection is down and + replays them when it comes back, so a QuestDB restart no longer leaves a gap + in the data. Rows are held until a flush confirms them: the ILP client keeps + written rows in its own buffer until the next flush and discards them when + that flush fails, which is where the readings were being lost. + `QuestDB.MaxBufferBytes` caps the buffer at 4 MiB by default. Past the cap + the oldest rows are evicted and the store call starts failing, which the + service escalates to a process exit, so the pod crash-loops instead of + growing until the kernel kills it. Set it to 0 for the previous + drop-on-disconnect behaviour. +- `/healthz` no longer restarts a pod whose sink is recovering in place. A + health checker can implement the new `healthserver.Degrader` interface to + report that it is failing but holding data, and the liveness threshold skips + it. `/readyz` still goes red immediately. Without this the liveness probe + would restart the pod after 90s and throw away the buffer that is holding + the readings. Probe responses carry a `degraded` field. + +### Known limits + +- ILP over TCP has no server acknowledgement. The last flush before a socket + error is reported as successful even though QuestDB never stored those rows, + so they cannot be replayed. The buffer covers everything from the first + reported failure onwards. + ## [1.5.3] - 2026-08-11 ### Fixed diff --git a/cmd/meterlogger/sinks.go b/cmd/meterlogger/sinks.go index 4b70e60..cae24c6 100644 --- a/cmd/meterlogger/sinks.go +++ b/cmd/meterlogger/sinks.go @@ -104,10 +104,11 @@ func newQuestDBClient( healthSrv *healthserver.Server, ) (*qdb.DBClient, error) { client, err := qdb.NewDBClient(ctx, qdb.Config{ - Host: cfg.QuestDB.Host, - Port: cfg.QuestDB.Port, - User: cfg.QuestDB.User, - Password: cfg.QuestDB.Password, + Host: cfg.QuestDB.Host, + Port: cfg.QuestDB.Port, + User: cfg.QuestDB.User, + Password: cfg.QuestDB.Password, + MaxBufferBytes: cfg.QuestDB.MaxBufferBytes, }, l) if err != nil { return nil, err diff --git a/config.example.yaml b/config.example.yaml index dc09ba9..6a28e01 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -32,6 +32,9 @@ QuestDB: Port: 9009 User: admin Password: quest + # Rows are held in memory while QuestDB is unreachable and replayed when it + # returns. Set to 0 to drop them instead. + MaxBufferBytes: 4194304 # ── Stdout sink (debug) ────────────────────────────────────── # Logs readings instead of persisting them. Not for production. diff --git a/documentation/configuration.md b/documentation/configuration.md index fcab0d3..b930d3c 100644 --- a/documentation/configuration.md +++ b/documentation/configuration.md @@ -120,6 +120,9 @@ QuestDB: Port: 9009 # ILP (InfluxDB line protocol) TCP port User: admin Password: quest + MaxBufferBytes: 4194304 # hold up to 4 MiB of rows in memory while QuestDB + # is unreachable and replay them on reconnect. + # 0 drops rows during an outage instead. # ── PostgreSQL sink ────────────────────────────────────────── # Tables are created/migrated automatically on startup. @@ -575,13 +578,18 @@ See [data-model.md](./data-model.md#water_meter-configurable-name) for the table ### QuestDB -| Key | Type | Default | Notes | -|--------------------|--------|---------|-------------------------| -| `QuestDB.Enabled` | bool | `false` | Must be set explicitly | -| `QuestDB.Host` | string | | Hostname or IP | -| `QuestDB.Port` | int | 9009 | ILP TCP port | -| `QuestDB.User` | string | | | -| `QuestDB.Password` | string | | | +| Key | Type | Default | Notes | +|--------------------------|--------|-----------|--------------------------------------------------------| +| `QuestDB.Enabled` | bool | `false` | Must be set explicitly | +| `QuestDB.Host` | string | | Hostname or IP | +| `QuestDB.Port` | int | 9009 | ILP TCP port | +| `QuestDB.User` | string | | | +| `QuestDB.Password` | string | | | +| `QuestDB.MaxBufferBytes` | int | `4194304` | Write buffer held during an outage; `0` disables it | + +The write buffer holds rows that QuestDB has not confirmed, so an outage does not lose data. +See [observability.md - QuestDB connection loss](./observability.md#questdb-connection-loss) +for what happens when it fills up. ### PostgreSQL diff --git a/documentation/observability.md b/documentation/observability.md index 0d773de..b0c13c1 100644 --- a/documentation/observability.md +++ b/documentation/observability.md @@ -27,10 +27,29 @@ any sink fails, the endpoint returns `503 Service Unavailable` with details in t ### QuestDB connection loss QuestDB ILP runs over a single long-lived TCP connection. When the server closes it (restart, upgrade, host -reboot) the sink closes the dead socket and redials with exponential backoff, from 1s up to 60s. Rows handed -to the sink while the connection is down are dropped and counted; the count and the total downtime are -logged on the successful reconnect. Because a flush with an empty buffer writes no bytes, it cannot detect a -closed peer, so the loss surfaces on the first row written after the server went away. +reboot) the sink closes the dead socket and redials with exponential backoff, from 1s up to 60s. + +Rows are held in memory until a flush confirms them, so an outage does not lose the readings taken during +it. On reconnect the held rows are replayed in order and the downtime is logged. `QuestDB.MaxBufferBytes` +caps the buffer, 4 MiB by default. Past the cap the oldest rows are evicted, the drop is logged, and the +store call starts returning errors, which the service escalates to a process exit after five in a row. The +pod then crash-loops until QuestDB is reachable again. Set `MaxBufferBytes` to 0 to drop rows during an +outage instead of holding them. + +| State | `/readyz` | `/healthz` | Result | +|-------------------------------|-----------|------------|-------------------------------------------| +| Connected | 200 | 200 | Normal | +| Disconnected, buffer has room | 503 | 200 | Pod keeps running, readings held in memory | +| Disconnected, buffer full | 503 | 503 | Pod restarts, readings dropped | + +A buffering sink deliberately keeps `/healthz` green. Restarting the pod would throw away the buffer that is +holding the data, so liveness leaves it alone until the buffer overflows. + +Two limits are worth knowing. A flush with an empty buffer writes no bytes, so it cannot detect a closed +peer; the loss surfaces on the next flush that actually carries rows. And ILP over TCP has no server +acknowledgement, so the last flush before the socket error is reported as successful even though QuestDB +never stored it. Those rows are not recoverable. The buffer protects everything from the first reported +failure onwards. ### Liveness detail diff --git a/documentation/troubleshooting.md b/documentation/troubleshooting.md index 710b9f9..69dd65a 100644 --- a/documentation/troubleshooting.md +++ b/documentation/troubleshooting.md @@ -156,10 +156,15 @@ QuestDB uses ILP/TCP and reports failures through `/readyz` and the write-error below. If QuestDB restarts while meterlogger is running, expect one `questdb: connection lost, will reconnect` -error followed by `questdb: reconnecting` and `questdb: reconnected` with the downtime and the number of -rows dropped in the gap. Repeated `questdb: reconnect failed` lines mean the server is still unreachable; -the retry interval grows to a maximum of 60s. A continuous stream of `broken pipe` write errors on the same -source port is the pre-1.5.3 behaviour and means the pod is running an old image. +error followed by `questdb: reconnecting`, then `questdb: reconnected` with the downtime and +`questdb: replayed buffered rows` with the number of readings recovered from the write buffer. Repeated +`questdb: reconnect failed` lines mean the server is still unreachable; the retry interval grows to a +maximum of 60s. A continuous stream of `broken pipe` write errors on the same source port is the pre-1.5.3 +behaviour and means the pod is running an old image. + +`questdb: write buffer full, dropping oldest rows` means the outage outlasted `QuestDB.MaxBufferBytes` of +readings. From that point data is being lost and the pod will restart itself. Raise `MaxBufferBytes` if the +deployment needs to ride out longer outages, keeping in mind that it is process memory. --- diff --git a/internal/adapters/sink/qdb/buffer.go b/internal/adapters/sink/qdb/buffer.go new file mode 100644 index 0000000..44e8ec2 --- /dev/null +++ b/internal/adapters/sink/qdb/buffer.go @@ -0,0 +1,174 @@ +package qdb + +import ( + "context" + "math/big" + "time" + + qdbclient "github.com/questdb/go-questdb-client/v3" +) + +// RowBuilder writes one ILP row into sender. It receives its own context +// because a buffered row is replayed long after the store call that produced +// it has returned and that call's timeout has expired. +type RowBuilder func(ctx context.Context, sender qdbclient.LineSender) error + +// bufferedRow is one row waiting for the connection to come back, together +// with the estimated size of its ILP encoding. +type bufferedRow struct { + build RowBuilder + size int +} + +// rowBuffer holds rows written while the ILP connection is down, oldest first, +// under a byte cap. Once the cap is reached the oldest rows are evicted: the +// newest data is the most useful, and an unbounded buffer would trade a data +// gap for an OOM kill. +type rowBuffer struct { + maxBytes int + + rows []bufferedRow + bytes int + dropped int64 +} + +func newRowBuffer(maxBytes int) *rowBuffer { + return &rowBuffer{maxBytes: maxBytes} +} + +// enabled reports whether rows are buffered at all. A zero cap restores the +// drop-on-disconnect behaviour. +func (b *rowBuffer) enabled() bool { return b.maxBytes > 0 } + +// add appends a row, evicting the oldest rows if it does not fit. It reports +// whether the row was stored without evicting anything. +func (b *rowBuffer) add(row bufferedRow) bool { + if !b.enabled() || row.size > b.maxBytes { + b.dropped++ + return false + } + + evicted := false + for b.bytes+row.size > b.maxBytes && len(b.rows) > 0 { + b.bytes -= b.rows[0].size + b.rows = b.rows[1:] + b.dropped++ + evicted = true + } + + b.rows = append(b.rows, row) + b.bytes += row.size + return !evicted +} + +// take removes and returns up to n rows from the front. +func (b *rowBuffer) take(n int) []bufferedRow { + if n > len(b.rows) { + n = len(b.rows) + } + batch := b.rows[:n] + for _, row := range batch { + b.bytes -= row.size + } + // Copy so the retained tail does not keep the batch alive through the + // shared backing array. + b.rows = append([]bufferedRow(nil), b.rows[n:]...) + return batch +} + +// pushFront puts an unsent batch back at the head, keeping insertion order. +func (b *rowBuffer) pushFront(batch []bufferedRow) { + for _, row := range batch { + b.bytes += row.size + } + b.rows = append(batch, b.rows...) +} + +func (b *rowBuffer) len() int { return len(b.rows) } + +// reset clears the buffer and the drop count after a successful drain. +func (b *rowBuffer) reset() { + b.rows = nil + b.bytes = 0 + b.dropped = 0 +} + +// Estimated ILP encoding sizes. Symbols and strings are measured exactly; +// numeric values use the widest formatting they can produce, so the estimate +// errs towards over-counting and the memory cap is never exceeded in practice. +const ( + sizeSeparators = 3 // table/symbol-set/column-set/timestamp separators plus newline + sizeInt64Value = 21 // -9223372036854775808 plus the 'i' suffix + sizeFloat64Value = 24 // widest strconv 'G' rendering + sizeBoolValue = 1 + sizeTimestampVal = 21 // microseconds since epoch plus the 't' suffix + sizeLong256Value = 66 // "0x" plus 64 hex digits + sizeStringQuotes = 2 + sizeFieldName = 2 // '=' and the ',' that separates fields +) + +// sizingSender implements qdbclient.LineSender and measures the ILP encoding +// of a row without sending anything. The client does not expose the encoded +// length of a message, so the buffer estimates it here. +type sizingSender struct { + size int +} + +func (s *sizingSender) Table(name string) qdbclient.LineSender { + s.size += len(name) + sizeSeparators + return s +} + +func (s *sizingSender) Symbol(name, val string) qdbclient.LineSender { + s.size += len(name) + len(val) + sizeFieldName + return s +} + +func (s *sizingSender) Int64Column(name string, _ int64) qdbclient.LineSender { + s.size += len(name) + sizeInt64Value + sizeFieldName + return s +} + +func (s *sizingSender) Long256Column(name string, _ *big.Int) qdbclient.LineSender { + s.size += len(name) + sizeLong256Value + sizeFieldName + return s +} + +func (s *sizingSender) TimestampColumn(name string, _ time.Time) qdbclient.LineSender { + s.size += len(name) + sizeTimestampVal + sizeFieldName + return s +} + +func (s *sizingSender) Float64Column(name string, _ float64) qdbclient.LineSender { + s.size += len(name) + sizeFloat64Value + sizeFieldName + return s +} + +func (s *sizingSender) StringColumn(name, val string) qdbclient.LineSender { + s.size += len(name) + len(val) + sizeStringQuotes + sizeFieldName + return s +} + +func (s *sizingSender) BoolColumn(name string, _ bool) qdbclient.LineSender { + s.size += len(name) + sizeBoolValue + sizeFieldName + return s +} + +func (s *sizingSender) At(_ context.Context, _ time.Time) error { + s.size += sizeTimestampVal + return nil +} + +func (s *sizingSender) AtNow(_ context.Context) error { return nil } +func (s *sizingSender) Flush(_ context.Context) error { return nil } +func (s *sizingSender) Close(_ context.Context) error { return nil } + +// measure runs build against a sizing sender and reports the estimated ILP +// size of the row it writes. +func measure(ctx context.Context, build RowBuilder) (int, error) { + s := &sizingSender{} + if err := build(ctx, s); err != nil { + return 0, err + } + return s.size, nil +} diff --git a/internal/adapters/sink/qdb/buffer_test.go b/internal/adapters/sink/qdb/buffer_test.go new file mode 100644 index 0000000..27b5b0f --- /dev/null +++ b/internal/adapters/sink/qdb/buffer_test.go @@ -0,0 +1,302 @@ +package qdb + +import ( + "context" + "errors" + "testing" + "time" + + qdbclient "github.com/questdb/go-questdb-client/v3" +) + +// rowBuilder returns a builder that writes one small, recognisable row. +func rowBuilder(power int64) RowBuilder { + return func(ctx context.Context, sender qdbclient.LineSender) error { + return sender.Table("heat").Int64Column("power", power).At(ctx, time.Unix(power, 0)) + } +} + +// rowSize is the measured size of a rowBuilder row, so tests can express a cap +// in rows without hardcoding the estimator's arithmetic. +func rowSize(t *testing.T) int { + t.Helper() + size, err := measure(t.Context(), rowBuilder(1)) + if err != nil { + t.Fatalf("measure: %v", err) + } + if size <= 0 { + t.Fatalf("measured size = %d, want a positive estimate", size) + } + return size +} + +func TestSizingSender_GrowsWithRowContent(t *testing.T) { + small, err := measure(t.Context(), func(ctx context.Context, s qdbclient.LineSender) error { + return s.Table("t").Symbol("a", "b").At(ctx, time.Unix(0, 0)) + }) + if err != nil { + t.Fatalf("measure small: %v", err) + } + large, err := measure(t.Context(), func(ctx context.Context, s qdbclient.LineSender) error { + return s.Table("t"). + Symbol("a", "b"). + StringColumn("long", "a considerably longer value than the small row carries"). + Float64Column("f", 1.5). + BoolColumn("ok", true). + At(ctx, time.Unix(0, 0)) + }) + if err != nil { + t.Fatalf("measure large: %v", err) + } + if large <= small { + t.Errorf("large row measured %d bytes, small row %d; want the estimate to grow", large, small) + } +} + +// A row whose builder fails is a data problem, not a connection problem, so it +// must not be buffered for a replay that would fail the same way. +func TestBufferRow_BuilderErrorIsNotBuffered(t *testing.T) { + want := errors.New("bad row") + c := newTestDBClientBuffered(&mockLineSender{}, 1<<20) + c.now = func() time.Time { return time.Unix(0, 0) } + + err := c.Write(t.Context(), func(context.Context, qdbclient.LineSender) error { return want }) + + c.senderMu.Lock() + buffered := c.buffer.len() + c.senderMu.Unlock() + + if !errors.Is(err, want) { + t.Errorf("Write() with a failing builder = %v, want %v", err, want) + } + if buffered != 0 { + t.Errorf("buffered %d rows, want 0", buffered) + } +} + +// The point of the buffer: rows written during an outage reach QuestDB once it +// comes back, in the order they were produced. +func TestDBClient_BuffersAndReplaysAcrossAnOutage(t *testing.T) { + dead := &mockLineSender{atErr: errors.New("write: broken pipe")} + fresh := &mockLineSender{} + now := time.Unix(0, 0) + + c := newTestDBClientBuffered(dead, 1<<20) + c.now = func() time.Time { return now } + c.dial = func(context.Context, Config) (qdbclient.LineSender, error) { return fresh, nil } + + // The write that discovers the dead socket is itself buffered, not lost. + if err := c.Write(t.Context(), rowBuilder(1)); err != nil { + t.Fatalf("Write() that hit the dead connection = %v, want nil (buffered)", err) + } + for _, power := range []int64{2, 3} { + if err := c.Write(t.Context(), rowBuilder(power)); err != nil { + t.Fatalf("Write() while disconnected = %v, want nil (buffered)", err) + } + } + + c.senderMu.Lock() + buffered := c.buffer.len() + c.senderMu.Unlock() + if buffered != 3 { + t.Fatalf("buffered %d rows, want 3", buffered) + } + if !c.Degraded() { + t.Error("Degraded() while buffering = false, want true") + } + + now = now.Add(maxReconnectDelay) + if err := c.Flush(t.Context()); err != nil { + t.Fatalf("Flush() after the backoff window = %v, want nil", err) + } + + rows := requireRows(t, fresh, 3) + for i, want := range []int64{1, 2, 3} { + if got := rows[i].columns["power"]; got != want { + t.Errorf("replayed row %d power = %v, want %d", i, got, want) + } + } + c.senderMu.Lock() + remaining := c.buffer.len() + c.senderMu.Unlock() + if remaining != 0 { + t.Errorf("%d rows left buffered after the replay, want 0", remaining) + } + if c.Degraded() { + t.Error("Degraded() after the replay = true, want false") + } + if err := c.Check(t.Context()); err != nil { + t.Errorf("Check() after the replay = %v, want nil", err) + } +} + +// The buffer must not grow without limit. Past the cap the oldest rows go, the +// caller starts seeing errors, and the sink stops asking liveness for mercy so +// the pod is restarted. +func TestDBClient_BufferOverflowDropsOldestAndStopsShieldingLiveness(t *testing.T) { + size := rowSize(t) + const capRows = 3 + + dead := &mockLineSender{atErr: errors.New("write: broken pipe")} + fresh := &mockLineSender{} + now := time.Unix(0, 0) + + c := newTestDBClientBuffered(dead, size*capRows) + c.now = func() time.Time { return now } + c.dial = func(context.Context, Config) (qdbclient.LineSender, error) { return fresh, nil } + + for power := int64(1); power <= capRows; power++ { + if err := c.Write(t.Context(), rowBuilder(power)); err != nil { + t.Fatalf("Write() of row %d = %v, want nil (buffered)", power, err) + } + } + if !c.Degraded() { + t.Error("Degraded() with a buffer that still has room = false, want true") + } + + // One row past the cap. + err := c.Write(t.Context(), rowBuilder(capRows+1)) + if err == nil { + t.Fatal("Write() past the buffer cap = nil, want an error so the service can escalate") + } + if c.Degraded() { + t.Error("Degraded() once rows are being dropped = true, want false so liveness restarts the pod") + } + + c.senderMu.Lock() + buffered, dropped, held := c.buffer.len(), c.buffer.dropped, c.buffer.bytes + c.senderMu.Unlock() + if buffered != capRows { + t.Errorf("buffered %d rows, want the cap of %d", buffered, capRows) + } + if dropped != 1 { + t.Errorf("dropped %d rows, want 1", dropped) + } + if held > size*capRows { + t.Errorf("holding %d bytes, want at most %d", held, size*capRows) + } + + // What survives is the newest data, oldest evicted. + now = now.Add(maxReconnectDelay) + if flushErr := c.Flush(t.Context()); flushErr != nil { + t.Fatalf("Flush() after the backoff window = %v, want nil", flushErr) + } + rows := requireRows(t, fresh, capRows) + for i, want := range []int64{2, 3, 4} { + if got := rows[i].columns["power"]; got != want { + t.Errorf("replayed row %d power = %v, want %d", i, got, want) + } + } +} + +// With buffering off the sink keeps the pre-1.6.0 behaviour: rows written +// during an outage are dropped and the caller is told immediately. +func TestDBClient_BufferingDisabledDropsRows(t *testing.T) { + dead := &mockLineSender{atErr: errors.New("write: broken pipe")} + c := newTestDBClientWith(dead) + c.now = func() time.Time { return time.Unix(0, 0) } + + if err := c.Write(t.Context(), rowBuilder(1)); err == nil { + t.Fatal("Write() with buffering disabled = nil, want an error") + } + if err := c.Write(t.Context(), rowBuilder(2)); !errors.Is(err, ErrDisconnected) { + t.Fatalf("Write() with buffering disabled = %v, want ErrDisconnected", err) + } + if c.Degraded() { + t.Error("Degraded() with buffering disabled = true, want false") + } + c.senderMu.Lock() + buffered := c.buffer.len() + c.senderMu.Unlock() + if buffered != 0 { + t.Errorf("buffered %d rows with buffering disabled, want 0", buffered) + } +} + +// A replay that fails partway keeps the rows it never handed over, so a second +// outage during recovery does not throw away everything still in hand. +func TestDBClient_ReplayFailureKeepsUnsentRows(t *testing.T) { + dead := &mockLineSender{atErr: errors.New("write: broken pipe")} + now := time.Unix(0, 0) + + c := newTestDBClientBuffered(dead, 1<<20) + c.now = func() time.Time { return now } + + const buffered = 4 + for power := int64(1); power <= buffered; power++ { + if err := c.Write(t.Context(), rowBuilder(power)); err != nil { + t.Fatalf("Write() of row %d = %v, want nil (buffered)", power, err) + } + } + + // The redial succeeds but the connection dies again on the replay flush. + stillBroken := &mockLineSender{flushErr: errors.New("write: broken pipe")} + c.dial = func(context.Context, Config) (qdbclient.LineSender, error) { return stillBroken, nil } + now = now.Add(maxReconnectDelay) + + if err := c.Flush(t.Context()); err == nil { + t.Fatal("Flush() with a replay that fails = nil, want an error") + } + + c.senderMu.Lock() + remaining, dropped := c.buffer.len(), c.buffer.dropped + sender := c.sender + c.senderMu.Unlock() + + // One chunk was handed over and lost with the failed flush. There are + // fewer rows than a chunk here, so all of them went in one batch. + if remaining != 0 { + t.Errorf("%d rows kept after the whole batch was handed over, want 0", remaining) + } + if dropped != buffered { + t.Errorf("dropped %d rows, want %d", dropped, buffered) + } + if sender != nil { + t.Error("sender kept after a failed replay, want it torn down for another redial") + } +} + +func TestRowBuffer_TakeAndPushFrontPreserveOrderAndAccounting(t *testing.T) { + b := newRowBuffer(1000) + for i := range 5 { + if !b.add(bufferedRow{build: rowBuilder(int64(i)), size: 10}) { + t.Fatalf("add(%d) evicted, want it to fit", i) + } + } + if b.bytes != 50 { + t.Fatalf("bytes = %d, want 50", b.bytes) + } + + batch := b.take(2) + if len(batch) != 2 || b.len() != 3 || b.bytes != 30 { + t.Fatalf("after take(2): batch=%d left=%d bytes=%d, want 2/3/30", len(batch), b.len(), b.bytes) + } + + b.pushFront(batch) + if b.len() != 5 || b.bytes != 50 { + t.Fatalf("after pushFront: left=%d bytes=%d, want 5/50", b.len(), b.bytes) + } + + // take never asks for more than it holds. + if got := len(b.take(99)); got != 5 { + t.Errorf("take(99) returned %d rows, want 5", got) + } + if b.bytes != 0 { + t.Errorf("bytes = %d after draining, want 0", b.bytes) + } +} + +// A single row larger than the whole cap can never be stored, and must be +// reported rather than silently discarded. +func TestRowBuffer_RowLargerThanCapIsRejected(t *testing.T) { + b := newRowBuffer(10) + if b.add(bufferedRow{build: rowBuilder(1), size: 11}) { + t.Error("add() of a row larger than the cap reported success, want failure") + } + if b.len() != 0 || b.bytes != 0 { + t.Errorf("buffer holds %d rows / %d bytes, want empty", b.len(), b.bytes) + } + if b.dropped != 1 { + t.Errorf("dropped = %d, want 1", b.dropped) + } +} diff --git a/internal/adapters/sink/qdb/checker_test.go b/internal/adapters/sink/qdb/checker_test.go index 9f9d9d0..0d8a78b 100644 --- a/internal/adapters/sink/qdb/checker_test.go +++ b/internal/adapters/sink/qdb/checker_test.go @@ -122,8 +122,8 @@ func TestDBClient_Write_RedialsAfterConnectionLoss(t *testing.T) { } writeRow := func() error { - return c.Write(t.Context(), func(sender qdbclient.LineSender) error { - return sender.Table("t").Symbol("s", "v").At(t.Context(), now) + return c.Write(t.Context(), func(ctx context.Context, sender qdbclient.LineSender) error { + return sender.Table("t").Symbol("s", "v").At(ctx, now) }) } @@ -142,10 +142,10 @@ func TestDBClient_Write_RedialsAfterConnectionLoss(t *testing.T) { t.Fatalf("dialled %d times during the backoff window, want 0", dials) } c.senderMu.Lock() - dropped := c.droppedRows + dropped := c.buffer.dropped c.senderMu.Unlock() - if dropped != 1 { - t.Errorf("droppedRows = %d, want 1", dropped) + if dropped != 2 { + t.Errorf("dropped rows = %d, want 2 (buffering is off in this test)", dropped) } now = now.Add(initialReconnectDelay) diff --git a/internal/adapters/sink/qdb/common.go b/internal/adapters/sink/qdb/common.go index 0e3f812..b444322 100644 --- a/internal/adapters/sink/qdb/common.go +++ b/internal/adapters/sink/qdb/common.go @@ -33,11 +33,16 @@ const ( // before Check reports unhealthy. One failed flush followed by a successful // redial is normal during a QuestDB restart and must not flap /readyz. unhealthyAfterFailures = 5 + + // replayChunkRows is how many buffered rows are flushed per batch on + // reconnect. The ILP client discards its buffer when a flush fails, so a + // failed chunk is lost; chunking bounds that loss. + replayChunkRows = 1000 ) // ErrDisconnected is returned by Write and Flush while the ILP connection is -// down and the next redial is not due yet. Rows handed to Write in that window -// are dropped. +// down and the next redial is not due yet. When buffering is enabled, Write +// returns it only once the buffer has overflowed and rows are being dropped. var ErrDisconnected = errors.New("questdb: ILP connection is down") // DBClient wraps a QuestDB line sender and owns its reconnect state. @@ -63,13 +68,14 @@ type DBClient struct { sender qdbclient.LineSender nextDialAt time.Time reconnectDelay time.Duration - droppedRows int64 downSince time.Time + buffer *rowBuffer // stateMu guards the health fields only. stateMu sync.RWMutex consecutiveFailures int lastErr error + buffering bool } // Config holds the connection parameters for a QuestDB ILP client. @@ -78,6 +84,11 @@ type Config struct { Port int User string Password string + + // MaxBufferBytes caps the estimated ILP payload held in memory while the + // connection is down. Zero disables buffering, and rows written during an + // outage are dropped. + MaxBufferBytes int } // NewDBClient opens a persistent ILP/TCP line sender to QuestDB. @@ -94,6 +105,7 @@ func NewDBClient(ctx context.Context, cfg Config, logger *slog.Logger) (*DBClien now: time.Now, sender: sender, reconnectDelay: initialReconnectDelay, + buffer: newRowBuffer(cfg.MaxBufferBytes), }, nil } @@ -128,30 +140,99 @@ func (c *DBClient) Check(_ context.Context) error { if c.consecutiveFailures < unhealthyAfterFailures { return nil } + if c.buffering { + return fmt.Errorf("%d consecutive QuestDB failures, buffering rows: %w", c.consecutiveFailures, c.lastErr) + } return fmt.Errorf("%d consecutive QuestDB failures: %w", c.consecutiveFailures, c.lastErr) } +// Degraded implements healthserver.Degrader. While rows are being buffered the +// sink is unhealthy but recovering in place, and restarting the process would +// throw away exactly the data the buffer exists to protect. Once the buffer +// overflows there is nothing left to protect and the restart is allowed. +func (c *DBClient) Degraded() bool { + c.stateMu.RLock() + defer c.stateMu.RUnlock() + return c.buffering +} + +func (c *DBClient) setBuffering(buffering bool) { + c.stateMu.Lock() + defer c.stateMu.Unlock() + c.buffering = buffering +} + // Write runs build against the current line sender to buffer one row. It is the // only way writers may reach the sender: DBClient swaps the sender on reconnect // and serialises access, which a cached reference would defeat. // -// While the connection is down and the next redial is not due, the row is -// dropped and ErrDisconnected is returned. -func (c *DBClient) Write(ctx context.Context, build func(sender qdbclient.LineSender) error) error { +// While the connection is down the row is held in memory and replayed on the +// next successful redial. Write returns an error only once the buffer has +// overflowed and rows are being dropped, which lets the service escalate to a +// process restart instead of quietly losing data. +func (c *DBClient) Write(ctx context.Context, build RowBuilder) error { c.senderMu.Lock() defer c.senderMu.Unlock() - sender, err := c.ensureSender(ctx) + size, err := measure(ctx, build) if err != nil { - c.droppedRows++ + // The builder rejected the row, so replaying it would fail the same + // way. This is a data error and never touches the connection. return err } + row := bufferedRow{build: build, size: size} - if buildErr := build(sender); buildErr != nil { + sender, connErr := c.ensureSender(ctx) + if connErr != nil { + return c.hold(ctx, row, connErr) + } + + if buildErr := build(ctx, sender); buildErr != nil { c.connectionFailed(ctx, buildErr) - return buildErr + return c.hold(ctx, row, buildErr) } - return nil + return c.hold(ctx, row, nil) +} + +// hold keeps a row until a flush confirms it reached QuestDB. +// +// Rows handed to the ILP client sit in its buffer until the next flush, and it +// discards that buffer when a flush fails. So a row that Write accepted is not +// safe yet: the outage is usually discovered by the flush, after the rows it +// would have carried are already gone. Holding every row until the flush +// succeeds is what makes those rows replayable. +// +// connErr is non-nil when the row never reached a live sender. Callers must +// hold senderMu. +func (c *DBClient) hold(ctx context.Context, row bufferedRow, connErr error) error { + if !c.buffer.enabled() { + if connErr != nil { + c.buffer.dropped++ + } + return connErr + } + + if c.buffer.add(row) { + if connErr != nil { + c.setBuffering(true) + } + return nil + } + + // Evicting while connected only costs the replay copy of a row the sender + // already has. Evicting while disconnected loses the row itself. + if connErr == nil { + return nil + } + + c.logger.WarnContext( + ctx, + "questdb: write buffer full, dropping oldest rows", + slog.Int("max_bytes", c.buffer.maxBytes), + slog.Int64("dropped_rows", c.buffer.dropped), + ) + c.setBuffering(false) + return fmt.Errorf("questdb: write buffer full after %d dropped rows: %w", c.buffer.dropped, connErr) } // Flush flushes the underlying line sender. On failure the connection is @@ -167,9 +248,12 @@ func (c *DBClient) Flush(ctx context.Context) error { } if flushErr := sender.Flush(ctx); flushErr != nil { + // Everything still held is unconfirmed and stays queued for the replay. c.connectionFailed(ctx, flushErr) return flushErr } + // The rows are in QuestDB, so the replay copies can go. + c.buffer.reset() c.recordSuccess() return nil } @@ -207,14 +291,65 @@ func (c *DBClient) ensureSender(ctx context.Context) (qdbclient.LineSender, erro ctx, "questdb: reconnected", slog.Duration("down_for", now.Sub(c.downSince)), - slog.Int64("dropped_rows", c.droppedRows), + slog.Int("buffered_rows", c.buffer.len()), + slog.Int64("dropped_rows", c.buffer.dropped), ) c.sender = sender - c.droppedRows = 0 c.reconnectDelay = initialReconnectDelay c.nextDialAt = time.Time{} c.recordSuccess() - return sender, nil + + if replayErr := c.replay(ctx); replayErr != nil { + return nil, replayErr + } + c.setBuffering(false) + return c.sender, nil +} + +// replay writes the buffered rows to the fresh connection, oldest first, and +// flushes every chunk. Callers must hold senderMu. +func (c *DBClient) replay(ctx context.Context) error { + replayed := 0 + for c.buffer.len() > 0 { + batch := c.buffer.take(replayChunkRows) + sent, err := c.replayBatch(ctx, batch) + if err != nil { + // Rows before sent were handed to the ILP client, which discards + // its buffer on a failed flush, so they are gone. Keep the rest. + c.buffer.pushFront(batch[sent:]) + c.buffer.dropped += int64(sent) + c.connectionFailed(ctx, err) + c.logger.ErrorContext( + ctx, + "questdb: replay failed", + slog.Any("error", err), + slog.Int("replayed_rows", replayed), + slog.Int("remaining_rows", c.buffer.len()), + ) + return err + } + replayed += sent + } + + if replayed > 0 { + c.logger.InfoContext(ctx, "questdb: replayed buffered rows", slog.Int("replayed_rows", replayed)) + } + c.buffer.reset() + return nil +} + +// replayBatch writes one chunk and flushes it. It returns how many rows were +// handed to the ILP client, which are lost if the error came from the flush. +func (c *DBClient) replayBatch(ctx context.Context, batch []bufferedRow) (int, error) { + for i, row := range batch { + if err := row.build(ctx, c.sender); err != nil { + return i, err + } + } + if err := c.sender.Flush(ctx); err != nil { + return len(batch), err + } + return len(batch), nil } // connectionFailed tears down the dead sender and schedules a redial. Callers @@ -270,6 +405,17 @@ func (c *DBClient) Close() { c.senderMu.Lock() defer c.senderMu.Unlock() + + // Buffered rows only live in memory. Shutting down with rows still held is + // a real data loss and has to be visible in the logs, not silent. + if remaining := c.buffer.len(); remaining > 0 { + c.logger.Error( + "questdb: discarding buffered rows on shutdown", + slog.Int("rows", remaining), + slog.Int("bytes", c.buffer.bytes), + ) + } + if c.sender == nil { return } diff --git a/internal/adapters/sink/qdb/qdb_gas_writer.go b/internal/adapters/sink/qdb/qdb_gas_writer.go index e69d8ac..932a027 100644 --- a/internal/adapters/sink/qdb/qdb_gas_writer.go +++ b/internal/adapters/sink/qdb/qdb_gas_writer.go @@ -33,7 +33,7 @@ func (w *QuestDBGasWriter) StoreGasReading(ctx context.Context, r domain.GasRead slog.Float64("reading_m3", r.ReadingM3), slog.Time("captured_at", r.CapturedAt), ) - return w.client.Write(ctx, func(sender qdbclient.LineSender) error { + return w.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return sender. Table(w.measurement). Symbol("serial_no", r.SerialNo). diff --git a/internal/adapters/sink/qdb/qdb_grid_writer.go b/internal/adapters/sink/qdb/qdb_grid_writer.go index cac2589..b0efa75 100644 --- a/internal/adapters/sink/qdb/qdb_grid_writer.go +++ b/internal/adapters/sink/qdb/qdb_grid_writer.go @@ -18,7 +18,7 @@ type GridStore struct { func (w *GridStore) StoreGridTelegram(ctx context.Context, telegram domain.GridTelegram) error { w.logger.DebugContext(ctx, "qdb: buffering grid telegram", debuglog.GridAttrs(telegram)) - return w.client.Write(ctx, func(sender qdbclient.LineSender) error { + return w.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return w.buildRow(ctx, sender, telegram) }) } diff --git a/internal/adapters/sink/qdb/qdb_heat_writer.go b/internal/adapters/sink/qdb/qdb_heat_writer.go index 4f49e31..e0c5c63 100644 --- a/internal/adapters/sink/qdb/qdb_heat_writer.go +++ b/internal/adapters/sink/qdb/qdb_heat_writer.go @@ -40,7 +40,7 @@ func NewQuestDBHeatTelegramWriter( func (store *HeatTelegramStore) StoreHeatTelegram(ctx context.Context, telegram domain.HeatTelegram) error { store.logger.DebugContext(ctx, "qdb: buffering heat telegram", debuglog.HeatAttrs(telegram)) - return store.client.Write(ctx, func(sender qdbclient.LineSender) error { + return store.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return sender.Table(store.table). Symbol("device", fmt.Sprintf("Multical %s", telegram.MeterID)). Symbol("serial", telegram.SerialNo). diff --git a/internal/adapters/sink/qdb/qdb_solar_writer.go b/internal/adapters/sink/qdb/qdb_solar_writer.go index bb8d3bb..b4e8711 100644 --- a/internal/adapters/sink/qdb/qdb_solar_writer.go +++ b/internal/adapters/sink/qdb/qdb_solar_writer.go @@ -17,7 +17,7 @@ type SolarWriter struct { } func (w *SolarWriter) StoreEnvoySolarData(ctx context.Context, data domain.EnvoySolarData) error { - err := w.client.Write(ctx, func(sender qdbclient.LineSender) error { + err := w.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return sender.Table(w.table). Symbol("EnvoySerialNumber", data.EnvoySerial). Float64Column("ProductionWattHours", data.ProductionWh). @@ -39,7 +39,7 @@ func (w *SolarWriter) StoreEnvoySolarData(ctx context.Context, data domain.Envoy } func (w *SolarWriter) storeInverter(ctx context.Context, envoySerial string, inverter domain.InverterDetails) error { - return w.client.Write(ctx, func(sender qdbclient.LineSender) error { + return w.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return sender.Table(w.table+"_inverters"). Symbol("InverterSerialNumber", inverter.SerialNumber). StringColumn("EnvoySerialNumber", envoySerial). diff --git a/internal/adapters/sink/qdb/qdb_test.go b/internal/adapters/sink/qdb/qdb_test.go index f75b66e..271aa99 100644 --- a/internal/adapters/sink/qdb/qdb_test.go +++ b/internal/adapters/sink/qdb/qdb_test.go @@ -103,14 +103,21 @@ func newTestDBClient() (*DBClient, *mockLineSender) { // newTestDBClientWith builds a client around a given sender with the reconnect // machinery wired to fail loudly: the default dial returns an error, so a test // that unexpectedly loses its connection does not silently get a fresh one. +// Buffering is off; tests that exercise it call newTestDBClientBuffered. func newTestDBClientWith(sender qdbclient.LineSender) *DBClient { + return newTestDBClientBuffered(sender, 0) +} + +func newTestDBClientBuffered(sender qdbclient.LineSender, maxBufferBytes int) *DBClient { + cfg := Config{Host: "questdb.test", Port: 9009, MaxBufferBytes: maxBufferBytes} return &DBClient{ - cfg: Config{Host: "questdb.test", Port: 9009}, + cfg: cfg, logger: testLogger(), now: time.Now, dial: func(context.Context, Config) (qdbclient.LineSender, error) { return nil, errNoDial }, sender: sender, reconnectDelay: initialReconnectDelay, + buffer: newRowBuffer(maxBufferBytes), } } diff --git a/internal/adapters/sink/qdb/qdb_thermal_writer.go b/internal/adapters/sink/qdb/qdb_thermal_writer.go index ca532f3..9344f50 100644 --- a/internal/adapters/sink/qdb/qdb_thermal_writer.go +++ b/internal/adapters/sink/qdb/qdb_thermal_writer.go @@ -33,7 +33,7 @@ func (w *QuestDBThermalWriter) StoreThermalReading(ctx context.Context, r domain slog.Float64("reading_gj", r.ReadingGJ), slog.Time("captured_at", r.CapturedAt), ) - return w.client.Write(ctx, func(sender qdbclient.LineSender) error { + return w.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return sender. Table(w.measurement). Symbol("serial_no", r.SerialNo). diff --git a/internal/adapters/sink/qdb/qdb_ventilation_writer.go b/internal/adapters/sink/qdb/qdb_ventilation_writer.go index 26536ca..353551c 100644 --- a/internal/adapters/sink/qdb/qdb_ventilation_writer.go +++ b/internal/adapters/sink/qdb/qdb_ventilation_writer.go @@ -30,7 +30,7 @@ func NewDucoQuestDBRepository( } func (repo *DucoQuestDBRepository) StoreBoxStatus(ctx context.Context, boxStatus domain.DucoBoxStatus) error { - return repo.client.Write(ctx, func(sender qdbclient.LineSender) error { + return repo.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return repo.buildBoxRow(ctx, sender, boxStatus) }) } @@ -82,7 +82,7 @@ func (repo *DucoQuestDBRepository) buildBoxRow( } func (repo *DucoQuestDBRepository) StoreNodeData(ctx context.Context, nodeData domain.DucoNodeStatus) error { - return repo.client.Write(ctx, func(sender qdbclient.LineSender) error { + return repo.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return repo.buildNodeRow(ctx, sender, nodeData) }) } diff --git a/internal/adapters/sink/qdb/qdb_water_writer.go b/internal/adapters/sink/qdb/qdb_water_writer.go index 501cc57..433e14e 100644 --- a/internal/adapters/sink/qdb/qdb_water_writer.go +++ b/internal/adapters/sink/qdb/qdb_water_writer.go @@ -33,7 +33,7 @@ func (w *QuestDBWaterWriter) StoreWaterReading(ctx context.Context, r domain.Wat slog.Float64("reading_m3", r.ReadingM3), slog.Time("captured_at", r.CapturedAt), ) - return w.client.Write(ctx, func(sender qdbclient.LineSender) error { + return w.client.Write(ctx, func(ctx context.Context, sender qdbclient.LineSender) error { return sender. Table(w.measurement). Symbol("serial_no", r.SerialNo). diff --git a/internal/adapters/sink/qdb/reconnect_test.go b/internal/adapters/sink/qdb/reconnect_test.go index 8abbe19..fca8d19 100644 --- a/internal/adapters/sink/qdb/reconnect_test.go +++ b/internal/adapters/sink/qdb/reconnect_test.go @@ -1,6 +1,7 @@ package qdb import ( + "context" "errors" "io" "net" @@ -113,8 +114,8 @@ func TestDBClient_RedialsAfterServerClosesConnection(t *testing.T) { } writeRow := func(value int64) error { - return client.Write(t.Context(), func(sender qdbclient.LineSender) error { - return sender.Table("heat").Int64Column("power", value).At(t.Context(), time.Unix(value, 0)) + return client.Write(t.Context(), func(ctx context.Context, sender qdbclient.LineSender) error { + return sender.Table("heat").Int64Column("power", value).At(ctx, time.Unix(value, 0)) }) } @@ -169,6 +170,83 @@ func TestDBClient_RedialsAfterServerClosesConnection(t *testing.T) { } } +// TestDBClient_ReplaysBufferedRowsOverTheWire is the end-to-end proof that a +// reading taken while QuestDB was down actually reaches QuestDB afterwards. +func TestDBClient_ReplaysBufferedRowsOverTheWire(t *testing.T) { + srv := newILPServer(t) + host, port := srv.hostPort(t) + + client, err := NewDBClient( + t.Context(), + Config{Host: host, Port: port, MaxBufferBytes: 1 << 20}, + testLogger(), + ) + if err != nil { + t.Fatalf("NewDBClient: %v", err) + } + + writeRow := func(value int64) error { + return client.Write(t.Context(), func(ctx context.Context, sender qdbclient.LineSender) error { + return sender.Table("heat").Int64Column("power", value).At(ctx, time.Unix(value, 0)) + }) + } + + now := time.Now() + client.now = func() time.Time { return now } + waitFor(t, func() bool { return srv.connections() == 1 }, "the server to accept the first connection") + closeAcceptedConns(t, srv) + + // Write and flush the way a source does until the flush reports the loss. + // A write to a socket the peer closed can succeed once before the RST + // arrives, and ILP over TCP has no server acknowledgement, so the rows + // carried by that last falsely-successful flush are gone for good. From + // the first reported failure on, every row is held for the replay. + var flushErr error + for value := int64(1); value <= 50 && flushErr == nil; value++ { + if writeErr := writeRow(value); writeErr != nil { + t.Fatalf("write %d during the outage: %v", value, writeErr) + } + flushErr = client.Flush(t.Context()) + time.Sleep(5 * time.Millisecond) + } + if flushErr == nil { + t.Fatal("flushing into a closed connection never failed") + } + + for value := int64(51); value <= 55; value++ { + if writeErr := writeRow(value); writeErr != nil { + t.Fatalf("write %d while disconnected: %v", value, writeErr) + } + } + + client.senderMu.Lock() + buffered := client.buffer.len() + client.senderMu.Unlock() + if buffered == 0 { + t.Fatal("nothing was buffered during the outage") + } + if !client.Degraded() { + t.Error("Degraded() while buffering = false, want true") + } + + // QuestDB comes back. + now = now.Add(maxReconnectDelay) + if recoveryErr := client.Flush(t.Context()); recoveryErr != nil { + t.Fatalf("flush after the outage: %v", recoveryErr) + } + waitFor(t, func() bool { return srv.bytesOn(1) > 0 }, "buffered rows to reach the server") + + client.senderMu.Lock() + remaining := client.buffer.len() + client.senderMu.Unlock() + if remaining != 0 { + t.Errorf("%d rows still buffered after the replay, want 0", remaining) + } + if got := srv.bytesOn(1); got < buffered { + t.Errorf("second connection received %d bytes for %d replayed rows", got, buffered) + } +} + // closeAcceptedConns simulates a QuestDB restart by closing the listener's // accepted connection from the server side. func closeAcceptedConns(t *testing.T, srv *ilpServer) { diff --git a/internal/config/config.go b/internal/config/config.go index 54c5984..00a935d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -191,6 +191,11 @@ type VentilationConfig struct { Nodes []int } +// DefaultQuestDBMaxBufferBytes is the default size of the QuestDB write +// buffer that holds rows while the ILP connection is down. Roughly an hour of +// readings for a typical single-source deployment. +const DefaultQuestDBMaxBufferBytes = 4 << 20 // 4 MiB + // QuestDBConfig configures the QuestDB sink. type QuestDBConfig struct { Enabled bool @@ -198,6 +203,11 @@ type QuestDBConfig struct { Port int User string Password string + + // MaxBufferBytes caps the memory used to hold rows while the ILP + // connection is down. Rows are replayed when it comes back. Set to 0 to + // drop rows during an outage instead of buffering them. + MaxBufferBytes int } // PostgresConfig configures the PostgreSQL sink. diff --git a/internal/config/load.go b/internal/config/load.go index 73f52ed..68f7a8d 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -90,6 +90,7 @@ func setSourceDefaults() { func setSinkDefaults() { viper.SetDefault("QuestDB.Port", 9009) //nolint:mnd // documented default ILP port + viper.SetDefault("QuestDB.MaxBufferBytes", DefaultQuestDBMaxBufferBytes) viper.SetDefault("Postgres.Port", 5432) //nolint:mnd // documented default PostgreSQL port viper.SetDefault("Postgres.SSLMode", "disable") diff --git a/internal/config/load_test.go b/internal/config/load_test.go index 994431f..4a85843 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -56,6 +56,12 @@ func TestLoad_DefaultsNoFile(t *testing.T) { if cfg.HTTPServer.Port != 8080 { t.Errorf("HTTPServer.Port default = %d, want 8080", cfg.HTTPServer.Port) } + if cfg.QuestDB.MaxBufferBytes != DefaultQuestDBMaxBufferBytes { + t.Errorf( + "QuestDB.MaxBufferBytes default = %d, want %d", + cfg.QuestDB.MaxBufferBytes, DefaultQuestDBMaxBufferBytes, + ) + } if cfg.Grid.Gas.Enabled { t.Error("Grid.Gas.Enabled should default to false") } diff --git a/internal/config/validate.go b/internal/config/validate.go index 0228ce3..22159b0 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -46,6 +46,13 @@ func Validate(cfg Config, sourceFilter string) []string { ) } + if cfg.QuestDB.Enabled && cfg.QuestDB.MaxBufferBytes < 0 { + errs = append(errs, fmt.Sprintf( + "QuestDB.MaxBufferBytes is %d; use 0 to disable buffering or a positive byte count", + cfg.QuestDB.MaxBufferBytes, + )) + } + errs = append(errs, sinkFieldErrors(cfg)...) errs = append(errs, sourceFieldErrors(cfg)...) diff --git a/internal/config/validate_test.go b/internal/config/validate_test.go index 346932e..7d7678a 100644 --- a/internal/config/validate_test.go +++ b/internal/config/validate_test.go @@ -53,6 +53,28 @@ func TestValidate_NoSources(t *testing.T) { } } +func TestValidate_NegativeQuestDBBuffer(t *testing.T) { + cfg := Config{ + QuestDB: QuestDBConfig{ + Enabled: true, + Host: testQuestDBHost, + User: testAdminUser, + MaxBufferBytes: -1, + }, + } + + errs := Validate(cfg, "") + if !containsSubstring(errs, "QuestDB.MaxBufferBytes") { + t.Errorf("Validate() = %v, want a MaxBufferBytes error", errs) + } + + cfg.QuestDB.MaxBufferBytes = 0 + disabled := Validate(cfg, "") + if containsSubstring(disabled, "QuestDB.MaxBufferBytes") { + t.Errorf("Validate() rejected 0, which disables buffering: %v", disabled) + } +} + func TestValidate_InvalidSourceFilter(t *testing.T) { cfg := Config{QuestDB: QuestDBConfig{Enabled: true, Host: testQuestDBHost, User: testAdminUser}} diff --git a/internal/healthserver/server.go b/internal/healthserver/server.go index 1a830c3..a82d063 100644 --- a/internal/healthserver/server.go +++ b/internal/healthserver/server.go @@ -36,6 +36,17 @@ type Checker interface { Check(ctx context.Context) error } +// Degrader is an optional Checker extension for components that can be +// unhealthy without being stuck. A checker that reports Degraded is failing, +// so it fails /readyz, but it is recovering in place and /healthz will not +// count it towards a restart no matter how long it stays that way. +// +// The QuestDB sink uses this while it buffers rows through an outage: a +// restart would throw away the buffer that is holding the data. +type Degrader interface { + Degraded() bool +} + // Server is a small HTTP server exposing /healthz, /readyz, and /metrics. // // /readyz reflects the current state of every registered checker. /healthz @@ -43,7 +54,9 @@ type Checker interface { // kubelet does not restart pods on every short outage, but flips to 503 once // any checker has been continuously unhealthy for livenessThreshold. That // turns a stuck Running-but-NotReady pod into a CrashLoopBackOff that the -// orchestrator can act on. +// orchestrator can act on. A checker that implements Degrader is exempt from +// the liveness threshold while it reports Degraded, because restarting it +// would destroy the in-process state it is using to recover. type Server struct { addr string checkers []Checker @@ -139,6 +152,7 @@ func (s *Server) Wait() { type checkResult struct { Name string `json:"name"` Healthy bool `json:"healthy"` + Degraded bool `json:"degraded,omitempty"` Error string `json:"error,omitempty"` FailingFor string `json:"failingFor,omitempty"` @@ -164,6 +178,9 @@ func (s *Server) runChecks(ctx context.Context) []checkResult { if checkErr := s.runCheck(ctx, c); checkErr != nil { res.Healthy = false res.Error = checkErr.Error() + if d, ok := c.(Degrader); ok { + res.Degraded = d.Degraded() + } } results = append(results, res) } @@ -201,7 +218,10 @@ func (s *Server) handleLiveness(w http.ResponseWriter, r *http.Request) { stuck := make([]string, 0) for _, res := range results { - if !res.Healthy && res.failingDur >= s.livenessThreshold { + if res.Healthy || res.Degraded { + continue + } + if res.failingDur >= s.livenessThreshold { stuck = append(stuck, res.Name) } } diff --git a/internal/healthserver/server_test.go b/internal/healthserver/server_test.go index 56bbbdc..a37a6d9 100644 --- a/internal/healthserver/server_test.go +++ b/internal/healthserver/server_test.go @@ -83,9 +83,9 @@ func newGetRequest(t *testing.T, target string) *http.Request { return req } -func newServerWithClock(t *testing.T, threshold time.Duration, now func() time.Time) *healthserver.Server { +func newServerWithClock(t *testing.T, now func() time.Time) *healthserver.Server { t.Helper() - srv := healthserver.New(":0", testLogger(), prometheus.NewRegistry(), threshold) + srv := healthserver.New(":0", testLogger(), prometheus.NewRegistry(), testThreshold) if now != nil { healthserver.SetNow(srv, now) } @@ -122,7 +122,7 @@ func TestLiveness_HealthyChecker(t *testing.T) { func TestLiveness_TransientFailureStaysGreen(t *testing.T) { now := time.Now() clock := func() time.Time { return now } - srv := newServerWithClock(t, testThreshold, clock) + srv := newServerWithClock(t, clock) srv.Register(&unhealthyChecker{name: testCheckerName}) w := httptest.NewRecorder() @@ -138,7 +138,7 @@ func TestLiveness_TransientFailureStaysGreen(t *testing.T) { func TestLiveness_SustainedFailureTrips(t *testing.T) { current := time.Now() clock := func() time.Time { return current } - srv := newServerWithClock(t, testThreshold, clock) + srv := newServerWithClock(t, clock) srv.Register(&unhealthyChecker{name: testCheckerName}) w := httptest.NewRecorder() @@ -162,13 +162,62 @@ func TestLiveness_SustainedFailureTrips(t *testing.T) { } } +// degradingChecker is unhealthy and reports whether it is recovering in place. +type degradingChecker struct { + name string + degraded bool +} + +func (d *degradingChecker) Name() string { return d.name } +func (d *degradingChecker) Check(_ context.Context) error { return errors.New("buffering") } +func (d *degradingChecker) Degraded() bool { return d.degraded } + +// A degraded checker is failing, so it must fail readiness, but restarting it +// would destroy the state it is recovering with. Liveness has to leave it +// alone no matter how long it stays that way. +func TestLiveness_DegradedCheckerIsNotStuck(t *testing.T) { + current := time.Now() + clock := func() time.Time { return current } + srv := newServerWithClock(t, clock) + + checker := °radingChecker{name: testCheckerName, degraded: true} + srv.Register(checker) + + current = current.Add(10 * testThreshold) + + w := httptest.NewRecorder() + srv.ServeHTTP(w, newGetRequest(t, "/healthz")) + if w.Code != http.StatusOK { + t.Errorf("degraded checker well past the threshold: want 200, got %d; body: %s", w.Code, w.Body.String()) + } + + w = httptest.NewRecorder() + srv.ServeHTTP(w, newGetRequest(t, "/readyz")) + if w.Code != http.StatusServiceUnavailable { + t.Errorf("degraded checker readiness: want 503, got %d", w.Code) + } + if !strings.Contains(w.Body.String(), `"degraded":true`) { + t.Errorf("readiness body should mark the checker degraded: %s", w.Body.String()) + } + + // Once it stops recovering in place, the restart is allowed. + checker.degraded = false + current = current.Add(testThreshold + time.Second) + + w = httptest.NewRecorder() + srv.ServeHTTP(w, newGetRequest(t, "/healthz")) + if w.Code != http.StatusServiceUnavailable { + t.Errorf("no longer degraded: want 503, got %d; body: %s", w.Code, w.Body.String()) + } +} + // TestLiveness_RecoveryClearsState ensures that once a checker recovers, the // failure timer resets so a fresh blip later does not immediately trip // liveness. func TestLiveness_RecoveryClearsState(t *testing.T) { current := time.Now() clock := func() time.Time { return current } - srv := newServerWithClock(t, testThreshold, clock) + srv := newServerWithClock(t, clock) flaky := &flakyChecker{name: testCheckerName} flaky.unhealthy = true