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/08/23 14:49:31 UTC
(skywalking-banyandb) branch main updated: Add liaison write-path latency metrics (#1299)
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 7638c9fb2 Add liaison write-path latency metrics (#1299)
7638c9fb2 is described below
commit 7638c9fb275bab51b444fb8e483fe049b4f65e69
Author: Owen Willison <41...@users.noreply.github.com>
AuthorDate: Sun Aug 23 22:49:25 2026 +0800
Add liaison write-path latency metrics (#1299)
---
CHANGES.md | 1 +
banyand/measure/file_sync_queue_latency_test.go | 106 ++++++
banyand/measure/introducer.go | 13 +
banyand/measure/merger.go | 1 +
banyand/measure/metadata.go | 1 +
banyand/measure/metrics.go | 18 +
banyand/measure/part_metadata.go | 16 +-
banyand/measure/syncer.go | 23 ++
banyand/queue/pub/batch.go | 145 +++++++-
banyand/queue/pub/metrics_test.go | 423 +++++++++++++++++++++-
banyand/queue/pub/migration_metrics.go | 38 +-
banyand/queue/pub/migration_metrics_test.go | 2 +-
banyand/queue/pub/pub.go | 40 +-
banyand/stream/file_sync_queue_latency_test.go | 106 ++++++
banyand/stream/introducer.go | 13 +
banyand/stream/merger.go | 1 +
banyand/stream/metrics.go | 20 +
banyand/stream/part_metadata.go | 16 +-
banyand/stream/syncer.go | 23 ++
banyand/trace/file_sync_queue_latency_test.go | 106 ++++++
banyand/trace/introducer.go | 12 +
banyand/trace/merger.go | 1 +
banyand/trace/metrics.go | 18 +
banyand/trace/part_metadata.go | 16 +-
banyand/trace/syncer.go | 29 +-
docs/operation/grafana-fodc-workload.json | 198 ++++++++--
docs/operation/observability/metrics.md | 25 +-
pkg/meter/meter.go | 8 +
test/e2e-v2/cases/fodc/README.md | 12 +-
test/e2e-v2/cases/fodc/check-metrics.sh | 15 +
test/e2e-v2/cases/fodc/metrics/documented_gap.txt | 1 +
test/e2e-v2/cases/fodc/metrics/non_empty.txt | 5 +
32 files changed, 1356 insertions(+), 96 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 283f72253..f8328fa72 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -5,6 +5,7 @@ Release Notes.
## 0.11.0
### Features
+- Add batch-write phase metrics for client-held-open time, post-seal admission time, and admission deadlines, plus per-part liaison-wqueue-to-data-node file-sync latency for measure, stream, and trace.
- Add bounded trace plugin overhead metrics: chain decision batches by result, per-plugin execution count and duration, batch trace-count distribution, and per-plugin fail-open bypasses by reason. The configured `plugin_name` distinguishes links in a multi-plugin chain; segment, shard, and schema labels remain absent. The catalog works with `pipeline_traces_evaluated` for per-plugin time/calls per trace, latency percentiles, timeout/circuit-open rates, and link-health monitoring. The Pha [...]
- Add trace **finalization sampling** (`PIPELINE_EVENT_FINALIZE`), a best-effort background backstop for the in-merge trace retention filter. A single node-wide, concurrency-1 scanner periodically sweeps cooled segments (`segEnd < now − finalize_grace`, reopening idle-closed segments via `SelectSegments(reopenClosed=true)`) and force-merges each shard's un-finalized parts through the group's registered sampler chain, reusing the hot merge path (`mergeParts` + per-sidx `Merge` + the shard [...]
- Bound the trace merge **drop-set** so a large sampled merge cannot exhaust the heap. The merge-wide dropped-trace-ID collector was unbounded: because a shard's first finalize round selects every cooled part into one merge, an 18M-drop set reaches ~1.3 GiB live and ~2.6 GiB reserved heap (append doubling plus a single contiguous lookup index), inside a process also serving queries. The bound is applied to the sampling **decision**, not to the pruning predicate: once a merge's drop set i [...]
diff --git a/banyand/measure/file_sync_queue_latency_test.go b/banyand/measure/file_sync_queue_latency_test.go
new file mode 100644
index 000000000..cf71f127e
--- /dev/null
+++ b/banyand/measure/file_sync_queue_latency_test.go
@@ -0,0 +1,106 @@
+// 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 measure
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+type queueLatencyObservation struct {
+ labels []string
+ value float64
+}
+
+type queueLatencyHistogram struct {
+ observations []queueLatencyObservation
+}
+
+func (h *queueLatencyHistogram) Observe(value float64, labelValues ...string) {
+ h.observations = append(h.observations, queueLatencyObservation{value: value, labels: append([]string(nil), labelValues...)})
+}
+
+func (h *queueLatencyHistogram) Delete(...string) bool {
+ return true
+}
+
+func TestFileSyncQueuedAtUnixNanoPersistsAndMergeKeepsOldest(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ partPath := t.TempDir()
+ wantQueuedAtUnixNano := time.Now().Add(-time.Hour).UnixNano()
+ metadata := &partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano}
+ metadata.mustWriteMetadata(fileSystem, partPath)
+
+ var restored partMetadata
+ restored.mustReadMetadata(fileSystem, partPath)
+ require.Equal(t, wantQueuedAtUnixNano, restored.QueuedAtUnixNano)
+
+ parts := []*partWrapper{
+ {p: &part{partMetadata: partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano + int64(time.Minute)}}},
+ {p: &part{partMetadata: partMetadata{}}},
+ {p: &part{partMetadata: partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano}}},
+ }
+ require.Equal(t, wantQueuedAtUnixNano, oldestQueuedAtUnixNano(parts))
+}
+
+func TestStampQueuedAtUnixNanoOnlyAtWriteQueueIntroduction(t *testing.T) {
+ queuedAtUnixNano := time.Now().UnixNano()
+ queuedMemPart := &memPart{}
+ wrappedPart := &partWrapper{mp: queuedMemPart, p: &part{}}
+ writeQueue := &tsTable{getNodes: func() []string { return nil }}
+ writeQueue.stampQueuedAtUnixNano(wrappedPart, queuedAtUnixNano)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.p.partMetadata.QueuedAtUnixNano)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.mp.partMetadata.QueuedAtUnixNano)
+
+ writeQueue.stampQueuedAtUnixNano(wrappedPart, queuedAtUnixNano+1)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.p.partMetadata.QueuedAtUnixNano)
+
+ storagePart := &partWrapper{mp: &memPart{}, p: &part{}}
+ (&tsTable{}).stampQueuedAtUnixNano(storagePart, queuedAtUnixNano)
+ require.Zero(t, storagePart.p.partMetadata.QueuedAtUnixNano)
+}
+
+func TestObserveAcknowledgedFileSyncPartsSelectsSuccessfulParts(t *testing.T) {
+ histogram := &queueLatencyHistogram{}
+ tst := &tsTable{metrics: &metrics{fileSyncQueueLatency: histogram}}
+ acknowledgedAt := time.Unix(1_000, 0)
+ succeeded := &part{partMetadata: partMetadata{ID: 1, QueuedAtUnixNano: acknowledgedAt.Add(-10 * time.Second).UnixNano()}}
+ failed := &part{partMetadata: partMetadata{ID: 2, QueuedAtUnixNano: acknowledgedAt.Add(-20 * time.Second).UnixNano()}}
+ legacy := &part{partMetadata: partMetadata{ID: 3}}
+ future := &part{partMetadata: partMetadata{ID: 4, QueuedAtUnixNano: acknowledgedAt.Add(time.Second).UnixNano()}}
+
+ tst.observeAcknowledgedFileSyncParts([]*part{succeeded, failed, legacy, future}, "data-1", &queue.SyncResult{
+ Success: true,
+ FailedParts: []queue.FailedPart{{PartID: "2"}},
+ }, acknowledgedAt)
+ require.Equal(t, []queueLatencyObservation{{value: 10, labels: []string{"data-1"}}}, histogram.observations)
+
+ tst.observeAcknowledgedFileSyncParts([]*part{failed}, "data-1", &queue.SyncResult{Success: true}, acknowledgedAt)
+ require.Equal(t, []queueLatencyObservation{
+ {value: 10, labels: []string{"data-1"}},
+ {value: 20, labels: []string{"data-1"}},
+ }, histogram.observations)
+
+ tst.observeAcknowledgedFileSyncParts([]*part{succeeded}, "data-1", &queue.SyncResult{}, acknowledgedAt)
+ require.Len(t, histogram.observations, 2)
+}
diff --git a/banyand/measure/introducer.go b/banyand/measure/introducer.go
index 53fb8df5b..d502ddae8 100644
--- a/banyand/measure/introducer.go
+++ b/banyand/measure/introducer.go
@@ -18,6 +18,8 @@
package measure
import (
+ "time"
+
"github.com/apache/skywalking-banyandb/pkg/pool"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -227,6 +229,7 @@ func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64)
}
next := nextIntroduction.part
+ tst.stampQueuedAtUnixNano(next, time.Now().UnixNano())
if next.mp != nil {
tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
}
@@ -241,6 +244,16 @@ func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64)
}
}
+func (tst *tsTable) stampQueuedAtUnixNano(part *partWrapper, queuedAtUnixNano int64) {
+ if tst.getNodes == nil || part == nil || part.p == nil || part.p.partMetadata.QueuedAtUnixNano > 0 {
+ return
+ }
+ part.p.partMetadata.QueuedAtUnixNano = queuedAtUnixNano
+ if part.mp != nil {
+ part.mp.partMetadata.QueuedAtUnixNano = queuedAtUnixNano
+ }
+}
+
func (tst *tsTable) introduceFlushed(nextIntroduction *flusherIntroduction, epoch uint64) {
cur := tst.currentSnapshot()
if cur == nil {
diff --git a/banyand/measure/merger.go b/banyand/measure/merger.go
index a335bd0c6..588cc0111 100644
--- a/banyand/measure/merger.go
+++ b/banyand/measure/merger.go
@@ -274,6 +274,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
if err != nil {
return nil, err
}
+ pm.QueuedAtUnixNano = oldestQueuedAtUnixNano(parts)
mergedTagType.mustWriteTagType(fileSystem, dstPath)
pm.mustWriteMetadata(fileSystem, dstPath)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 2c196f405..d5805d922 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -902,6 +902,7 @@ func (s *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observab
totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalSyncLoopLatency: factory.NewCounter("total_sync_loop_latency"),
totalSyncLoopBytes: factory.NewCounter("total_sync_loop_bytes"),
+ fileSyncQueueLatency: factory.NewHistogram("file_sync_queue_latency_seconds", meter.QueueLatencyBuckets, "remote_node"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
diff --git a/banyand/measure/metrics.go b/banyand/measure/metrics.go
index 9e4e00dc3..7c9fd508b 100644
--- a/banyand/measure/metrics.go
+++ b/banyand/measure/metrics.go
@@ -19,6 +19,7 @@ package measure
import (
"context"
+ "sync"
"github.com/pkg/errors"
@@ -54,6 +55,7 @@ type metrics struct {
totalSyncLoopErr meter.Counter
totalSyncLoopLatency meter.Counter
totalSyncLoopBytes meter.Counter
+ fileSyncQueueLatency meter.Histogram
totalFlushLoopProgress meter.Counter
totalFlushed meter.Counter
@@ -68,6 +70,7 @@ type metrics struct {
totalMerged meter.Counter
tbMetrics
+ fileSyncTargets sync.Map
}
func (tst *tsTable) incTotalWritten(delta int) {
@@ -182,6 +185,14 @@ func (tst *tsTable) incTotalSyncLoopBytes(delta uint64) {
tst.metrics.totalSyncLoopBytes.Inc(float64(delta))
}
+func (tst *tsTable) observeFileSyncQueueLatency(delta float64, node string) {
+ if tst == nil || tst.metrics == nil || tst.metrics.fileSyncQueueLatency == nil {
+ return
+ }
+ tst.metrics.fileSyncTargets.Store(node, struct{}{})
+ tst.metrics.fileSyncQueueLatency.Observe(delta, node)
+}
+
func (tst *tsTable) incTotalFlushLoopProgress(delta int) {
if tst == nil || tst.metrics == nil {
return
@@ -292,6 +303,12 @@ func (m *metrics) DeleteAll() {
m.totalSyncLoopErr.Delete()
m.totalSyncLoopLatency.Delete()
m.totalSyncLoopBytes.Delete()
+ m.fileSyncTargets.Range(func(target, _ any) bool {
+ node := target.(string)
+ m.fileSyncQueueLatency.Delete(node)
+ m.fileSyncTargets.Delete(node)
+ return true
+ })
m.totalFlushLoopProgress.Delete()
m.totalFlushed.Delete()
@@ -333,6 +350,7 @@ func (s *supplier) newMetrics(p common.Position) (storage.Metrics, observability
totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalSyncLoopLatency: factory.NewCounter("total_sync_loop_latency"),
totalSyncLoopBytes: factory.NewCounter("total_sync_loop_bytes"),
+ fileSyncQueueLatency: factory.NewHistogram("file_sync_queue_latency_seconds", meter.QueueLatencyBuckets, "remote_node"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
diff --git a/banyand/measure/part_metadata.go b/banyand/measure/part_metadata.go
index fdad43001..8184df08a 100644
--- a/banyand/measure/part_metadata.go
+++ b/banyand/measure/part_metadata.go
@@ -41,7 +41,9 @@ type partMetadata struct {
BlocksCount uint64 `json:"blocksCount"`
MinTimestamp int64 `json:"minTimestamp"`
MaxTimestamp int64 `json:"maxTimestamp"`
- ID uint64 `json:"-"`
+ // QueuedAtUnixNano is the liaison write-queue introduction commit time.
+ QueuedAtUnixNano int64 `json:"queuedAtUnixNano,omitempty"`
+ ID uint64 `json:"-"`
}
func (pm *partMetadata) reset() {
@@ -51,9 +53,21 @@ func (pm *partMetadata) reset() {
pm.BlocksCount = 0
pm.MinTimestamp = 0
pm.MaxTimestamp = 0
+ pm.QueuedAtUnixNano = 0
pm.ID = 0
}
+func oldestQueuedAtUnixNano(parts []*partWrapper) int64 {
+ var oldest int64
+ for _, part := range parts {
+ queuedAtUnixNano := part.p.partMetadata.QueuedAtUnixNano
+ if queuedAtUnixNano > 0 && (oldest == 0 || queuedAtUnixNano < oldest) {
+ oldest = queuedAtUnixNano
+ }
+ }
+ return oldest
+}
+
func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
pm.CompressedSizeBytes = ctx.CompressedSizeBytes
pm.UncompressedSizeBytes = ctx.UncompressedSizeBytes
diff --git a/banyand/measure/syncer.go b/banyand/measure/syncer.go
index a2bd45bf3..d1e8e8efc 100644
--- a/banyand/measure/syncer.go
+++ b/banyand/measure/syncer.go
@@ -236,9 +236,11 @@ func (tst *tsTable) syncPartsToNodesHelper(ctx context.Context, parts []*part, n
}
result, err := chunkedClient.SyncStreamingParts(ctx, streamingParts)
+ acknowledgedAt := time.Now()
if err != nil {
return nil, fmt.Errorf("failed to sync streaming parts to node %s: %w", node, err)
}
+ tst.observeAcknowledgedFileSyncParts(parts, node, result, acknowledgedAt)
tst.incTotalSyncLoopBytes(result.TotalBytes)
if dl := tst.l.Debug(); dl.Enabled() {
@@ -259,6 +261,27 @@ func (tst *tsTable) syncPartsToNodesHelper(ctx context.Context, parts []*part, n
return allFailedParts, nil
}
+func (tst *tsTable) observeAcknowledgedFileSyncParts(parts []*part, node string, result *queue.SyncResult, acknowledgedAt time.Time) {
+ if result == nil || !result.Success {
+ return
+ }
+ failedPartIDs := make(map[string]struct{}, len(result.FailedParts))
+ for _, failedPart := range result.FailedParts {
+ failedPartIDs[failedPart.PartID] = struct{}{}
+ }
+ acknowledgedAtNanos := acknowledgedAt.UnixNano()
+ for _, part := range parts {
+ if _, failed := failedPartIDs[strconv.FormatUint(part.partMetadata.ID, 10)]; failed {
+ continue
+ }
+ queuedAtUnixNano := part.partMetadata.QueuedAtUnixNano
+ if queuedAtUnixNano <= 0 || queuedAtUnixNano > acknowledgedAtNanos {
+ continue
+ }
+ tst.observeFileSyncQueueLatency(float64(acknowledgedAtNanos-queuedAtUnixNano)/float64(time.Second), node)
+ }
+}
+
func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntroduction) error {
startTime := time.Now()
defer func() {
diff --git a/banyand/queue/pub/batch.go b/banyand/queue/pub/batch.go
index 575b12f00..5aae4f250 100644
--- a/banyand/queue/pub/batch.go
+++ b/banyand/queue/pub/batch.go
@@ -61,9 +61,58 @@ type writeStream struct {
ctxDoneCh <-chan struct{}
cancel context.CancelFunc
batchStart time.Time
+ firstFrameAt time.Time
+ timing *batchTiming
+ group string
firstFrameSent bool // false until the first frame is sent; the first frame carries the sender_* identity labels
}
+type batchTiming struct {
+ sealedAt time.Time
+ terminalAt time.Time
+ mu sync.Mutex
+ admissionObserved bool
+}
+
+func (bt *batchTiming) seal(sealedAt time.Time) (duration time.Duration, admissionReady, sealed bool) {
+ if bt == nil {
+ return 0, false, false
+ }
+ bt.mu.Lock()
+ defer bt.mu.Unlock()
+ if !bt.sealedAt.IsZero() {
+ return 0, false, false
+ }
+ bt.sealedAt = sealedAt
+ duration, admissionReady = bt.readyAdmissionDurationLocked()
+ return duration, admissionReady, true
+}
+
+func (bt *batchTiming) recordTerminal(terminalAt time.Time) (duration time.Duration, admissionReady, recorded bool) {
+ if bt == nil {
+ return 0, false, false
+ }
+ bt.mu.Lock()
+ defer bt.mu.Unlock()
+ if !bt.terminalAt.IsZero() {
+ return 0, false, false
+ }
+ bt.terminalAt = terminalAt
+ duration, admissionReady = bt.readyAdmissionDurationLocked()
+ return duration, admissionReady, true
+}
+
+func (bt *batchTiming) readyAdmissionDurationLocked() (time.Duration, bool) {
+ if bt.sealedAt.IsZero() || bt.terminalAt.IsZero() || bt.admissionObserved {
+ return 0, false
+ }
+ bt.admissionObserved = true
+ if bt.terminalAt.Before(bt.sealedAt) {
+ return 0, true
+ }
+ return bt.terminalAt.Sub(bt.sealedAt), true
+}
+
type batchPublisher struct {
pub *pub
streams map[string]writeStream
@@ -137,6 +186,7 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
if !stream.firstFrameSent {
ws := bp.streams[node]
ws.firstFrameSent = true
+ ws.firstFrameAt = time.Now()
bp.streams[node] = ws
}
// Record success for circuit breaker
@@ -194,11 +244,15 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
continue
}
streamBatchStart := time.Now()
+ batchGroup := m.Group()
+ streamTiming := &batchTiming{}
bp.streams[node] = writeStream{
client: stream,
ctxDoneCh: streamCtx.Done(),
cancel: cancel,
batchStart: streamBatchStart,
+ group: batchGroup,
+ timing: streamTiming,
}
var batchOp string
var batchInfo nodeInfo
@@ -208,14 +262,13 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
if bp.pub != nil {
batchInfo = bp.pub.getNodeInfo(node)
}
- batchGroup := m.Group()
if bp.hasMetrics() {
bp.pub.metrics.totalBatchStarted.Inc(1, batchOp, batchGroup, node, batchInfo.role, batchInfo.tier)
}
if bp.hasMigrationMetrics() {
bp.pub.migrationMetrics.totalBatchStarted.Inc(1, batchOp, batchGroup, node, batchInfo.role, batchInfo.tier)
}
- bp.f.events = append(bp.f.events, make(chan batchEvent))
+ bp.f.events = append(bp.f.events, make(chan batchEvent, 1))
_ = sendData()
nodeName := node
recvStream := stream
@@ -223,8 +276,9 @@ func (bp *batchPublisher) Publish(ctx context.Context, topic bus.Topic, messages
recvBC := bp.f.events[len(bp.f.events)-1]
recvBatchStart := streamBatchStart
recvGroup := batchGroup
+ recvTiming := streamTiming
run.Go(ctx, "batch-stream-recv", bp.pub.log, func(runCtx context.Context) {
- bp.listenBatchResponse(runCtx, recvStream, recvDeferFn, recvBC, nodeName, recvBatchStart, recvGroup)
+ bp.listenBatchResponse(runCtx, recvStream, recvDeferFn, recvBC, nodeName, recvBatchStart, recvGroup, recvTiming)
})
}
return nil, err
@@ -238,11 +292,75 @@ func (bp *batchPublisher) hasMigrationMetrics() bool {
return bp.pub != nil && bp.pub.migrationMetrics != nil
}
+func (bp *batchPublisher) batchMetricMetadata(nodeName string) (operation string, info nodeInfo) {
+ if bp.topic != nil {
+ operation = apidata.OperationOf(*bp.topic)
+ }
+ if bp.pub != nil {
+ info = bp.pub.getNodeInfo(nodeName)
+ }
+ return operation, info
+}
+
+func (bp *batchPublisher) observeBatchOpenDuration(nodeName string, stream writeStream, sealedAt time.Time) {
+ if !stream.firstFrameSent || stream.firstFrameAt.IsZero() || sealedAt.Before(stream.firstFrameAt) {
+ return
+ }
+ operation, info := bp.batchMetricMetadata(nodeName)
+ durationSeconds := sealedAt.Sub(stream.firstFrameAt).Seconds()
+ if bp.hasMetrics() && bp.pub.metrics.batchOpenDuration != nil {
+ bp.pub.metrics.batchOpenDuration.Observe(durationSeconds, operation, stream.group, nodeName, info.role, info.tier)
+ }
+ if bp.hasMigrationMetrics() && bp.pub.migrationMetrics.batchOpenDuration != nil {
+ bp.pub.migrationMetrics.batchOpenDuration.Observe(durationSeconds, operation, stream.group, nodeName, info.role, info.tier)
+ }
+}
+
+func (bp *batchPublisher) observeBatchAdmissionDuration(nodeName, group string, duration time.Duration) {
+ operation, info := bp.batchMetricMetadata(nodeName)
+ if bp.hasMetrics() && bp.pub.metrics.batchAdmissionDuration != nil {
+ bp.pub.metrics.batchAdmissionDuration.Observe(duration.Seconds(), operation, group, nodeName, info.role, info.tier)
+ }
+ if bp.hasMigrationMetrics() && bp.pub.migrationMetrics.batchAdmissionDuration != nil {
+ bp.pub.migrationMetrics.batchAdmissionDuration.Observe(duration.Seconds(), operation, group, nodeName, info.role, info.tier)
+ }
+}
+
+func (bp *batchPublisher) sealBatchStream(nodeName string, stream writeStream, sealedAt time.Time) {
+ duration, ready, sealed := stream.timing.seal(sealedAt)
+ if !sealed {
+ return
+ }
+ if ready {
+ bp.observeBatchAdmissionDuration(nodeName, stream.group, duration)
+ }
+ bp.observeBatchOpenDuration(nodeName, stream, sealedAt)
+}
+
+func (bp *batchPublisher) recordBatchTerminal(nodeName, group string, timing *batchTiming, terminalAt time.Time) bool {
+ duration, ready, recorded := timing.recordTerminal(terminalAt)
+ if ready {
+ bp.observeBatchAdmissionDuration(nodeName, group, duration)
+ }
+ return recorded
+}
+
+func (bp *batchPublisher) recordBatchAdmissionTimeout(nodeName string, stream writeStream) {
+ operation, info := bp.batchMetricMetadata(nodeName)
+ if bp.hasMetrics() && bp.pub.metrics.batchAdmissionTimeoutTotal != nil {
+ bp.pub.metrics.batchAdmissionTimeoutTotal.Inc(1, operation, stream.group, nodeName, info.role, info.tier)
+ }
+ if bp.hasMigrationMetrics() && bp.pub.migrationMetrics.batchAdmissionTimeoutTotal != nil {
+ bp.pub.migrationMetrics.batchAdmissionTimeoutTotal.Inc(1, operation, stream.group, nodeName, info.role, info.tier)
+ }
+}
+
// 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, batchStart time.Time, group string,
+ bc chan batchEvent, curNode string, batchStart time.Time, group string, timing *batchTiming,
) {
defer func() {
+ bp.recordBatchTerminal(curNode, group, timing, time.Now())
close(bc)
deferFn()
}()
@@ -264,6 +382,7 @@ func (bp *batchPublisher) listenBatchResponse(ctx context.Context, s clusterv1.S
}
resp, errRecv := s.Recv()
+ bp.recordBatchTerminal(curNode, group, timing, time.Now())
if errRecv != nil {
if bp.hasMetrics() {
bp.pub.metrics.totalErr.Inc(1, operation, group, curNode, info.role, info.tier, sendErrReasonRecvError)
@@ -322,6 +441,7 @@ func (bp *batchPublisher) listenBatchResponse(ctx context.Context, s clusterv1.S
func (bp *batchPublisher) Close() (cee map[string]*common.Error, err error) {
for nodeName := range bp.streams {
stream := bp.streams[nodeName]
+ bp.sealBatchStream(nodeName, stream, time.Now())
err = multierr.Append(err, stream.client.CloseSend())
}
if len(bp.streams) > 0 {
@@ -332,9 +452,20 @@ func (bp *batchPublisher) Close() (cee map[string]*common.Error, err error) {
select {
case <-bp.streams[nodeName].ctxDoneCh:
case <-admissionTimer.C:
- for pendingNodeName := range bp.streams {
- if bp.streams[pendingNodeName].cancel != nil {
- bp.streams[pendingNodeName].cancel()
+ timedOutAt := time.Now()
+ for pendingNodeName, pendingStream := range bp.streams {
+ select {
+ case <-pendingStream.ctxDoneCh:
+ continue
+ default:
+ }
+ timedOut := bp.recordBatchTerminal(pendingNodeName, pendingStream.group, pendingStream.timing, timedOutAt)
+ if !timedOut {
+ continue
+ }
+ bp.recordBatchAdmissionTimeout(pendingNodeName, pendingStream)
+ if pendingStream.cancel != nil {
+ pendingStream.cancel()
}
}
admissionTimedOut = true
diff --git a/banyand/queue/pub/metrics_test.go b/banyand/queue/pub/metrics_test.go
index a821c3e19..46eee338e 100644
--- a/banyand/queue/pub/metrics_test.go
+++ b/banyand/queue/pub/metrics_test.go
@@ -19,6 +19,7 @@ package pub
import (
"context"
+ "sync"
"testing"
"time"
@@ -267,7 +268,7 @@ func TestListenBatchResponseRecordsRecvError(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
require.Equal(t, float64(1), sendErrCap.sum(sendErrReasonRecvError))
}
@@ -288,7 +289,7 @@ func TestListenBatchResponseRecvNonFailoverStillRecordsRecvError(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
require.Equal(t, float64(1), sendErrCap.sum(sendErrReasonRecvError))
}
@@ -312,7 +313,7 @@ func TestListenBatchResponseServerRejectedWithoutFailover(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
require.Equal(t, float64(1), sendErrCap.sum(sendErrReasonServerRejected))
require.Equal(t, float64(0), sendErrCap.sum(sendErrReasonRecvError))
@@ -348,7 +349,7 @@ func TestListenBatchResponseDiskFullSendsFailoverEvent(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
require.Equal(t, float64(1), sendErrCap.sum(sendErrReasonServerRejected))
@@ -378,7 +379,9 @@ func TestPublishRecordsStartedAndFinished(t *testing.T) {
ctx := context.Background()
mockStream := NewMockSendClient(ctx)
+ var sendSucceededAt time.Time
mockStream.SetSendFunc(func(*clusterv1.SendRequest) error {
+ sendSucceededAt = time.Now()
return nil
})
@@ -401,14 +404,38 @@ 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")
+ stream := bp.streams[nodeName]
+ require.True(t, stream.firstFrameSent)
+ require.False(t, stream.firstFrameAt.Before(sendSucceededAt), "batch open time must start only after the first frame is sent successfully")
}
type countingHistogram struct {
- count int
+ values []float64
+ labels [][]string
+ mu sync.Mutex
+ count int
}
-func (h *countingHistogram) Observe(_ float64, _ ...string) { h.count++ }
-func (*countingHistogram) Delete(_ ...string) bool { return true }
+func (h *countingHistogram) Observe(value float64, labels ...string) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.count++
+ h.values = append(h.values, value)
+ h.labels = append(h.labels, append([]string(nil), labels...))
+}
+
+func (*countingHistogram) Delete(_ ...string) bool { return true }
+
+func (h *countingHistogram) snapshot() ([]float64, [][]string) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ values := append([]float64(nil), h.values...)
+ labels := make([][]string, len(h.labels))
+ for idx := range h.labels {
+ labels[idx] = append([]string(nil), h.labels[idx]...)
+ }
+ return values, labels
+}
// TestListenBatchResponseCtxDoneTicksNoBatchFinished verifies that the ctx.Done() early-return
// path in listenBatchResponse does NOT increment total_batch_finished or total_batch_latency.
@@ -437,7 +464,7 @@ func TestListenBatchResponseCtxDoneTicksNoBatchFinished(t *testing.T) {
mockStream := NewMockSendClient(context.Background())
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
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")
@@ -472,8 +499,386 @@ func TestListenBatchResponseSuccessTicksBatchFinished(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
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")
}
+
+func TestBatchTimingRecordsResponseBeforeSealOnce(t *testing.T) {
+ timing := &batchTiming{}
+ terminalAt := time.Unix(1_000, 0)
+
+ duration, ready, recorded := timing.recordTerminal(terminalAt)
+ require.True(t, recorded)
+ require.False(t, ready)
+ require.Zero(t, duration)
+
+ duration, ready, sealed := timing.seal(terminalAt.Add(5 * time.Second))
+ require.True(t, sealed)
+ require.True(t, ready)
+ require.Zero(t, duration, "a response received before sealing must have zero post-seal admission duration")
+
+ _, ready, recorded = timing.recordTerminal(terminalAt.Add(10 * time.Second))
+ require.False(t, recorded, "the first terminal outcome must win")
+ require.False(t, ready, "admission duration must be emitted exactly once")
+ _, ready, sealed = timing.seal(terminalAt.Add(20 * time.Second))
+ require.False(t, sealed, "a batch target must be sealed exactly once")
+ require.False(t, ready, "admission duration must be emitted exactly once")
+}
+
+func TestBatchLifecycleMetricsSeparateHeldOpenFromFastAdmission(t *testing.T) {
+ openDuration := &countingHistogram{}
+ admissionDuration := &countingHistogram{}
+ p := &pub{ //nolint:exhaustruct
+ metrics: &pubMetrics{ //nolint:exhaustruct
+ batchOpenDuration: openDuration,
+ batchAdmissionDuration: admissionDuration,
+ },
+ nodeCache: map[string]nodeInfo{"node-a": {role: "data", tier: "hot"}},
+ }
+ bp := &batchPublisher{pub: p, topic: topicPtr(data.TopicMeasureWrite)}
+ firstFrameAt := time.Unix(2_000, 0)
+ sealedAt := firstFrameAt.Add(6 * time.Hour)
+ stream := writeStream{ //nolint:exhaustruct
+ firstFrameAt: firstFrameAt,
+ firstFrameSent: true,
+ group: "test-group",
+ timing: &batchTiming{},
+ }
+
+ bp.sealBatchStream("node-a", stream, sealedAt)
+ bp.sealBatchStream("node-a", stream, sealedAt.Add(time.Second))
+ require.True(t, bp.recordBatchTerminal("node-a", stream.group, stream.timing, sealedAt.Add(50*time.Millisecond)))
+ require.False(t, bp.recordBatchTerminal("node-a", stream.group, stream.timing, sealedAt.Add(time.Second)))
+
+ openValues, openLabels := openDuration.snapshot()
+ require.Equal(t, []float64{21_600}, openValues, "the open phase must retain the long client-held duration")
+ require.Len(t, openLabels, 1)
+ require.Equal(t, "test-group", openLabels[0][1])
+ require.Equal(t, "node-a", openLabels[0][2])
+ require.Equal(t, "data", openLabels[0][3])
+ require.Equal(t, "hot", openLabels[0][4])
+ admissionValues, admissionLabels := admissionDuration.snapshot()
+ require.Equal(t, []float64{0.05}, admissionValues, "a fast ACK must only contribute its short post-seal duration")
+ require.Equal(t, openLabels, admissionLabels)
+}
+
+func TestBatchAdmissionTimeoutRecordsTerminalDuration(t *testing.T) {
+ openDuration := &countingHistogram{}
+ admissionDuration := &countingHistogram{}
+ timeoutTotal := &countingCounter{}
+ p := &pub{ //nolint:exhaustruct
+ metrics: &pubMetrics{ //nolint:exhaustruct
+ batchOpenDuration: openDuration,
+ batchAdmissionDuration: admissionDuration,
+ batchAdmissionTimeoutTotal: timeoutTotal,
+ },
+ nodeCache: make(map[string]nodeInfo),
+ }
+ streamCtx, streamCancel := context.WithCancel(context.Background())
+ defer streamCancel()
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ timeout: 20 * time.Millisecond,
+ streams: map[string]writeStream{
+ "node-a": {
+ client: NewMockSendClient(streamCtx),
+ ctxDoneCh: streamCtx.Done(),
+ cancel: streamCancel,
+ firstFrameAt: time.Now().Add(-time.Second),
+ firstFrameSent: true,
+ group: "test-group",
+ timing: &batchTiming{},
+ },
+ },
+ }
+
+ cee, closeErr := bp.Close()
+ require.NoError(t, closeErr)
+ require.Empty(t, cee)
+ require.Equal(t, float64(1), timeoutTotal.count)
+ openValues, _ := openDuration.snapshot()
+ require.Len(t, openValues, 1)
+ admissionValues, _ := admissionDuration.snapshot()
+ require.Len(t, admissionValues, 1, "the timeout must also be the target's terminal admission outcome")
+ require.Positive(t, admissionValues[0])
+}
+
+func TestBatchAdmissionTimeoutCountsOnlyPendingTargets(t *testing.T) {
+ openDuration := &countingHistogram{}
+ admissionDuration := &countingHistogram{}
+ timeoutTotal := &countingCounter{}
+ migrationOpenDuration := &countingHistogram{}
+ migrationAdmissionDuration := &countingHistogram{}
+ migrationTimeoutTotal := &countingCounter{}
+ p := &pub{ //nolint:exhaustruct
+ metrics: &pubMetrics{ //nolint:exhaustruct
+ batchOpenDuration: openDuration,
+ batchAdmissionDuration: admissionDuration,
+ batchAdmissionTimeoutTotal: timeoutTotal,
+ },
+ migrationMetrics: &pubMigrationMetrics{ //nolint:exhaustruct
+ batchOpenDuration: migrationOpenDuration,
+ batchAdmissionDuration: migrationAdmissionDuration,
+ batchAdmissionTimeoutTotal: migrationTimeoutTotal,
+ },
+ nodeCache: make(map[string]nodeInfo),
+ }
+ completedCtx, completedCancel := context.WithCancel(context.Background())
+ completedTiming := &batchTiming{}
+ _, _, recorded := completedTiming.recordTerminal(time.Now())
+ require.True(t, recorded)
+ completedCancel()
+ pendingCtx, pendingCancel := context.WithCancel(context.Background())
+ defer pendingCancel()
+ firstFrameAt := time.Now().Add(-time.Second)
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ timeout: 20 * time.Millisecond,
+ streams: map[string]writeStream{
+ "completed": {
+ client: NewMockSendClient(completedCtx),
+ ctxDoneCh: completedCtx.Done(),
+ cancel: completedCancel,
+ firstFrameAt: firstFrameAt,
+ firstFrameSent: true,
+ group: "test-group",
+ timing: completedTiming,
+ },
+ "pending": {
+ client: NewMockSendClient(pendingCtx),
+ ctxDoneCh: pendingCtx.Done(),
+ cancel: pendingCancel,
+ firstFrameAt: firstFrameAt,
+ firstFrameSent: true,
+ group: "test-group",
+ timing: &batchTiming{},
+ },
+ },
+ }
+
+ cee, closeErr := bp.Close()
+ require.NoError(t, closeErr)
+ require.Empty(t, cee)
+ require.Equal(t, float64(1), timeoutTotal.count, "only the still-pending target should time out")
+ require.Equal(t, timeoutTotal.count, migrationTimeoutTotal.count, "migration timeout count must mirror queue_pub")
+ openValues, _ := openDuration.snapshot()
+ migrationOpenValues, _ := migrationOpenDuration.snapshot()
+ require.Len(t, openValues, 2, "open duration must be recorded once for every successfully opened target")
+ require.Equal(t, openValues, migrationOpenValues)
+ admissionValues, _ := admissionDuration.snapshot()
+ migrationAdmissionValues, _ := migrationAdmissionDuration.snapshot()
+ require.Len(t, admissionValues, 2, "every terminal outcome, including timeout, must record admission duration")
+ require.Equal(t, admissionValues, migrationAdmissionValues)
+ select {
+ case <-pendingCtx.Done():
+ default:
+ t.Fatal("timed-out stream context was not canceled")
+ }
+}
+
+func TestBatchCloseSealsEachTargetBeforeClosingItsStream(t *testing.T) {
+ admissionDuration := &countingHistogram{}
+ p := &pub{ //nolint:exhaustruct
+ metrics: &pubMetrics{ //nolint:exhaustruct
+ batchOpenDuration: &countingHistogram{},
+ batchAdmissionDuration: admissionDuration,
+ },
+ nodeCache: make(map[string]nodeInfo),
+ }
+ const (
+ group = "test-group"
+ closeDelay = 20 * time.Millisecond
+ )
+ bp := &batchPublisher{ //nolint:exhaustruct
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ timeout: time.Second,
+ streams: make(map[string]writeStream),
+ }
+ newStream := func(nodeName string) writeStream {
+ streamCtx, streamCancel := context.WithCancel(context.Background())
+ timing := &batchTiming{}
+ mockStream := NewMockSendClient(streamCtx)
+ mockStream.SetCloseSendFunc(func() error {
+ time.Sleep(closeDelay)
+ bp.recordBatchTerminal(nodeName, group, timing, time.Now())
+ streamCancel()
+ time.Sleep(closeDelay)
+ return nil
+ })
+ return writeStream{ //nolint:exhaustruct
+ client: mockStream,
+ ctxDoneCh: streamCtx.Done(),
+ cancel: streamCancel,
+ firstFrameAt: time.Now().Add(-time.Second),
+ firstFrameSent: true,
+ group: group,
+ timing: timing,
+ }
+ }
+ bp.streams["node-a"] = newStream("node-a")
+ bp.streams["node-b"] = newStream("node-b")
+
+ cee, closeErr := bp.Close()
+ require.NoError(t, closeErr)
+ require.Empty(t, cee)
+ admissionValues, _ := admissionDuration.snapshot()
+ require.Len(t, admissionValues, 2)
+ for _, duration := range admissionValues {
+ require.GreaterOrEqual(t, duration, (closeDelay / 2).Seconds(), "each target's admission phase must begin before its own CloseSend call")
+ }
+}
+
+func newCloseTriggeredResponseStream(ctx context.Context, response *clusterv1.SendResponse) (*MockSendClient, <-chan struct{}) {
+ sealed := make(chan struct{})
+ listenerStarted := make(chan struct{})
+ mockStream := NewMockSendClient(ctx)
+ mockStream.SetCloseSendFunc(func() error {
+ close(sealed)
+ return nil
+ })
+ mockStream.SetRecvFunc(func() (*clusterv1.SendResponse, error) {
+ close(listenerStarted)
+ <-sealed
+ return response, nil
+ })
+ return mockStream, listenerStarted
+}
+
+func waitForBatchListener(t *testing.T, listenerStarted <-chan struct{}) {
+ t.Helper()
+ select {
+ case <-listenerStarted:
+ case <-time.After(time.Second):
+ t.Fatal("batch response listener did not start")
+ }
+}
+
+func TestBatchHeldOpenFastAckSeparatesDurations(t *testing.T) {
+ legacyBatchLatency := &countingHistogram{}
+ openDuration := &countingHistogram{}
+ admissionDuration := &countingHistogram{}
+ timeoutTotal := &countingCounter{}
+ pm := newPubMetricsWithErrCapture(newErrReasonCapturer())
+ pm.totalBatchLatency = legacyBatchLatency
+ pm.batchOpenDuration = openDuration
+ pm.batchAdmissionDuration = admissionDuration
+ pm.batchAdmissionTimeoutTotal = timeoutTotal
+ p := newPubWithConnMgrForMetrics(t, pm)
+ t.Cleanup(p.GracefulStop)
+
+ streamCtx, streamCancel := context.WithCancel(context.Background())
+ mockStream, listenerStarted := newCloseTriggeredResponseStream(streamCtx, &clusterv1.SendResponse{})
+ const nodeName = "node-a"
+ const group = "test-group"
+ timing := &batchTiming{}
+ eventCh := make(chan batchEvent, 1)
+ firstFrameAt := time.Now().Add(-6 * time.Hour)
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ timeout: time.Second,
+ streams: map[string]writeStream{
+ nodeName: {
+ client: mockStream,
+ ctxDoneCh: streamCtx.Done(),
+ cancel: streamCancel,
+ batchStart: firstFrameAt,
+ firstFrameAt: firstFrameAt,
+ firstFrameSent: true,
+ group: group,
+ timing: timing,
+ },
+ },
+ f: batchFuture{
+ events: []chan batchEvent{eventCh},
+ errNodes: make(map[string]struct{}),
+ errors: make(map[string]batchEvent),
+ l: p.log,
+ },
+ }
+ go bp.listenBatchResponse(streamCtx, mockStream, streamCancel, eventCh, nodeName, firstFrameAt, group, timing)
+ waitForBatchListener(t, listenerStarted)
+
+ cee, closeErr := bp.Close()
+ require.NoError(t, closeErr)
+ require.Empty(t, cee)
+ require.Zero(t, timeoutTotal.count)
+ openValues, _ := openDuration.snapshot()
+ require.Len(t, openValues, 1)
+ require.GreaterOrEqual(t, openValues[0], (6 * time.Hour).Seconds())
+ admissionValues, _ := admissionDuration.snapshot()
+ require.Len(t, admissionValues, 1)
+ require.Less(t, admissionValues[0], time.Second.Seconds())
+ legacyLatencyValues, _ := legacyBatchLatency.snapshot()
+ require.Len(t, legacyLatencyValues, 1, "the legacy end-to-end batch latency must remain available")
+}
+
+func TestBatchPromptRejectionReturnsErrorWithoutAdmissionTimeout(t *testing.T) {
+ batchFinished := &countingCounter{}
+ legacyBatchLatency := &countingHistogram{}
+ openDuration := &countingHistogram{}
+ admissionDuration := &countingHistogram{}
+ timeoutTotal := &countingCounter{}
+ pm := newPubMetricsWithErrCapture(newErrReasonCapturer())
+ pm.totalBatchFinished = batchFinished
+ pm.totalBatchLatency = legacyBatchLatency
+ pm.batchOpenDuration = openDuration
+ pm.batchAdmissionDuration = admissionDuration
+ pm.batchAdmissionTimeoutTotal = timeoutTotal
+ p := newPubWithConnMgrForMetrics(t, pm)
+ t.Cleanup(p.GracefulStop)
+
+ streamCtx, streamCancel := context.WithCancel(context.Background())
+ mockStream, listenerStarted := newCloseTriggeredResponseStream(streamCtx, &clusterv1.SendResponse{
+ Error: "rejected",
+ Status: modelv1.Status_STATUS_INTERNAL_ERROR,
+ })
+
+ const nodeName = "node-a"
+ const group = "test-group"
+ timing := &batchTiming{}
+ eventCh := make(chan batchEvent, 1)
+ bp := &batchPublisher{
+ pub: p,
+ topic: topicPtr(data.TopicMeasureWrite),
+ timeout: 100 * time.Millisecond,
+ streams: map[string]writeStream{
+ nodeName: {
+ client: mockStream,
+ ctxDoneCh: streamCtx.Done(),
+ cancel: streamCancel,
+ batchStart: time.Now(),
+ firstFrameAt: time.Now().Add(-time.Second),
+ firstFrameSent: true,
+ group: group,
+ timing: timing,
+ },
+ },
+ f: batchFuture{
+ events: []chan batchEvent{eventCh},
+ errNodes: make(map[string]struct{}),
+ errors: make(map[string]batchEvent),
+ l: p.log,
+ },
+ }
+ go bp.listenBatchResponse(streamCtx, mockStream, streamCancel, eventCh, nodeName, bp.streams[nodeName].batchStart, group, timing)
+ waitForBatchListener(t, listenerStarted)
+
+ cee, closeErr := bp.Close()
+ require.NoError(t, closeErr)
+ require.Contains(t, cee, nodeName)
+ require.Equal(t, modelv1.Status_STATUS_INTERNAL_ERROR, cee[nodeName].Status())
+ require.Zero(t, timeoutTotal.count, "a prompt rejection must not be misclassified as an admission timeout")
+ require.Equal(t, float64(1), batchFinished.count)
+ legacyLatencyValues, _ := legacyBatchLatency.snapshot()
+ require.Len(t, legacyLatencyValues, 1, "the existing total_batch_latency observation must be preserved")
+ openValues, _ := openDuration.snapshot()
+ require.Len(t, openValues, 1)
+ admissionValues, _ := admissionDuration.snapshot()
+ require.Len(t, admissionValues, 1)
+}
diff --git a/banyand/queue/pub/migration_metrics.go b/banyand/queue/pub/migration_metrics.go
index 3054b0ad3..581ed5c8b 100644
--- a/banyand/queue/pub/migration_metrics.go
+++ b/banyand/queue/pub/migration_metrics.go
@@ -32,26 +32,32 @@ 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
- totalBatchStarted meter.Counter
- totalBatchFinished meter.Counter
- totalBatchLatency meter.Histogram
+ totalStarted meter.Counter
+ totalFinished meter.Counter
+ totalLatency meter.Histogram
+ totalErr meter.Counter
+ sentBytes meter.Counter
+ totalBatchStarted meter.Counter
+ totalBatchFinished meter.Counter
+ totalBatchLatency meter.Histogram
+ batchOpenDuration meter.Histogram
+ batchAdmissionDuration meter.Histogram
+ batchAdmissionTimeoutTotal meter.Counter
}
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...),
- totalBatchStarted: factory.NewCounter("total_batch_started", labels...),
- totalBatchFinished: factory.NewCounter("total_batch_finished", labels...),
- totalBatchLatency: factory.NewHistogram("total_batch_latency", meter.BatchBuckets, 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...),
+ batchOpenDuration: factory.NewHistogram("batch_open_duration_seconds", batchOpenDurationBuckets, labels...),
+ batchAdmissionDuration: factory.NewHistogram("batch_admission_duration_seconds", meter.BatchBuckets, labels...),
+ batchAdmissionTimeoutTotal: factory.NewCounter("batch_admission_timeout_total", labels...),
}
}
diff --git a/banyand/queue/pub/migration_metrics_test.go b/banyand/queue/pub/migration_metrics_test.go
index d9f8f06da..0708af835 100644
--- a/banyand/queue/pub/migration_metrics_test.go
+++ b/banyand/queue/pub/migration_metrics_test.go
@@ -169,7 +169,7 @@ func TestMigrationMirrorBatchFinishedAndLatency(t *testing.T) {
})
bc := make(chan batchEvent, 1)
- bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group")
+ bp.listenBatchResponse(ctx, mockStream, func() {}, bc, "node-a", time.Now(), "test-group", nil)
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")
diff --git a/banyand/queue/pub/pub.go b/banyand/queue/pub/pub.go
index 8c13a1729..f062128c7 100644
--- a/banyand/queue/pub/pub.go
+++ b/banyand/queue/pub/pub.go
@@ -54,6 +54,8 @@ import (
var queuePubScope = observability.RootScope.SubScope("queue_pub")
+var batchOpenDurationBuckets = meter.Buckets{1, 5, 10, 30, 60, 120, 300, 600, 1800, 3600, 21600, 86400}
+
// defaultHealthCheckInterval is how often the queue client re-checks the nodes it considers
// active, matching the property schema client's default. Without it the active set is only
// validated on admission and then by a request failing on it, so a dead node keeps absorbing
@@ -108,27 +110,33 @@ type nodeInfo struct {
}
type pubMetrics struct {
- totalStarted meter.Counter
- totalFinished meter.Counter
- totalLatency meter.Histogram
- totalErr meter.Counter
- sentBytes meter.Counter
- totalBatchStarted meter.Counter
- totalBatchFinished meter.Counter
- totalBatchLatency meter.Histogram
+ totalStarted meter.Counter
+ totalFinished meter.Counter
+ totalLatency meter.Histogram
+ totalErr meter.Counter
+ sentBytes meter.Counter
+ totalBatchStarted meter.Counter
+ totalBatchFinished meter.Counter
+ totalBatchLatency meter.Histogram
+ batchOpenDuration meter.Histogram
+ batchAdmissionDuration meter.Histogram
+ batchAdmissionTimeoutTotal meter.Counter
}
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...),
- totalBatchStarted: factory.NewCounter("total_batch_started", labels...),
- totalBatchFinished: factory.NewCounter("total_batch_finished", labels...),
- totalBatchLatency: factory.NewHistogram("total_batch_latency", meter.BatchBuckets, 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...),
+ batchOpenDuration: factory.NewHistogram("batch_open_duration_seconds", batchOpenDurationBuckets, labels...),
+ batchAdmissionDuration: factory.NewHistogram("batch_admission_duration_seconds", meter.BatchBuckets, labels...),
+ batchAdmissionTimeoutTotal: factory.NewCounter("batch_admission_timeout_total", labels...),
}
}
diff --git a/banyand/stream/file_sync_queue_latency_test.go b/banyand/stream/file_sync_queue_latency_test.go
new file mode 100644
index 000000000..b3fd13105
--- /dev/null
+++ b/banyand/stream/file_sync_queue_latency_test.go
@@ -0,0 +1,106 @@
+// 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 stream
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+type queueLatencyObservation struct {
+ labels []string
+ value float64
+}
+
+type queueLatencyHistogram struct {
+ observations []queueLatencyObservation
+}
+
+func (h *queueLatencyHistogram) Observe(value float64, labelValues ...string) {
+ h.observations = append(h.observations, queueLatencyObservation{value: value, labels: append([]string(nil), labelValues...)})
+}
+
+func (h *queueLatencyHistogram) Delete(...string) bool {
+ return true
+}
+
+func TestFileSyncQueuedAtUnixNanoPersistsAndMergeKeepsOldest(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ partPath := t.TempDir()
+ wantQueuedAtUnixNano := time.Now().Add(-time.Hour).UnixNano()
+ metadata := &partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano}
+ metadata.mustWriteMetadata(fileSystem, partPath)
+
+ var restored partMetadata
+ restored.mustReadMetadata(fileSystem, partPath)
+ require.Equal(t, wantQueuedAtUnixNano, restored.QueuedAtUnixNano)
+
+ parts := []*partWrapper{
+ {p: &part{partMetadata: partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano + int64(time.Minute)}}},
+ {p: &part{partMetadata: partMetadata{}}},
+ {p: &part{partMetadata: partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano}}},
+ }
+ require.Equal(t, wantQueuedAtUnixNano, oldestQueuedAtUnixNano(parts))
+}
+
+func TestStampQueuedAtUnixNanoOnlyAtWriteQueueIntroduction(t *testing.T) {
+ queuedAtUnixNano := time.Now().UnixNano()
+ queuedMemPart := &memPart{}
+ wrappedPart := &partWrapper{mp: queuedMemPart, p: &part{}}
+ writeQueue := &tsTable{getNodes: func() []string { return nil }}
+ writeQueue.stampQueuedAtUnixNano(wrappedPart, queuedAtUnixNano)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.p.partMetadata.QueuedAtUnixNano)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.mp.partMetadata.QueuedAtUnixNano)
+
+ writeQueue.stampQueuedAtUnixNano(wrappedPart, queuedAtUnixNano+1)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.p.partMetadata.QueuedAtUnixNano)
+
+ storagePart := &partWrapper{mp: &memPart{}, p: &part{}}
+ (&tsTable{}).stampQueuedAtUnixNano(storagePart, queuedAtUnixNano)
+ require.Zero(t, storagePart.p.partMetadata.QueuedAtUnixNano)
+}
+
+func TestObserveAcknowledgedFileSyncPartsSelectsSuccessfulParts(t *testing.T) {
+ histogram := &queueLatencyHistogram{}
+ tst := &tsTable{metrics: &metrics{fileSyncQueueLatency: histogram}}
+ acknowledgedAt := time.Unix(1_000, 0)
+ succeeded := &part{partMetadata: partMetadata{ID: 1, QueuedAtUnixNano: acknowledgedAt.Add(-10 * time.Second).UnixNano()}}
+ failed := &part{partMetadata: partMetadata{ID: 2, QueuedAtUnixNano: acknowledgedAt.Add(-20 * time.Second).UnixNano()}}
+ legacy := &part{partMetadata: partMetadata{ID: 3}}
+ future := &part{partMetadata: partMetadata{ID: 4, QueuedAtUnixNano: acknowledgedAt.Add(time.Second).UnixNano()}}
+
+ tst.observeAcknowledgedFileSyncParts([]*part{succeeded, failed, legacy, future}, "data-1", &queue.SyncResult{
+ Success: true,
+ FailedParts: []queue.FailedPart{{PartID: "2"}},
+ }, acknowledgedAt)
+ require.Equal(t, []queueLatencyObservation{{value: 10, labels: []string{"data-1"}}}, histogram.observations)
+
+ tst.observeAcknowledgedFileSyncParts([]*part{failed}, "data-1", &queue.SyncResult{Success: true}, acknowledgedAt)
+ require.Equal(t, []queueLatencyObservation{
+ {value: 10, labels: []string{"data-1"}},
+ {value: 20, labels: []string{"data-1"}},
+ }, histogram.observations)
+
+ tst.observeAcknowledgedFileSyncParts([]*part{succeeded}, "data-1", &queue.SyncResult{}, acknowledgedAt)
+ require.Len(t, histogram.observations, 2)
+}
diff --git a/banyand/stream/introducer.go b/banyand/stream/introducer.go
index 621ea4238..7b7534c88 100644
--- a/banyand/stream/introducer.go
+++ b/banyand/stream/introducer.go
@@ -18,6 +18,8 @@
package stream
import (
+ "time"
+
"github.com/apache/skywalking-banyandb/pkg/pool"
"github.com/apache/skywalking-banyandb/pkg/watcher"
)
@@ -256,6 +258,7 @@ func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64)
}
next := nextIntroduction.part
+ tst.stampQueuedAtUnixNano(next, time.Now().UnixNano())
if next.mp != nil {
tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
}
@@ -273,6 +276,16 @@ func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64)
}
}
+func (tst *tsTable) stampQueuedAtUnixNano(part *partWrapper, queuedAtUnixNano int64) {
+ if tst.getNodes == nil || part == nil || part.p == nil || part.p.partMetadata.QueuedAtUnixNano > 0 {
+ return
+ }
+ part.p.partMetadata.QueuedAtUnixNano = queuedAtUnixNano
+ if part.mp != nil {
+ part.mp.partMetadata.QueuedAtUnixNano = queuedAtUnixNano
+ }
+}
+
func (tst *tsTable) introduceFlushed(nextIntroduction *flusherIntroduction, epoch uint64) {
cur := tst.currentSnapshot()
if cur == nil {
diff --git a/banyand/stream/merger.go b/banyand/stream/merger.go
index a55989cbb..8ace9532d 100644
--- a/banyand/stream/merger.go
+++ b/banyand/stream/merger.go
@@ -272,6 +272,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
if err != nil {
return nil, err
}
+ pm.QueuedAtUnixNano = oldestQueuedAtUnixNano(parts)
mergedTagType.mustWriteTagType(fileSystem, dstPath)
pm.mustWriteMetadata(fileSystem, dstPath)
// No SyncPath: mustWriteMetadata goes through WriteAtomic which already
diff --git a/banyand/stream/metrics.go b/banyand/stream/metrics.go
index 9a148ba99..3d26cdea1 100644
--- a/banyand/stream/metrics.go
+++ b/banyand/stream/metrics.go
@@ -18,6 +18,8 @@
package stream
import (
+ "sync"
+
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/observability"
@@ -54,6 +56,7 @@ type metrics struct {
totalSyncLoopErr meter.Counter
totalSyncLoopLatency meter.Counter
totalSyncLoopBytes meter.Counter
+ fileSyncQueueLatency meter.Histogram
totalFlushLoopProgress meter.Counter
totalFlushed meter.Counter
@@ -68,6 +71,7 @@ type metrics struct {
totalMerged meter.Counter
tbMetrics
+ fileSyncTargets sync.Map
}
func (tst *tsTable) incTotalWritten(delta int) {
@@ -182,6 +186,14 @@ func (tst *tsTable) incTotalSyncLoopBytes(delta uint64) {
tst.metrics.totalSyncLoopBytes.Inc(float64(delta))
}
+func (tst *tsTable) observeFileSyncQueueLatency(delta float64, node string) {
+ if tst == nil || tst.metrics == nil || tst.metrics.fileSyncQueueLatency == nil {
+ return
+ }
+ tst.metrics.fileSyncTargets.Store(node, struct{}{})
+ tst.metrics.fileSyncQueueLatency.Observe(delta, node)
+}
+
func (tst *tsTable) incTotalFlushLoopProgress(delta int) {
if tst == nil || tst.metrics == nil {
return
@@ -292,6 +304,12 @@ func (m *metrics) DeleteAll() {
m.totalSyncLoopErr.Delete()
m.totalSyncLoopLatency.Delete()
m.totalSyncLoopBytes.Delete()
+ m.fileSyncTargets.Range(func(target, _ any) bool {
+ node := target.(string)
+ m.fileSyncQueueLatency.Delete(node)
+ m.fileSyncTargets.Delete(node)
+ return true
+ })
m.totalFlushLoopProgress.Delete()
m.totalFlushed.Delete()
@@ -327,6 +345,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalSyncLoopLatency: factory.NewCounter("total_sync_loop_latency"),
totalSyncLoopBytes: factory.NewCounter("total_sync_loop_bytes"),
+ fileSyncQueueLatency: factory.NewHistogram("file_sync_queue_latency_seconds", meter.QueueLatencyBuckets, "remote_node"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
@@ -373,6 +392,7 @@ func (s *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observab
totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalSyncLoopLatency: factory.NewCounter("total_sync_loop_latency"),
totalSyncLoopBytes: factory.NewCounter("total_sync_loop_bytes"),
+ fileSyncQueueLatency: factory.NewHistogram("file_sync_queue_latency_seconds", meter.QueueLatencyBuckets, "remote_node"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
diff --git a/banyand/stream/part_metadata.go b/banyand/stream/part_metadata.go
index d6dde20e0..e7bddca75 100644
--- a/banyand/stream/part_metadata.go
+++ b/banyand/stream/part_metadata.go
@@ -41,7 +41,9 @@ type partMetadata struct {
BlocksCount uint64 `json:"blocksCount"`
MinTimestamp int64 `json:"minTimestamp"`
MaxTimestamp int64 `json:"maxTimestamp"`
- ID uint64 `json:"-"`
+ // QueuedAtUnixNano is the liaison write-queue introduction commit time.
+ QueuedAtUnixNano int64 `json:"queuedAtUnixNano,omitempty"`
+ ID uint64 `json:"-"`
}
func (pm *partMetadata) reset() {
@@ -51,9 +53,21 @@ func (pm *partMetadata) reset() {
pm.BlocksCount = 0
pm.MinTimestamp = 0
pm.MaxTimestamp = 0
+ pm.QueuedAtUnixNano = 0
pm.ID = 0
}
+func oldestQueuedAtUnixNano(parts []*partWrapper) int64 {
+ var oldest int64
+ for _, part := range parts {
+ queuedAtUnixNano := part.p.partMetadata.QueuedAtUnixNano
+ if queuedAtUnixNano > 0 && (oldest == 0 || queuedAtUnixNano < oldest) {
+ oldest = queuedAtUnixNano
+ }
+ }
+ return oldest
+}
+
func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
pm.CompressedSizeBytes = ctx.CompressedSizeBytes
pm.UncompressedSizeBytes = ctx.UncompressedSizeBytes
diff --git a/banyand/stream/syncer.go b/banyand/stream/syncer.go
index fc5dcfb2d..db0ee17a8 100644
--- a/banyand/stream/syncer.go
+++ b/banyand/stream/syncer.go
@@ -209,9 +209,11 @@ func (tst *tsTable) syncPartsToNodesHelper(ctx context.Context, parts []*part, n
}
result, err := chunkedClient.SyncStreamingParts(ctx, streamingParts)
+ acknowledgedAt := time.Now()
if err != nil {
return nil, fmt.Errorf("failed to sync streaming parts to node %s: %w", node, err)
}
+ tst.observeAcknowledgedFileSyncParts(parts, node, result, acknowledgedAt)
tst.incTotalSyncLoopBytes(result.TotalBytes)
if dl := tst.l.Debug(); dl.Enabled() {
@@ -232,6 +234,27 @@ func (tst *tsTable) syncPartsToNodesHelper(ctx context.Context, parts []*part, n
return allFailedParts, nil
}
+func (tst *tsTable) observeAcknowledgedFileSyncParts(parts []*part, node string, result *queue.SyncResult, acknowledgedAt time.Time) {
+ if result == nil || !result.Success {
+ return
+ }
+ failedPartIDs := make(map[string]struct{}, len(result.FailedParts))
+ for _, failedPart := range result.FailedParts {
+ failedPartIDs[failedPart.PartID] = struct{}{}
+ }
+ acknowledgedAtNanos := acknowledgedAt.UnixNano()
+ for _, part := range parts {
+ if _, failed := failedPartIDs[strconv.FormatUint(part.partMetadata.ID, 10)]; failed {
+ continue
+ }
+ queuedAtUnixNano := part.partMetadata.QueuedAtUnixNano
+ if queuedAtUnixNano <= 0 || queuedAtUnixNano > acknowledgedAtNanos {
+ continue
+ }
+ tst.observeFileSyncQueueLatency(float64(acknowledgedAtNanos-queuedAtUnixNano)/float64(time.Second), node)
+ }
+}
+
func (tst *tsTable) syncSnapshot(curSnapshot *snapshot, syncCh chan *syncIntroduction) error {
startTime := time.Now()
defer func() {
diff --git a/banyand/trace/file_sync_queue_latency_test.go b/banyand/trace/file_sync_queue_latency_test.go
new file mode 100644
index 000000000..799c93852
--- /dev/null
+++ b/banyand/trace/file_sync_queue_latency_test.go
@@ -0,0 +1,106 @@
+// 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 trace
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+type queueLatencyObservation struct {
+ labels []string
+ value float64
+}
+
+type queueLatencyHistogram struct {
+ observations []queueLatencyObservation
+}
+
+func (h *queueLatencyHistogram) Observe(value float64, labelValues ...string) {
+ h.observations = append(h.observations, queueLatencyObservation{value: value, labels: append([]string(nil), labelValues...)})
+}
+
+func (h *queueLatencyHistogram) Delete(...string) bool {
+ return true
+}
+
+func TestFileSyncQueuedAtUnixNanoPersistsAndMergeKeepsOldest(t *testing.T) {
+ fileSystem := fs.NewLocalFileSystem()
+ partPath := t.TempDir()
+ wantQueuedAtUnixNano := time.Now().Add(-time.Hour).UnixNano()
+ metadata := &partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano}
+ metadata.mustWriteMetadata(fileSystem, partPath)
+
+ var restored partMetadata
+ restored.mustReadMetadata(fileSystem, partPath)
+ require.Equal(t, wantQueuedAtUnixNano, restored.QueuedAtUnixNano)
+
+ parts := []*partWrapper{
+ {p: &part{partMetadata: partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano + int64(time.Minute)}}},
+ {p: &part{partMetadata: partMetadata{}}},
+ {p: &part{partMetadata: partMetadata{QueuedAtUnixNano: wantQueuedAtUnixNano}}},
+ }
+ require.Equal(t, wantQueuedAtUnixNano, oldestQueuedAtUnixNano(parts))
+}
+
+func TestStampQueuedAtUnixNanoOnlyAtWriteQueueIntroduction(t *testing.T) {
+ queuedAtUnixNano := time.Now().UnixNano()
+ queuedMemPart := &memPart{}
+ wrappedPart := &partWrapper{mp: queuedMemPart, p: &part{}}
+ writeQueue := &tsTable{getNodes: func() []string { return nil }}
+ writeQueue.stampQueuedAtUnixNano(wrappedPart, queuedAtUnixNano)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.p.partMetadata.QueuedAtUnixNano)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.mp.partMetadata.QueuedAtUnixNano)
+
+ writeQueue.stampQueuedAtUnixNano(wrappedPart, queuedAtUnixNano+1)
+ require.Equal(t, queuedAtUnixNano, wrappedPart.p.partMetadata.QueuedAtUnixNano)
+
+ storagePart := &partWrapper{mp: &memPart{}, p: &part{}}
+ (&tsTable{}).stampQueuedAtUnixNano(storagePart, queuedAtUnixNano)
+ require.Zero(t, storagePart.p.partMetadata.QueuedAtUnixNano)
+}
+
+func TestObserveAcknowledgedFileSyncPartsSelectsSuccessfulParts(t *testing.T) {
+ histogram := &queueLatencyHistogram{}
+ tst := &tsTable{metrics: &metrics{fileSyncQueueLatency: histogram}}
+ acknowledgedAt := time.Unix(1_000, 0)
+ succeeded := &part{partMetadata: partMetadata{ID: 1, QueuedAtUnixNano: acknowledgedAt.Add(-10 * time.Second).UnixNano()}}
+ failed := &part{partMetadata: partMetadata{ID: 2, QueuedAtUnixNano: acknowledgedAt.Add(-20 * time.Second).UnixNano()}}
+ legacy := &part{partMetadata: partMetadata{ID: 3}}
+ future := &part{partMetadata: partMetadata{ID: 4, QueuedAtUnixNano: acknowledgedAt.Add(time.Second).UnixNano()}}
+
+ tst.observeAcknowledgedFileSyncParts([]*part{succeeded, failed, legacy, future}, "data-1", &queue.SyncResult{
+ Success: true,
+ FailedParts: []queue.FailedPart{{PartID: "2"}},
+ }, acknowledgedAt)
+ require.Equal(t, []queueLatencyObservation{{value: 10, labels: []string{"data-1"}}}, histogram.observations)
+
+ tst.observeAcknowledgedFileSyncParts([]*part{failed}, "data-1", &queue.SyncResult{Success: true}, acknowledgedAt)
+ require.Equal(t, []queueLatencyObservation{
+ {value: 10, labels: []string{"data-1"}},
+ {value: 20, labels: []string{"data-1"}},
+ }, histogram.observations)
+
+ tst.observeAcknowledgedFileSyncParts([]*part{succeeded}, "data-1", &queue.SyncResult{}, acknowledgedAt)
+ require.Len(t, histogram.observations, 2)
+}
diff --git a/banyand/trace/introducer.go b/banyand/trace/introducer.go
index 454156852..d6c290a35 100644
--- a/banyand/trace/introducer.go
+++ b/banyand/trace/introducer.go
@@ -19,6 +19,7 @@ package trace
import (
"fmt"
+ "time"
"github.com/apache/skywalking-banyandb/banyand/internal/sidx"
snapshotpkg "github.com/apache/skywalking-banyandb/banyand/internal/snapshot"
@@ -279,6 +280,7 @@ func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64)
// Prepare trace snapshot transition
next := nextIntroduction.part
+ tst.stampQueuedAtUnixNano(next, time.Now().UnixNano())
if next.mp != nil {
tst.addPendingDataCount(-int64(next.mp.partMetadata.TotalCount))
}
@@ -335,6 +337,16 @@ func (tst *tsTable) introducePart(nextIntroduction *introduction, epoch uint64)
}
}
+func (tst *tsTable) stampQueuedAtUnixNano(part *partWrapper, queuedAtUnixNano int64) {
+ if tst.getNodes == nil || part == nil || part.p == nil || part.p.partMetadata.QueuedAtUnixNano > 0 {
+ return
+ }
+ part.p.partMetadata.QueuedAtUnixNano = queuedAtUnixNano
+ if part.mp != nil {
+ part.mp.partMetadata.QueuedAtUnixNano = queuedAtUnixNano
+ }
+}
+
// accountUnsampledFlushed is the arrival-based unsampled-bytes accounting for
// finalization sampling. Once this shard has been finalized at least once (cached
// generation > 0), a newly-flushed part is new unsampled data arriving into a cooled
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 4967c7dae..e16b4c7da 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -1274,6 +1274,7 @@ func (tst *tsTable) mergeParts(fileSystem fs.FileSystem, closeCh <-chan struct{}
tt.mustWriteTagType(fileSystem, dstPath)
pm.MinTimestamp = minTimestamp
pm.MaxTimestamp = maxTimestamp
+ pm.QueuedAtUnixNano = oldestQueuedAtUnixNano(parts)
// Finalization-sampling generation stamp, applied BEFORE the on-disk metadata write
// (and the subsequent re-open below) so it survives restart. finalizeGenOverride set
// => this is a finalize round: stamp the output at the round's generation. nil => any
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 1ac7bdfcd..0964c4781 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -54,6 +54,7 @@ type metrics struct {
totalSyncLoopErr meter.Counter
totalSyncLoopLatency meter.Counter
totalSyncLoopBytes meter.Counter
+ fileSyncQueueLatency meter.Histogram
totalFlushLoopProgress meter.Counter
totalFlushed meter.Counter
totalFlushedMemParts meter.Counter
@@ -98,6 +99,7 @@ type metrics struct {
totalBatch meter.Counter
indexMetrics *inverted.Metrics
pipelinePluginNames sync.Map
+ fileSyncTargets sync.Map
pipelinePluginLifecycleMu sync.RWMutex
pipelinePluginClosed bool
}
@@ -230,6 +232,14 @@ func (tst *tsTable) incTotalSyncLoopBytes(delta uint64) {
tst.metrics.totalSyncLoopBytes.Inc(float64(delta))
}
+func (tst *tsTable) observeFileSyncQueueLatency(delta float64, node string) {
+ if tst == nil || tst.metrics == nil || tst.metrics.fileSyncQueueLatency == nil {
+ return
+ }
+ tst.metrics.fileSyncTargets.Store(node, struct{}{})
+ tst.metrics.fileSyncQueueLatency.Observe(delta, node)
+}
+
func (tst *tsTable) incTotalFlushLoopProgress(delta int) {
if tst == nil || tst.metrics == nil {
return
@@ -546,6 +556,12 @@ func (m *metrics) DeleteAll() {
m.totalSyncLoopErr.Delete()
m.totalSyncLoopLatency.Delete()
m.totalSyncLoopBytes.Delete()
+ m.fileSyncTargets.Range(func(target, _ any) bool {
+ node := target.(string)
+ m.fileSyncQueueLatency.Delete(node)
+ m.fileSyncTargets.Delete(node)
+ return true
+ })
m.totalFlushLoopProgress.Delete()
m.totalFlushed.Delete()
@@ -628,6 +644,7 @@ func (s *supplier) newMetrics(p common.Position) storage.Metrics {
totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalSyncLoopLatency: factory.NewCounter("total_sync_loop_latency"),
totalSyncLoopBytes: factory.NewCounter("total_sync_loop_bytes"),
+ fileSyncQueueLatency: factory.NewHistogram("file_sync_queue_latency_seconds", meter.QueueLatencyBuckets, "remote_node"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
@@ -702,6 +719,7 @@ func (qs *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observa
totalSyncLoopErr: factory.NewCounter("total_sync_loop_err"),
totalSyncLoopLatency: factory.NewCounter("total_sync_loop_latency"),
totalSyncLoopBytes: factory.NewCounter("total_sync_loop_bytes"),
+ fileSyncQueueLatency: factory.NewHistogram("file_sync_queue_latency_seconds", meter.QueueLatencyBuckets, "remote_node"),
totalFlushLoopProgress: factory.NewCounter("total_flush_loop_progress"),
totalFlushed: factory.NewCounter("total_flushed"),
totalFlushedMemParts: factory.NewCounter("total_flushed_mem_parts"),
diff --git a/banyand/trace/part_metadata.go b/banyand/trace/part_metadata.go
index 97ec09bdc..661df2c8f 100644
--- a/banyand/trace/part_metadata.go
+++ b/banyand/trace/part_metadata.go
@@ -41,7 +41,9 @@ type partMetadata struct {
BlocksCount uint64 `json:"blocksCount"`
MinTimestamp int64 `json:"minTimestamp"`
MaxTimestamp int64 `json:"maxTimestamp"`
- ID uint64 `json:"-"`
+ // QueuedAtUnixNano is the liaison write-queue introduction commit time.
+ QueuedAtUnixNano int64 `json:"queuedAtUnixNano,omitempty"`
+ ID uint64 `json:"-"`
// FinalizeGen is the finalization-sampling generation stamp: 0/absent means the
// part was never touched by a finalize round. On merge it is min-propagated from
// the inputs; a finalize round overrides it to the round's generation. Unlike ID
@@ -56,10 +58,22 @@ func (pm *partMetadata) reset() {
pm.BlocksCount = 0
pm.MinTimestamp = 0
pm.MaxTimestamp = 0
+ pm.QueuedAtUnixNano = 0
pm.ID = 0
pm.FinalizeGen = 0
}
+func oldestQueuedAtUnixNano(parts []*partWrapper) int64 {
+ var oldest int64
+ for _, part := range parts {
+ queuedAtUnixNano := part.p.partMetadata.QueuedAtUnixNano
+ if queuedAtUnixNano > 0 && (oldest == 0 || queuedAtUnixNano < oldest) {
+ oldest = queuedAtUnixNano
+ }
+ }
+ return oldest
+}
+
func (pm *partMetadata) fillFromSyncContext(ctx *queue.ChunkedSyncPartContext) {
pm.CompressedSizeBytes = ctx.CompressedSizeBytes
pm.UncompressedSpanSizeBytes = ctx.UncompressedSizeBytes
diff --git a/banyand/trace/syncer.go b/banyand/trace/syncer.go
index b0c522247..485306b74 100644
--- a/banyand/trace/syncer.go
+++ b/banyand/trace/syncer.go
@@ -281,7 +281,7 @@ func (tst *tsTable) syncPartsToNodesHelper(
return streamingParts[i].ID < streamingParts[j].ID
})
- failedParts, err := tst.syncStreamingPartsToNode(ctx, node, streamingParts)
+ failedParts, err := tst.syncStreamingPartsToNode(ctx, node, parts, streamingParts)
if err != nil {
return allFailedParts, err
}
@@ -481,7 +481,9 @@ func (tst *tsTable) handleSyncIntroductions(partsToSync []*part, syncCh chan *sy
// syncStreamingPartsToNode synchronizes streaming parts to a node.
// Returns the list of failed parts.
-func (tst *tsTable) syncStreamingPartsToNode(ctx context.Context, node string, streamingParts []queue.StreamingPartData) ([]queue.FailedPart, error) {
+func (tst *tsTable) syncStreamingPartsToNode(
+ ctx context.Context, node string, parts []*part, streamingParts []queue.StreamingPartData,
+) ([]queue.FailedPart, error) {
// Get chunked sync client for this node
chunkedClient, err := tst.option.tire2Client.NewChunkedSyncClient(node, 1024*1024)
if err != nil {
@@ -491,9 +493,11 @@ func (tst *tsTable) syncStreamingPartsToNode(ctx context.Context, node string, s
// Sync parts using chunked transfer with streaming
result, err := chunkedClient.SyncStreamingParts(ctx, streamingParts)
+ acknowledgedAt := time.Now()
if err != nil {
return nil, fmt.Errorf("failed to sync streaming parts to node %s: %w", node, err)
}
+ tst.observeAcknowledgedFileSyncParts(parts, node, result, acknowledgedAt)
tst.incTotalSyncLoopBytes(result.TotalBytes)
if dl := tst.l.Debug(); dl.Enabled() {
@@ -510,3 +514,24 @@ func (tst *tsTable) syncStreamingPartsToNode(ctx context.Context, node string, s
return result.FailedParts, nil
}
+
+func (tst *tsTable) observeAcknowledgedFileSyncParts(parts []*part, node string, result *queue.SyncResult, acknowledgedAt time.Time) {
+ if result == nil || !result.Success {
+ return
+ }
+ failedPartIDs := make(map[string]struct{}, len(result.FailedParts))
+ for _, failedPart := range result.FailedParts {
+ failedPartIDs[failedPart.PartID] = struct{}{}
+ }
+ acknowledgedAtNanos := acknowledgedAt.UnixNano()
+ for _, part := range parts {
+ if _, failed := failedPartIDs[strconv.FormatUint(part.partMetadata.ID, 10)]; failed {
+ continue
+ }
+ queuedAtUnixNano := part.partMetadata.QueuedAtUnixNano
+ if queuedAtUnixNano <= 0 || queuedAtUnixNano > acknowledgedAtNanos {
+ continue
+ }
+ tst.observeFileSyncQueueLatency(float64(acknowledgedAtNanos-queuedAtUnixNano)/float64(time.Second), node)
+ }
+}
diff --git a/docs/operation/grafana-fodc-workload.json b/docs/operation/grafana-fodc-workload.json
index bd095b12a..c1f461e10 100644
--- a/docs/operation/grafana-fodc-workload.json
+++ b/docs/operation/grafana-fodc-workload.json
@@ -1188,7 +1188,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "description": "p99 latency of publish sends from liaison to data nodes, by operation.",
+ "description": "p99 publisher RPC latency by operation, plus core-part latency from liaison wqueue commit through a successful data-node file-sync acknowledgement. The latter includes queue residence, merge/flush scheduling, transfer, and retries.",
"fieldConfig": {
"defaults": {
"color": {
@@ -1279,9 +1279,45 @@
"legendFormat": "p99 {{operation}} {{group}}",
"range": true,
"refId": "A"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr": "histogram_quantile(0.99, sum(rate(banyandb_measure_file_sync_queue_latency_seconds_bucket{job=~\"$job\", container_name=\"liaison\", pod_name=~\"$pod\", group=~\"$group\"}[$__rate_interval])) by (le, group, remote_node))",
+ "instant": false,
+ "legendFormat": "wqueue measure {{group}} \u2192 {{remote_node}}",
+ "range": true,
+ "refId": "B"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr": "histogram_quantile(0.99, sum(rate(banyandb_stream_tst_file_sync_queue_latency_seconds_bucket{job=~\"$job\", container_name=\"liaison\", pod_name=~\"$pod\", group=~\"$group\"}[$__rate_interval])) by (le, group, remote_node))",
+ "instant": false,
+ "legendFormat": "wqueue stream {{group}} \u2192 {{remote_node}}",
+ "range": true,
+ "refId": "C"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr": "histogram_quantile(0.99, sum(rate(banyandb_trace_tst_file_sync_queue_latency_seconds_bucket{job=~\"$job\", container_name=\"liaison\", pod_name=~\"$pod\", group=~\"$group\"}[$__rate_interval])) by (le, group, remote_node))",
+ "instant": false,
+ "legendFormat": "wqueue trace {{group}} \u2192 {{remote_node}}",
+ "range": true,
+ "refId": "D"
}
],
- "title": "Publish Send Latency p99",
+ "title": "Publish & Wqueue-to-data Latency p99",
"type": "timeseries"
},
{
@@ -3159,7 +3195,7 @@
"type": "prometheus",
"uid": "${DS_PROMETHEUS}"
},
- "description": "p99 end-to-end latency of a single batch stream from open to close, for pub and sub.",
+ "description": "p99 batch-stream latency split into client-held-open and post-seal admission phases. The legacy pub/sub end-to-end series remain for compatibility.",
"fieldConfig": {
"defaults": {
"color": {
@@ -3208,34 +3244,9 @@
}
]
},
- "unit": "ops"
+ "unit": "s"
},
- "overrides": [
- {
- "matcher": {
- "id": "byFrameRefID",
- "options": "A"
- },
- "properties": [
- {
- "id": "unit",
- "value": "s"
- }
- ]
- },
- {
- "matcher": {
- "id": "byFrameRefID",
- "options": "B"
- },
- "properties": [
- {
- "id": "unit",
- "value": "s"
- }
- ]
- }
- ]
+ "overrides": []
},
"gridPos": {
"h": 7,
@@ -3287,9 +3298,33 @@
"legendFormat": "sub p99 {{group}}",
"range": true,
"refId": "B"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr": "histogram_quantile(0.99, sum(rate(banyandb_queue_pub_batch_open_duration_seconds_bucket{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\", operation=~\"$operation\", group=~\"$group\"}[$__rate_interval])) by (le, group))",
+ "instant": false,
+ "legendFormat": "open p99 {{group}}",
+ "range": true,
+ "refId": "C"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr": "histogram_quantile(0.99, sum(rate(banyandb_queue_pub_batch_admission_duration_seconds_bucket{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\", operation=~\"$operation\", group=~\"$group\"}[$__rate_interval])) by (le, group))",
+ "instant": false,
+ "legendFormat": "admission p99 {{group}}",
+ "range": true,
+ "refId": "D"
}
],
- "title": "Batch-write \u2014 Batch Stream Latency p99",
+ "title": "Batch-write \u2014 Stream Phase Latency p99",
"type": "timeseries"
},
{
@@ -3405,6 +3440,107 @@
"title": "Batch-write \u2014 Message Throughput",
"type": "timeseries"
},
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "Post-seal admission deadlines reached per target node. The counter is registered lazily, so an empty panel means no timeout has occurred.",
+ "fieldConfig": {
+ "defaults": {
+ "color": {
+ "mode": "palette-classic"
+ },
+ "custom": {
+ "axisBorderShow": false,
+ "axisCenteredZero": false,
+ "axisColorMode": "text",
+ "axisLabel": "",
+ "axisPlacement": "auto",
+ "barAlignment": 0,
+ "barWidthFactor": 0.6,
+ "drawStyle": "line",
+ "fillOpacity": 10,
+ "gradientMode": "none",
+ "hideFrom": {
+ "legend": false,
+ "tooltip": false,
+ "viz": false
+ },
+ "insertNulls": false,
+ "lineInterpolation": "linear",
+ "lineWidth": 1,
+ "pointSize": 5,
+ "scaleDistribution": {
+ "type": "linear"
+ },
+ "showPoints": "auto",
+ "spanNulls": false,
+ "stacking": {
+ "group": "A",
+ "mode": "none"
+ },
+ "thresholdsStyle": {
+ "mode": "off"
+ }
+ },
+ "mappings": [],
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "green",
+ "value": null
+ }
+ ]
+ },
+ "unit": "ops"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 7,
+ "w": 12,
+ "x": 12,
+ "y": 81
+ },
+ "id": 52,
+ "options": {
+ "legend": {
+ "calcs": [
+ "lastNotNull",
+ "max",
+ "mean"
+ ],
+ "displayMode": "table",
+ "placement": "bottom",
+ "showLegend": true,
+ "sortBy": "Last *",
+ "sortDesc": true
+ },
+ "tooltip": {
+ "mode": "multi",
+ "sort": "desc"
+ }
+ },
+ "pluginVersion": "11.2.0",
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr": "sum(rate(banyandb_queue_pub_batch_admission_timeout_total{job=~\"$job\", container_name=~\"$role\", pod_name=~\"$pod\", operation=~\"$operation\", group=~\"$group\"}[$__rate_interval])) by (group, remote_node)",
+ "instant": false,
+ "legendFormat": "{{group}} \u2192 {{remote_node}}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "Batch-write \u2014 Admission Timeout Rate",
+ "type": "timeseries"
+ },
{
"collapsed": false,
"gridPos": {
diff --git a/docs/operation/observability/metrics.md b/docs/operation/observability/metrics.md
index b35862915..2d7493d4e 100644
--- a/docs/operation/observability/metrics.md
+++ b/docs/operation/observability/metrics.md
@@ -233,17 +233,21 @@ The tier-2 publisher's success throughput by `operation`, alongside file-sync by
The liaison buffers writes in an **on-disk write queue (wqueue)** before syncing each part to the data nodes. These gauges (scoped with the literal `container_name="liaison"`) expose the queue depth per `pod_name`/`group`. The wqueue accumulates on disk — it does not drop data after a fixed window — so a sustained rise in any of these, especially `pending`, signals a slow or unavailable downstream data node and can eventually fill the liaison disk (`STATUS_DISK_FULL`). The matching `*_to [...]
+The `banyandb_{measure,stream_tst,trace_tst}_file_sync_queue_latency_seconds` histograms measure a different boundary: from a core part's commit into the liaison wqueue until that part receives a successful file-sync acknowledgement from each target data node. This includes queue residence, merge/flush scheduling, transfer, and retries. A merged part retains the oldest enqueue time of its inputs. Samples are labeled by `remote_node`; failed acknowledgements and legacy parts without an en [...]
+
**Expression1 (file parts)**: `sum(banyandb_{measure,stream_tst,trace_tst}_total_file_parts{job=~"$job", container_name="liaison", pod_name=~"$pod"}) by (pod_name, group)`
**Expression2 (mem parts)**: `sum(banyandb_{measure,stream_tst,trace_tst}_total_mem_part{job=~"$job", container_name="liaison", pod_name=~"$pod"}) by (pod_name, group)`
**Expression3 (pending)**: `sum(banyandb_{measure,stream_tst,trace_tst}_pending_data_count{job=~"$job", container_name="liaison", pod_name=~"$pod"}) by (pod_name, group)`
+**Expression4 (queue-to-data p99)**: `histogram_quantile(0.99, sum(rate(banyandb_{measure,stream_tst,trace_tst}_file_sync_queue_latency_seconds_bucket{job=~"$job", container_name="liaison", pod_name=~"$pod"}[$__rate_interval])) by (le, pod_name, group, remote_node))`
-> The panel charts all three families for `measure`, `stream_tst` and `trace_tst` as separate series (nine queries); the `{measure,stream_tst,trace_tst}` brace above is shorthand for the three concrete metric names.
+> The backlog panel charts all three gauge families for `measure`, `stream_tst` and `trace_tst` as separate series (nine queries). The neighboring latency panel adds the three queue-to-data histograms. The `{measure,stream_tst,trace_tst}` brace above is shorthand for the three concrete metric names.
-### Publish Send Latency p99
+### Publish & Wqueue-to-data Latency p99
-The 99th-percentile tier-2 send latency, per `operation`, from the publisher latency histogram.
+The 99th-percentile tier-2 publisher RPC latency is shown beside the full queue-to-data latency for measure, stream and trace core parts. The first series starts when an RPC is sent; the other three start when the part enters the liaison wqueue.
-**Expression**: `histogram_quantile(0.99, sum(rate(banyandb_queue_pub_total_latency_bucket{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (le, operation))`
+**Expression1 (publisher RPC)**: `histogram_quantile(0.99, sum(rate(banyandb_queue_pub_total_latency_bucket{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (le, operation, group))`
+**Expression2 (queue-to-data)**: `histogram_quantile(0.99, sum(rate(banyandb_{measure,stream_tst,trace_tst}_file_sync_queue_latency_seconds_bucket{job=~"$job", container_name="liaison", pod_name=~"$pod"}[$__rate_interval])) by (le, group, remote_node))`
## Data: Storage
@@ -480,6 +484,8 @@ Each panel uses the same three expressions, with `operation` pinned to the panel
> `batch-write` covers both plain writes and the secondary-index sync (measure/stream series-index, stream local-index, trace sidx-series). All of these carry their business `group`, so the per-group breakdown is complete for every operation.
+The batch-write section also exposes per-batch and per-message throughput. Its phase-latency panel keeps the legacy publisher/subscriber `total_batch_latency` p99 and adds the publisher's first-frame-to-seal `batch_open_duration_seconds` p99 and seal-to-terminal `batch_admission_duration_seconds` p99. The admission-timeout panel uses `rate(banyandb_queue_pub_batch_admission_timeout_total[$__rate_interval])` by `group` and `remote_node`; it is empty until the first deadline is reached.
+
---
# Internal queue metrics reference (`queue_sub` / `queue_pub`)
@@ -507,8 +513,13 @@ 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 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` | Counter | `operation`, `group`, `remote_node`, `remote_role`, `remote_tier` | Batch streams opened / terminal responses received. |
+| `total_batch_latency` | Histogram | `operation`, `group`, `remote_node`, `remote_role`, `remote_tier` | Legacy stream-creation-to-terminal-response duration. It includes both time that the client holds the stream open and post-seal admission time. |
+| `batch_open_duration_seconds` | Histogram | `operation`, `group`, `remote_node`, `remote_role`, `remote_tier` | First successfully sent frame to the batch seal. Long-lived client streams are expected and use buckets through 24 hours. |
+| `batch_admission_duration_seconds` | Histogram | `operation`, `group`, `remote_node`, `remote_role`, `remote_tier` | Batch seal to terminal response, transport failure, or admission deadline. This is the phase governed by the batch publisher timeout. |
+| `batch_admission_timeout_total` | Counter | `operation`, `group`, `remote_node`, `remote_role`, `remote_tier` | Admission deadlines reached, once per still-pending target. Lazily registered and therefore absent on a healthy cluster. |
-**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.
+**Troubleshooting:** `total_err` by `error_type` separates transport failures (`recv_error`) from application-level `SendResponse` errors (`server_rejected`) and exhausted retries (`retry_exhausted`). For `batch-write`, use `batch_open_duration_seconds` to identify client-side stream lifetime and `batch_admission_duration_seconds` plus `batch_admission_timeout_total` to isolate post-seal liaison admission. Do not use the legacy `total_batch_latency` alone for this distinction. For `file- [...]
Metrics are only registered when `metadata` implements `metadata.Service` and `MetricsRegistry()` is non-nil (e.g. after `SetMetricsRegistry` in bootstrap). `NewWithoutMetadata()` leaves `queue_pub` metrics disabled and logs a warning (`queue_pub metrics disabled: ...`). The `total_err` counters above are registered lazily on first occurrence, so they are simply absent (not zero) on a healthy cluster.
@@ -522,10 +533,14 @@ Saturation (scope by node with the proxy labels):
- **Publisher success rate:** `sum(rate(banyandb_queue_pub_total_finished{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (operation)`
- **Publisher errors by type:** `sum(rate(banyandb_queue_pub_total_err{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (operation, error_type)`
- **Publisher file-sync bytes sent (by tier):** `sum(rate(banyandb_queue_pub_sent_bytes{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (remote_tier)`
+- **Batch-open p99:** `histogram_quantile(0.99, sum(rate(banyandb_queue_pub_batch_open_duration_seconds_bucket{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (le, group))`
+- **Post-seal admission p99:** `histogram_quantile(0.99, sum(rate(banyandb_queue_pub_batch_admission_duration_seconds_bucket{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (le, group))`
+- **Admission timeouts by target:** `sum(rate(banyandb_queue_pub_batch_admission_timeout_total{job=~"$job", container_name=~"$role", pod_name=~"$pod"}[$__rate_interval])) by (group, remote_node)`
**Suggested alerts (tune thresholds per cluster):**
- Non-zero sustained `rate(banyandb_queue_pub_total_err{error_type="retry_exhausted"}[5m])` on liaison.
+- Non-zero `increase(banyandb_queue_pub_batch_admission_timeout_total[5m])`, or post-seal admission p99 approaching the configured write timeout.
- A sustained `rate(banyandb_queue_sub_total_started[5m]) - rate(banyandb_queue_sub_total_finished[5m])` gap (subscribe backlog), or `total_latency` p99 above an environment-specific ceiling, for a single `group` / `operation`.
### Aggregate pipeline error rate (optional)
diff --git a/pkg/meter/meter.go b/pkg/meter/meter.go
index 66382302a..5b9fa5672 100644
--- a/pkg/meter/meter.go
+++ b/pkg/meter/meter.go
@@ -36,6 +36,14 @@ var DefBuckets = Buckets{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10}
// 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}
+// QueueLatencyBuckets is the bucket set for durable queue-residence latency.
+// A downstream outage can leave data queued for hours, so the upper bound
+// extends to 24 hours while retaining the short-latency resolution of DefBuckets.
+var QueueLatencyBuckets = Buckets{
+ .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, 30, 60, 120, 300,
+ 600, 900, 1800, 3600, 10_800, 21_600, 43_200, 86_400,
+}
+
// 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))
diff --git a/test/e2e-v2/cases/fodc/README.md b/test/e2e-v2/cases/fodc/README.md
index 66fd96d8d..5be8a3fb5 100644
--- a/test/e2e-v2/cases/fodc/README.md
+++ b/test/e2e-v2/cases/fodc/README.md
@@ -40,17 +40,17 @@ consumer ─POST /users─▶ provider ──▶ OAP ──┬─ writes ─▶
## What is asserted
-The metric universe (106 distinct names, extracted mechanically from dashboard
+The metric universe (115 distinct names, extracted mechanically from dashboard
panel `expr` fields) is partitioned into four checked-in lists under `metrics/`:
| List | Meaning | check-metrics.sh action |
|------|---------|-------------------------|
| `presence.txt` (44) | reliably exported by a 1:1 OAP cluster; value may be 0 | assert ≥ 1 exported series |
-| `non_empty.txt` (14) | always-on core through OAP traffic | assert a sample value > 0 (histogram families via the companion `_count`) |
-| `documented_gap.txt` (20) | need a specific event or an undeployed lifecycle service | reported only, never asserted |
-| `optional_plugin.txt` (28) | used only by the separately installed trace-plugin dashboard | excluded from the base run; validate in a plugin-enabled cluster |
+| `non_empty.txt` (19) | always-on core through OAP traffic | assert a sample value > 0 (histogram families via the companion `_count`) |
+| `documented_gap.txt` (21) | need a specific event or an undeployed lifecycle service | reported only, never asserted |
+| `optional_plugin.txt` (31) | used only by the separately installed trace-plugin dashboard | excluded from the base run; validate in a plugin-enabled cluster |
-The four lists are disjoint and their union is exactly the 106 dashboard metrics —
+The four lists are disjoint and their union is exactly the 115 dashboard metrics —
nothing is silently dropped. `check-metrics.sh` additionally asserts that the two
agents registered with **distinct `node_role` labels** and that the dashboard's
query-throughput series `banyandb_queue_sub_*{operation="query"}` is present on the
@@ -61,7 +61,7 @@ the OAP→liaison→data query fan-out is observable.
The split was calibrated against a live run; the recorded findings:
-- **`documented_gap.txt` holds event-driven and undeployed-service families (20 names), reported but never asserted.**
+- **`documented_gap.txt` holds event-driven and undeployed-service families (21 names), reported but never asserted.**
*Error counters* (`*_grpc_total_err`, `*_grpc_total_stream_msg_received_err`, four
`*_total_sync_loop_err`, `queue_pub_total_err`): a labelled Prometheus counter is
not exported until first incremented, so on a healthy run they never appear.
diff --git a/test/e2e-v2/cases/fodc/check-metrics.sh b/test/e2e-v2/cases/fodc/check-metrics.sh
index b3e87b18c..7b3caa72a 100644
--- a/test/e2e-v2/cases/fodc/check-metrics.sh
+++ b/test/e2e-v2/cases/fodc/check-metrics.sh
@@ -176,6 +176,9 @@ QUEUE_LABEL_METRICS=(
banyandb_queue_pub_total_batch_started
banyandb_queue_pub_total_batch_finished
banyandb_queue_pub_total_batch_latency_bucket
+ banyandb_queue_pub_batch_open_duration_seconds_bucket
+ banyandb_queue_pub_batch_admission_duration_seconds_bucket
+ banyandb_queue_pub_batch_admission_timeout_total
banyandb_queue_sub_total_started
banyandb_queue_sub_total_finished
banyandb_queue_sub_total_latency_bucket
@@ -193,6 +196,18 @@ for m in "${QUEUE_LABEL_METRICS[@]}"; do
done
done
+# Liaison wqueue-to-data histograms use group as a database-level const label
+# and remote_node as the per-target metric label consumed by panel 31.
+for m in \
+ banyandb_measure_file_sync_queue_latency_seconds_bucket \
+ banyandb_stream_tst_file_sync_queue_latency_seconds_bucket \
+ banyandb_trace_tst_file_sync_queue_latency_seconds_bucket; do
+ series_present "${m}" || continue
+ for lbl in group remote_node; do
+ label_exists "${m}" "${lbl}" || MISSING_LABELS="${MISSING_LABELS} ${m}(${lbl})"
+ done
+done
+
# System metrics: nodes dashboard filters banyandb_system_{memory_state,disk,net_state}
# by {kind=...}. The kind dimension is metric-level, not an infra label.
for m in banyandb_system_memory_state banyandb_system_disk banyandb_system_net_state; do
diff --git a/test/e2e-v2/cases/fodc/metrics/documented_gap.txt b/test/e2e-v2/cases/fodc/metrics/documented_gap.txt
index f4bea13b3..b2df3e7f7 100644
--- a/test/e2e-v2/cases/fodc/metrics/documented_gap.txt
+++ b/test/e2e-v2/cases/fodc/metrics/documented_gap.txt
@@ -24,6 +24,7 @@
banyandb_liaison_grpc_total_err
banyandb_liaison_grpc_total_stream_msg_received_err
banyandb_queue_pub_total_err
+banyandb_queue_pub_batch_admission_timeout_total
banyandb_queue_sub_total_err
#
# Group 2 -- part-merge completion metrics. These increment only when a
diff --git a/test/e2e-v2/cases/fodc/metrics/non_empty.txt b/test/e2e-v2/cases/fodc/metrics/non_empty.txt
index 60c16624b..c83b52694 100644
--- a/test/e2e-v2/cases/fodc/metrics/non_empty.txt
+++ b/test/e2e-v2/cases/fodc/metrics/non_empty.txt
@@ -34,3 +34,8 @@ banyandb_queue_pub_total_latency_bucket
banyandb_queue_sub_total_latency_bucket
banyandb_queue_pub_total_batch_latency_bucket
banyandb_queue_sub_total_batch_latency_bucket
+banyandb_queue_pub_batch_open_duration_seconds_bucket
+banyandb_queue_pub_batch_admission_duration_seconds_bucket
+banyandb_measure_file_sync_queue_latency_seconds_bucket
+banyandb_stream_tst_file_sync_queue_latency_seconds_bucket
+banyandb_trace_tst_file_sync_queue_latency_seconds_bucket