enhance: expose more metrics data (milvus-io#39456)

jaime0815 · web-flow · commit 8a4ac8cccd44 · 2025-02-07T11:50:50.000+08:00
issue: milvus-io#36621 milvus-io#39417 1. Adjust the server-side cache size. 2. Add source information for configurations. 3. Add node ID for compaction and indexing tasks. 4. Resolve localhost access issues to fix health check failures for etcd. Signed-off-by: jaime <yun.zhang@zilliz.com>
diff --git a/deployments/docker/standalone/docker-compose.yml b/deployments/docker/standalone/docker-compose.yml
@@ -11,7 +11,7 @@ services:
       - ETCD_SNAPSHOT_COUNT=50000
     volumes:
       - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd
-    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
+    command: etcd -advertise-client-urls=http://etcd:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
     healthcheck:
       test: ["CMD", "etcdctl", "endpoint", "health"]
       interval: 30s
diff --git a/internal/datacoord/compaction_task_meta.go b/internal/datacoord/compaction_task_meta.go
@@ -52,6 +52,7 @@ func newCompactionTaskStats(task *datapb.CompactionTask) *metricsinfo.Compaction
 		ResultSegments: lo.Map(task.ResultSegments, func(t int64, i int) string {
 			return strconv.FormatInt(t, 10)
 		}),
+		NodeID: task.NodeID,
 	}
 }
 
@@ -70,7 +71,7 @@ func newCompactionTaskMeta(ctx context.Context, catalog metastore.DataCoordCatal
 		ctx:             ctx,
 		catalog:         catalog,
 		compactionTasks: make(map[int64]map[int64]*datapb.CompactionTask, 0),
-		taskStats:       expirable.NewLRU[UniqueID, *metricsinfo.CompactionTask](32, nil, time.Minute*15),
+		taskStats:       expirable.NewLRU[UniqueID, *metricsinfo.CompactionTask](512, nil, time.Minute*15),
 	}
 	if err := csm.reloadFromKV(); err != nil {
 		return nil, err
diff --git a/internal/datacoord/import_meta.go b/internal/datacoord/import_meta.go
@@ -53,7 +53,7 @@ type importTasks struct {
 func newImportTasks() *importTasks {
 	return &importTasks{
 		tasks:     make(map[int64]ImportTask),
-		taskStats: expirable.NewLRU[UniqueID, ImportTask](64, nil, time.Minute*30),
+		taskStats: expirable.NewLRU[UniqueID, ImportTask](512, nil, time.Minute*30),
 	}
 }
 
diff --git a/internal/datacoord/index_meta.go b/internal/datacoord/index_meta.go
@@ -82,6 +82,7 @@ func newIndexTaskStats(s *model.SegmentIndex) *metricsinfo.IndexTaskStats {
 		IndexVersion:    s.IndexVersion,
 		CreatedUTCTime:  typeutil.TimestampToString(s.CreatedUTCTime * 1000),
 		FinishedUTCTime: typeutil.TimestampToString(s.FinishedUTCTime * 1000),
+		NodeID:          s.NodeID,
 	}
 }
 
@@ -98,7 +99,7 @@ func newSegmentIndexBuildInfo() *segmentBuildInfo {
 		// build ID -> segment index
 		buildID2SegmentIndex: make(map[UniqueID]*model.SegmentIndex),
 		// build ID -> task stats
-		taskStats: expirable.NewLRU[UniqueID, *metricsinfo.IndexTaskStats](64, nil, time.Minute*30),
+		taskStats: expirable.NewLRU[UniqueID, *metricsinfo.IndexTaskStats](1024, nil, time.Minute*30),
 	}
 }
 
diff --git a/internal/datacoord/job_manager_test.go b/internal/datacoord/job_manager_test.go
@@ -104,7 +104,7 @@ func (s *jobManagerSuite) TestJobManager_triggerStatsTaskLoop() {
 			allocator: alloc,
 			tasks:     make(map[int64]Task),
 			meta:      mt,
-			taskStats: expirable.NewLRU[UniqueID, Task](64, nil, time.Minute*5),
+			taskStats: expirable.NewLRU[UniqueID, Task](512, nil, time.Minute*5),
 		},
 		allocator: alloc,
 	}
