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 ac1ee2a84 refactor(cluster): consolidate duplicated endpoint ID suffix
algorithm (#969) (#972)
ac1ee2a84 is described below
commit ac1ee2a84770d56245cf818bd44deb16a098d17a
Author: 承潜 <[email protected]>
AuthorDate: Sun Jul 26 13:38:04 2026 +0800
refactor(cluster): consolidate duplicated endpoint ID suffix algorithm
(#969) (#972)
nextStableEndpointID (pkg/server) and uniqueSnapshotEndpointID (pkg/cluster)
were byte-for-byte identical implementations of the -2/-3 collision-suffix
algorithm, kept in sync only by convention. Since the endpoint ID is the
runtime health/cooldown key, a future edit to one and not the other would
silently assign different IDs to the same endpoint, splitting its health
state across a snapshot rebuild.
Lift the algorithm into model.StableUniqueEndpointID next to
GenerateEndpointID
(which both already call) and route all three call sites through it. No
behavior
change. Add a test asserting config assembly and snapshot rebuild produce
identical IDs for a duplicate-ID cluster, locking the two paths together.
---
pkg/cluster/cluster.go | 28 +---------------
pkg/model/cluster.go | 35 ++++++++++++++++++++
pkg/server/cluster_manager.go | 32 ++-----------------
pkg/server/cluster_manager_test.go | 65 ++++++++++++++++++++++++++++++++++++++
4 files changed, 104 insertions(+), 56 deletions(-)
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index 4dc783820..6608e546b 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -18,7 +18,6 @@
package cluster
import (
- "fmt"
"sync"
"sync/atomic"
)
@@ -246,7 +245,7 @@ func newEndpointSnapshot(config *model.ClusterConfig,
previous *EndpointSnapshot
continue
}
snapshotEndpoint := model.CloneEndpoint(endpoint)
- snapshotEndpoint.ID = uniqueSnapshotEndpointID(clusterName,
snapshotEndpoint, endpointIDs)
+ snapshotEndpoint.ID = model.StableUniqueEndpointID(clusterName,
snapshotEndpoint, endpointIDs)
endpointIDs[snapshotEndpoint.ID] = struct{}{}
address := snapshotEndpoint.Address.GetAddress()
healthy := endpointSnapshotHealth(snapshotEndpoint, address,
previous, inheritRuntimeHealth)
@@ -268,31 +267,6 @@ func newEndpointSnapshotIndex(endpointCount int)
*EndpointSnapshot {
}
}
-// uniqueSnapshotEndpointID resolves a stable runtime ID for one endpoint in
-// the snapshot's per-cluster dedup set. The operator's explicit endpoint.ID
-// wins unless it collides; collisions append -2, -3, ... so an operator who
-// wrote id: foo twice sees foo and foo-2 (not generated-<hash>-2). When the
-// operator did not supply an ID, the deterministic hash from PR-2 is used as
-// the base and collisions on that synthesized base also append -2, -3, ...
-func uniqueSnapshotEndpointID(clusterName string, endpoint *model.Endpoint,
endpointIDs map[string]struct{}) string {
- id := ""
- if endpoint != nil {
- id = endpoint.ID
- }
- if id == "" {
- id = model.GenerateEndpointID(clusterName, endpoint)
- }
- if _, exists := endpointIDs[id]; !exists {
- return id
- }
- for suffix := 2; ; suffix++ {
- candidate := fmt.Sprintf("%s-%d", id, suffix)
- if _, exists := endpointIDs[candidate]; !exists {
- return candidate
- }
- }
-}
-
func endpointSnapshotHealth(
endpoint *model.Endpoint,
address string,
diff --git a/pkg/model/cluster.go b/pkg/model/cluster.go
index dda7e7983..6238f6001 100644
--- a/pkg/model/cluster.go
+++ b/pkg/model/cluster.go
@@ -216,6 +216,41 @@ func endpointIDMaterialField(name, value string) string {
return fmt.Sprintf("%s:%d:%s\n", name, len(value), value)
}
+// StableUniqueEndpointID resolves a stable runtime ID for one endpoint within
+// a per-cluster dedup set. The operator's explicit endpoint.ID wins unless it
+// collides with an entry already in taken; on collision it appends -2, -3, ...
+// so an operator who wrote id: foo twice sees foo and foo-2 (not
+// generated-<hash>-2). When the operator supplied no ID, GenerateEndpointID's
+// deterministic hash is the base and collisions on that base also append
+// -2, -3, ...
+//
+// The endpoint ID is the runtime health/cooldown key. Static/dynamic config
+// assembly (server.ClusterStore.assembleClusterEndpoints) and snapshot rebuild
+// (cluster.newEndpointSnapshot) must agree on it, or the same endpoint would
+// split its health state across a snapshot rebuild. Both call this so the two
+// paths can never drift.
+//
+// taken is read only: the caller records the returned ID before resolving the
+// next endpoint.
+func StableUniqueEndpointID(clusterName string, endpoint *Endpoint, taken
map[string]struct{}) string {
+ baseID := ""
+ if endpoint != nil {
+ baseID = endpoint.ID
+ }
+ if baseID == "" {
+ baseID = GenerateEndpointID(clusterName, endpoint)
+ }
+ if _, exists := taken[baseID]; !exists {
+ return baseID
+ }
+ for suffix := 2; ; suffix++ {
+ candidate := fmt.Sprintf("%s-%d", baseID, suffix)
+ if _, exists := taken[candidate]; !exists {
+ return candidate
+ }
+ }
+}
+
// CloneEndpoints returns a deep copy of endpoints. Nil input is preserved.
func CloneEndpoints(endpoints []*Endpoint) []*Endpoint {
if endpoints == nil {
diff --git a/pkg/server/cluster_manager.go b/pkg/server/cluster_manager.go
index fb3afb12b..31c907872 100644
--- a/pkg/server/cluster_manager.go
+++ b/pkg/server/cluster_manager.go
@@ -409,10 +409,10 @@ func (s *ClusterStore) assembleClusterEndpoints(c
*model.ClusterConfig) {
}
// Endpoint IDs are runtime health keys, so keep them unique
per cluster.
if endpoint.ID == "" {
- endpoint.ID = nextStableEndpointID(c.Name, endpoint,
endpointIDs)
+ endpoint.ID = model.StableUniqueEndpointID(c.Name,
endpoint, endpointIDs)
} else if _, exists := endpointIDs[endpoint.ID]; exists {
duplicateID := endpoint.ID
- endpoint.ID = nextStableEndpointID(c.Name, endpoint,
endpointIDs)
+ endpoint.ID = model.StableUniqueEndpointID(c.Name,
endpoint, endpointIDs)
logger.Warnf(
"[dubbo-go-pixiu] duplicate endpoint ID %s in
cluster %s, assigned endpoint ID %s",
duplicateID,
@@ -431,32 +431,6 @@ func (s *ClusterStore) assembleClusterEndpoints(c
*model.ClusterConfig) {
}
}
-// nextStableEndpointID returns a unique endpoint ID for the cluster's dedup
-// set. If endpoint.ID is set (operator-supplied) and only collides with a
-// sibling, it appends -2, -3, ... to preserve the operator's choice. If
-// endpoint.ID is empty, it derives a deterministic generated-* base via
-// model.GenerateEndpointID and suffixes that on collision. This matches
-// uniqueSnapshotEndpointID in pkg/cluster and keeps the same dashboard/log
-// identity post-rebuild.
-func nextStableEndpointID(clusterName string, endpoint *model.Endpoint,
endpointIDs map[string]struct{}) string {
- baseID := ""
- if endpoint != nil {
- baseID = endpoint.ID
- }
- if baseID == "" {
- baseID = model.GenerateEndpointID(clusterName, endpoint)
- }
- if _, exists := endpointIDs[baseID]; !exists {
- return baseID
- }
- for suffix := 2; ; suffix++ {
- candidate := fmt.Sprintf("%s-%d", baseID, suffix)
- if _, exists := endpointIDs[candidate]; !exists {
- return candidate
- }
- }
-}
-
// replaceClusterRuntime returns the old runtime so callers decide when to
stop it.
func (s *ClusterStore) replaceClusterRuntime(name string, config
*model.ClusterConfig) *cluster.Cluster {
var previous *cluster.EndpointSnapshot
@@ -818,7 +792,7 @@ func resolveSetEndpointSlotByHash(clusterName string,
incoming *model.Endpoint,
if endpointContentEqualForSet(e, incoming) {
return setEndpointOutcome{targetID: e.ID, action:
setEndpointIdempotent, replaceIdx: -1}
}
- suffixedID := nextStableEndpointID(clusterName, incoming,
existingEndpointIDs(existing))
+ suffixedID := model.StableUniqueEndpointID(clusterName,
incoming, existingEndpointIDs(existing))
logSetEndpointSuffix(clusterName, incomingHash, suffixedID)
return setEndpointOutcome{targetID: suffixedID, action:
setEndpointAppend, replaceIdx: -1}
}
diff --git a/pkg/server/cluster_manager_test.go
b/pkg/server/cluster_manager_test.go
index 87ece3de5..c7834aeb0 100644
--- a/pkg/server/cluster_manager_test.go
+++ b/pkg/server/cluster_manager_test.go
@@ -1111,6 +1111,71 @@ func TestAssembleEndpointsDeduplicatesExplicitID(t
*testing.T) {
"not the generated- hash, so the operator's choice
stays readable")
}
+// TestEndpointIDAssemblyAndSnapshotRebuildAgree locks issue #969: the config
+// assembly path (ClusterStore.assembleClusterEndpoints, used for
static/dynamic
+// config) and the snapshot-rebuild path (cluster.NewCluster ->
newEndpointSnapshot)
+// must assign byte-identical endpoint IDs for the same cluster. The endpoint
ID
+// is the runtime health/cooldown key, so if the two -2/-3 suffix algorithms
ever
+// drift, the same endpoint would get a different ID on a snapshot rebuild and
+// split its health state. Both paths now route through
model.StableUniqueEndpointID;
+// this test fails if a future change reintroduces a second, diverging copy.
+func TestEndpointIDAssemblyAndSnapshotRebuildAgree(t *testing.T) {
+ const clusterName = "id-agreement"
+
+ idsOf := func(endpoints []*model.Endpoint) []string {
+ ids := make([]string, len(endpoints))
+ for i, endpoint := range endpoints {
+ ids[i] = endpoint.ID
+ }
+ return ids
+ }
+
+ tests := []struct {
+ name string
+ endpoints func() []*model.Endpoint
+ }{
+ {
+ // Operator wrote the same id: twice — base is the
operator ID.
+ name: "explicit duplicate ID",
+ endpoints: func() []*model.Endpoint {
+ return []*model.Endpoint{
+ {ID: "foo", Address:
model.SocketAddress{Address: "127.0.0.1", Port: 22001}},
+ {ID: "foo", Address:
model.SocketAddress{Address: "127.0.0.1", Port: 22002}},
+ }
+ },
+ },
+ {
+ // No IDs and identical hash material — base is the
generated hash.
+ name: "anonymous endpoints with identical hash
material",
+ endpoints: func() []*model.Endpoint {
+ return []*model.Endpoint{
+ {Address: model.SocketAddress{Address:
"127.0.0.1", Port: 22010}},
+ {Address: model.SocketAddress{Address:
"127.0.0.1", Port: 22010}},
+ }
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ assembled := &model.ClusterConfig{Name: clusterName,
Endpoints: tt.endpoints()}
+ (&ClusterStore{}).assembleClusterEndpoints(assembled)
+ assembledIDs := idsOf(assembled.Endpoints)
+
+ runtime :=
cluster.NewCluster(&model.ClusterConfig{Name: clusterName, Endpoints:
tt.endpoints()})
+ snapshotIDs :=
idsOf(runtime.EndpointSnapshot().AllEndpoints())
+
+ assert.Equal(t, assembledIDs, snapshotIDs,
+ "config assembly and snapshot rebuild must
assign identical endpoint IDs")
+
+ if assert.Len(t, assembledIDs, 2) {
+ assert.Equal(t, assembledIDs[0]+"-2",
assembledIDs[1],
+ "both paths must suffix the colliding
second endpoint with -2, not collapse it")
+ }
+ })
+ }
+}
+
func testClusterManager(clusters ...*model.ClusterConfig) *ClusterManager {
return CreateDefaultClusterManager(&model.Bootstrap{
StaticResources: model.StaticResources{
