This is an automated email from the ASF dual-hosted git repository. mrproliu pushed a commit to branch bydbql-topk-ttl-and-pub-health-check in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 53482f89ce5ac5e61e5ad9d5bb1affdbd1b23bf8 Author: mrproliu <[email protected]> AuthorDate: Tue Jul 21 18:41:03 2026 +0800 Add TTL to top-k query and add health check for pub --- CHANGES.md | 2 + banyand/liaison/grpc/bydbql.go | 27 ++++++-- banyand/liaison/grpc/bydbql_reparse_test.go | 2 +- banyand/liaison/grpc/bydbql_test.go | 21 +++++- banyand/liaison/grpc/server.go | 8 ++- banyand/liaison/grpc/topk.go | 61 +++++++++++++---- banyand/liaison/grpc/topk_test.go | 95 ++++++++++++++++++++++++--- banyand/queue/pub/health_check_wiring_test.go | 75 +++++++++++++++++++++ banyand/queue/pub/pub.go | 62 ++++++++++------- docs/operation/configuration.md | 16 ++++- 10 files changed, 316 insertions(+), 53 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 95dbdbf6d..f78e52aaf 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -79,6 +79,7 @@ Release Notes. - Add a Claude/Codex plugin packaging the BanyanDB MCP server with a parse-only `validate_bydbql` tool (backed by the `mcp/tools/bydbql-parse` Go validator) and a BydbQL skill for read-only natural-language-to-BydbQL generation over STREAM/MEASURE/TRACE/PROPERTY resources. - Introduce positional parameter binding (`?` placeholders) into BydbQL to eliminate QL injection. - Add reusable BydbQL binding: Prepare a query once, then Bind it many times without re-parsing or mutating the template. The liaison caches prepared statements on the gRPC query path (LRU bounded by entry count and bytes, on by default; `--bydbql-prepared-cache-size`/`--bydbql-prepared-cache-max-bytes`, `bydbql_prepared_cache_*` metrics) so repeated templates skip parsing. To pinpoint un-cacheable and slow queries without high-cardinality labels, the query access log tags each entry by [...] +- Expire BydbQL top-K entries not seen within their TTL (`--bydbql-topk-slow-ttl` / `--bydbql-topk-reparse-ttl`, default `24h`), and log `last_seen` / `max_latency_at`. ### Bug Fixes @@ -138,6 +139,7 @@ Release Notes. - Deleting one TopN aggregation no longer tears down sibling aggregations on the same source measure. - Purge a deleted group's resource, index-rule and binding cache entries to avoid dangling references. - Clear a trace subject's index when its last index rule or binding is removed. +- Enable periodic health checks on the queue client (`--<prefix>-client-health-check-interval`, default `10s`), evicting dead data nodes proactively. ### Document diff --git a/banyand/liaison/grpc/bydbql.go b/banyand/liaison/grpc/bydbql.go index 7a6422f2f..4d89a6943 100644 --- a/banyand/liaison/grpc/bydbql.go +++ b/banyand/liaison/grpc/bydbql.go @@ -177,7 +177,7 @@ func (b *bydbQLService) Query(ctx context.Context, req *bydbqlv1.QueryRequest) ( } // topKDumper tracks the top re-parsed and slow queries and, on a supervised -// goroutine, periodically logs the cumulative top-K. All methods are nil-safe, so the +// goroutine, periodically logs the top-K within each tracker's TTL window. All methods are nil-safe, so the // call sites need no guards when the top-K log is disabled (the dumper is nil). type topKDumper struct { // reparse holds only templates the cache had already compiled once and had to @@ -191,13 +191,19 @@ type topKDumper struct { } // newTopKDumper starts the trackers and the dump goroutine; a non-positive interval -// disables the feature and returns nil. -func newTopKDumper(interval time.Duration, l *logger.Logger) *topKDumper { +// disables the feature and returns nil. Each tracker expires entries on its own TTL: +// a template nobody has re-parsed, or a query nobody has run slowly, for that long stops +// being reported. A non-positive TTL keeps that tracker's entries for the process lifetime. +func newTopKDumper(interval, reparseTTL, slowTTL time.Duration, l *logger.Logger) *topKDumper { if interval <= 0 { return nil } ctx, cancel := context.WithCancel(context.Background()) - d := &topKDumper{reparse: newTopK(bydbqlTopKSize), slow: newTopK(bydbqlTopKSize), l: l, cancel: cancel} + d := &topKDumper{ + reparse: newTopK(bydbqlTopKSize, reparseTTL, nil), + slow: newTopK(bydbqlTopKSize, slowTTL, nil), + l: l, cancel: cancel, + } run.Go(ctx, "liaison.grpc.bydbql.topk-dump", l, func(ctx context.Context) { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -245,13 +251,22 @@ func (d *topKDumper) dump() { // root; keeping the threshold on top of it would only re-hide the real re-parses the // exclusion just made visible. d.logTopK(d.reparse.snapshot(), 1, "top bydbql cache-miss queries", func(s topKSlot) string { - return fmt.Sprintf("%q count=%d", s.key, s.count) + return fmt.Sprintf("%q count=%d last_seen=%s", s.key, s.count, formatTopKTime(s.lastSeen)) }) + // max_latency is a running peak, so it can outlive the condition that caused it by + // as long as the TTL allows. max_latency_at dates it, and last_seen says when the + // query last ran at all, which together separate a live problem from a stale peak. d.logTopK(d.slow.snapshotByLatency(), 1, "top bydbql slow queries", func(s topKSlot) string { - return fmt.Sprintf("%q count=%d max_latency=%s", s.key, s.count, s.maxDur) + return fmt.Sprintf("%q count=%d max_latency=%s max_latency_at=%s last_seen=%s", + s.key, s.count, s.maxDur, formatTopKTime(s.maxDurAt), formatTopKTime(s.lastSeen)) }) } +// formatTopKTime renders a tracker timestamp in the same layout the surrounding logs use. +func formatTopKTime(t time.Time) string { + return t.UTC().Format(time.RFC3339) +} + // logTopK logs the entries with at least minCount occurrences, formatted by line. func (d *topKDumper) logTopK(entries []topKSlot, minCount uint64, msg string, line func(topKSlot) string) { lines := formatTopK(entries, minCount, line) diff --git a/banyand/liaison/grpc/bydbql_reparse_test.go b/banyand/liaison/grpc/bydbql_reparse_test.go index a17675968..e678389c2 100644 --- a/banyand/liaison/grpc/bydbql_reparse_test.go +++ b/banyand/liaison/grpc/bydbql_reparse_test.go @@ -31,7 +31,7 @@ import ( // change that broke it. run() returns the cache result for assertions. func newReparseProbe(t *testing.T, size, maxBytes int) (*preparedCache, *topK, func(query string) string) { t.Helper() - tk := newTopK(bydbqlTopKSize) + tk := newTopK(bydbqlTopKSize, 0, nil) c := newPreparedCache(size, maxBytes, nil) run := func(q string) string { _, result, err := c.getOrPrepare(q) diff --git a/banyand/liaison/grpc/bydbql_test.go b/banyand/liaison/grpc/bydbql_test.go index ae92bb8a6..a9ecb4bf3 100644 --- a/banyand/liaison/grpc/bydbql_test.go +++ b/banyand/liaison/grpc/bydbql_test.go @@ -95,7 +95,7 @@ func TestBydbQLQuery_ParamTypeMismatch_ReturnsInvalidArgument(t *testing.T) { // newTestDumper builds a topKDumper without starting the dump goroutine. func newTestDumper(l *logger.Logger) *topKDumper { - return &topKDumper{reparse: newTopK(bydbqlTopKSize), slow: newTopK(bydbqlTopKSize), l: l} + return &topKDumper{reparse: newTopK(bydbqlTopKSize, 0, nil), slow: newTopK(bydbqlTopKSize, 0, nil), l: l} } // attachTestDumper gives svc a dumper without starting its dump goroutine, so the @@ -157,6 +157,25 @@ func TestBydbQLDumpTopK(t *testing.T) { assert.NotEmpty(t, d.slow.snapshot()) } +// newTopKDumper takes the two TTLs as adjacent time.Duration parameters, so swapping them +// would compile and behave plausibly while silently applying each tracker's TTL to the +// other. Pin each one to its own tracker. The other tests build the dumper struct directly, +// so this is the only coverage of the real constructor. +func TestNewTopKDumperWiresEachTTLToItsTracker(t *testing.T) { + const reparseTTL, slowTTL = 3 * time.Hour, 7 * time.Hour + d := newTopKDumper(time.Hour, reparseTTL, slowTTL, logger.GetLogger("test-bydbql")) + require.NotNil(t, d) + defer d.close() + + assert.Equal(t, reparseTTL, d.reparse.ttl, "the reparse tracker must get the reparse TTL") + assert.Equal(t, slowTTL, d.slow.ttl, "the slow tracker must get the slow TTL") +} + +func TestNewTopKDumperDisabledByNonPositiveInterval(t *testing.T) { + assert.Nil(t, newTopKDumper(0, time.Hour, time.Hour, logger.GetLogger("test-bydbql")), + "a non-positive interval disables the feature, and the nil dumper's observers no-op") +} + func TestFormatTopKAppliesItsMinCount(t *testing.T) { entries := []topKSlot{ {key: "frequent", count: 5}, diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go index a9425a632..71af0bb9a 100644 --- a/banyand/liaison/grpc/server.go +++ b/banyand/liaison/grpc/server.go @@ -139,6 +139,8 @@ type server struct { grpcBufferMemoryRatio float64 bydbqlSlowThreshold time.Duration bydbqlTopKLogInterval time.Duration + bydbqlTopKSlowTTL time.Duration + bydbqlTopKReparseTTL time.Duration bydbqlCacheSize int bydbqlCacheMaxBytes int port uint32 @@ -372,7 +374,7 @@ func (s *server) PreRun(ctx context.Context) error { s.bydbQLSVC.slowThreshold = s.bydbqlSlowThreshold // The dump goroutine lives for the server's lifetime and is stopped by Close(), // so it is rooted at a background context rather than PreRun's setup context. - s.bydbQLSVC.dumper = newTopKDumper(s.bydbqlTopKLogInterval, s.bydbQLSVC.l) //nolint:contextcheck + s.bydbQLSVC.dumper = newTopKDumper(s.bydbqlTopKLogInterval, s.bydbqlTopKReparseTTL, s.bydbqlTopKSlowTTL, s.bydbQLSVC.l) //nolint:contextcheck s.propertyServer.metrics = metrics if s.barrierSVC != nil { s.barrierSVC.metrics = metrics @@ -466,6 +468,10 @@ func (s *server) FlagSet() *run.FlagSet { "end-to-end latency above which a BydbQL query is counted as slow; 0 disables slow-query tracking") fs.DurationVar(&s.bydbqlTopKLogInterval, "bydbql-topk-log-interval", 5*time.Minute, "how often to log the top BydbQL cache-miss and slow queries; 0 disables the top-K log") + fs.DurationVar(&s.bydbqlTopKSlowTTL, "bydbql-topk-slow-ttl", 24*time.Hour, + "drop a slow-query top-K entry whose query has not been slow again for this long; 0 keeps it for the process lifetime") + fs.DurationVar(&s.bydbqlTopKReparseTTL, "bydbql-topk-reparse-ttl", 24*time.Hour, + "drop a cache-miss top-K entry whose template has not been re-parsed again for this long; 0 keeps it for the process lifetime") s.grpcBufferMemoryRatio = 0.1 fs.Float64Var(&s.grpcBufferMemoryRatio, "grpc-buffer-memory-ratio", 0.1, "ratio of memory limit to use for gRPC buffer size calculation (0.0 < ratio <= 1.0)") diff --git a/banyand/liaison/grpc/topk.go b/banyand/liaison/grpc/topk.go index f6ef17773..087e93fdf 100644 --- a/banyand/liaison/grpc/topk.go +++ b/banyand/liaison/grpc/topk.go @@ -30,11 +30,15 @@ import ( // a normal workload, keeping the reported counts exact while still bounding memory. const bydbqlTopKSize = 128 -// topKSlot is one tracked query and its accumulated statistics. +// topKSlot is one tracked query and its accumulated statistics. lastSeen drives TTL +// expiry; maxDurAt pins when the peak latency happened, so a consumer can tell a live +// problem from a peak the tracker has merely been carrying since a startup incident. type topKSlot struct { - key string - count uint64 - maxDur time.Duration + lastSeen time.Time + maxDurAt time.Time + key string + count uint64 + maxDur time.Duration } // topK is a bounded approximate heavy-hitters tracker (Space-Saving): it keeps at @@ -42,17 +46,28 @@ type topKSlot struct { // inherit that entry's count + 1 so a fresh key gets a fair chance instead of being // evicted again immediately. With k modest (128) the min scan and the snapshot sort are // cheap, and observing an existing key needs no reordering at all. +// +// Entries also expire: a key not observed for ttl is dropped, so the tracker reports +// what is happening now rather than everything since process start. A ttl <= 0 keeps +// entries for the process lifetime. type topK struct { slots map[string]*topKSlot + now func() time.Time k int + ttl time.Duration mu sync.Mutex } -func newTopK(k int) *topK { +// newTopK builds a tracker holding at most k entries, expiring any entry not observed +// for ttl. now supplies the clock; passing nil uses time.Now. +func newTopK(k int, ttl time.Duration, now func() time.Time) *topK { if k < 1 { k = 1 } - return &topK{slots: make(map[string]*topKSlot, k), k: k} + if now == nil { + now = time.Now + } + return &topK{slots: make(map[string]*topKSlot, k), k: k, ttl: ttl, now: now} } // observe records one occurrence of key; dur is the query latency (0 when latency is @@ -60,15 +75,22 @@ func newTopK(k int) *topK { func (t *topK) observe(key string, dur time.Duration) { t.mu.Lock() defer t.mu.Unlock() + now := t.now() if s, ok := t.slots[key]; ok { s.count++ + s.lastSeen = now if dur > s.maxDur { - s.maxDur = dur + s.maxDur, s.maxDurAt = dur, now } return } + if len(t.slots) >= t.k { + // Reclaim expired entries before falling back to evicting a live one: a key + // nobody has asked about in ttl is a better victim than a merely infrequent one. + t.purgeExpiredLocked(now) + } if len(t.slots) < t.k { - t.slots[key] = &topKSlot{key: key, count: 1, maxDur: dur} + t.slots[key] = &topKSlot{key: key, count: 1, maxDur: dur, lastSeen: now, maxDurAt: now} return } // Full: evict the least-frequent entry and let the new key inherit its count. @@ -80,12 +102,25 @@ func (t *topK) observe(key string, dur time.Duration) { } } delete(t.slots, minKey) - t.slots[key] = &topKSlot{key: key, count: minCount + 1, maxDur: dur} + t.slots[key] = &topKSlot{key: key, count: minCount + 1, maxDur: dur, lastSeen: now, maxDurAt: now} +} + +// purgeExpiredLocked drops every entry not observed within ttl. Call with the lock held. +func (t *topK) purgeExpiredLocked(now time.Time) { + if t.ttl <= 0 { + return + } + for k, s := range t.slots { + if now.Sub(s.lastSeen) > t.ttl { + delete(t.slots, k) + } + } } // snapshot returns the tracked entries ranked by frequency: (count desc, maxDur desc, -// key asc). The tracker is cumulative, so each dump reflects the hottest queries since -// process start. The full tie-break makes the order deterministic across dumps. +// key asc). Counts accumulate over an entry's lifetime, which the TTL bounds: a dump +// reflects the hottest queries within the TTL window, not since process start. The full +// tie-break makes the order deterministic across dumps. func (t *topK) snapshot() []topKSlot { out := t.copyOut() sort.Slice(out, func(i, j int) bool { return lessByCount(out[i], out[j]) }) @@ -101,8 +136,12 @@ func (t *topK) snapshotByLatency() []topKSlot { return out } +// copyOut snapshots the live entries, expiring stale ones first. Purging here rather +// than only in observe is what lets a tracker that has gone quiet drain: an entry no +// one observes again would otherwise never be revisited, and would be reported forever. func (t *topK) copyOut() []topKSlot { t.mu.Lock() + t.purgeExpiredLocked(t.now()) out := make([]topKSlot, 0, len(t.slots)) for _, s := range t.slots { out = append(out, *s) diff --git a/banyand/liaison/grpc/topk_test.go b/banyand/liaison/grpc/topk_test.go index db864ee03..8c3cd3718 100644 --- a/banyand/liaison/grpc/topk_test.go +++ b/banyand/liaison/grpc/topk_test.go @@ -24,6 +24,8 @@ import ( "time" "github.com/stretchr/testify/assert" + + "github.com/apache/skywalking-banyandb/pkg/timestamp" ) func topKByKey(slots []topKSlot) map[string]topKSlot { @@ -35,7 +37,7 @@ func topKByKey(slots []topKSlot) map[string]topKSlot { } func TestTopKCountsAndMaxDur(t *testing.T) { - tk := newTopK(4) + tk := newTopK(4, 0, nil) tk.observe("a", 5*time.Millisecond) tk.observe("a", 8*time.Millisecond) // largest, observed in the middle tk.observe("a", 3*time.Millisecond) // smaller and last: maxDur must stay 8ms, not 3ms @@ -48,7 +50,7 @@ func TestTopKCountsAndMaxDur(t *testing.T) { } func TestTopKEvictsMinAndInheritsCount(t *testing.T) { - tk := newTopK(2) + tk := newTopK(2, 0, nil) tk.observe("a", 0) tk.observe("a", 0) tk.observe("a", 0) // a.count = 3 @@ -67,7 +69,7 @@ func TestTopKEvictsMinAndInheritsCount(t *testing.T) { } func TestTopKSnapshotSortedAndCumulative(t *testing.T) { - tk := newTopK(4) + tk := newTopK(4, 0, nil) tk.observe("hi", 0) tk.observe("hi", 0) tk.observe("lo", 0) @@ -80,7 +82,7 @@ func TestTopKSnapshotSortedAndCumulative(t *testing.T) { } func TestTopKSnapshotByLatency(t *testing.T) { - tk := newTopK(bydbqlTopKSize) + tk := newTopK(bydbqlTopKSize, 0, nil) tk.observe("frequent", time.Millisecond) // high count, low latency tk.observe("frequent", time.Millisecond) tk.observe("frequent", time.Millisecond) @@ -92,9 +94,11 @@ func TestTopKSnapshotByLatency(t *testing.T) { } func TestTopKDeterministicTieBreak(t *testing.T) { - // Equal counts and durations must order by key, not by map iteration. - a := newTopK(bydbqlTopKSize) - b := newTopK(bydbqlTopKSize) + // Equal counts and durations must order by key, not by map iteration. Both trackers run + // on mock clocks frozen at the same instant, so the compared slots differ only in order. + clockA, clockB := timestamp.NewMockClock(), timestamp.NewMockClock() + a := newTopK(bydbqlTopKSize, 0, clockA.Now) + b := newTopK(bydbqlTopKSize, 0, clockB.Now) for _, k := range []string{"c", "a", "b"} { a.observe(k, 0) } @@ -106,8 +110,83 @@ func TestTopKDeterministicTieBreak(t *testing.T) { assert.Equal(t, []string{"a", "b", "c"}, []string{snap[0].key, snap[1].key, snap[2].key}) } +func TestTopKExpiresUnobservedEntry(t *testing.T) { + mc := timestamp.NewMockClock() + tk := newTopK(4, time.Hour, mc.Now) + tk.observe("stale", time.Second) + + mc.Add(time.Hour - time.Minute) + assert.Len(t, tk.snapshot(), 1, "still within the TTL") + + mc.Add(2 * time.Minute) // now past the TTL + assert.Empty(t, tk.snapshot(), "an entry not seen within the TTL is dropped") +} + +func TestTopKObserveRefreshesTTL(t *testing.T) { + mc := timestamp.NewMockClock() + tk := newTopK(4, time.Hour, mc.Now) + tk.observe("recurring", 0) + + // Re-observing before expiry must reset the clock on the entry, so a query that + // keeps happening is never dropped no matter how long the process has run. + for i := 0; i < 5; i++ { + mc.Add(50 * time.Minute) + tk.observe("recurring", 0) + } + mc.Add(50 * time.Minute) + + snap := tk.snapshot() + assert.Len(t, snap, 1, "a repeatedly observed entry survives indefinitely") + assert.Equal(t, uint64(6), snap[0].count) +} + +func TestTopKZeroTTLKeepsEntriesForever(t *testing.T) { + mc := timestamp.NewMockClock() + tk := newTopK(4, 0, mc.Now) + tk.observe("kept", 0) + + mc.Add(365 * 24 * time.Hour) + assert.Len(t, tk.snapshot(), 1, "ttl <= 0 restores the cumulative behavior") +} + +func TestTopKPurgesExpiredBeforeEvictingLiveEntry(t *testing.T) { + mc := timestamp.NewMockClock() + tk := newTopK(2, time.Hour, mc.Now) + tk.observe("expiring", 0) + tk.observe("expiring", 0) + tk.observe("expiring", 0) // count 3: the most frequent, so never the evict-min victim + + mc.Add(2 * time.Hour) // "expiring" is now stale + tk.observe("fresh", 0) // fills the second slot + tk.observe("newcomer", 0) + + byKey := topKByKey(tk.snapshot()) + assert.Len(t, byKey, 2) + _, hasExpiring := byKey["expiring"] + assert.False(t, hasExpiring, "the stale entry is reclaimed even though it had the highest count") + assert.Equal(t, uint64(1), byKey["newcomer"].count, + "reclaiming a slot lets the new key start at 1 instead of inheriting an evicted count") +} + +func TestTopKMaxDurAtTracksPeakNotLastObserve(t *testing.T) { + mc := timestamp.NewMockClock() + tk := newTopK(4, time.Hour, mc.Now) + + tk.observe("q", time.Millisecond) + mc.Add(time.Minute) + tk.observe("q", time.Second) // the peak + peakAt := mc.Now() + mc.Add(time.Minute) + tk.observe("q", time.Millisecond) // slower observation must not move maxDurAt + + snap := tk.snapshot() + assert.Equal(t, time.Second, snap[0].maxDur) + assert.Equal(t, peakAt, snap[0].maxDurAt, "maxDurAt dates the peak, not the latest observation") + assert.Equal(t, mc.Now(), snap[0].lastSeen, "lastSeen tracks the latest observation") +} + func TestTopKConcurrentObserve(t *testing.T) { - tk := newTopK(bydbqlTopKSize) + tk := newTopK(bydbqlTopKSize, 0, nil) const workers, iters = 16, 500 var wg sync.WaitGroup for w := 0; w < workers; w++ { diff --git a/banyand/queue/pub/health_check_wiring_test.go b/banyand/queue/pub/health_check_wiring_test.go new file mode 100644 index 000000000..c2fb017ba --- /dev/null +++ b/banyand/queue/pub/health_check_wiring_test.go @@ -0,0 +1,75 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package pub + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + grpc_health_v1 "google.golang.org/grpc/health/grpc_health_v1" + + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" + modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1" + "github.com/apache/skywalking-banyandb/pkg/logger" +) + +// The periodic prober itself has long been implemented and tested in pkg/grpchelper; what +// was missing is that PreRun never passed an interval, so for the queue client the prober +// simply never started. The active set was then only ever corrected by a request failing on +// a dead node, which is why a node that disappeared stayed routable until some query paid +// the full timeout to discover it. +// +// These tests therefore pin the WIRING, not the mechanism. Both fail before the fix: the +// default assertion because no flag existed, and the eviction assertion because a zero +// interval leaves the prober unstarted, so the node stays active forever. +func TestFlagSetSuppliesNonZeroHealthCheckInterval(t *testing.T) { + p := New(nil, databasev1.Role_ROLE_DATA).(*pub) + p.FlagSet() // DurationVar writes the default through immediately + + assert.NotZero(t, p.healthCheckInterval, + "a zero interval silently disables the prober, which is the bug this flag exists to prevent") + assert.Equal(t, defaultHealthCheckInterval, p.healthCheckInterval) +} + +func TestPreRunStartsPeriodicHealthCheck(t *testing.T) { + addr := getAddress() + healthSrv, stopServer := setupWithStatus(addr, modelv1.Status_STATUS_SUCCEED) + defer stopServer() + + p := New(nil, databasev1.Role_ROLE_DATA).(*pub) + p.log = logger.GetLogger("queue-client") + p.FlagSet() + p.healthCheckInterval = 100 * time.Millisecond // keep the test short + require.NoError(t, p.PreRun(context.Background())) + defer p.connMgr.GracefulStop() + + p.OnAddOrUpdate(getDataNode("node1", addr)) + require.Eventually(t, func() bool { return p.connMgr.ActiveCount() == 1 }, + 10*time.Second, 20*time.Millisecond, "the healthy node should be admitted") + + // The node stops serving. Nothing publishes to it, so only a background prober can + // notice — which is exactly what a zero interval would fail to do. + healthSrv.SetServingStatus("", grpc_health_v1.HealthCheckResponse_NOT_SERVING) + + assert.Eventually(t, func() bool { return p.connMgr.ActiveCount() == 0 }, + 10*time.Second, 20*time.Millisecond, + "the prober must evict the unreachable node without any request being sent to it") +} diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go index a51996888..8c13a1729 100644 --- a/banyand/queue/pub/pub.go +++ b/banyand/queue/pub/pub.go @@ -54,6 +54,12 @@ import ( var queuePubScope = observability.RootScope.SubScope("queue_pub") +// defaultHealthCheckInterval is how often the queue client re-checks the nodes it considers +// active, matching the property schema client's default. Without it the active set is only +// validated on admission and then by a request failing on it, so a dead node keeps absorbing +// queries — each paying the full broadcast timeout — until one of them evicts it. +const defaultHealthCheckInterval = 10 * time.Second + // ChunkedSyncClientConfig configures chunked sync client behavior. type ChunkedSyncClientConfig struct { ChunkSize uint32 // Size of each chunk in bytes @@ -72,26 +78,27 @@ var ( type pub struct { schema.UnimplementedOnInitHandler - metadata metadata.Repo - handlers map[bus.Topic]schema.EventHandler - log *logger.Logger - metrics *pubMetrics - migrationMetrics *pubMigrationMetrics - connMgr *grpchelper.ConnManager[*client] - closer *run.Closer - writableProbe map[string]map[string]struct{} - nodeCache map[string]nodeInfo - caCertPath string - caCertReloader *pkgtls.Reloader - prefix string - retryPolicy string - selfNode string - selfRole string - selfTier string - allowedRoles []databasev1.Role - writableProbeMu sync.Mutex - nodeCacheMu sync.RWMutex - tlsEnabled bool + metadata metadata.Repo + handlers map[bus.Topic]schema.EventHandler + log *logger.Logger + metrics *pubMetrics + migrationMetrics *pubMigrationMetrics + connMgr *grpchelper.ConnManager[*client] + closer *run.Closer + writableProbe map[string]map[string]struct{} + nodeCache map[string]nodeInfo + caCertPath string + caCertReloader *pkgtls.Reloader + prefix string + retryPolicy string + selfNode string + selfRole string + selfTier string + allowedRoles []databasev1.Role + healthCheckInterval time.Duration + writableProbeMu sync.Mutex + nodeCacheMu sync.RWMutex + tlsEnabled bool } // nodeInfo caches the resolved role and tier for a remote node. @@ -227,6 +234,8 @@ func (p *pub) FlagSet() *run.FlagSet { fs := run.NewFlagSet("queue-client") fs.BoolVar(&p.tlsEnabled, prefixFlag("client-tls"), false, fmt.Sprintf("enable client TLS for %s", p.prefix)) fs.StringVar(&p.caCertPath, prefixFlag("client-ca-cert"), "", fmt.Sprintf("CA certificate file to verify the %s server", p.prefix)) + fs.DurationVar(&p.healthCheckInterval, prefixFlag("client-health-check-interval"), defaultHealthCheckInterval, + fmt.Sprintf("how often to re-check active %s nodes and evict the unreachable ones; 0 disables the periodic check", p.prefix)) return fs } @@ -489,6 +498,10 @@ func NewWithoutMetadata(omr observability.MetricsRegistry) queue.Client { p := New(nil, databasev1.Role_ROLE_DATA) pp := p.(*pub) pp.log = logger.GetLogger("queue-client") + // No HealthCheckInterval here, unlike PreRun: this constructor skips FlagSet, and its + // callers (lifecycle migration, tests) are short-lived and do not all reach GracefulStop. + // Starting a background prober they never stop would leak a goroutine, and the reactive + // failover path already covers a batch job's much narrower exposure. pp.connMgr = grpchelper.NewConnManager(grpchelper.ConnManagerConfig[*client]{ Handler: pp, Logger: pp.log, @@ -526,10 +539,11 @@ func (p *pub) PreRun(context.Context) error { // Initialize connection manager with the pub as the handler p.connMgr = grpchelper.NewConnManager(grpchelper.ConnManagerConfig[*client]{ //nolint:contextcheck // health check runs in background goroutine - Handler: p, - Logger: p.log, - RetryPolicy: p.retryPolicy, - MaxRecvMsgSize: maxReceiveMessageSize, + Handler: p, + Logger: p.log, + RetryPolicy: p.retryPolicy, + MaxRecvMsgSize: maxReceiveMessageSize, + HealthCheckInterval: p.healthCheckInterval, }) // Initialize CA certificate reloader if TLS is enabled and CA cert path is provided diff --git a/docs/operation/configuration.md b/docs/operation/configuration.md index a4d193329..3675cb649 100644 --- a/docs/operation/configuration.md +++ b/docs/operation/configuration.md @@ -97,7 +97,13 @@ The following flags tune the BydbQL prepared-statement cache on the query path. These flags surface the queries behind cache misses and slow responses without exposing high-cardinality query text as metric labels: Prometheus gets only two counters (`bydbql_prepared_cache_total{result="miss"}` and `bydbql_slow_query_total`), while the specific hot queries are logged. - `--bydbql-slow-query-threshold duration`: End-to-end latency above which a BydbQL query is counted as slow (increments `bydbql_slow_query_total`) and tracked in the slow-query top-K; `0` disables slow-query tracking (default: `1s`). -- `--bydbql-topk-log-interval duration`: How often to log the hottest cache-miss and slow queries. Counts are cumulative since process start. The cache-miss list only shows templates re-parsed at least twice (`count>=2`): every template misses once on its cold-start lookup, so a `count==1` entry is benign and is filtered out — only repeatedly evicted-and-re-parsed (thrashing) templates are surfaced. The slow-query list is ranked by peak latency (`max_latency`) so a rarely-but-catastrophi [...] +- `--bydbql-topk-log-interval duration`: How often to log the hottest cache-miss and slow queries. The cache-miss list holds only templates that were compiled more than once — either evicted and compiled again (thrashing), or too large for the byte bound and therefore re-compiled on every single request; a template's unavoidable first-ever compile is excluded at the source, so every entry is actionable. The slow-query list is ranked by peak latency (`max_latency`) so a rarely-but-catastr [...] +- `--bydbql-topk-slow-ttl duration`: Drop a slow-query top-K entry whose query has not been slow again for this long; `0` keeps entries for the process lifetime (default: `24h`). +- `--bydbql-topk-reparse-ttl duration`: Drop a cache-miss top-K entry whose template has not been re-parsed again for this long; `0` keeps entries for the process lifetime (default: `24h`). + +Each logged entry carries `last_seen`, and slow entries additionally carry `max_latency_at`. Both trackers accumulate, so `max_latency` is a running peak that can long outlive the incident that produced it: without these timestamps a one-off startup spike keeps being reported as if it were current, and a reader cannot tell a live problem from a stale one. The TTLs bound how long that can happen — an entry whose query stops recurring disappears — while `max_latency_at` dates the peak itse [...] + +Keep each TTL comfortably above `--bydbql-topk-log-interval`. An entry is only ever reported by a dump, so a TTL shorter than the interval lets an entry expire in the gap between two dumps and never be logged at all — the tracker would go quiet not because nothing is wrong but because it forgets faster than it reports. The defaults (`24h` against `5m`) leave a wide margin. #### Diagnosing an ineffective BydbQL cache @@ -238,6 +244,14 @@ These flags configure how every node talks to the schema server. - `--schema-property-client-tls`: Enable TLS for property schema client connections. - `--schema-property-client-ca-cert string`: CA certificate file to verify the property schema server. +### Queue client (liaison to data nodes) + +These flags configure the connections a liaison holds to the data nodes it queries and writes to. `<prefix>` is `data` or `liaison` depending on which peer set the client serves. + +- `--<prefix>-client-health-check-interval duration`: How often to re-check the nodes the client currently considers active, evicting any that no longer answer; `0` disables the periodic check (default: `10s`). The active set is otherwise only validated when a node is admitted and, after that, when a request happens to fail on it — so a node that dies stays routable until some query picks it and pays that query's full timeout to discover it. Keep this well below the distributed query tim [...] +- `--<prefix>-client-tls`: Enable client TLS. +- `--<prefix>-client-ca-cert string`: CA certificate file to verify the server. + ### Other - `-n, --name string`: Name of this service.
