You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/06/11 05:15:42 UTC
(skywalking-banyandb) branch main updated: feat(queue): add message + batch unit catalogs to batch-write metrics (#1169)
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 <ha...@gmail.com>
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))