This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new c2d925e4e feat(queue): add message + batch unit catalogs to
batch-write metrics (#1169)
c2d925e4e is described below
commit c2d925e4eae4d77edda94e1fd438243483960150
Author: Gao Hongtao <[email protected]>
AuthorDate: Thu Jun 11 13:15:36 2026 +0800
feat(queue): add message + batch unit catalogs to batch-write metrics
(#1169)
* feat(queue): add message + batch unit catalogs to batch-write metrics
The internal queue's batch-write traffic was counted in incomparable
units on the two ends: the publisher's banyandb_queue_pub_* counted per
message, while the subscriber's banyandb_queue_sub_* counted per
collected batch (one batch-mode wire stream). For the liaison tier-1
write routing this produced a ~950:1 Pub:Sub ratio that looked like loss
but was just a unit mismatch.
---
CHANGES.md | 1 +
banyand/observability/services/metrics_system.go | 65 +++--
banyand/queue/pub/batch.go | 41 +++-
banyand/queue/pub/metrics_test.go | 96 +++++++-
banyand/queue/pub/migration_metrics.go | 26 +-
banyand/queue/pub/migration_metrics_test.go | 64 ++++-
banyand/queue/pub/pub.go | 26 +-
banyand/queue/sub/batch_metrics_test.go | 298 +++++++++++++++++++++++
banyand/queue/sub/helpers.go | 15 +-
banyand/queue/sub/server.go | 30 ++-
banyand/queue/sub/server_metrics_test.go | 15 +-
banyand/queue/sub/sub.go | 8 +-
banyand/trace/metadata_test.go | 2 +-
banyand/trace/snapshot_test.go | 34 ++-
docs/operation/grafana-fodc-nodes.json | 4 +-
docs/operation/observability/metrics.md | 13 +-
pkg/meter/meter.go | 5 +
17 files changed, 652 insertions(+), 91 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index d76e09f5f..d0917c5a1 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -5,6 +5,7 @@ Release Notes.
## 0.11.0
### Features
+- Add two catalogs to the queue batch-write metrics so traffic is comparable
on both ends: a per-batch-stream **batch catalog**
(`total_batch_started`/`total_batch_finished`/`total_batch_latency`, buckets to
~300s) on `queue_pub`/`queue_sub` and the `lifecycle_migration` mirror, and a
per-message **message catalog**
(`total_message_started`/`total_message_finished`) on `queue_sub` (the
publisher's existing `total_*` already counts per message). All existing
`total_*` series are unchanged [...]
- Redesign the queue (`queue_pub`/`queue_sub`) metrics around a uniform model:
keep only `total_started`, `total_finished`, `total_latency` (now a histogram)
and `total_err`, plus file-sync-only `sent_bytes` (pub) / `received_bytes`
(sub). Replace the `topic` label with `operation`
(`batch-write`/`file-sync`/`query`/`control`) and `group`, add an `error_type`
label on `total_err`, and add remote-endpoint labels
(`remote_node`/`remote_role`/`remote_tier`) so the liaison↔data (hot/warm/col
[...]
- Stamp the lifecycle's tier-migration publisher's identity onto the wire so
the receiving data node records a non-empty
`remote_node`/`remote_role`/`remote_tier` on its
`banyandb_queue_sub_total_finished` series. The lifecycle's `parseGroup`
resolves the lifecycle's self identity by matching its `--grpc-addr` (the
co-located data node's gRPC address) against the data-node registry —
`Metadata.Name` becomes `remote_node`, `Labels["type"]` becomes `remote_tier` —
and calls `SetSelfNode(se [...]
- Add `banyandb_lifecycle_last_run_timestamp_seconds` and
`banyandb_lifecycle_last_run_success` gauges to the lifecycle service for
at-a-glance health monitoring. `last_run_timestamp_seconds` records the
wall-clock epoch (in seconds) of the most recent migration cycle;
`last_run_success` is `1` on a nil error and `0` otherwise. Both are stamped by
a `defer` at the end of `action()` so every return path (success, error,
recovered panic) updates the pair atomically — dashboards can pin an [...]
diff --git a/banyand/observability/services/metrics_system.go
b/banyand/observability/services/metrics_system.go
index c9b23d9c7..3de2256bf 100644
--- a/banyand/observability/services/metrics_system.go
+++ b/banyand/observability/services/metrics_system.go
@@ -44,6 +44,11 @@ var (
startTime = time.Now()
)
+// gaugesMu protects the process-global gauge vars below. Multiple
metricService
+// instances can exist in the same process (e.g. distributed integration
tests),
+// so initMetrics() (writer) and collect*() (readers) must be synchronized.
+var gaugesMu sync.RWMutex
+
var (
cpuStateGauge meter.Gauge
cpuNumGauge meter.Gauge
@@ -85,6 +90,8 @@ func init() {
func (p *metricService) initMetrics() {
factory := p.With(observability.SystemScope)
+ gaugesMu.Lock()
+ defer gaugesMu.Unlock()
cpuStateGauge = factory.NewGauge("cpu_state", "kind")
cpuNumGauge = factory.NewGauge("cpu_num")
memoryStateGauge = factory.NewGauge("memory_state", "kind")
@@ -101,7 +108,6 @@ func collectCPU() {
cpuCount = c
}
})
- cpuNumGauge.Set(float64(cpuCount))
s, err := cpuTimesFunc(false)
if err != nil {
log.Error().Err(err).Msg("cannot get cpu stat")
@@ -114,6 +120,12 @@ func collectCPU() {
allStat := s[0]
total := allStat.User + allStat.System + allStat.Idle + allStat.Nice +
allStat.Iowait + allStat.Irq +
allStat.Softirq + allStat.Steal + allStat.Guest +
allStat.GuestNice
+ gaugesMu.RLock()
+ defer gaugesMu.RUnlock()
+ if cpuNumGauge == nil || cpuStateGauge == nil {
+ return
+ }
+ cpuNumGauge.Set(float64(cpuCount))
cpuStateGauge.Set(allStat.User/total, "user")
cpuStateGauge.Set(allStat.System/total, "system")
cpuStateGauge.Set(allStat.Idle/total, "idle")
@@ -125,30 +137,32 @@ func collectCPU() {
}
func collectMemory() {
- if memoryStateGauge == nil {
- log.Error().Msg("memoryStateGauge is not registered")
- return
- }
m, err := mem.VirtualMemory()
if err != nil {
log.Error().Err(err).Msg("cannot get memory stat")
return
}
+ gaugesMu.RLock()
+ defer gaugesMu.RUnlock()
+ if memoryStateGauge == nil {
+ return
+ }
memoryStateGauge.Set(m.UsedPercent/100, "used_percent")
memoryStateGauge.Set(float64(m.Used), "used")
memoryStateGauge.Set(float64(m.Total), "total")
}
func collectNet() {
- if netStateGauge == nil {
- log.Error().Msg("netStateGauge is not registered")
- return
- }
stats, err := getNetStat(context.Background())
if err != nil {
log.Error().Err(err).Msg("cannot get net stat")
return
}
+ gaugesMu.RLock()
+ defer gaugesMu.RUnlock()
+ if netStateGauge == nil {
+ return
+ }
for _, stat := range stats {
netStateGauge.Set(float64(stat.BytesRecv), "bytes_recv",
stat.Name)
netStateGauge.Set(float64(stat.BytesSent), "bytes_sent",
stat.Name)
@@ -188,23 +202,42 @@ func getNetStat(ctx context.Context)
([]net.IOCountersStat, error) {
}
func collectUpTime() {
+ gaugesMu.RLock()
+ defer gaugesMu.RUnlock()
+ if upTimeGauge == nil {
+ return
+ }
upTimeGauge.Set(time.Since(startTime).Seconds())
}
func collectDisk() {
- for _, path := range getPath() {
+ paths := getPath()
+ type usageStat struct {
+ usage *disk.UsageStat
+ path string
+ }
+ var stats []usageStat
+ for _, path := range paths {
usage, err := disk.Usage(path)
if err != nil {
- if _, err = os.Stat(path); err != nil {
- if !os.IsNotExist(err) {
- log.Error().Err(err).Msgf("failed to
get stat for path: %s", path)
+ if _, statErr := os.Stat(path); statErr != nil {
+ if !os.IsNotExist(statErr) {
+ log.Error().Err(statErr).Msgf("failed
to get stat for path: %s", path)
}
}
return
}
diskMap.Store(path, int(usage.UsedPercent))
- diskStateGauge.Set(usage.UsedPercent/100, path, "used_percent")
- diskStateGauge.Set(float64(usage.Used), path, "used")
- diskStateGauge.Set(float64(usage.Total), path, "total")
+ stats = append(stats, usageStat{path: path, usage: usage})
+ }
+ gaugesMu.RLock()
+ defer gaugesMu.RUnlock()
+ if diskStateGauge == nil {
+ return
+ }
+ for _, s := range stats {
+ diskStateGauge.Set(s.usage.UsedPercent/100, s.path,
"used_percent")
+ diskStateGauge.Set(float64(s.usage.Used), s.path, "used")
+ diskStateGauge.Set(float64(s.usage.Total), s.path, "total")
}
}
diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index 0cdf12686..bcb227d10 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -59,6 +59,7 @@ const (
type writeStream struct {
client clusterv1.Service_SendClient
ctxDoneCh <-chan struct{}
+ batchStart time.Time
firstFrameSent bool // false until the first frame is sent; the first
frame carries the sender_* identity labels
}
@@ -192,9 +193,25 @@ func (bp *batchPublisher) Publish(ctx context.Context,
topic bus.Topic, messages
err = multierr.Append(err, fmt.Errorf("failed to get
stream for node %s: %w", node, errCreateStream))
continue
}
+ streamBatchStart := time.Now()
bp.streams[node] = writeStream{
- client: stream,
- ctxDoneCh: streamCtx.Done(),
+ client: stream,
+ ctxDoneCh: streamCtx.Done(),
+ batchStart: streamBatchStart,
+ }
+ var batchOp string
+ var batchInfo nodeInfo
+ if bp.topic != nil {
+ batchOp = apidata.OperationOf(*bp.topic)
+ }
+ if bp.pub != nil {
+ batchInfo = bp.pub.getNodeInfo(node)
+ }
+ if bp.hasMetrics() {
+ bp.pub.metrics.totalBatchStarted.Inc(1, batchOp, "",
node, batchInfo.role, batchInfo.tier)
+ }
+ if bp.hasMigrationMetrics() {
+ bp.pub.migrationMetrics.totalBatchStarted.Inc(1,
batchOp, "", node, batchInfo.role, batchInfo.tier)
}
bp.f.events = append(bp.f.events, make(chan batchEvent))
_ = sendData()
@@ -202,8 +219,9 @@ func (bp *batchPublisher) Publish(ctx context.Context,
topic bus.Topic, messages
recvStream := stream
recvDeferFn := deferFn
recvBC := bp.f.events[len(bp.f.events)-1]
+ recvBatchStart := streamBatchStart
run.Go(ctx, "batch-stream-recv", bp.pub.log, func(runCtx
context.Context) {
- bp.listenBatchResponse(runCtx, recvStream, recvDeferFn,
recvBC, nodeName)
+ bp.listenBatchResponse(runCtx, recvStream, recvDeferFn,
recvBC, nodeName, recvBatchStart)
})
}
return nil, err
@@ -218,11 +236,14 @@ func (bp *batchPublisher) hasMigrationMetrics() bool {
}
// listenBatchResponse receives the server response and records failover
events and end-to-end failure metrics.
-func (bp *batchPublisher) listenBatchResponse(ctx context.Context, s
clusterv1.Service_SendClient, deferFn func(), bc chan batchEvent, curNode
string) {
+func (bp *batchPublisher) listenBatchResponse(ctx context.Context, s
clusterv1.Service_SendClient, deferFn func(),
+ bc chan batchEvent, curNode string, batchStart time.Time,
+) {
defer func() {
close(bc)
deferFn()
}()
+ // ctx.Done() fires before any Recv; do NOT tick batch finished/latency
here.
select {
case <-ctx.Done():
return
@@ -258,13 +279,25 @@ func (bp *batchPublisher) listenBatchResponse(ctx
context.Context, s clusterv1.S
return
}
if resp == nil || resp.Error == "" || resp.Status ==
modelv1.Status_STATUS_SUCCEED {
+ if bp.hasMetrics() {
+ bp.pub.metrics.totalBatchFinished.Inc(1, operation, "",
curNode, info.role, info.tier)
+
bp.pub.metrics.totalBatchLatency.Observe(time.Since(batchStart).Seconds(),
operation, "", curNode, info.role, info.tier)
+ }
+ if bp.hasMigrationMetrics() {
+ bp.pub.migrationMetrics.totalBatchFinished.Inc(1,
operation, "", curNode, info.role, info.tier)
+
bp.pub.migrationMetrics.totalBatchLatency.Observe(time.Since(batchStart).Seconds(),
operation, "", curNode, info.role, info.tier)
+ }
return
}
if bp.hasMetrics() {
bp.pub.metrics.totalErr.Inc(1, operation, "", curNode,
info.role, info.tier, sendErrReasonServerRejected)
+ bp.pub.metrics.totalBatchFinished.Inc(1, operation, "",
curNode, info.role, info.tier)
+
bp.pub.metrics.totalBatchLatency.Observe(time.Since(batchStart).Seconds(),
operation, "", curNode, info.role, info.tier)
}
if bp.hasMigrationMetrics() {
bp.pub.migrationMetrics.totalErr.Inc(1, operation, "", curNode,
info.role, info.tier, sendErrReasonServerRejected)
+ bp.pub.migrationMetrics.totalBatchFinished.Inc(1, operation,
"", curNode, info.role, info.tier)
+
bp.pub.migrationMetrics.totalBatchLatency.Observe(time.Since(batchStart).Seconds(),
operation, "", curNode, info.role, info.tier)
}
ce := common.NewErrorWithStatus(resp.Status, resp.Error)
// Only failover statuses trigger circuit-breaker accounting; other
server-side
diff --git a/banyand/queue/pub/metrics_test.go
b/banyand/queue/pub/metrics_test.go
index c088628cb..f68e89663 100644
--- a/banyand/queue/pub/metrics_test.go
+++ b/banyand/queue/pub/metrics_test.go
@@ -102,11 +102,14 @@ func (*noopHistogram) Delete(_ ...string) bool {
return true }
func newPubMetricsWithErrCapture(totalErr *errReasonCapturerImpl) *pubMetrics
{ //nolint:exhaustruct
return &pubMetrics{
- totalStarted: &countingCounter{},
- totalFinished: &countingCounter{},
- totalLatency: &noopHistogram{},
- totalErr: totalErr,
- sentBytes: &countingCounter{},
+ totalStarted: &countingCounter{},
+ totalFinished: &countingCounter{},
+ totalLatency: &noopHistogram{},
+ totalErr: totalErr,
+ sentBytes: &countingCounter{},
+ totalBatchStarted: &countingCounter{},
+ totalBatchFinished: &countingCounter{},
+ totalBatchLatency: &noopHistogram{},
}
}
@@ -264,7 +267,7 @@ func TestListenBatchResponseRecordsRecvError(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
require.Equal(t, float64(1), sendErrCap.sum(sendErrReasonRecvError))
}
@@ -285,7 +288,7 @@ func
TestListenBatchResponseRecvNonFailoverStillRecordsRecvError(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
require.Equal(t, float64(1), sendErrCap.sum(sendErrReasonRecvError))
}
@@ -309,7 +312,7 @@ func TestListenBatchResponseServerRejectedWithoutFailover(t
*testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
require.Equal(t, float64(1),
sendErrCap.sum(sendErrReasonServerRejected))
require.Equal(t, float64(0), sendErrCap.sum(sendErrReasonRecvError))
@@ -345,7 +348,7 @@ func TestListenBatchResponseDiskFullSendsFailoverEvent(t
*testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
require.Equal(t, float64(1),
sendErrCap.sum(sendErrReasonServerRejected))
@@ -399,3 +402,78 @@ func TestPublishRecordsStartedAndFinished(t *testing.T) {
require.Equal(t, float64(1), started.count, "totalStarted must be 1 on
success")
require.Equal(t, float64(1), finished.count, "totalFinished must be 1
on success")
}
+
+type countingHistogram struct {
+ count int
+}
+
+func (h *countingHistogram) Observe(_ float64, _ ...string) { h.count++ }
+func (*countingHistogram) Delete(_ ...string) bool { return true }
+
+// TestListenBatchResponseCtxDoneTicksNoBatchFinished verifies that the
ctx.Done() early-return
+// path in listenBatchResponse does NOT increment total_batch_finished or
total_batch_latency.
+func TestListenBatchResponseCtxDoneTicksNoBatchFinished(t *testing.T) {
+ batchFinished := &countingCounter{}
+ batchLatency := &countingHistogram{}
+ pm := &pubMetrics{ //nolint:exhaustruct
+ totalStarted: &countingCounter{},
+ totalFinished: &countingCounter{},
+ totalLatency: &noopHistogram{},
+ totalErr: newErrReasonCapturer(),
+ sentBytes: &countingCounter{},
+ totalBatchStarted: &countingCounter{},
+ totalBatchFinished: batchFinished,
+ totalBatchLatency: batchLatency,
+ }
+ p := newPubWithConnMgrForMetrics(t, pm)
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ }
+
+ // Cancel the context before calling listenBatchResponse — the
ctx.Done() select fires immediately.
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ mockStream := NewMockSendClient(context.Background())
+ bc := make(chan batchEvent, 1)
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
+
+ require.Equal(t, float64(0), batchFinished.count, "total_batch_finished
must NOT be ticked on ctx.Done() early-return")
+ require.Equal(t, 0, batchLatency.count, "total_batch_latency must NOT
be observed on ctx.Done() early-return")
+}
+
+// TestListenBatchResponseSuccessTicksBatchFinished verifies that a successful
single-response path
+// increments total_batch_finished exactly once and observes
total_batch_latency exactly once with group="".
+func TestListenBatchResponseSuccessTicksBatchFinished(t *testing.T) {
+ batchFinished := &countingCounter{}
+ batchLatency := &countingHistogram{}
+ pm := &pubMetrics{ //nolint:exhaustruct
+ totalStarted: &countingCounter{},
+ totalFinished: &countingCounter{},
+ totalLatency: &noopHistogram{},
+ totalErr: newErrReasonCapturer(),
+ sentBytes: &countingCounter{},
+ totalBatchStarted: &countingCounter{},
+ totalBatchFinished: batchFinished,
+ totalBatchLatency: batchLatency,
+ }
+ p := newPubWithConnMgrForMetrics(t, pm)
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ }
+
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+ // Return a successful response (nil error, empty Error field → success
path).
+ mockStream.SetRecvFunc(func() (*clusterv1.SendResponse, error) {
+ return &clusterv1.SendResponse{}, nil
+ })
+
+ bc := make(chan batchEvent, 1)
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
+
+ require.Equal(t, float64(1), batchFinished.count, "total_batch_finished
must be 1 on success")
+ require.Equal(t, 1, batchLatency.count, "total_batch_latency must be
observed once on success")
+}
diff --git a/banyand/queue/pub/migration_metrics.go
b/banyand/queue/pub/migration_metrics.go
index 0be8e4320..3054b0ad3 100644
--- a/banyand/queue/pub/migration_metrics.go
+++ b/banyand/queue/pub/migration_metrics.go
@@ -32,20 +32,26 @@ var lifecycleMigrationScope =
observability.RootScope.SubScope("lifecycle_migrat
// from an explicitly supplied MetricsRegistry instead, on its own scope so the
// migration layer can be queried independently of the write-pipeline pub
family.
type pubMigrationMetrics struct {
- totalStarted meter.Counter
- totalFinished meter.Counter
- totalLatency meter.Histogram
- totalErr meter.Counter
- sentBytes meter.Counter
+ totalStarted meter.Counter
+ totalFinished meter.Counter
+ totalLatency meter.Histogram
+ totalErr meter.Counter
+ sentBytes meter.Counter
+ totalBatchStarted meter.Counter
+ totalBatchFinished meter.Counter
+ totalBatchLatency meter.Histogram
}
func newPubMigrationMetrics(factory observability.Factory)
*pubMigrationMetrics {
labels := []string{"operation", "group", "remote_node", "remote_role",
"remote_tier"}
return &pubMigrationMetrics{
- totalStarted: factory.NewCounter("total_started", labels...),
- totalFinished: factory.NewCounter("total_finished", labels...),
- totalLatency: factory.NewHistogram("total_latency",
meter.DefBuckets, labels...),
- totalErr: factory.NewCounter("total_err", append(labels,
"error_type")...),
- sentBytes: factory.NewCounter("sent_bytes", labels...),
+ totalStarted: factory.NewCounter("total_started",
labels...),
+ totalFinished: factory.NewCounter("total_finished",
labels...),
+ totalLatency: factory.NewHistogram("total_latency",
meter.DefBuckets, labels...),
+ totalErr: factory.NewCounter("total_err",
append(labels, "error_type")...),
+ sentBytes: factory.NewCounter("sent_bytes", labels...),
+ totalBatchStarted: factory.NewCounter("total_batch_started",
labels...),
+ totalBatchFinished: factory.NewCounter("total_batch_finished",
labels...),
+ totalBatchLatency: factory.NewHistogram("total_batch_latency",
meter.BatchBuckets, labels...),
}
}
diff --git a/banyand/queue/pub/migration_metrics_test.go
b/banyand/queue/pub/migration_metrics_test.go
index 7c600bf81..4b18d02e0 100644
--- a/banyand/queue/pub/migration_metrics_test.go
+++ b/banyand/queue/pub/migration_metrics_test.go
@@ -34,11 +34,14 @@ import (
func newPubMigrationMetricsWithErrCapture(totalErr *errReasonCapturerImpl)
*pubMigrationMetrics { //nolint:exhaustruct
return &pubMigrationMetrics{
- totalStarted: &countingCounter{},
- totalFinished: &countingCounter{},
- totalLatency: &noopHistogram{},
- totalErr: totalErr,
- sentBytes: &countingCounter{},
+ totalStarted: &countingCounter{},
+ totalFinished: &countingCounter{},
+ totalLatency: &noopHistogram{},
+ totalErr: totalErr,
+ sentBytes: &countingCounter{},
+ totalBatchStarted: &countingCounter{},
+ totalBatchFinished: &countingCounter{},
+ totalBatchLatency: &noopHistogram{},
}
}
@@ -122,3 +125,54 @@ func TestPublishMirrorsStartedFinishedToMigrationMetrics(t
*testing.T) {
require.Equal(t, float64(1), pubFinished.count)
require.Equal(t, float64(1), migFinished.count, "migration finished
must mirror pub finished")
}
+
+// TestMigrationMirrorBatchFinishedAndLatency verifies that the migration
mirror's
+// totalBatchFinished and totalBatchLatency are ticked from
listenBatchResponse on a
+// terminal (success) response path — the first latency observation in that
function for the mirror.
+func TestMigrationMirrorBatchFinishedAndLatency(t *testing.T) {
+ pubBatchFinished := &countingCounter{}
+ migBatchFinished := &countingCounter{}
+ pubBatchLatency := &countingHistogram{}
+ migBatchLatency := &countingHistogram{}
+
+ pm := &pubMetrics{ //nolint:exhaustruct
+ totalStarted: &countingCounter{},
+ totalFinished: &countingCounter{},
+ totalLatency: &noopHistogram{},
+ totalErr: newErrReasonCapturer(),
+ sentBytes: &countingCounter{},
+ totalBatchStarted: &countingCounter{},
+ totalBatchFinished: pubBatchFinished,
+ totalBatchLatency: pubBatchLatency,
+ }
+ p := newPubWithConnMgrForMetrics(t, pm)
+ p.migrationMetrics = &pubMigrationMetrics{ //nolint:exhaustruct
+ totalStarted: &countingCounter{},
+ totalFinished: &countingCounter{},
+ totalLatency: &noopHistogram{},
+ totalErr: newErrReasonCapturer(),
+ sentBytes: &countingCounter{},
+ totalBatchStarted: &countingCounter{},
+ totalBatchFinished: migBatchFinished,
+ totalBatchLatency: migBatchLatency,
+ }
+
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ }
+
+ ctx := context.Background()
+ mockStream := NewMockSendClient(ctx)
+ mockStream.SetRecvFunc(func() (*clusterv1.SendResponse, error) {
+ return &clusterv1.SendResponse{}, nil
+ })
+
+ bc := make(chan batchEvent, 1)
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a",
time.Now())
+
+ require.Equal(t, float64(1), pubBatchFinished.count, "pub
total_batch_finished must be 1 on success")
+ require.Equal(t, float64(1), migBatchFinished.count, "migration
total_batch_finished must mirror pub on success")
+ require.Equal(t, 1, pubBatchLatency.count, "pub total_batch_latency
must be observed once on success")
+ require.Equal(t, 1, migBatchLatency.count, "migration
total_batch_latency must mirror pub on success")
+}
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index c0805d254..a51996888 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -101,21 +101,27 @@ type nodeInfo struct {
}
type pubMetrics struct {
- totalStarted meter.Counter
- totalFinished meter.Counter
- totalLatency meter.Histogram
- totalErr meter.Counter
- sentBytes meter.Counter
+ totalStarted meter.Counter
+ totalFinished meter.Counter
+ totalLatency meter.Histogram
+ totalErr meter.Counter
+ sentBytes meter.Counter
+ totalBatchStarted meter.Counter
+ totalBatchFinished meter.Counter
+ totalBatchLatency meter.Histogram
}
func newPubMetrics(factory observability.Factory) *pubMetrics {
labels := []string{"operation", "group", "remote_node", "remote_role",
"remote_tier"}
return &pubMetrics{
- totalStarted: factory.NewCounter("total_started", labels...),
- totalFinished: factory.NewCounter("total_finished", labels...),
- totalLatency: factory.NewHistogram("total_latency",
meter.DefBuckets, labels...),
- totalErr: factory.NewCounter("total_err", append(labels,
"error_type")...),
- sentBytes: factory.NewCounter("sent_bytes", labels...),
+ totalStarted: factory.NewCounter("total_started",
labels...),
+ totalFinished: factory.NewCounter("total_finished",
labels...),
+ totalLatency: factory.NewHistogram("total_latency",
meter.DefBuckets, labels...),
+ totalErr: factory.NewCounter("total_err",
append(labels, "error_type")...),
+ sentBytes: factory.NewCounter("sent_bytes", labels...),
+ totalBatchStarted: factory.NewCounter("total_batch_started",
labels...),
+ totalBatchFinished: factory.NewCounter("total_batch_finished",
labels...),
+ totalBatchLatency: factory.NewHistogram("total_batch_latency",
meter.BatchBuckets, labels...),
}
}
diff --git a/banyand/queue/sub/batch_metrics_test.go
b/banyand/queue/sub/batch_metrics_test.go
new file mode 100644
index 000000000..53e8e0ef8
--- /dev/null
+++ b/banyand/queue/sub/batch_metrics_test.go
@@ -0,0 +1,298 @@
+// 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 sub
+
+import (
+ "context"
+ "errors"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+ "google.golang.org/grpc/metadata"
+
+ "github.com/apache/skywalking-banyandb/api/data"
+ clusterv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// mockSendServer implements clusterv1.Service_SendServer for sub batch tests.
+type mockSendServer struct {
+ ctx context.Context
+ sendErr error
+ sent []*clusterv1.SendResponse
+ mu sync.Mutex
+}
+
+func newMockSendServer() *mockSendServer {
+ return &mockSendServer{
+ ctx: metadata.NewIncomingContext(context.Background(),
metadata.MD{}),
+ }
+}
+
+func (m *mockSendServer) Send(resp *clusterv1.SendResponse) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.sent = append(m.sent, resp)
+ return m.sendErr
+}
+
+func (m *mockSendServer) Recv() (*clusterv1.SendRequest, error) { return nil,
nil }
+func (m *mockSendServer) Context() context.Context { return m.ctx
}
+func (m *mockSendServer) SetHeader(metadata.MD) error { return nil }
+func (m *mockSendServer) SendHeader(metadata.MD) error { return nil }
+func (m *mockSendServer) SetTrailer(metadata.MD) {}
+func (m *mockSendServer) SendMsg(_ any) error { return nil }
+func (m *mockSendServer) RecvMsg(_ any) error { return nil }
+
+// echoListener is a healthy MessageListener that echoes messages back.
+type echoListener struct {
+ bus.UnImplementedHealthyListener
+}
+
+func (*echoListener) Rev(_ context.Context, msg bus.Message) bus.Message {
return msg }
+
+func newBatchTestServer(m *metrics) *server { //nolint:exhaustruct
+ logInitErr := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ if logInitErr != nil {
+ panic(logInitErr)
+ }
+ return &server{
+ log: logger.GetLogger("batch-metrics-test"),
+ metrics: m,
+ listeners: make(map[bus.Topic][]bus.MessageListener),
+ listenersLock: sync.RWMutex{},
+ topicMap: make(map[string]bus.Topic),
+ }
+}
+
+func newBatchMetrics() (m *metrics, msgStarted, msgFinished, batchStarted,
batchFinished *fakeCounter, batchLatency *fakeHistogram) {
+ msgStarted = &fakeCounter{}
+ msgFinished = &fakeCounter{}
+ batchStarted = &fakeCounter{}
+ batchFinished = &fakeCounter{}
+ batchLatency = &fakeHistogram{}
+ m = &metrics{ //nolint:exhaustruct
+ totalStarted: &fakeCounter{},
+ totalFinished: &fakeCounter{},
+ totalLatency: &fakeHistogram{},
+ totalErr: &fakeCounter{},
+ receivedBytes: &fakeCounter{},
+ totalMessageStarted: msgStarted,
+ totalMessageFinished: msgFinished,
+ totalBatchStarted: batchStarted,
+ totalBatchFinished: batchFinished,
+ totalBatchLatency: batchLatency,
+ }
+ return m, msgStarted, msgFinished, batchStarted, batchFinished,
batchLatency
+}
+
+func makeBatchReq(group, senderNode string) *clusterv1.SendRequest {
//nolint:exhaustruct
+ return &clusterv1.SendRequest{
+ Topic: data.TopicMeasureWrite.String(),
+ Group: group,
+ SenderNode: senderNode,
+ SenderRole: "liaison",
+ SenderTier: "hot",
+ BatchMod: true,
+ Body: []byte("data"),
+ }
+}
+
+// TestBatchModNMessageMetrics asserts that for an N-message BatchMod batch:
+// total_message_started==N, total_message_finished==N,
total_batch_started==1, total_batch_finished==1,
+// and total_batch_latency is observed exactly once.
+func TestBatchModNMessageMetrics(t *testing.T) {
+ const N = 3
+ logInitErr := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ require.NoError(t, logInitErr)
+
+ m, msgStarted, msgFinished, batchStarted, batchFinished, batchLatency
:= newBatchMetrics()
+ s := newBatchTestServer(m)
+
+ topic := data.TopicMeasureWrite
+ require.NoError(t, s.Subscribe(topic, &echoListener{}))
+
+ stream := newMockSendServer()
+ req := makeBatchReq("g1", "liaison-0:17912")
+
+ identity := &streamIdentity{}
+ var dataCollection []any
+ start := time.Now()
+
+ // Mirror the production Send path: pin identity before the first
handleBatch call.
+ s.pinIdentity(identity, req, topic)
+
+ // Simulate N batch messages (handleBatch called N times with the same
identity).
+ for range N {
+ s.handleBatch(&dataCollection, req, &start, identity)
+ }
+
+ // Simulate EOF: handleEOF dispatches the collected batch.
+ s.handleEOF(stream, &topic, dataCollection, req, identity, start)
+
+ require.Equal(t, float64(N), msgStarted.total, "total_message_started
must equal N")
+ require.Equal(t, float64(N), msgFinished.total, "total_message_finished
must equal N")
+ require.Equal(t, float64(1), batchStarted.total, "total_batch_started
must equal 1")
+ require.Equal(t, float64(1), batchFinished.total, "total_batch_finished
must equal 1")
+ require.Equal(t, 1, batchLatency.count, "total_batch_latency must be
observed exactly once")
+}
+
+// TestBatchModSendFailureMetrics verifies that a Send failure on the response
does NOT affect
+// the four metric counts — they are pinned after listener.Rev, independent of
Send outcome.
+func TestBatchModSendFailureMetrics(t *testing.T) {
+ const N = 4
+ logInitErr := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ require.NoError(t, logInitErr)
+
+ m, msgStarted, msgFinished, batchStarted, batchFinished, batchLatency
:= newBatchMetrics()
+ s := newBatchTestServer(m)
+
+ topic := data.TopicMeasureWrite
+ require.NoError(t, s.Subscribe(topic, &echoListener{}))
+
+ // Wire a Send that always fails (simulates client disconnect before
reading EOF response).
+ stream := newMockSendServer()
+ stream.sendErr = errors.New("client disconnect")
+
+ req := makeBatchReq("g2", "liaison-1:17912")
+ identity := &streamIdentity{}
+ var dataCollection []any
+ start := time.Now()
+
+ // Mirror the production Send path: pin identity before the first
handleBatch call.
+ s.pinIdentity(identity, req, topic)
+
+ for range N {
+ s.handleBatch(&dataCollection, req, &start, identity)
+ }
+ s.handleEOF(stream, &topic, dataCollection, req, identity, start)
+
+ require.Equal(t, float64(N), msgStarted.total, "total_message_started
must equal N even when Send fails")
+ require.Equal(t, float64(N), msgFinished.total, "total_message_finished
must equal N even when Send fails")
+ require.Equal(t, float64(1), batchStarted.total, "total_batch_started
must equal 1 even when Send fails")
+ require.Equal(t, float64(1), batchFinished.total, "total_batch_finished
must equal 1 even when Send fails")
+ require.Equal(t, 1, batchLatency.count, "total_batch_latency must be
observed once even when Send fails")
+}
+
+// TestHandleEOFEmptyCollectionTicksNothing verifies that handleEOF with an
empty dataCollection
+// returns early without ticking any of the new instruments.
+func TestHandleEOFEmptyCollectionTicksNothing(t *testing.T) {
+ logInitErr := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ require.NoError(t, logInitErr)
+
+ m, msgStarted, msgFinished, batchStarted, batchFinished, batchLatency
:= newBatchMetrics()
+ s := newBatchTestServer(m)
+
+ topic := data.TopicMeasureWrite
+ require.NoError(t, s.Subscribe(topic, &echoListener{}))
+
+ stream := newMockSendServer()
+ identity := &streamIdentity{
+ senderNode: "liaison-0:17912",
+ senderRole: "liaison",
+ senderTier: "hot",
+ group: "g1",
+ operation: "batch-write",
+ pinned: true,
+ }
+
+ // Call handleEOF with an empty collection — must return immediately
without ticking.
+ s.handleEOF(stream, &topic, []any{}, nil, identity, time.Now())
+
+ require.Equal(t, float64(0), msgStarted.total, "no message_started on
empty collection")
+ require.Equal(t, float64(0), msgFinished.total, "no message_finished on
empty collection")
+ require.Equal(t, float64(0), batchStarted.total, "no batch_started on
empty collection")
+ require.Equal(t, float64(0), batchFinished.total, "no batch_finished on
empty collection")
+ require.Equal(t, 0, batchLatency.count, "no batch_latency on empty
collection")
+}
+
+// TestHandleEOFNoListenerTicksNothing verifies that handleEOF with no
registered listener
+// returns (via replyWithErrType) without ticking any of the new instruments.
+func TestHandleEOFNoListenerTicksNothing(t *testing.T) {
+ logInitErr := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ require.NoError(t, logInitErr)
+
+ m, msgStarted, msgFinished, batchStarted, batchFinished, batchLatency
:= newBatchMetrics()
+ s := newBatchTestServer(m)
+
+ // No listener registered for this topic.
+ topic := data.TopicMeasureWrite
+ stream := newMockSendServer()
+ req := makeBatchReq("g1", "liaison-0:17912")
+ identity := &streamIdentity{
+ senderNode: "liaison-0:17912",
+ senderRole: "liaison",
+ senderTier: "hot",
+ group: "g1",
+ operation: "batch-write",
+ pinned: true,
+ }
+
+ // Non-empty dataCollection but no listener — must not tick
finished/batch.
+ s.handleEOF(stream, &topic, []any{req.Body}, req, identity, time.Now())
+
+ require.Equal(t, float64(0), msgStarted.total, "no message_started when
no listener")
+ require.Equal(t, float64(0), msgFinished.total, "no message_finished
when no listener")
+ require.Equal(t, float64(0), batchStarted.total, "no batch_started when
no listener")
+ require.Equal(t, float64(0), batchFinished.total, "no batch_finished
when no listener")
+ require.Equal(t, 0, batchLatency.count, "no batch_latency when no
listener")
+}
+
+// TestDispatchMessageNonBatchMetrics verifies that a single dispatchMessage
call increments
+// total_message_started==1 and total_message_finished==1.
+func TestDispatchMessageNonBatchMetrics(t *testing.T) {
+ logInitErr := logger.Init(logger.Logging{Env: "dev", Level: "info"})
+ require.NoError(t, logInitErr)
+
+ m, msgStarted, msgFinished, batchStarted, batchFinished, batchLatency
:= newBatchMetrics()
+ s := newBatchTestServer(m)
+
+ topic := data.TopicMeasureWrite
+ require.NoError(t, s.Subscribe(topic, &echoListener{}))
+
+ stream := newMockSendServer()
+ req := &clusterv1.SendRequest{ //nolint:exhaustruct
+ Topic: data.TopicMeasureWrite.String(),
+ Group: "g1",
+ SenderNode: "liaison-0:17912",
+ SenderRole: "liaison",
+ SenderTier: "hot",
+ BatchMod: false,
+ Body: []byte("payload"),
+ }
+ identity := &streamIdentity{
+ senderNode: "liaison-0:17912",
+ senderRole: "liaison",
+ senderTier: "hot",
+ group: "g1",
+ operation: "batch-write",
+ pinned: true,
+ }
+ msg := bus.NewMessage(bus.MessageID(1), req.Body)
+ start := time.Now()
+ s.dispatchMessage(stream, req, topic, msg, identity, &start)
+
+ require.Equal(t, float64(1), msgStarted.total, "total_message_started
must be 1 on dispatchMessage")
+ require.Equal(t, float64(1), msgFinished.total, "total_message_finished
must be 1 on dispatchMessage")
+ require.Zero(t, batchStarted.total, "the non-batch path must not tick
the batch catalog")
+ require.Zero(t, batchFinished.total, "the non-batch path must not tick
the batch catalog")
+ require.Zero(t, batchLatency.count, "the non-batch path must not
observe batch latency")
+}
diff --git a/banyand/queue/sub/helpers.go b/banyand/queue/sub/helpers.go
index b8b77e93d..bd16f7f86 100644
--- a/banyand/queue/sub/helpers.go
+++ b/banyand/queue/sub/helpers.go
@@ -29,7 +29,9 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
)
-func (s *server) handleEOF(stream clusterv1.Service_SendServer, topic
*bus.Topic, dataCollection []any, writeEntity *clusterv1.SendRequest, identity
*streamIdentity) {
+func (s *server) handleEOF(stream clusterv1.Service_SendServer, topic
*bus.Topic, dataCollection []any,
+ writeEntity *clusterv1.SendRequest, identity *streamIdentity, start
time.Time,
+) {
if len(dataCollection) < 1 {
return
}
@@ -47,6 +49,13 @@ func (s *server) handleEOF(stream
clusterv1.Service_SendServer, topic *bus.Topic
return
}
message := listener.Rev(stream.Context(),
bus.NewMessage(bus.MessageID(0), dataCollection))
+ // Tick batch + message finished immediately after Rev returns,
independent of the response Send outcome.
+ // Pre-Rev early returns (len<1, no-listener, CheckHealth fail) are
excluded — they return before this point.
+ if s.metrics != nil {
+
s.metrics.totalMessageFinished.Inc(float64(len(dataCollection)),
identity.operation, identity.group, identity.senderNode, identity.senderRole,
identity.senderTier)
+ s.metrics.totalBatchFinished.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
+
s.metrics.totalBatchLatency.Observe(time.Since(start).Seconds(),
identity.operation, identity.group, identity.senderNode, identity.senderRole,
identity.senderTier)
+ }
// writeEntity is nil when the stream closed (Recv returned io.EOF), so
guard the ID access.
var msgID uint64
if writeEntity != nil {
@@ -77,8 +86,12 @@ func (s *server) handleBatch(dataCollection *[]any,
writeEntity *clusterv1.SendR
if len(*dataCollection) == 0 {
if s.metrics != nil {
s.metrics.totalStarted.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
+ s.metrics.totalBatchStarted.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
}
*start = time.Now()
}
+ if s.metrics != nil {
+ s.metrics.totalMessageStarted.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
+ }
*dataCollection = append(*dataCollection, writeEntity.Body)
}
diff --git a/banyand/queue/sub/server.go b/banyand/queue/sub/server.go
index 72ca46d71..f19236733 100644
--- a/banyand/queue/sub/server.go
+++ b/banyand/queue/sub/server.go
@@ -444,20 +444,30 @@ func (s *server) SetNodeSchemaStatusRepo(svc
metadata.Service) {
}
type metrics struct {
- totalStarted meter.Counter
- totalFinished meter.Counter
- totalLatency meter.Histogram
- totalErr meter.Counter
- receivedBytes meter.Counter
+ totalStarted meter.Counter
+ totalFinished meter.Counter
+ totalLatency meter.Histogram
+ totalErr meter.Counter
+ receivedBytes meter.Counter
+ totalBatchStarted meter.Counter
+ totalBatchFinished meter.Counter
+ totalBatchLatency meter.Histogram
+ totalMessageStarted meter.Counter
+ totalMessageFinished meter.Counter
}
func newMetrics(factory observability.Factory) *metrics {
labels := []string{"operation", "group", "remote_node", "remote_role",
"remote_tier"}
return &metrics{
- totalStarted: factory.NewCounter("total_started", labels...),
- totalFinished: factory.NewCounter("total_finished", labels...),
- totalLatency: factory.NewHistogram("total_latency",
meter.DefBuckets, labels...),
- totalErr: factory.NewCounter("total_err", append(labels,
"error_type")...),
- receivedBytes: factory.NewCounter("received_bytes", labels...),
+ totalStarted: factory.NewCounter("total_started",
labels...),
+ totalFinished: factory.NewCounter("total_finished",
labels...),
+ totalLatency: factory.NewHistogram("total_latency",
meter.DefBuckets, labels...),
+ totalErr: factory.NewCounter("total_err",
append(labels, "error_type")...),
+ receivedBytes: factory.NewCounter("received_bytes",
labels...),
+ totalBatchStarted: factory.NewCounter("total_batch_started",
labels...),
+ totalBatchFinished:
factory.NewCounter("total_batch_finished", labels...),
+ totalBatchLatency:
factory.NewHistogram("total_batch_latency", meter.BatchBuckets, labels...),
+ totalMessageStarted:
factory.NewCounter("total_message_started", labels...),
+ totalMessageFinished:
factory.NewCounter("total_message_finished", labels...),
}
}
diff --git a/banyand/queue/sub/server_metrics_test.go
b/banyand/queue/sub/server_metrics_test.go
index 91638af00..c47130a55 100644
--- a/banyand/queue/sub/server_metrics_test.go
+++ b/banyand/queue/sub/server_metrics_test.go
@@ -43,11 +43,16 @@ func newTestMetrics() *metrics { //nolint:exhaustruct
c := &noopCounter{}
h := &noopHistogram{}
return &metrics{
- totalStarted: c,
- totalFinished: c,
- totalLatency: h,
- totalErr: c,
- receivedBytes: c,
+ totalStarted: c,
+ totalFinished: c,
+ totalLatency: h,
+ totalErr: c,
+ receivedBytes: c,
+ totalMessageStarted: c,
+ totalMessageFinished: c,
+ totalBatchStarted: c,
+ totalBatchFinished: c,
+ totalBatchLatency: h,
}
}
diff --git a/banyand/queue/sub/sub.go b/banyand/queue/sub/sub.go
index 37307e327..f4e53e018 100644
--- a/banyand/queue/sub/sub.go
+++ b/banyand/queue/sub/sub.go
@@ -125,7 +125,7 @@ func (s *server) Send(stream clusterv1.Service_SendServer)
error {
}
writeEntity, err := stream.Recv()
if errors.Is(err, io.EOF) {
- s.handleEOF(stream, topic, dataCollection, writeEntity,
identity)
+ s.handleEOF(stream, topic, dataCollection, writeEntity,
identity, start)
return nil
}
if err != nil {
@@ -258,6 +258,7 @@ func (s *server) dispatchMessage(
) {
if s.metrics != nil {
s.metrics.totalStarted.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
+ s.metrics.totalMessageStarted.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
}
listeners := s.getListeners(topic)
if len(listeners) == 0 {
@@ -268,6 +269,11 @@ func (s *server) dispatchMessage(
logger.Panicf("multiple listeners found for topic %s", topic)
}
m = listeners[0].Rev(stream.Context(), m)
+ // The BatchMod fork in Send routes each SendRequest to exactly one of
+ // handleBatch→handleEOF (batch) or dispatchMessage (non-batch), so no
message is double-counted.
+ if s.metrics != nil {
+ s.metrics.totalMessageFinished.Inc(1, identity.operation,
identity.group, identity.senderNode, identity.senderRole, identity.senderTier)
+ }
if m.Data() == nil {
if errSend := stream.Send(&clusterv1.SendResponse{MessageId:
writeEntity.MessageId}); errSend != nil {
s.log.Error().Stringer("request",
writeEntity).Err(errSend).Msg("failed to send empty response")
diff --git a/banyand/trace/metadata_test.go b/banyand/trace/metadata_test.go
index a6e76bc91..80faf468f 100644
--- a/banyand/trace/metadata_test.go
+++ b/banyand/trace/metadata_test.go
@@ -392,7 +392,7 @@ var _ = Describe("Metadata", func() {
// part count is stable (no further writes), so
slow polling never misses it.
Eventually(func() int64 {
return getTotalPartCount(svcs,
groupName)
- }, 3*flags.EventuallyTimeout,
500*time.Millisecond).Should(BeNumerically("<", partCountBeforeMerge))
+ }, 10*flags.EventuallyTimeout,
500*time.Millisecond).Should(BeNumerically("<", partCountBeforeMerge))
Eventually(func(innerGm Gomega) {
spans :=
querySchemaChangeTraceData(svcs, traceName, groupName,
diff --git a/banyand/trace/snapshot_test.go b/banyand/trace/snapshot_test.go
index f8909c9dd..f2787731b 100644
--- a/banyand/trace/snapshot_test.go
+++ b/banyand/trace/snapshot_test.go
@@ -528,6 +528,24 @@ func TestSnapshotRemove(t *testing.T) {
}
}
+// waitForPersistedSnapshot blocks until a flush has been introduced and its
+// ".snp" manifest persisted. Waiting for the part directory alone is racy: the
+// directory is created by the first line of memPart.mustFlush, before the
+// mem->file introduction reaches the in-memory snapshot and before the
manifest
+// is written, so the directory can exist while the snapshot still holds only
+// mem parts and no ".snp" has been persisted.
+func waitForPersistedSnapshot(t *testing.T, fileSystem fs.FileSystem, tabDir
string) {
+ t.Helper()
+ require.Eventually(t, func() bool {
+ for _, e := range fileSystem.ReadDir(tabDir) {
+ if !e.IsDir() && filepath.Ext(e.Name()) ==
snapshotSuffix {
+ return true
+ }
+ }
+ return false
+ }, flags.EventuallyTimeout, time.Millisecond, "wait for snapshot
manifest to be persisted")
+}
+
func TestSnapshotFunctionality(t *testing.T) {
fileSystem := fs.NewLocalFileSystem()
@@ -557,18 +575,10 @@ func TestSnapshotFunctionality(t *testing.T) {
tst.mustAddTraces(tsTS1, nil)
tst.mustAddTraces(tsTS2, nil)
- time.Sleep(100 * time.Millisecond) // allow time for flushing
-
- require.Eventually(t, func() bool {
- dd := fileSystem.ReadDir(tabDir)
- partNum := 0
- for _, d := range dd {
- if d.IsDir() {
- partNum++
- }
- }
- return partNum >= 1
- }, flags.EventuallyTimeout, time.Millisecond, "wait for file parts to
be created")
+ // Wait until the flush has been introduced and its ".snp" manifest
persisted,
+ // not merely until a part directory appears (which happens at the
start of
+ // mustFlush, before the mem->file introduction reaches the snapshot).
+ waitForPersistedSnapshot(t, fileSystem, tabDir)
snapshotPath := filepath.Join(tmpPath, "snapshot")
fileSystem.MkdirIfNotExist(snapshotPath, 0o755)
diff --git a/docs/operation/grafana-fodc-nodes.json
b/docs/operation/grafana-fodc-nodes.json
index a7145ffa3..7b4d0552f 100644
--- a/docs/operation/grafana-fodc-nodes.json
+++ b/docs/operation/grafana-fodc-nodes.json
@@ -659,7 +659,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "description": "Each row is one directed flow source→target (bare pod
names) per operation, with the PUBLISHER's view and the SUBSCRIBER's view of
the SAME traffic side by side — never sum the two. Request edges pair
banyandb_queue_pub_* (sender) with banyandb_queue_sub_* (receiver);
tier-migration edges pair banyandb_lifecycle_migration_* (lifecycle sidecar,
sharing its data pod's pod_name) with the receiver's queue_sub
remote_role=\"lifecycle\" series. The two sides are joined on [...]
+ "description": "Each row is one directed flow source→target (bare pod
names) per operation, with the PUBLISHER's view and the SUBSCRIBER's view of
the SAME traffic side by side — never sum the two. Request edges pair
banyandb_queue_pub_* (sender) with banyandb_queue_sub_* (receiver);
tier-migration edges pair banyandb_lifecycle_migration_* (lifecycle sidecar,
sharing its data pod's pod_name) with the receiver's queue_sub
remote_role=\"lifecycle\" series. The two sides are joined on [...]
"fieldConfig": {
"defaults": {
"color": {
@@ -909,7 +909,7 @@
},
"editorMode": "code",
"exemplar": false,
- "expr": "sum by (source, target, operation)
(label_replace(label_replace(rate(banyandb_queue_sub_total_started{job=~\"$job\",
container_name=~\"$role\", pod_name=~\"$pod\"}[$__rate_interval]) or
rate(banyandb_queue_sub_total_started{job=~\"$job\", remote_role=~\"$role\",
remote_node=~\"($pod)\\\\..*\"}[$__rate_interval]), \"source\", \"$1\",
\"remote_node\", \"([^.:]+).*\"), \"target\", \"$1\", \"pod_name\", \"(.*)\"))",
+ "expr": "sum by (source, target, operation)
(label_replace(label_replace(rate(banyandb_queue_sub_total_message_finished{job=~\"$job\",
container_name=~\"$role\", pod_name=~\"$pod\"}[$__rate_interval]) or
rate(banyandb_queue_sub_total_message_finished{job=~\"$job\",
remote_role=~\"$role\", remote_node=~\"($pod)\\\\..*\"}[$__rate_interval]),
\"source\", \"$1\", \"remote_node\", \"([^.:]+).*\"), \"target\", \"$1\",
\"pod_name\", \"(.*)\"))",
"format": "table",
"instant": true,
"legendFormat": "__auto",
diff --git a/docs/operation/observability/metrics.md
b/docs/operation/observability/metrics.md
index f05703931..b0048bf6d 100644
--- a/docs/operation/observability/metrics.md
+++ b/docs/operation/observability/metrics.md
@@ -92,12 +92,12 @@ sum by (source, target, operation)
(label_replace(label_replace(
"source", "$1", "pod_name", "(.*)"), "target", "$1", "remote_node",
"([^.:]+).*"))
```
-**Sub side (receiver view; inverted mapping — `total_started` is the
receiver's per-dispatch counter, the closest match to the publisher's
per-message count):**
+**Sub side (receiver view; inverted mapping — `total_message_finished` is the
receiver's per-message catalog, the unit-for-unit match to the publisher's
per-message `total_finished`):**
```promql
sum by (source, target, operation) (label_replace(label_replace(
- rate(banyandb_queue_sub_total_started{job=~"$job", container_name=~"$role",
pod_name=~"$pod"}[$__rate_interval])
- or rate(banyandb_queue_sub_total_started{job=~"$job",
remote_role=~"$role", remote_node=~"($pod)\\..*"}[$__rate_interval]),
+ rate(banyandb_queue_sub_total_message_finished{job=~"$job",
container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])
+ or rate(banyandb_queue_sub_total_message_finished{job=~"$job",
remote_role=~"$role", remote_node=~"($pod)\\..*"}[$__rate_interval]),
"source", "$1", "remote_node", "([^.:]+).*"), "target", "$1", "pod_name",
"(.*)"))
```
@@ -105,7 +105,7 @@ The p99 columns apply the same relabeling inside
`histogram_quantile(0.99, sum b
Reading the table:
-- The two sides describe the **same traffic — never add them**. A healthy
steady-state edge shows roughly equal `Pub msg/s` and `Sub msg/s` — except for
**batch-mode** streams: the liaison→liaison **tier-1 write routing**, where
each write the SkyWalking OAP sends to its connected liaison is re-published
**round-robin to the owner liaison** (including itself, over real gRPC), one
wire stream per OAP write stream, and the receiver collects the whole stream
and dispatches it **once at stre [...]
+- The two sides describe the **same traffic — never add them**. Both `Pub
msg/s` (`queue_pub_total_finished`) and `Sub msg/s`
(`queue_sub_total_message_finished`) now count the **same per-message unit**,
so a healthy steady-state edge shows them ≈ equal on every edge — including the
liaison→liaison **tier-1 write routing**, where each write the SkyWalking OAP
sends to its connected liaison is re-published **round-robin to the owner
liaison** (including itself), one wire stream per OAP wr [...]
- A populated Pub cell with an **empty Sub cell** (or vice versa) is also
signal: an uninstrumented side (older servers record `query`/`control` only on
the receiver) or a one-sided counter (`sent_bytes` exists only on
pub/migration, `received_bytes` only on sub; both are file-sync-only).
- Empty `err` cells mean **no errors** (the counters are lazily registered).
- `$pod` / `$role` select flows **involving** a matching node on **either
end**. A naive `pod_name=~"$pod"` filter would hit the two sides asymmetrically
(pub rows are scraped from the sender, sub rows from the receiver) and split
the merged rows; instead every query unions two selectors — one matching the
scrape-target side (`pod_name` / `container_name`) and one matching the remote
side (`remote_node=~"($pod)\\..*"`, a prefix match on the full DNS node name,
and `remote_role=~"$role"`, [...]
@@ -386,6 +386,8 @@ Both namespaces share one model: the base metrics
`total_started`, `total_finish
| `total_latency` | Histogram | `operation`, `group`, `remote_node`,
`remote_role`, `remote_tier` | Handling latency (`_bucket` / `_sum` /
`_count`); use `histogram_quantile` for p50/p99. |
| `total_err` | Counter | …, `error_type` | Errors by type. Lazily registered,
so absent (not zero) on a healthy cluster. |
| `received_bytes` | Counter | `operation`, `group`, `remote_node`,
`remote_role`, `remote_tier` | Bytes received, **file-sync only**
(`operation="file-sync"`). |
+| `total_message_started`, `total_message_finished` | Counter | `operation`,
`group`, `remote_node`, `remote_role`, `remote_tier` | **Message catalog**
(`queue_sub` only) — one count per dispatched message on every path: per
message for ordinary traffic, per part for file-sync, and per message *within*
a batch-mode stream (where `total_started`/`total_finished` count the whole
batch once). This is the unit-for-unit match to the publisher's per-message
`total_finished`, so the two sides c [...]
+| `total_batch_started`, `total_batch_finished`, `total_batch_latency` |
Counter / Histogram | `operation`, `group`, `remote_node`, `remote_role`,
`remote_tier` | **Batch catalog** — one count per batch-mode wire stream:
started at stream open, finished + latency observed at EOF dispatch.
`total_batch_latency` buckets extend to ~300s for minutes-long migration/replay
streams. Also present on `queue_pub` and the `lifecycle_migration` mirror. |
**Troubleshooting:** a growing `total_started − total_finished` gap (or rising
`total_latency` p99) for a `group`/`operation` points at slow or stuck
consumers. `total_err` broken down by `error_type` distinguishes transport
issues (`stream_error`, `recv_error`, `checksum_mismatch`, `out_of_order`) from
completion issues (`finish_sync_err`, `part_failed`). For file-sync,
`received_bytes` together with `total_latency` separates partial completion
from healthy throughput. On data nodes, se [...]
@@ -397,6 +399,7 @@ Both namespaces share one model: the base metrics
`total_started`, `total_finish
| `total_latency` | Histogram | `operation`, `group`, `remote_node`,
`remote_role`, `remote_tier` | Send latency (`_bucket` / `_sum` / `_count`);
use `histogram_quantile` for p99. |
| `total_err` | Counter | …, `error_type` | Send errors by type. `error_type`
is one of `non_transient`, `canceled`, `stream_canceled`, `retry_exhausted`,
`recv_error`, `server_rejected`, `send_error`, `decode_error`, `invalid_topic`
(Send/publish path) or `stream_error`, `recv_error`, `checksum_mismatch`,
`out_of_order`, `session_not_found`, `completion_error` (file-sync). Lazily
registered. |
| `sent_bytes` | Counter | `operation`, `group`, `remote_node`, `remote_role`,
`remote_tier` | Bytes sent, **file-sync only**. |
+| `total_batch_started`, `total_batch_finished`, `total_batch_latency` |
Counter / Histogram | `operation`, `group`, `remote_node`, `remote_role`,
`remote_tier` | **Batch catalog** — one count per batch-mode wire stream:
started at stream open (`group=""`), finished + latency observed when the
stream's final response is received. Note `total_batch_finished` means the
final response arrived, **not** that every message was acked (per-message acks
await a future `cluster.v1.SendResponse.acc [...]
**Troubleshooting:** `total_err` by `error_type` separates transport failures
(`recv_error`) from application-level `SendResponse` errors (`server_rejected`)
and exhausted retries (`retry_exhausted`). A persistent `total_started −
total_finished` gap, or rising `total_latency` p99 for a given `remote_node` /
`remote_tier`, indicates slow or unavailable data nodes.
@@ -404,7 +407,7 @@ Metrics are only registered when `metadata` implements
`metadata.Service` and `M
### `lifecycle_migration` — the tier-migration mirror of `queue_pub`
-The **lifecycle** service migrates data hot→warm→cold through a queue client
built without a metadata service (`pub.NewWithoutMetadata`), so its traffic
does **not** appear under `banyandb_queue_pub_*`. Instead the same five
instruments — `total_started`, `total_finished`, `total_latency`, `total_err`,
`sent_bytes` — are registered under **`banyandb_lifecycle_migration_*`**, with
the same labels (`operation`, `group`, `remote_node`, `remote_role`,
`remote_tier`, plus `error_type` on `tot [...]
+The **lifecycle** service migrates data hot→warm→cold through a queue client
built without a metadata service (`pub.NewWithoutMetadata`), so its traffic
does **not** appear under `banyandb_queue_pub_*`. Instead the same instruments
— `total_started`, `total_finished`, `total_latency`, `total_err`,
`sent_bytes`, plus the batch catalog
`total_batch_started`/`total_batch_finished`/`total_batch_latency` — are
registered under **`banyandb_lifecycle_migration_*`**, with the same labels
(`opera [...]
The lifecycle sidecar serves these series (plus the run-health gauges below)
at `/metrics` on its own HTTP port (`--lifecycle-http-port`, default `17915`)
instead of the `2121` observability listener; the co-located FODC agent polls
that port, so the series arrive through the proxy scrape with
`container_name="lifecycle"` and the `pod_name` shared with the co-located data
node. On the **receiving** data node the same flows are mirrored as
`banyandb_queue_sub_*` series with `remote_role=" [...]
diff --git a/pkg/meter/meter.go b/pkg/meter/meter.go
index a52aef9de..66382302a 100644
--- a/pkg/meter/meter.go
+++ b/pkg/meter/meter.go
@@ -31,6 +31,11 @@ type (
// DefBuckets is the default buckets for histograms.
var DefBuckets = Buckets{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10}
+// BatchBuckets is the bucket set for batch-stream latency histograms.
Batch-mode
+// streams and lifecycle migration replays can run for minutes, so the upper
bound
+// extends to 300s instead of the 10s ceiling of DefBuckets.
+var BatchBuckets = Buckets{0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300}
+
// Merge merges the given label pairs with the current label pairs.
func (p LabelPairs) Merge(other LabelPairs) LabelPairs {
result := make(LabelPairs, len(p)+len(other))
