This is an automated email from the ASF dual-hosted git repository.
Similarityoung pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go-pixiu.git
The following commit(s) were added to refs/heads/develop by this push:
new e7691370f feat(cluster): add metrics for endpoint snapshot publication
(#962)
e7691370f is described below
commit e7691370f89fc2f394546dfdc9a79f1a53f573d0
Author: 承潜 <[email protected]>
AuthorDate: Thu Jul 23 21:02:08 2026 +0800
feat(cluster): add metrics for endpoint snapshot publication (#962)
* feat(cluster): add metrics for endpoint snapshot publication
Add observability for endpoint snapshot publication so operators can
diagnose registry churn, unexpected cluster size, or excessive health
update publication.
Emit three OpenTelemetry instruments, labeled only by cluster name, on
each successful snapshot CompareAndSwap:
- pixiu_cluster_snapshot_publish_total (counter)
- pixiu_cluster_snapshot_endpoint_count (gauge)
- pixiu_cluster_snapshot_healthy_endpoint_count (gauge)
Instruments bind lazily to the global MeterProvider, so they stay no-op
when metrics are disabled and do not alter snapshot publication behavior.
Closes #944
* fix(cluster): ensure snapshot metrics recorded at static cluster startup
Move registerOtelMetricMeter before cluster construction so static clusters'
initial snapshot publish lands on the real meter provider rather than the
no-op default. Without this fix, static clusters with no health transitions
show empty counter/gauges in steady state — defeating the PR's goal of
diagnosing unexpected cluster sizes.
Changes:
- Move registerOtelMetricMeter call from (*Server).Start() to Start(bs),
positioned before server.initialize(bs) which constructs clusters
- Add TestStaticClusterSnapshotMetricsRecordedAtStartup in pkg/server to
guard against ordering regression (models production: provider → clusters)
- Expand installSnapshotMetricsReader doc to note all cluster construction
triggers global instrument init (t.Parallel constraint)
Issue: #944
* docs(cluster): fix installClusterSnapshotMetricsReader comment accuracy
Update the helper's documentation to accurately reflect its behavior: it
installs a ManualReader provider and restores the previous provider on
cleanup, but does not reset pkg/cluster's instrument variables (which are
in a different package and not accessible from pkg/server tests).
The comment previously claimed it "resets the global instrument state" and
mentioned "pkg/cluster instrument variables," but the implementation only
swaps otel.MeterProvider without touching snapshotPublishTotal,
snapshotEndpointCount, or snapshotHealthyCount.
Addresses review feedback on PR #962.
---
pkg/cluster/cluster.go | 3 +
pkg/cluster/metrics.go | 87 +++++++++++++++++++
pkg/cluster/metrics_test.go | 174 +++++++++++++++++++++++++++++++++++++
pkg/server/cluster_manager_test.go | 122 ++++++++++++++++++++++++++
pkg/server/pixiu_start.go | 10 ++-
5 files changed, 395 insertions(+), 1 deletion(-)
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index c6c27a70f..4868107e6 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -101,6 +101,7 @@ func (c *Cluster) RefreshEndpointsFrom(previous
*EndpointSnapshot) {
}
next := newEndpointSnapshot(c.Config, source,
len(c.Config.HealthChecks) != 0)
if c.endpoints.CompareAndSwap(current, next) {
+ recordSnapshotPublish(c.clusterName(), next)
return
}
}
@@ -128,6 +129,7 @@ func (c *Cluster) UpdateEndpointHealth(endpointID,
endpointAddress string, healt
return true
}
if c.endpoints.CompareAndSwap(current, next) {
+ recordSnapshotPublish(c.clusterName(), next)
return true
}
}
@@ -158,6 +160,7 @@ func (c *Cluster)
UpdateEndpointAddressHealth(endpointAddress string, healthy bo
return true
}
if c.endpoints.CompareAndSwap(current, next) {
+ recordSnapshotPublish(c.clusterName(), next)
return true
}
}
diff --git a/pkg/cluster/metrics.go b/pkg/cluster/metrics.go
new file mode 100644
index 000000000..7701f5b92
--- /dev/null
+++ b/pkg/cluster/metrics.go
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 cluster
+
+import (
+ "context"
+ "sync"
+)
+
+import (
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ "go.opentelemetry.io/otel/metric"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+// Endpoint snapshot publication metrics. Instruments are bound lazily to the
+// global MeterProvider on first publish; when metrics are disabled the
provider
+// is a no-op, so recording stays a cheap no-op and never blocks publication.
+var (
+ snapshotMetricsOnce sync.Once
+
+ snapshotPublishTotal metric.Int64Counter
+ snapshotEndpointCount metric.Int64Gauge
+ snapshotHealthyCount metric.Int64Gauge
+)
+
+func initSnapshotMetrics() {
+ snapshotMetricsOnce.Do(func() {
+ meter := otel.GetMeterProvider().Meter("pixiu")
+
+ var err error
+ if snapshotPublishTotal, err =
meter.Int64Counter("pixiu_cluster_snapshot_publish_total",
+ metric.WithDescription("Total number of cluster
endpoint snapshots successfully published.")); err != nil {
+ logger.Errorf("[dubbo-go-pixiu] register
pixiu_cluster_snapshot_publish_total failed: %v", err)
+ }
+ if snapshotEndpointCount, err =
meter.Int64Gauge("pixiu_cluster_snapshot_endpoint_count",
+ metric.WithDescription("Total endpoints in the latest
published cluster snapshot.")); err != nil {
+ logger.Errorf("[dubbo-go-pixiu] register
pixiu_cluster_snapshot_endpoint_count failed: %v", err)
+ }
+ if snapshotHealthyCount, err =
meter.Int64Gauge("pixiu_cluster_snapshot_healthy_endpoint_count",
+ metric.WithDescription("Healthy endpoints in the latest
published cluster snapshot.")); err != nil {
+ logger.Errorf("[dubbo-go-pixiu] register
pixiu_cluster_snapshot_healthy_endpoint_count failed: %v", err)
+ }
+ })
+}
+
+// recordSnapshotPublish reports a successful snapshot publication for a
cluster.
+// Call exactly once per successful CompareAndSwap of the published snapshot.
The
+// only label is the cluster name, which is bounded by configuration rather
than
+// request traffic, so cardinality stays low. The health-update callers invoke
+// this while holding the cluster's healthMu, so the body must stay
+// allocation-light and must not block.
+func recordSnapshotPublish(clusterName string, snapshot *EndpointSnapshot) {
+ initSnapshotMetrics()
+
+ attrs := metric.WithAttributes(attribute.String("cluster", clusterName))
+ ctx := context.Background()
+
+ if snapshotPublishTotal != nil {
+ snapshotPublishTotal.Add(ctx, 1, attrs)
+ }
+ if snapshotEndpointCount != nil {
+ snapshotEndpointCount.Record(ctx,
int64(snapshot.EndpointCount()), attrs)
+ }
+ if snapshotHealthyCount != nil {
+ snapshotHealthyCount.Record(ctx,
int64(snapshot.HealthyEndpointCount()), attrs)
+ }
+}
diff --git a/pkg/cluster/metrics_test.go b/pkg/cluster/metrics_test.go
new file mode 100644
index 000000000..c03dd33f9
--- /dev/null
+++ b/pkg/cluster/metrics_test.go
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 cluster
+
+import (
+ "context"
+ "sync"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+ "go.opentelemetry.io/otel/sdk/metric/metricdata"
+)
+
+// installSnapshotMetricsReader binds the snapshot instruments to a fresh
+// ManualReader so a test can inspect the recorded measurements, and resets the
+// global provider and instrument state on cleanup so later callers rebind.
+//
+// It mutates package-global instruments and the process-global MeterProvider,
+// so tests that use it must not call t.Parallel(). Additionally, any test in
+// this package that constructs a cluster (even without calling this helper)
+// will trigger recordSnapshotPublish → initSnapshotMetrics, so tests that
+// construct clusters must also not call t.Parallel() to avoid racing on the
+// sync.Once and global instrument variables.
+func installSnapshotMetricsReader(t *testing.T) *sdkmetric.ManualReader {
+ t.Helper()
+
+ reader := sdkmetric.NewManualReader()
+ provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+
+ prevProvider := otel.GetMeterProvider()
+
+ otel.SetMeterProvider(provider)
+ snapshotMetricsOnce = sync.Once{}
+ snapshotPublishTotal, snapshotEndpointCount, snapshotHealthyCount =
nil, nil, nil
+
+ t.Cleanup(func() {
+ otel.SetMeterProvider(prevProvider)
+ snapshotMetricsOnce = sync.Once{}
+ snapshotPublishTotal, snapshotEndpointCount,
snapshotHealthyCount = nil, nil, nil
+ })
+
+ return reader
+}
+
+func collectSnapshotMetrics(t *testing.T, reader *sdkmetric.ManualReader)
map[string]metricdata.Metrics {
+ t.Helper()
+
+ var rm metricdata.ResourceMetrics
+ require.NoError(t, reader.Collect(context.Background(), &rm))
+
+ out := make(map[string]metricdata.Metrics)
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ out[m.Name] = m
+ }
+ }
+ return out
+}
+
+func sumForCluster(t *testing.T, m metricdata.Metrics, cluster string) int64 {
+ t.Helper()
+ data, ok := m.Data.(metricdata.Sum[int64])
+ require.True(t, ok, "metric %s is not an int64 Sum", m.Name)
+ for _, dp := range data.DataPoints {
+ if v, ok := dp.Attributes.Value(attribute.Key("cluster")); ok
&& v.AsString() == cluster {
+ return dp.Value
+ }
+ }
+ t.Fatalf("no data point for cluster %q in metric %s", cluster, m.Name)
+ return 0
+}
+
+func gaugeForCluster(t *testing.T, m metricdata.Metrics, cluster string) int64
{
+ t.Helper()
+ data, ok := m.Data.(metricdata.Gauge[int64])
+ require.True(t, ok, "metric %s is not an int64 Gauge", m.Name)
+ for _, dp := range data.DataPoints {
+ if v, ok := dp.Attributes.Value(attribute.Key("cluster")); ok
&& v.AsString() == cluster {
+ return dp.Value
+ }
+ }
+ t.Fatalf("no data point for cluster %q in metric %s", cluster, m.Name)
+ return 0
+}
+
+func TestSnapshotMetricsRecordPublishCountAndSizes(t *testing.T) {
+ reader := installSnapshotMetricsReader(t)
+
+ healthy := testEndpoint("ep-1", "127.0.0.1", 18080)
+ unhealthy := testEndpoint("ep-2", "127.0.0.2", 18081)
+ runtimeCluster := NewCluster(testCluster("snapshot-metrics", healthy,
unhealthy))
+
+ // One publish from NewCluster's initial RefreshEndpointsFrom.
+ require.True(t, runtimeCluster.UpdateEndpointHealth(unhealthy.ID,
unhealthy.Address.GetAddress(), false))
+
+ metrics := collectSnapshotMetrics(t, reader)
+
+ publish, ok := metrics["pixiu_cluster_snapshot_publish_total"]
+ require.True(t, ok, "publish total metric missing")
+ assert.Equal(t, int64(2), sumForCluster(t, publish, "snapshot-metrics"))
+
+ count, ok := metrics["pixiu_cluster_snapshot_endpoint_count"]
+ require.True(t, ok, "endpoint count metric missing")
+ assert.Equal(t, int64(2), gaugeForCluster(t, count, "snapshot-metrics"))
+
+ healthyCount, ok :=
metrics["pixiu_cluster_snapshot_healthy_endpoint_count"]
+ require.True(t, ok, "healthy endpoint count metric missing")
+ assert.Equal(t, int64(1), gaugeForCluster(t, healthyCount,
"snapshot-metrics"))
+}
+
+func TestSnapshotMetricsDoNotCountNoOpHealthUpdate(t *testing.T) {
+ reader := installSnapshotMetricsReader(t)
+
+ endpoint := testEndpoint("ep-1", "127.0.0.1", 18082)
+ runtimeCluster := NewCluster(testCluster("snapshot-metrics-noop",
endpoint))
+
+ // Endpoint already healthy; setting it healthy again is a no-op that
does
+ // not swap the snapshot and so must not increment the publish counter.
+ require.True(t, runtimeCluster.UpdateEndpointHealth(endpoint.ID,
endpoint.Address.GetAddress(), true))
+
+ metrics := collectSnapshotMetrics(t, reader)
+ publish, ok := metrics["pixiu_cluster_snapshot_publish_total"]
+ require.True(t, ok, "publish total metric missing")
+ assert.Equal(t, int64(1), sumForCluster(t, publish,
"snapshot-metrics-noop"))
+}
+
+func TestSnapshotMetricsRecordAddressHealthPublish(t *testing.T) {
+ reader := installSnapshotMetricsReader(t)
+
+ // Two endpoints share one address, so an address-keyed health flip
marks
+ // both unhealthy in a single publish.
+ first := testEndpoint("ep-1", "127.0.0.1", 18083)
+ second := testEndpoint("ep-2", "127.0.0.1", 18083)
+ runtimeCluster := NewCluster(testCluster("snapshot-metrics-address",
first, second))
+
+ // One publish from NewCluster's initial RefreshEndpointsFrom.
+ require.True(t,
runtimeCluster.UpdateEndpointAddressHealth(first.Address.GetAddress(), false))
+
+ metrics := collectSnapshotMetrics(t, reader)
+
+ publish, ok := metrics["pixiu_cluster_snapshot_publish_total"]
+ require.True(t, ok, "publish total metric missing")
+ assert.Equal(t, int64(2), sumForCluster(t, publish,
"snapshot-metrics-address"))
+
+ count, ok := metrics["pixiu_cluster_snapshot_endpoint_count"]
+ require.True(t, ok, "endpoint count metric missing")
+ assert.Equal(t, int64(2), gaugeForCluster(t, count,
"snapshot-metrics-address"))
+
+ healthyCount, ok :=
metrics["pixiu_cluster_snapshot_healthy_endpoint_count"]
+ require.True(t, ok, "healthy endpoint count metric missing")
+ assert.Equal(t, int64(0), gaugeForCluster(t, healthyCount,
"snapshot-metrics-address"))
+}
diff --git a/pkg/server/cluster_manager_test.go
b/pkg/server/cluster_manager_test.go
index b8cef223b..67f038c60 100644
--- a/pkg/server/cluster_manager_test.go
+++ b/pkg/server/cluster_manager_test.go
@@ -18,6 +18,7 @@
package server
import (
+ "context"
"fmt"
"reflect"
"sync"
@@ -27,6 +28,12 @@ import (
import (
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "go.opentelemetry.io/otel"
+ "go.opentelemetry.io/otel/attribute"
+ sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+ "go.opentelemetry.io/otel/sdk/metric/metricdata"
)
import (
@@ -1111,3 +1118,118 @@ func healthCheckerAddresses(runtime *cluster.Cluster)
[]string {
}
return addrs
}
+
+// TestStaticClusterSnapshotMetricsRecordedAtStartup verifies that snapshot
+// publication metrics emitted during static cluster initialization land on the
+// real meter provider. This test models production startup ordering: clusters
+// are created during CreateDefaultClusterManager (called from initialize), and
+// the OTel provider must be installed BEFORE that point so the initial
snapshot
+// publish (the only guaranteed emission for steady-state clusters) is
recorded.
+//
+// Regression guard for: if registerOtelMetricMeter is moved back after cluster
+// construction, static clusters' initial publish lands on the no-op delegating
+// provider, and the counter/gauges remain empty in steady state.
+func TestStaticClusterSnapshotMetricsRecordedAtStartup(t *testing.T) {
+ // Step 1: Install a ManualReader meter provider BEFORE cluster
construction.
+ // This models the corrected startup order:
registerOtelMetricMeter(bs.Metric)
+ // is called in Start(bs) before server.initialize(bs).
+ reader := installClusterSnapshotMetricsReader(t)
+
+ // Step 2: Construct a cluster manager with static clusters, simulating
what
+ // happens during initialize → CreateDefaultClusterManager.
+ staticCluster := testCluster("static-metrics-test",
model.LoadBalancerRoundRobin, []*model.Endpoint{
+ testEndpoint("ep-1", "127.0.0.1", 19001),
+ testEndpoint("ep-2", "127.0.0.1", 19002),
+ })
+ _ = CreateDefaultClusterManager(&model.Bootstrap{
+ StaticResources: model.StaticResources{
+ Clusters: []*model.ClusterConfig{staticCluster},
+ },
+ })
+
+ // Step 3: Verify the initial snapshot publish was recorded. NewCluster
calls
+ // RefreshEndpointsFrom, which publishes once. That single emission
must land
+ // on the real provider for steady-state clusters (those with no health
flips
+ // or registry churn) to have any recorded metrics at all.
+ metrics := collectClusterSnapshotMetrics(t, reader)
+
+ publishTotal, ok := metrics["pixiu_cluster_snapshot_publish_total"]
+ assert.True(t, ok, "publish total metric missing")
+ assert.Equal(t, int64(1), sumForClusterMetric(t, publishTotal,
"static-metrics-test"),
+ "static cluster initial publish must increment the counter")
+
+ endpointCount, ok := metrics["pixiu_cluster_snapshot_endpoint_count"]
+ assert.True(t, ok, "endpoint count gauge missing")
+ assert.Equal(t, int64(2), gaugeForClusterMetric(t, endpointCount,
"static-metrics-test"),
+ "endpoint count gauge must reflect the initial snapshot size")
+
+ healthyCount, ok :=
metrics["pixiu_cluster_snapshot_healthy_endpoint_count"]
+ assert.True(t, ok, "healthy endpoint count gauge missing")
+ assert.Equal(t, int64(2), gaugeForClusterMetric(t, healthyCount,
"static-metrics-test"),
+ "healthy endpoint count gauge must reflect the initial snapshot
size")
+}
+
+// installClusterSnapshotMetricsReader installs a ManualReader meter provider
+// for snapshot metrics testing and restores the previous provider on cleanup.
+// This helper mutates the process-global MeterProvider, so tests using it must
+// not call t.Parallel().
+func installClusterSnapshotMetricsReader(t *testing.T) *sdkmetric.ManualReader
{
+ t.Helper()
+
+ reader := sdkmetric.NewManualReader()
+ provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+ prevProvider := otel.GetMeterProvider()
+
+ otel.SetMeterProvider(provider)
+
+ t.Cleanup(func() {
+ otel.SetMeterProvider(prevProvider)
+ })
+
+ return reader
+}
+
+// collectClusterSnapshotMetrics collects metrics from the reader and returns
+// them as a map keyed by metric name.
+func collectClusterSnapshotMetrics(t *testing.T, reader
*sdkmetric.ManualReader) map[string]metricdata.Metrics {
+ t.Helper()
+
+ var rm metricdata.ResourceMetrics
+ require.NoError(t, reader.Collect(context.Background(), &rm))
+
+ out := make(map[string]metricdata.Metrics)
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ out[m.Name] = m
+ }
+ }
+ return out
+}
+
+// sumForClusterMetric extracts the counter value for the given cluster label.
+func sumForClusterMetric(t *testing.T, m metricdata.Metrics, clusterName
string) int64 {
+ t.Helper()
+ data, ok := m.Data.(metricdata.Sum[int64])
+ require.True(t, ok, "metric %s is not an int64 Sum", m.Name)
+ for _, dp := range data.DataPoints {
+ if v, ok := dp.Attributes.Value(attribute.Key("cluster")); ok
&& v.AsString() == clusterName {
+ return dp.Value
+ }
+ }
+ t.Fatalf("no data point for cluster %q in metric %s", clusterName,
m.Name)
+ return 0
+}
+
+// gaugeForClusterMetric extracts the gauge value for the given cluster label.
+func gaugeForClusterMetric(t *testing.T, m metricdata.Metrics, clusterName
string) int64 {
+ t.Helper()
+ data, ok := m.Data.(metricdata.Gauge[int64])
+ require.True(t, ok, "metric %s is not an int64 Gauge", m.Name)
+ for _, dp := range data.DataPoints {
+ if v, ok := dp.Attributes.Value(attribute.Key("cluster")); ok
&& v.AsString() == clusterName {
+ return dp.Value
+ }
+ }
+ t.Fatalf("no data point for cluster %q in metric %s", clusterName,
m.Name)
+ return 0
+}
diff --git a/pkg/server/pixiu_start.go b/pkg/server/pixiu_start.go
index 3f04746d7..38b882e57 100644
--- a/pkg/server/pixiu_start.go
+++ b/pkg/server/pixiu_start.go
@@ -93,7 +93,6 @@ func (s *Server) Start() {
}
}()
- registerOtelMetricMeter(conf.Metric)
s.listenerManager.StartListen()
s.adapterManager.Start()
@@ -127,6 +126,15 @@ func Start(bs *model.Bootstrap) {
logger.Infof("[dubbo-go-pixiu] start by config : %+v", bs)
// global variable
server = NewServer()
+
+ // Register the OTel meter provider BEFORE cluster construction so that
+ // snapshot publication metrics emitted during cluster initialization
land
+ // on the real provider rather than the no-op default. Static clusters
+ // publish their initial snapshot in initialize →
CreateDefaultClusterManager,
+ // and that emission is the only guaranteed recording for steady-state
+ // clusters with no health transitions.
+ registerOtelMetricMeter(bs.Metric)
+
server.initialize(bs)
server.Start()
server.startWG.Wait()
