add locks for registerred metric fetcher

kubewharf · Jul 30, 2023 · 7fc1ddb · 7fc1ddb
1 parent 9d3580c
commit 7fc1ddb
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 4 deletions.
diff --git a/pkg/metaserver/agent/metric/fake_metric.go b/pkg/metaserver/agent/metric/fake_metric.go
@@ -32,6 +32,7 @@ package metric
 
 import (
 	"context"
+	"sync"
 
 	v1 "k8s.io/api/core/v1"
 
@@ -49,19 +50,31 @@ func NewFakeMetricsFetcher(emitter metrics.MetricEmitter) MetricsFetcher {
 }
 
 type FakeMetricsFetcher struct {
-	metricStore *metric.MetricStore
-	emitter     metrics.MetricEmitter
+	sync.RWMutex
+	metricStore      *metric.MetricStore
+	emitter          metrics.MetricEmitter
+	registeredMetric []func(store *metric.MetricStore)
 }
 
-func (f *FakeMetricsFetcher) Run(ctx context.Context) {}
+func (f *FakeMetricsFetcher) Run(ctx context.Context) {
+	f.RLock()
+	defer f.RUnlock()
+	for _, fu := range f.registeredMetric {
+		fu(f.metricStore)
+	}
+}
 
 func (f *FakeMetricsFetcher) RegisterNotifier(scope MetricsScope, req NotifiedRequest, response chan NotifiedResponse) string {
 	return ""
 }
 
 func (f *FakeMetricsFetcher) DeRegisterNotifier(scope MetricsScope, key string) {}
 
-func (f *FakeMetricsFetcher) RegisterExternalMetric(_ func(store *metric.MetricStore)) {}
+func (f *FakeMetricsFetcher) RegisterExternalMetric(fu func(store *metric.MetricStore)) {
+	f.Lock()
+	defer f.Unlock()
+	f.registeredMetric = append(f.registeredMetric, fu)
+}
 
 func (f *FakeMetricsFetcher) GetNodeMetric(metricName string) (metric.MetricData, error) {
 	return f.metricStore.GetNodeMetric(metricName)

diff --git a/pkg/metaserver/agent/metric/malachite/fetcher.go b/pkg/metaserver/agent/metric/malachite/fetcher.go
@@ -113,6 +113,8 @@ func (m *MalachiteMetricsFetcher) DeRegisterNotifier(scope metric.MetricsScope,
 }
 
 func (m *MalachiteMetricsFetcher) RegisterExternalMetric(f func(store *utilmetric.MetricStore)) {
+	m.Lock()
+	defer m.Unlock()
 	m.registeredMetric = append(m.registeredMetric, f)
 }
 
@@ -176,9 +178,12 @@ func (m *MalachiteMetricsFetcher) sample(ctx context.Context) {
 	m.updateCgroupData()
 
 	// after sampling, we should call the registered function to get external metric
+	m.RLock()
 	for _, f := range m.registeredMetric {
 		f(m.metricStore)
 	}
+	m.RUnlock()
+
 	m.notifySystem()
 	m.notifyPods()
 }