From 066c5c6934af104390083d28fb1bb1eba824fd4d Mon Sep 17 00:00:00 2001 From: Jeroen Date: Tue, 11 Aug 2026 10:01:35 +0200 Subject: [PATCH] Buffer QuestDB writes through an outage and replay them Reconnecting stopped the permanent wedge but still lost every reading taken while QuestDB was down. The sink now holds rows in memory and replays them in order once the connection is back. Rows are held until a flush confirms them, not just while the client knows it is disconnected. The ILP client keeps written rows in its own buffer until the next flush and throws that buffer away when the flush fails, so by the time the sink learns the socket is dead the readings it accepted are already gone. Holding them until the flush succeeds is what makes them recoverable. QuestDB.MaxBufferBytes caps the buffer, 4 MiB by default, sized from an estimate of the ILP encoding of each row. Past the cap the oldest rows are evicted and the store call starts failing, which the service escalates to a process exit. The pod then crash-loops until QuestDB is back, instead of growing until the kernel kills it. Zero restores the previous drop-on-disconnect behaviour. That needed a health-server change. Liveness would have restarted the pod after 90s and thrown away the buffer holding the data. A checker can now implement Degrader to say it is failing but recovering in place, and the liveness threshold skips it. Readiness still goes red immediately, so the outage stays visible. One limit stays: ILP over TCP has no server acknowledgement, so the last flush before a socket error reports success even though QuestDB never stored those rows. They cannot be replayed. Everything from the first reported failure onwards is covered. --- CHANGELOG.md | 28 ++ cmd/meterlogger/sinks.go | 9 +- config.example.yaml | 3 + documentation/configuration.md | 22 +- documentation/observability.md | 27 +- documentation/troubleshooting.md | 13 +- internal/adapters/sink/qdb/buffer.go | 174 ++++++++++ internal/adapters/sink/qdb/buffer_test.go | 302 ++++++++++++++++++ internal/adapters/sink/qdb/checker_test.go | 10 +- internal/adapters/sink/qdb/common.go | 174 +++++++++- internal/adapters/sink/qdb/qdb_gas_writer.go | 2 +- internal/adapters/sink/qdb/qdb_grid_writer.go | 2 +- internal/adapters/sink/qdb/qdb_heat_writer.go | 2 +- .../adapters/sink/qdb/qdb_solar_writer.go | 4 +- internal/adapters/sink/qdb/qdb_test.go | 9 +- .../adapters/sink/qdb/qdb_thermal_writer.go | 2 +- .../sink/qdb/qdb_ventilation_writer.go | 4 +- .../adapters/sink/qdb/qdb_water_writer.go | 2 +- internal/adapters/sink/qdb/reconnect_test.go | 82 ++++- internal/config/config.go | 10 + internal/config/load.go | 1 + internal/config/load_test.go | 6 + internal/config/validate.go | 7 + internal/config/validate_test.go | 22 ++ internal/healthserver/server.go | 24 +- internal/healthserver/server_test.go | 59 +++- 26 files changed, 943 insertions(+), 57 deletions(-) create mode 100644 internal/adapters/sink/qdb/buffer.go create mode 100644 internal/adapters/sink/qdb/buffer_test.go 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