diff --git a/client/pkg/circuitbreaker/circuit_breaker.go b/client/pkg/circuitbreaker/circuit_breaker.go index 9b97baafe6..f49dc8f4a1 100644 --- a/client/pkg/circuitbreaker/circuit_breaker.go +++ b/client/pkg/circuitbreaker/circuit_breaker.go @@ -19,6 +19,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "time" "github.com/prometheus/client_golang/prometheus" @@ -70,7 +71,12 @@ type CircuitBreaker struct { sync.RWMutex state *State + // Metrics are rebound by a registration callback that can run while the + // breaker is serving requests, so publish the related handles together. + metrics atomic.Pointer[circuitBreakerMetrics] +} +type circuitBreakerMetrics struct { successCounter prometheus.Counter errorCounter prometheus.Counter overloadCounter prometheus.Counter @@ -110,18 +116,18 @@ func NewCircuitBreaker(name string, st Settings) *CircuitBreaker { cb.config = &st cb.state = cb.newState(time.Now(), StateClosed) - m.RegisterConsumer(func() { - registerMetrics(cb) - }) + m.RegisterConsumer(cb.initMetrics) return cb } -func registerMetrics(cb *CircuitBreaker) { +func (cb *CircuitBreaker) initMetrics() { metricName := replacer.Replace(cb.name) - cb.successCounter = m.CircuitBreakerCounters.WithLabelValues(metricName, "success") - cb.errorCounter = m.CircuitBreakerCounters.WithLabelValues(metricName, "error") - cb.overloadCounter = m.CircuitBreakerCounters.WithLabelValues(metricName, "overload") - cb.fastFailCounter = m.CircuitBreakerCounters.WithLabelValues(metricName, "fast_fail") + cb.metrics.Store(&circuitBreakerMetrics{ + successCounter: m.CircuitBreakerCounters.WithLabelValues(metricName, "success"), + errorCounter: m.CircuitBreakerCounters.WithLabelValues(metricName, "error"), + overloadCounter: m.CircuitBreakerCounters.WithLabelValues(metricName, "overload"), + fastFailCounter: m.CircuitBreakerCounters.WithLabelValues(metricName, "fast_fail"), + }) } // IsEnabled returns true if the circuit breaker is enabled. @@ -147,7 +153,7 @@ func (cb *CircuitBreaker) ChangeSettings(apply func(config *Settings)) { func (cb *CircuitBreaker) Execute(call func() (Overloading, error)) error { state, err := cb.onRequest() if err != nil { - cb.fastFailCounter.Inc() + cb.metrics.Load().fastFailCounter.Inc() return err } @@ -185,16 +191,17 @@ func (cb *CircuitBreaker) onResult(state *State, overloaded Overloading) { } func (cb *CircuitBreaker) emitMetric(overloaded Overloading, err error) { + metrics := cb.metrics.Load() switch overloaded { case No: - cb.successCounter.Inc() + metrics.successCounter.Inc() case Yes: - cb.overloadCounter.Inc() + metrics.overloadCounter.Inc() default: panic("unknown state") } if err != nil { - cb.errorCounter.Inc() + metrics.errorCounter.Inc() } } diff --git a/client/pkg/circuitbreaker/circuit_breaker_test.go b/client/pkg/circuitbreaker/circuit_breaker_test.go index 63178020ca..ab6fcdad86 100644 --- a/client/pkg/circuitbreaker/circuit_breaker_test.go +++ b/client/pkg/circuitbreaker/circuit_breaker_test.go @@ -16,6 +16,8 @@ package circuitbreaker import ( "errors" + "sync" + "sync/atomic" "testing" "time" @@ -246,6 +248,38 @@ func TestCircuitBreakerEnabled(t *testing.T) { re.True(cb.IsEnabled()) } +func TestCircuitBreakerMetricsInitializationIsConcurrentSafe(t *testing.T) { + breaker := NewCircuitBreaker("test_cb_concurrent_metrics", AlwaysClosedSettings) + before := breaker.metrics.Load() + const producerCount = 32 + var ( + ready sync.WaitGroup + wg sync.WaitGroup + stop atomic.Bool + ) + ready.Add(producerCount) + wg.Add(producerCount) + for range producerCount { + go func() { + defer wg.Done() + _ = breaker.Execute(func() (Overloading, error) { + return No, nil + }) + ready.Done() + for !stop.Load() { + _ = breaker.Execute(func() (Overloading, error) { + return No, nil + }) + } + }() + } + ready.Wait() + breaker.initMetrics() + stop.Store(true) + wg.Wait() + require.NotSame(t, before, breaker.metrics.Load()) +} + func newCircuitBreakerMovedToHalfOpenState(re *require.Assertions) *CircuitBreaker { cb := NewCircuitBreaker("test_cb", settings) re.Equal(StateClosed, cb.state.stateType) diff --git a/client/servicediscovery/metrics.go b/client/servicediscovery/metrics.go new file mode 100644 index 0000000000..969b2f1bf9 --- /dev/null +++ b/client/servicediscovery/metrics.go @@ -0,0 +1,50 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package servicediscovery + +import ( + "sync/atomic" + + "github.com/prometheus/client_golang/prometheus" + + clientmetrics "github.com/tikv/pd/client/metrics" +) + +type serviceDiscoveryMetrics struct { + getClusterInfo prometheus.Observer + getClusterInfoFailed prometheus.Observer + getMembers prometheus.Observer + getMembersFailed prometheus.Observer +} + +var currentServiceDiscoveryMetrics atomic.Pointer[serviceDiscoveryMetrics] + +func init() { + // An HTTP client can start service discovery before the first RPC client + // rebuilds and registers client metrics. Publish the four handles together + // so an active discovery loop never reads globals while they are replaced. + clientmetrics.RegisterConsumer(func() { + currentServiceDiscoveryMetrics.Store(&serviceDiscoveryMetrics{ + getClusterInfo: clientmetrics.InternalCmdDurationGetClusterInfo, + getClusterInfoFailed: clientmetrics.InternalCmdFailedDurationGetClusterInfo, + getMembers: clientmetrics.InternalCmdDurationGetMembers, + getMembersFailed: clientmetrics.InternalCmdFailedDurationGetMembers, + }) + }) +} + +func loadServiceDiscoveryMetrics() *serviceDiscoveryMetrics { + return currentServiceDiscoveryMetrics.Load() +} diff --git a/client/servicediscovery/metrics_test.go b/client/servicediscovery/metrics_test.go new file mode 100644 index 0000000000..59e359d5fb --- /dev/null +++ b/client/servicediscovery/metrics_test.go @@ -0,0 +1,58 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package servicediscovery + +import ( + "sync" + "sync/atomic" + "testing" + + "github.com/prometheus/client_golang/prometheus" + + clientmetrics "github.com/tikv/pd/client/metrics" +) + +func TestServiceDiscoveryMetricsInitializationIsConcurrentSafe(t *testing.T) { + const producerCount = 32 + var ( + ready sync.WaitGroup + wg sync.WaitGroup + stop atomic.Bool + ) + t.Cleanup(func() { + stop.Store(true) + wg.Wait() + }) + ready.Add(producerCount) + wg.Add(producerCount) + for range producerCount { + go func() { + defer wg.Done() + loadServiceDiscoveryMetrics().getMembers.Observe(0) + ready.Done() + for !stop.Load() { + loadServiceDiscoveryMetrics().getClusterInfo.Observe(0) + loadServiceDiscoveryMetrics().getMembers.Observe(0) + } + }() + } + + ready.Wait() + clientmetrics.InitAndRegisterMetrics(prometheus.Labels{"instance": "test"}) + stop.Store(true) + wg.Wait() + + loadServiceDiscoveryMetrics().getMembers.Observe(0) +} diff --git a/client/servicediscovery/service_discovery.go b/client/servicediscovery/service_discovery.go index 5716c24125..c57cec462a 100644 --- a/client/servicediscovery/service_discovery.go +++ b/client/servicediscovery/service_discovery.go @@ -40,7 +40,6 @@ import ( "github.com/tikv/pd/client/constants" "github.com/tikv/pd/client/errs" - "github.com/tikv/pd/client/metrics" "github.com/tikv/pd/client/opt" "github.com/tikv/pd/client/pkg/retry" "github.com/tikv/pd/client/pkg/utils/grpcutil" @@ -929,7 +928,7 @@ func (c *serviceDiscovery) getClusterInfo(ctx context.Context, url string, timeo return nil, err } start := time.Now() - defer func() { metrics.InternalCmdDurationGetClusterInfo.Observe(time.Since(start).Seconds()) }() + defer func() { loadServiceDiscoveryMetrics().getClusterInfo.Observe(time.Since(start).Seconds()) }() key := "GetClusterInfo-" + url r := c.flight.DoChan(key, func() (any, error) { return pdpb.NewPDClient(cc).GetClusterInfo(ctx, &pdpb.GetClusterInfoRequest{}) @@ -938,21 +937,21 @@ func (c *serviceDiscovery) getClusterInfo(ctx context.Context, url string, timeo case res := <-r: err = res.Err if err != nil { - metrics.InternalCmdFailedDurationGetClusterInfo.Observe(time.Since(start).Seconds()) + loadServiceDiscoveryMetrics().getClusterInfoFailed.Observe(time.Since(start).Seconds()) attachErr := errors.Errorf("error:%s target:%s status:%s", err, cc.Target(), cc.GetState().String()) return nil, errs.ErrClientGetClusterInfo.Wrap(attachErr).GenWithStackByCause() } val := res.Val clusterInfo := val.(*pdpb.GetClusterInfoResponse) if clusterInfo.GetHeader().GetError() != nil { - metrics.InternalCmdFailedDurationGetClusterInfo.Observe(time.Since(start).Seconds()) + loadServiceDiscoveryMetrics().getClusterInfoFailed.Observe(time.Since(start).Seconds()) attachErr := errors.Errorf("error:%s target:%s status:%s", clusterInfo.GetHeader().GetError().String(), cc.Target(), cc.GetState().String()) return nil, errs.ErrClientGetClusterInfo.Wrap(attachErr).GenWithStackByCause() } return clusterInfo, nil case <-ctx.Done(): attachErr := errors.Errorf("error:%s target:%s status:%s", ctx.Err(), cc.Target(), cc.GetState().String()) - metrics.InternalCmdFailedDurationGetClusterInfo.Observe(time.Since(start).Seconds()) + loadServiceDiscoveryMetrics().getClusterInfoFailed.Observe(time.Since(start).Seconds()) return nil, errs.ErrClientGetClusterInfo.Wrap(attachErr).GenWithStackByCause() } } @@ -965,7 +964,7 @@ func (c *serviceDiscovery) getMembers(ctx context.Context, url string, timeout t return nil, err } start := time.Now() - defer func() { metrics.InternalCmdDurationGetMembers.Observe(time.Since(start).Seconds()) }() + defer func() { loadServiceDiscoveryMetrics().getMembers.Observe(time.Since(start).Seconds()) }() key := "GetMembers-" + url r := c.flight.DoChan(key, func() (any, error) { return pdpb.NewPDClient(cc).GetMembers(ctx, &pdpb.GetMembersRequest{}) @@ -974,21 +973,21 @@ func (c *serviceDiscovery) getMembers(ctx context.Context, url string, timeout t case res := <-r: err = res.Err if err != nil { - metrics.InternalCmdFailedDurationGetMembers.Observe(time.Since(start).Seconds()) + loadServiceDiscoveryMetrics().getMembersFailed.Observe(time.Since(start).Seconds()) attachErr := errors.Errorf("error:%s target:%s status:%s", err, cc.Target(), cc.GetState().String()) return nil, errs.ErrClientGetMember.Wrap(attachErr).GenWithStackByCause() } val := res.Val members := val.(*pdpb.GetMembersResponse) if members.GetHeader().GetError() != nil { - metrics.InternalCmdFailedDurationGetMembers.Observe(time.Since(start).Seconds()) + loadServiceDiscoveryMetrics().getMembersFailed.Observe(time.Since(start).Seconds()) attachErr := errors.Errorf("error:%s target:%s status:%s", members.GetHeader().GetError().String(), cc.Target(), cc.GetState().String()) return nil, errs.ErrClientGetMember.Wrap(attachErr).GenWithStackByCause() } return members, nil case <-ctx.Done(): attachErr := errors.Errorf("error:%s target:%s status:%s", ctx.Err(), cc.Target(), cc.GetState().String()) - metrics.InternalCmdFailedDurationGetMembers.Observe(time.Since(start).Seconds()) + loadServiceDiscoveryMetrics().getMembersFailed.Observe(time.Since(start).Seconds()) return nil, errs.ErrClientGetMember.Wrap(attachErr).GenWithStackByCause() } }