Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: [2.5] Remove frequently updating metric to avoid mutex contention #38778

Merged
merged 7 commits into from
Jan 16, 2025
26 changes: 9 additions & 17 deletions internal/datacoord/index_meta.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
"github.com/hashicorp/golang-lru/v2/expirable"
"github.com/prometheus/client_golang/prometheus"
"github.com/samber/lo"
"go.uber.org/atomic"
"go.uber.org/zap"
"google.golang.org/protobuf/proto"

Expand Down Expand Up @@ -65,6 +66,8 @@ type indexMeta struct {

// segmentID -> indexID -> segmentIndex
segmentIndexes map[UniqueID]map[UniqueID]*model.SegmentIndex

lastUpdateMetricTime atomic.Time
}

func newIndexTaskStats(s *model.SegmentIndex) *metricsinfo.IndexTaskStats {
Expand Down Expand Up @@ -205,6 +208,10 @@ func (m *indexMeta) updateSegIndexMeta(segIdx *model.SegmentIndex, updateFunc fu
}

func (m *indexMeta) updateIndexTasksMetrics() {
if time.Since(m.lastUpdateMetricTime.Load()) < 120*time.Second {
return
}
defer m.lastUpdateMetricTime.Store(time.Now())
taskMetrics := make(map[UniqueID]map[commonpb.IndexState]int)
for _, segIdx := range m.segmentBuildInfo.List() {
if segIdx.IsDeleted {
Expand Down Expand Up @@ -233,6 +240,7 @@ func (m *indexMeta) updateIndexTasksMetrics() {
}
}
}
log.Ctx(m.ctx).Info("update index metric", zap.Int("collectionNum", len(taskMetrics)))
}

func checkParams(fieldIndex *model.Index, req *indexpb.CreateIndexRequest) bool {
Expand Down Expand Up @@ -874,7 +882,7 @@ func (m *indexMeta) GetAllSegIndexes() map[int64]*model.SegmentIndex {
tasks := m.segmentBuildInfo.List()
segIndexes := make(map[int64]*model.SegmentIndex, len(tasks))
for buildID, segIndex := range tasks {
segIndexes[buildID] = model.CloneSegmentIndex(segIndex)
segIndexes[buildID] = segIndex
}
return segIndexes
}
Expand Down Expand Up @@ -971,22 +979,6 @@ func (m *indexMeta) CheckCleanSegmentIndex(buildID UniqueID) (bool, *model.Segme
return true, nil
}

func (m *indexMeta) GetMetasByNodeID(nodeID UniqueID) []*model.SegmentIndex {
m.RLock()
defer m.RUnlock()

metas := make([]*model.SegmentIndex, 0)
for _, segIndex := range m.segmentBuildInfo.List() {
if segIndex.IsDeleted {
continue
}
if nodeID == segIndex.NodeID {
metas = append(metas, model.CloneSegmentIndex(segIndex))
}
}
return metas
}

func (m *indexMeta) getSegmentsIndexStates(collectionID UniqueID, segmentIDs []UniqueID) map[int64]map[int64]*indexpb.SegmentIndexState {
m.RLock()
defer m.RUnlock()
Expand Down
8 changes: 7 additions & 1 deletion internal/querycoordv2/task/scheduler.go
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,9 @@ type taskScheduler struct {
channelTasks map[replicaChannelIndex]Task
processQueue *taskQueue
waitQueue *taskQueue
taskStats *expirable.LRU[UniqueID, Task]

taskStats *expirable.LRU[UniqueID, Task]
lastUpdateMetricTime atomic.Time
}

func NewScheduler(ctx context.Context,
Expand Down Expand Up @@ -292,6 +294,9 @@ func (scheduler *taskScheduler) Add(task Task) error {
}

func (scheduler *taskScheduler) updateTaskMetrics() {
if time.Since(scheduler.lastUpdateMetricTime.Load()) < 30*time.Second {
return
}
segmentGrowNum, segmentReduceNum, segmentMoveNum := 0, 0, 0
channelGrowNum, channelReduceNum, channelMoveNum := 0, 0, 0
for _, task := range scheduler.segmentTasks {
Expand Down Expand Up @@ -324,6 +329,7 @@ func (scheduler *taskScheduler) updateTaskMetrics() {
metrics.QueryCoordTaskNum.WithLabelValues(metrics.ChannelGrowTaskLabel).Set(float64(channelGrowNum))
metrics.QueryCoordTaskNum.WithLabelValues(metrics.ChannelReduceTaskLabel).Set(float64(channelReduceNum))
metrics.QueryCoordTaskNum.WithLabelValues(metrics.ChannelMoveTaskLabel).Set(float64(channelMoveNum))
scheduler.lastUpdateMetricTime.Store(time.Now())
}

// check whether the task is valid to add,
Expand Down
Loading