diff --git a/internal/datacoord/task_scheduler.go b/internal/datacoord/task_scheduler.go
@@ -93,7 +93,7 @@ func newTaskScheduler(
 		handler:                   handler,
 		indexEngineVersionManager: indexEngineVersionManager,
 		allocator:                 allocator,
-		taskStats:                 expirable.NewLRU[UniqueID, Task](64, nil, time.Minute*15),
+		taskStats:                 expirable.NewLRU[UniqueID, Task](512, nil, time.Minute*15),
 		compactionHandler:         compactionHandler,
 	}
 	ts.reloadFromMeta()
diff --git a/internal/datacoord/task_stats.go b/internal/datacoord/task_stats.go
@@ -142,7 +142,7 @@ func (st *statsTask) UpdateVersion(ctx context.Context, nodeID int64, meta *meta
 		log.Warn("segment is contains by l0 compaction, skip stats", zap.Int64("taskID", st.taskID),
 			zap.Int64("segmentID", st.segmentID))
 		st.SetState(indexpb.JobState_JobStateFailed, "segment is contains by l0 compaction")
-		//reset compacting
+		// reset compacting
 		meta.SetSegmentsCompacting(ctx, []UniqueID{st.segmentID}, false)
 		st.SetStartTime(time.Now())
 		return fmt.Errorf("segment is contains by l0 compaction")
diff --git a/internal/flushcommon/syncmgr/sync_manager.go b/internal/flushcommon/syncmgr/sync_manager.go
@@ -72,7 +72,7 @@ func NewSyncManager(chunkManager storage.ChunkManager) SyncManager {
 		keyLockDispatcher: dispatcher,
 		chunkManager:      chunkManager,
 		tasks:             typeutil.NewConcurrentMap[string, Task](),
-		taskStats:         expirable.NewLRU[string, Task](16, nil, time.Minute*15),
+		taskStats:         expirable.NewLRU[string, Task](64, nil, time.Minute*15),
 	}
 	// setup config update watcher
 	params.Watch(params.DataNodeCfg.MaxParallelSyncMgrTasks.Key, config.NewHandler("datanode.syncmgr.poolsize", syncMgr.resizeHandler))
diff --git a/internal/proxy/impl.go b/internal/proxy/impl.go
@@ -6753,7 +6753,7 @@ func DeregisterSubLabel(subLabel string) {
 func (node *Proxy) RegisterRestRouter(router gin.IRouter) {
 	// Cluster request that executed by proxy
 	router.GET(http.ClusterInfoPath, getClusterInfo(node))
-	router.GET(http.ClusterConfigsPath, getConfigs(paramtable.Get().GetAll()))
+	router.GET(http.ClusterConfigsPath, getConfigs(paramtable.Get().GetConfigsView()))
 	router.GET(http.ClusterClientsPath, getConnectedClients)
 	router.GET(http.ClusterDependenciesPath, getDependencies)
 
diff --git a/internal/querycoordv2/task/scheduler.go b/internal/querycoordv2/task/scheduler.go
@@ -284,7 +284,7 @@ func NewScheduler(ctx context.Context,
 		channelTasks:     NewConcurrentMap[replicaChannelIndex, Task](),
 		processQueue:     newTaskQueue(),
 		waitQueue:        newTaskQueue(),
-		taskStats:        expirable.NewLRU[UniqueID, Task](64, nil, time.Minute*15),
+		taskStats:        expirable.NewLRU[UniqueID, Task](256, nil, time.Minute*15),
 		segmentTaskDelta: NewExecutingTaskDelta(),
 		channelTaskDelta: NewExecutingTaskDelta(),
 	}
diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go
@@ -30,17 +30,17 @@ import (
 
 func TestConfigFromEnv(t *testing.T) {
 	mgr, _ := Init()
-	_, err := mgr.GetConfig("test.env")
+	_, _, err := mgr.GetConfig("test.env")
 	assert.ErrorIs(t, err, ErrKeyNotFound)
 
 	t.Setenv("TEST_ENV", "value")
 	mgr, _ = Init(WithEnvSource(formatKey))
 
-	v, err := mgr.GetConfig("test.env")
+	_, v, err := mgr.GetConfig("test.env")
 	assert.NoError(t, err)
 	assert.Equal(t, "value", v)
 
-	v, err = mgr.GetConfig("TEST_ENV")
+	_, v, err = mgr.GetConfig("TEST_ENV")
 	assert.NoError(t, err)
 	assert.Equal(t, "value", v)
 }
@@ -67,65 +67,65 @@ func TestConfigFromRemote(t *testing.T) {
 	ctx := context.Background()
 
 	t.Run("origin is empty", func(t *testing.T) {
-		_, err = mgr.GetConfig("test.etcd")
+		_, _, err = mgr.GetConfig("test.etcd")
 		assert.ErrorIs(t, err, ErrKeyNotFound)
 
 		client.KV.Put(ctx, "test/config/test/etcd", "value")
 
 		time.Sleep(100 * time.Millisecond)
 
-		v, err := mgr.GetConfig("test.etcd")
+		_, v, err := mgr.GetConfig("test.etcd")
 		assert.NoError(t, err)
 		assert.Equal(t, "value", v)
-		v, err = mgr.GetConfig("TEST_ETCD")
+		_, v, err = mgr.GetConfig("TEST_ETCD")
 		assert.NoError(t, err)
 		assert.Equal(t, "value", v)
 
 		client.KV.Delete(ctx, "test/config/test/etcd")
 		time.Sleep(100 * time.Millisecond)
 
-		_, err = mgr.GetConfig("TEST_ETCD")
+		_, _, err = mgr.GetConfig("TEST_ETCD")
 		assert.ErrorIs(t, err, ErrKeyNotFound)
 	})
 
 	t.Run("override origin value", func(t *testing.T) {
-		v, _ := mgr.GetConfig("tmp.key")
+		_, v, _ := mgr.GetConfig("tmp.key")
 		assert.Equal(t, "1", v)
 		client.KV.Put(ctx, "test/config/tmp/key", "2")
 
 		time.Sleep(100 * time.Millisecond)
 
-		v, _ = mgr.GetConfig("tmp.key")
+		_, v, _ = mgr.GetConfig("tmp.key")
 		assert.Equal(t, "2", v)
 
 		client.KV.Put(ctx, "test/config/tmp/key", "3")
 
 		time.Sleep(100 * time.Millisecond)
 
-		v, _ = mgr.GetConfig("tmp.key")
+		_, v, _ = mgr.GetConfig("tmp.key")
 		assert.Equal(t, "3", v)
 
 		client.KV.Delete(ctx, "test/config/tmp/key")
 		time.Sleep(100 * time.Millisecond)
 
-		v, _ = mgr.GetConfig("tmp.key")
+		_, v, _ = mgr.GetConfig("tmp.key")
 		assert.Equal(t, "1", v)
 	})
 
 	t.Run("multi priority", func(t *testing.T) {
-		v, _ := mgr.GetConfig("log.level")
+		_, v, _ := mgr.GetConfig("log.level")
 		assert.Equal(t, "info", v)
 		client.KV.Put(ctx, "test/config/log/level", "error")
 
 		time.Sleep(100 * time.Millisecond)
 
-		v, _ = mgr.GetConfig("log.level")
+		_, v, _ = mgr.GetConfig("log.level")
 		assert.Equal(t, "error", v)
 
 		client.KV.Delete(ctx, "test/config/log/level")
 		time.Sleep(100 * time.Millisecond)
 
-		v, _ = mgr.GetConfig("log.level")
+		_, v, _ = mgr.GetConfig("log.level")
 		assert.Equal(t, "info", v)
 	})
 
@@ -134,7 +134,7 @@ func TestConfigFromRemote(t *testing.T) {
 
 		client.KV.Put(ctx, "test/config/test/etcd", "value2")
 		assert.Eventually(t, func() bool {
-			_, err = mgr.GetConfig("test.etcd")
+			_, _, err = mgr.GetConfig("test.etcd")
 			return err != nil && errors.Is(err, ErrKeyNotFound)
 		}, 300*time.Millisecond, 10*time.Millisecond)
 	})
diff --git a/pkg/config/manager.go b/pkg/config/manager.go
@@ -28,7 +28,8 @@ import (
 )
 
 const (
-	TombValue = "TOMB_VAULE"
+	TombValue     = "TOMB_VAULE"
+	RuntimeSource = "RuntimeSource"
 )
 
 type Filter func(key string) (string, bool)
@@ -118,7 +119,7 @@ func (m *Manager) GetCachedValue(key string) (interface{}, bool) {
 func (m *Manager) CASCachedValue(key string, origin string, value interface{}) bool {
 	m.cacheMutex.Lock()
 	defer m.cacheMutex.Unlock()
-	current, err := m.GetConfig(key)
+	_, current, err := m.GetConfig(key)
 	if errors.Is(err, ErrKeyNotFound) {
 		m.configCache[key] = value
 		return true
@@ -147,28 +148,29 @@ func (m *Manager) EvictCacheValueByFormat(keys ...string) {
 	clear(m.configCache)
 }
 
-func (m *Manager) GetConfig(key string) (string, error) {
+func (m *Manager) GetConfig(key string) (string, string, error) {
 	realKey := formatKey(key)
 	v, ok := m.overlays.Get(realKey)
 	if ok {
 		if v == TombValue {
-			return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found %s", key)
+			return "", "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found %s", key)
 		}
-		return v, nil
+		return RuntimeSource, v, nil
 	}
 	sourceName, ok := m.keySourceMap.Get(realKey)
 	if !ok {
-		return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found: %s", key)
+		return "", "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found: %s", key)
 	}
-	return m.getConfigValueBySource(realKey, sourceName)
+	v, err := m.getConfigValueBySource(realKey, sourceName)
+	return sourceName, v, err
 }
 
 // GetConfigs returns all the key values
 func (m *Manager) GetConfigs() map[string]string {
 	config := make(map[string]string)
 
 	m.keySourceMap.Range(func(key, value string) bool {
-		sValue, err := m.GetConfig(key)
+		_, sValue, err := m.GetConfig(key)
 		if err != nil {
 			return true
 		}
@@ -185,15 +187,40 @@ func (m *Manager) GetConfigs() map[string]string {
 	return config
 }
 
+func (m *Manager) GetConfigsView() map[string]string {
+	config := make(map[string]string)
+
+	valueFmt := func(source, value string) string {
+		return fmt.Sprintf("%s[%s]", value, source)
+	}
+
+	m.keySourceMap.Range(func(key, value string) bool {
+		source, sValue, err := m.GetConfig(key)
+		if err != nil {
+			return true
+		}
+
+		config[key] = valueFmt(source, sValue)
+		return true
+	})
+
+	m.overlays.Range(func(key, value string) bool {
+		config[key] = valueFmt(RuntimeSource, value)
+		return true
+	})
+
+	return config
+}
+
 func (m *Manager) GetBy(filters ...Filter) map[string]string {
 	matchedConfig := make(map[string]string)
 
-	m.keySourceMap.Range(func(key, value string) bool {
+	m.keySourceMap.Range(func(key string, value string) bool {
 		newkey, ok := filterate(key, filters...)
 		if !ok {
 			return true
 		}
-		sValue, err := m.GetConfig(key)
+		_, sValue, err := m.GetConfig(key)
 		if err != nil {
 			return true
 		}
diff --git a/pkg/config/manager_test.go b/pkg/config/manager_test.go
diff --git a/pkg/util/metricsinfo/metrics_info.go b/pkg/util/metricsinfo/metrics_info.go
diff --git a/pkg/util/paramtable/base_table.go b/pkg/util/paramtable/base_table.go
diff --git a/pkg/util/paramtable/component_param.go b/pkg/util/paramtable/component_param.go
diff --git a/pkg/util/paramtable/param_item.go b/pkg/util/paramtable/param_item.go

Original file line number	Diff line number	Diff line change
`@@ -52,6 +52,7 @@ func newCompactionTaskStats(task datapb.CompactionTask) metricsinfo.Compaction`
`52`	`52`	`ResultSegments: lo.Map(task.ResultSegments, func(t int64, i int) string {`
`53`	`53`	`return strconv.FormatInt(t, 10)`
`54`	`54`	`}),`
	`55`	`+ NodeID: task.NodeID,`
`55`	`56`	`}`
`56`	`57`	`}`
`57`	`58`
`@@ -70,7 +71,7 @@ func newCompactionTaskMeta(ctx context.Context, catalog metastore.DataCoordCatal`
`70`	`71`	`ctx: ctx,`
`71`	`72`	`catalog: catalog,`
`72`	`73`	`compactionTasks: make(map[int64]map[int64]*datapb.CompactionTask, 0),`
`73`		`- taskStats: expirable.NewLRU[UniqueID, metricsinfo.CompactionTask](32, nil, time.Minute15),`
	`74`	`+ taskStats: expirable.NewLRU[UniqueID, metricsinfo.CompactionTask](512, nil, time.Minute15),`
`74`	`75`	`}`
`75`	`76`	`if err := csm.reloadFromKV(); err != nil {`
`76`	`77`	`return nil, err`
Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ type importTasks struct {`
`53`	`53`	`func newImportTasks() *importTasks {`
`54`	`54`	`return &importTasks{`
`55`	`55`	`tasks: make(map[int64]ImportTask),`
`56`		`- taskStats: expirable.NewLRU[UniqueID, ImportTask](64, nil, time.Minute*30),`
	`56`	`+ taskStats: expirable.NewLRU[UniqueID, ImportTask](512, nil, time.Minute*30),`
`57`	`57`	`}`
`58`	`58`	`}`
`59`	`59`
Original file line number	Diff line number	Diff line change
`@@ -82,6 +82,7 @@ func newIndexTaskStats(s model.SegmentIndex) metricsinfo.IndexTaskStats {`
`82`	`82`	`IndexVersion: s.IndexVersion,`
`83`	`83`	`CreatedUTCTime: typeutil.TimestampToString(s.CreatedUTCTime * 1000),`
`84`	`84`	`FinishedUTCTime: typeutil.TimestampToString(s.FinishedUTCTime * 1000),`
	`85`	`+ NodeID: s.NodeID,`
`85`	`86`	`}`
`86`	`87`	`}`
`87`	`88`
`@@ -98,7 +99,7 @@ func newSegmentIndexBuildInfo() *segmentBuildInfo {`
`98`	`99`	`// build ID -> segment index`
`99`	`100`	`buildID2SegmentIndex: make(map[UniqueID]*model.SegmentIndex),`
`100`	`101`	`// build ID -> task stats`
`101`		`- taskStats: expirable.NewLRU[UniqueID, metricsinfo.IndexTaskStats](64, nil, time.Minute30),`
	`102`	`+ taskStats: expirable.NewLRU[UniqueID, metricsinfo.IndexTaskStats](1024, nil, time.Minute30),`
`102`	`103`	`}`
`103`	`104`	`}`
`104`	`105`
Original file line number	Diff line number	Diff line change
`@@ -104,7 +104,7 @@ func (s *jobManagerSuite) TestJobManager_triggerStatsTaskLoop() {`
`104`	`104`	`allocator: alloc,`
`105`	`105`	`tasks: make(map[int64]Task),`
`106`	`106`	`meta: mt,`
`107`		`- taskStats: expirable.NewLRU[UniqueID, Task](64, nil, time.Minute*5),`
	`107`	`+ taskStats: expirable.NewLRU[UniqueID, Task](512, nil, time.Minute*5),`
`108`	`108`	`},`
`109`	`109`	`allocator: alloc,`
`110`	`110`	`}`
Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ func newTaskScheduler(`
`93`	`93`	`handler: handler,`
`94`	`94`	`indexEngineVersionManager: indexEngineVersionManager,`
`95`	`95`	`allocator: allocator,`
`96`		`- taskStats: expirable.NewLRU[UniqueID, Task](64, nil, time.Minute*15),`
	`96`	`+ taskStats: expirable.NewLRU[UniqueID, Task](512, nil, time.Minute*15),`
`97`	`97`	`compactionHandler: compactionHandler,`
`98`	`98`	`}`
`99`	`99`	`ts.reloadFromMeta()`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ func NewSyncManager(chunkManager storage.ChunkManager) SyncManager {`
`72`	`72`	`keyLockDispatcher: dispatcher,`
`73`	`73`	`chunkManager: chunkManager,`
`74`	`74`	`tasks: typeutil.NewConcurrentMap[string, Task](),`
`75`		`- taskStats: expirable.NewLRU[string, Task](16, nil, time.Minute*15),`
	`75`	`+ taskStats: expirable.NewLRU[string, Task](64, nil, time.Minute*15),`
`76`	`76`	`}`
`77`	`77`	`// setup config update watcher`
`78`	`78`	`params.Watch(params.DataNodeCfg.MaxParallelSyncMgrTasks.Key, config.NewHandler("datanode.syncmgr.poolsize", syncMgr.resizeHandler))`
Original file line number	Diff line number	Diff line change
`@@ -284,7 +284,7 @@ func NewScheduler(ctx context.Context,`
`284`	`284`	`channelTasks: NewConcurrentMap[replicaChannelIndex, Task](),`
`285`	`285`	`processQueue: newTaskQueue(),`
`286`	`286`	`waitQueue: newTaskQueue(),`
`287`		`- taskStats: expirable.NewLRU[UniqueID, Task](64, nil, time.Minute*15),`
	`287`	`+ taskStats: expirable.NewLRU[UniqueID, Task](256, nil, time.Minute*15),`
`288`	`288`	`segmentTaskDelta: NewExecutingTaskDelta(),`
`289`	`289`	`channelTaskDelta: NewExecutingTaskDelta(),`
`290`	`290`	`}`