furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 10:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

enhance: [2.6] add named C++ thread active metrics (#50064, #50216) (#50299)

Browse Source 
Cherry-pick from master
pr: https://github.com/milvus-io/milvus/pull/50064
pr: https://github.com/milvus-io/milvus/pull/50216
Related to https://github.com/milvus-io/milvus/issues/50066

## Summary
Add `milvus_thread_cpu_active_num_by_pool` to monitor active C++ threads
by named thread pool.

This backport includes classification for Knowhere, file write, Milvus
search/load, segcore, CGO, RocksDB high/low/bottom, and unclassified
threads. It also names file write worker threads and adds the Grafana
dashboard panel for the metric.

---------

Signed-off-by: CLiQing <2208529306@qq.com>
This commit is contained in:
ChenLiqing
2026-06-30 17:20:38 +08:00
committed by GitHub
parent da6965d7f8
commit 804fd1e12a
6 changed files with 256 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deployments/monitor/grafana/milvus-dashboard.json
+12 -3
View File
@@ -14362,7 +14362,7 @@
"datasource": {
"uid": "${datasource}"
},
"description": "Number of OS threads created.",
"description": "Process OS thread count and approximate active named C++ thread groups.",
"fill": 1,
"fillGradient": 0,
"gridPos": {
@@ -14400,12 +14400,21 @@
"targets": [
{
"exemplar": true,
"expr": "go_threads{app_kubernetes_io_name=\"$app_name\", app_kubernetes_io_instance=~\"$instance\"}",
"expr": "milvus_thread_num{app_kubernetes_io_name=\"$app_name\", app_kubernetes_io_instance=~\"$instance\", namespace=\"$namespace\"}",
"interval": "",
"intervalFactor": 2,
"legendFormat": "{{pod}}",
"legendFormat": "os_threads {{pod}}",
"queryType": "randomWalk",
"refId": "A"
},
{
"exemplar": true,
"expr": "sum(milvus_thread_cpu_active_num_by_pool{app_kubernetes_io_name=\"$app_name\", app_kubernetes_io_instance=~\"$instance\", namespace=\"$namespace\"}) by (pod, thread_pool)",
"interval": "",
"intervalFactor": 2,
"legendFormat": "active {{thread_pool}} {{pod}}",
"queryType": "randomWalk",
"refId": "B"
}
],
"thresholds": [],
internal/core/src/storage/FileWriter.h
+5 -2
View File
@@ -23,6 +23,7 @@
#include <cstring>
#include <fcntl.h>
#include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/executors/thread_factory/NamedThreadFactory.h>
#include <mutex>
#include <string>
#include <sys/mman.h>
@@ -330,8 +331,10 @@ class FileWriteWorkerPool {
std::lock_guard<std::mutex> lock(executor_mutex_);
old_executor = executor_;
if (nr_worker > 0) {
executor_ =
std::make_shared<folly::CPUThreadPoolExecutor>(nr_worker);
executor_ = std::make_shared<folly::CPUThreadPoolExecutor>(
nr_worker,
std::make_shared<folly::NamedThreadFactory>(
"MILVUS_FL_WR_"));
} else {
executor_ = nullptr;
}
internal/util/metrics/thread.go
+161 -2
View File
@@ -18,6 +18,10 @@ package metrics
import (
"os"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"time"
@@ -36,11 +40,52 @@ type threadWatcher struct {
stopOnce sync.Once
wg sync.WaitGroup
ch chan struct{}
samples map[int32]threadSample
}
type threadSample struct {
group string
cpu uint64
}
type threadStat struct {
tid int32
name string
state byte
cpu uint64
}
var threadNameGroupRules = []struct {
prefix string
group string
}{
{prefix: "knowhere_build", group: "knowhere_build"},
{prefix: "knowhere_search", group: "knowhere_search"},
{prefix: "knowhere_fetch", group: "knowhere_fetch"},
{prefix: "rocksdb:high", group: "rocksdb_high"},
{prefix: "rocksdb:low", group: "rocksdb_low"},
{prefix: "rocksdb:bottom", group: "rocksdb_bottom"},
{prefix: "MILVUS_FL_WR", group: "file_write"},
{prefix: "MILVUS_SEARCH", group: "milvus_search"},
{prefix: "MILVUS_LOAD", group: "milvus_load"},
{prefix: "HIGH_SEGC_POOL", group: "segcore_high"},
{prefix: "MIDD_SEGC_POOL", group: "segcore_middle"},
{prefix: "LOW_SEGC_POOL", group: "segcore_low"},
{prefix: "CGO_SQ", group: "cgo_sq"},
{prefix: "CGO_LOAD", group: "cgo_load"},
{prefix: "CGO_DYN", group: "cgo_dynamic"},
{prefix: "CGO_WARMUP", group: "cgo_warmup"},
}
const (
threadWatcherInterval = 5 * time.Second
unclassifiedThreadPool = "unclassified"
)
func NewThreadWatcher() *threadWatcher {
return &threadWatcher{
ch: make(chan struct{}),
ch: make(chan struct{}),
samples: make(map[int32]threadSample),
}
}
@@ -55,7 +100,7 @@ func (thw *threadWatcher) Start() {
}
func (thw *threadWatcher) watchThreadNum() {
ticker := time.NewTicker(time.Second * 30)
ticker := time.NewTicker(threadWatcherInterval)
defer ticker.Stop()
pid := os.Getpid()
p, err := process.NewProcess(int32(pid))
@@ -73,6 +118,7 @@ func (thw *threadWatcher) watchThreadNum() {
}
log.Debug("thread watcher observe thread num", zap.Int32("threadNum", threadNum))
metrics.ThreadNum.Set(float64(threadNum))
thw.updateNamedThreadCPUActiveNum()
case <-thw.ch:
log.Info("thread watcher exit")
return
@@ -80,6 +126,119 @@ func (thw *threadWatcher) watchThreadNum() {
}
}
func (thw *threadWatcher) updateNamedThreadCPUActiveNum() {
if runtime.GOOS != "linux" {
return
}
current, err := collectNamedThreadStats()
if err != nil {
log.Warn("thread watcher failed to collect named thread stats", zap.Error(err))
return
}
activeByGroup, nextSamples := collectActiveThreadGroups(current, thw.samples)
for _, rule := range threadNameGroupRules {
metrics.ThreadCPUActiveNumByPool.WithLabelValues(rule.group).Set(float64(activeByGroup[rule.group]))
}
metrics.ThreadCPUActiveNumByPool.WithLabelValues(unclassifiedThreadPool).Set(float64(activeByGroup[unclassifiedThreadPool]))
thw.samples = nextSamples
}
func collectActiveThreadGroups(current []threadStat, previous map[int32]threadSample) (map[string]int, map[int32]threadSample) {
activeByGroup := make(map[string]int)
nextSamples := make(map[int32]threadSample, len(current))
for _, stat := range current {
group, classified := classifyThreadName(stat.name)
if !classified {
group = unclassifiedThreadPool
}
nextSamples[stat.tid] = threadSample{
group: group,
cpu: stat.cpu,
}
previousSample, existed := previous[stat.tid]
if !existed {
continue
}
// Treat runnable and uninterruptible-sleep threads as active even
// without a CPU-time delta, so blocked I/O work is still visible.
if previousSample.group == group && (stat.cpu > previousSample.cpu || stat.state == 'R' || stat.state == 'D') {
activeByGroup[group]++
}
}
return activeByGroup, nextSamples
}
func collectNamedThreadStats() ([]threadStat, error) {
taskDir := "/proc/self/task"
entries, err := os.ReadDir(taskDir)
if err != nil {
return nil, err
}
stats := make([]threadStat, 0, len(entries))
for _, entry := range entries {
if !entry.IsDir() {
continue
}
tid64, err := strconv.ParseInt(entry.Name(), 10, 32)
if err != nil {
continue
}
tid := int32(tid64)
statBytes, err := os.ReadFile(filepath.Join(taskDir, entry.Name(), "stat"))
if err != nil {
continue
}
name, state, cpu, err := parseThreadStat(string(statBytes))
if err != nil {
continue
}
stats = append(stats, threadStat{
tid: tid,
name: name,
state: state,
cpu: cpu,
})
}
return stats, nil
}
func parseThreadStat(stat string) (string, byte, uint64, error) {
startComm := strings.IndexByte(stat, '(')
endComm := strings.LastIndexByte(stat, ')')
if startComm < 0 || startComm >= endComm || endComm+2 >= len(stat) {
return "", 0, 0, strconv.ErrSyntax
}
name := stat[startComm+1 : endComm]
fields := strings.Fields(stat[endComm+2:])
if len(fields) <= 12 || len(fields[0]) == 0 {
return "", 0, 0, strconv.ErrSyntax
}
utime, err := strconv.ParseUint(fields[11], 10, 64)
if err != nil {
return "", 0, 0, err
}
stime, err := strconv.ParseUint(fields[12], 10, 64)
if err != nil {
return "", 0, 0, err
}
return name, fields[0][0], utime + stime, nil
}
func classifyThreadName(name string) (string, bool) {
for _, rule := range threadNameGroupRules {
if strings.HasPrefix(name, rule.prefix) {
return rule.group, true
}
}
return unclassifiedThreadPool, false
}
func (thw *threadWatcher) Stop() {
thw.stopOnce.Do(func() {
close(thw.ch)
internal/util/metrics/thread_test.go
+66 -2
View File
@@ -20,7 +20,7 @@ import (
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestThreadWatcher(t *testing.T) {
@@ -36,6 +36,70 @@ func TestThreadWatcher(t *testing.T) {
select {
case <-ch:
case <-time.After(time.Second * 5):
assert.FailNow(t, "watcher failed to close after 5 seconds")
require.FailNow(t, "watcher failed to close after 5 seconds")
}
}
func TestClassifyThreadName(t *testing.T) {
group, ok := classifyThreadName("knowhere_search")
require.True(t, ok)
require.Equal(t, "knowhere_search", group)
group, ok = classifyThreadName("MILVUS_FL_WR_0")
require.True(t, ok)
require.Equal(t, "file_write", group)
group, ok = classifyThreadName("knowhere_fetch_")
require.True(t, ok)
require.Equal(t, "knowhere_fetch", group)
group, ok = classifyThreadName("rocksdb:high")
require.True(t, ok)
require.Equal(t, "rocksdb_high", group)
group, ok = classifyThreadName("rocksdb:low")
require.True(t, ok)
require.Equal(t, "rocksdb_low", group)
group, ok = classifyThreadName("rocksdb:bottom")
require.True(t, ok)
require.Equal(t, "rocksdb_bottom", group)
group, ok = classifyThreadName("grpc_global_tim")
require.False(t, ok)
require.Equal(t, unclassifiedThreadPool, group)
}
func TestParseThreadStat(t *testing.T) {
name, state, cpu, err := parseThreadStat("123 (knowhere search) R 1 2 3 4 5 6 7 8 9 10 34 56 0 0 0")
require.NoError(t, err)
require.Equal(t, "knowhere search", name)
require.Equal(t, byte('R'), state)
require.Equal(t, uint64(90), cpu)
}
func TestCollectActiveThreadGroups(t *testing.T) {
current := []threadStat{
{tid: 1, name: "rocksdb:high", state: 'S', cpu: 11},
{tid: 2, name: "arrow-worker", state: 'S', cpu: 6},
{tid: 3, name: "rocksdb:low", state: 'R', cpu: 1},
{tid: 4, name: "knowhere_search", state: 'S', cpu: 20},
{tid: 5, name: "rocksdb:bottom", state: 'D', cpu: 30},
}
previous := map[int32]threadSample{
1: {group: "rocksdb_high", cpu: 10},
2: {group: unclassifiedThreadPool, cpu: 5},
4: {group: "knowhere_fetch", cpu: 19},
5: {group: "rocksdb_bottom", cpu: 30},
}
activeByGroup, nextSamples := collectActiveThreadGroups(current, previous)
require.Equal(t, 1, activeByGroup["rocksdb_high"])
require.Equal(t, 1, activeByGroup[unclassifiedThreadPool])
require.Equal(t, 0, activeByGroup["rocksdb_low"])
require.Equal(t, 0, activeByGroup["knowhere_search"])
require.Equal(t, 1, activeByGroup["rocksdb_bottom"])
require.Equal(t, threadSample{group: unclassifiedThreadPool, cpu: 6}, nextSamples[2])
require.Len(t, nextSamples, len(current))
}
pkg/metrics/info_metrics.go
+11
View File
@@ -43,6 +43,17 @@ var (
Help: "the actual thread number of milvus process",
},
)
ThreadCPUActiveNumByPool = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: milvusNamespace,
Name: "thread_cpu_active_num_by_pool",
Help: "the approximate active thread number of milvus process by named OS thread pool, including threads with CPU time delta, runnable state, or uninterruptible-sleep state",
},
[]string{
"thread_pool",
},
)
)
// RegisterMQType registers the type of mq
pkg/metrics/metrics.go
+1
View File
@@ -229,5 +229,6 @@ func Register(r prometheus.Registerer) {
r.MustRegister(BuildInfo)
r.MustRegister(RuntimeInfo)
r.MustRegister(ThreadNum)
r.MustRegister(ThreadCPUActiveNumByPool)
metricRegisterer = r
}