add kvs range duration

internal/observability/metrics/agent/core/ngt/ngt.go

@@ -49,6 +49,9 @@ const (
 
 	brokenIndexStoreCountMetricsName        = "agent_core_ngt_broken_index_store_count"
 	brokenIndexStoreCountMetricsDescription = "How many broken index generations have been stored"
+
+	kvsRangeDurationMetricsName        = "agent_core_ngt_kvs_range_duration"
+	kvsRangeDurationMetricsDescription = "The duration of the kvs range method"
 )
 
 type ngtMetrics struct {
@@ -143,6 +146,15 @@ func (n *ngtMetrics) View() ([]*metrics.View, error) {
 		return nil, err
 	}
 
+	kvsRangeDuration, err := view.New(
+		view.MatchInstrumentName(kvsRangeDurationMetricsName),
+		view.WithSetDescription(kvsRangeDurationMetricsDescription),
+		view.WithSetAggregation(aggregation.LastValue{}),
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	return []*metrics.View{
 		&indexCount,
 		&uncommittedIndexCount,
@@ -153,6 +165,7 @@ func (n *ngtMetrics) View() ([]*metrics.View, error) {
 		&isIndexing,
 		&isSaving,
 		&brokenIndexCount,
+		&kvsRangeDuration,
 	}, nil
 }
 
@@ -238,6 +251,15 @@ func (n *ngtMetrics) Register(m metrics.Meter) error {
 		return err
 	}
 
+	kvsRangeDuration, err := m.AsyncInt64().Gauge(
+		kvsRangeDurationMetricsName,
+		metrics.WithDescription(kvsRangeDurationMetricsDescription),
+		metrics.WithUnit(metrics.Dimensionless),
+	)
+	if err != nil {
+		return err
+	}
+
 	return m.RegisterCallback(
 		[]metrics.AsynchronousInstrument{
 			indexCount,
@@ -249,6 +271,7 @@ func (n *ngtMetrics) Register(m metrics.Meter) error {
 			isIndexing,
 			isSaving,
 			brokenIndexCount,
+			kvsRangeDuration,
 		},
 		func(ctx context.Context) {
 			var indexing int64
@@ -270,6 +293,7 @@ func (n *ngtMetrics) Register(m metrics.Meter) error {
 			isIndexing.Observe(ctx, int64(indexing))
 			isSaving.Observe(ctx, int64(saving))
 			brokenIndexCount.Observe(ctx, int64(n.ngt.BrokenIndexCount()))
+			kvsRangeDuration.Observe(ctx, n.ngt.KvsRangeDuration())
 		},
 	)
 }

pkg/agent/core/ngt/service/kvs/kvs.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/vdaas/vald/internal/errgroup"
 	"github.com/vdaas/vald/internal/safety"
@@ -37,6 +38,7 @@ type BidiMap interface {
 	Range(ctx context.Context, f func(string, uint32, int64) bool)
 	Len() uint64
 	Close() error
+	RangeDuration() int64
 }
 
 type valueStructOu struct {
@@ -66,6 +68,11 @@ const (
 	// mask = 0xFFF.
 )
 
+var (
+	tmu      sync.RWMutex
+	rangeDur int64
+)
+
 // New returns the bidi that satisfies the BidiMap interface.
 func New(opts ...Option) BidiMap {
 	b := &bidi{
@@ -152,6 +159,8 @@ func (b *bidi) DeleteInverse(val uint32) (key string, ok bool) {
 
 // Range retrieves all set keys and values and calls the callback function f.
 func (b *bidi) Range(ctx context.Context, f func(string, uint32, int64) bool) {
+	start := time.Now()
+
 	var wg sync.WaitGroup
 	for i := range b.uo {
 		idx := i
@@ -171,6 +180,11 @@ func (b *bidi) Range(ctx context.Context, f func(string, uint32, int64) bool) {
 		}))
 	}
 	wg.Wait()
+
+	dur := time.Since(start).Nanoseconds()
+	tmu.Lock()
+	rangeDur = dur
+	tmu.Unlock()
 }
 
 // Len returns the length of the cache that is set in the bidi.
@@ -187,3 +201,9 @@ func (b *bidi) Close() error {
 	}
 	return b.eg.Wait()
 }
+
+func (b *bidi) RangeDuration() int64 {
+	tmu.RLock()
+	defer tmu.RUnlock()
+	return rangeDur
+}

pkg/agent/core/ngt/service/ngt.go

@@ -83,6 +83,7 @@ type NGT interface {
 	GetDimensionSize() int
 	Close(ctx context.Context) error
 	BrokenIndexCount() uint64
+	KvsRangeDuration() int64
 }
 
 type ngt struct {
@@ -1721,3 +1722,7 @@ func (n *ngt) BrokenIndexCount() uint64 {
 func (n *ngt) ListObjectFunc(ctx context.Context, f func(uuid string, oid uint32, ts int64) bool) {
 	n.kvs.Range(ctx, f)
 }
+
+func (n *ngt) KvsRangeDuration() int64 {
+	return n.kvs.RangeDuration()
+}