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 6ebc858b8 feat(cluster): decouple config from runtime — private
Config, RuntimeState, and RoundRobin cursor migration (#973)
6ebc858b8 is described below
commit 6ebc858b8d4857ed2281d7909537bb58918aaef9
Author: Yuqi Qiao <[email protected]>
AuthorDate: Thu Aug 20 14:38:06 2026 +0800
feat(cluster): decouple config from runtime — private Config, RuntimeState,
and RoundRobin cursor migration (#973)
* feat(model): add ConfigID identity and CloneClusterConfig for
config-runtime decoupling
- Add ConfigID field to ClusterConfig for stable config-runtime identity
(replaces fragile pointer comparison after deep copy)
- Add CloneClusterConfig deep-copy helper with independent Endpoints,
HealthChecks, and ConsistentHash (hash set to nil for runtime rebuild)
- Update ClusterConfig godoc: document immutability after publication,
runtime state ownership, and PrePickEndpointIndex legacy status
* feat(cluster): decouple config from runtime with private Config and
RuntimeState
- Make Cluster.Config private (config) with Config() getter
- Add RuntimeState struct with RoundRobinCursor; add accessors
- Add RoundRobinCursor to PickContext; update snapshot RoundRobin
to use runtime-owned cursor instead of ClusterConfig.PrePickEndpointIndex
- Add ConfigID for stable config-runtime identity; replace pointer
comparison with ConfigIsIdenticalTo
- Wire CloneClusterConfig into replaceClusterRuntimeWithSnapshot
- Remove dead getCluster method
- Adapt tests and benchmarks to new API
* fix(cluster): reconcile legacy RR cursor delta back to RoundRobinCursor
when present
When the legacy pickEndpoint fallback path reads cursorBefore from
RoundRobinCursor, the Handler advances config.PrePickEndpointIndex (a local
shallow copy). The delta was unconditionally written back to
context.Config.PrePickEndpointIndex, which left the runtime-owned
RoundRobinCursor stale. Apply the delta to RoundRobinCursor when it is
the source of cursorBefore.
* fix(cluster): add nil guard in NewClusterWithEndpointSnapshot for config
dereferences
Guard clusterConfig.ConfigID and HealthChecks access against nil config,
preserving the prior behaviour where a nil config built an empty snapshot
without panicking.
* fix(cluster): sync config endpoints to runtime before RefreshEndpoints
When SetEndpoint or replaceEndpointAt appends/modifies endpoints on the
store's ClusterConfig, the runtime holds an independent deep copy from
replaceClusterRuntimeWithSnapshot. RefreshEndpoints reads from the
runtime's config, missing the store-side mutations.
Add SyncConfigEndpoints to sync the store's authoritative endpoint slice
back to the runtime config before RefreshEndpoints.
* fix: gofmt indentation in cluster_manager.go
* fix(cluster): preserve custom hash across runtime config clone
---
admin/initialize/E2E_OPA.md | 2 -
pkg/cluster/cluster.go | 89 ++++++++++++++++++--
pkg/cluster/loadbalancer/load_balancer.go | 18 +++-
pkg/cluster/loadbalancer/roundrobin/round_robin.go | 8 +-
pkg/model/cluster.go | 51 ++++++++++-
pkg/model/cluster_test.go | 30 +++++++
pkg/server/cluster_manager.go | 66 +++++++--------
pkg/server/cluster_manager_bench_test.go | 5 +-
pkg/server/cluster_manager_test.go | 98 ++++++++++++++++++----
9 files changed, 301 insertions(+), 66 deletions(-)
diff --git a/admin/initialize/E2E_OPA.md b/admin/initialize/E2E_OPA.md
index 37b55caba..c1755a6a5 100644
--- a/admin/initialize/E2E_OPA.md
+++ b/admin/initialize/E2E_OPA.md
@@ -67,5 +67,3 @@ Expected output (verbose):
PASS
ok github.com/apache/dubbo-go-pixiu/admin/initialize 0.342s
```
-
-
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index a0adefbbe..24fa5a02a 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -28,15 +28,28 @@ import (
"github.com/apache/dubbo-go-pixiu/pkg/model"
)
+// Cluster is the runtime representation of an upstream cluster. It holds
+// a deep copy of the desired config and publishes immutable EndpointSnapshot
+// via CAS for the request path. Runtime state (cursors, health, snapshots)
+// lives here, not in model.ClusterConfig.
+//
+// RuntimeState holds operational state that belongs to the runtime, not the
+// configuration. Fields here are mutated only by the request path.
+type RuntimeState struct {
+ roundRobinCursor atomic.Uint32
+}
+
type Cluster struct {
HealthCheck *healthcheck.HealthChecker
- // Config is the desired cluster configuration. Runtime picks read the
- // published EndpointSnapshot, so direct edits to Config.Endpoints or
+ // config is the desired cluster configuration. Runtime picks read the
+ // published EndpointSnapshot, so direct edits to config.Endpoints or
// Endpoint.UnHealthy are not observed by PickEndpoint immediately.
Publish
// membership/address changes with RefreshEndpoints; publish runtime
health
// changes with UpdateEndpointHealth, or RefreshEndpoints for clusters
// without health checks.
- Config *model.ClusterConfig
+ config *model.ClusterConfig
+ configID uint64 // snapshot of config.ConfigID at
construction time
+ runtimeState *RuntimeState
healthMu sync.Mutex
acceptHealthEvents bool
endpoints atomic.Pointer[EndpointSnapshot]
@@ -47,17 +60,23 @@ func NewCluster(clusterConfig *model.ClusterConfig)
*Cluster {
}
func NewClusterWithEndpointSnapshot(clusterConfig *model.ClusterConfig,
previous *EndpointSnapshot) *Cluster {
+ configID := uint64(0)
+ if clusterConfig != nil {
+ configID = clusterConfig.ConfigID
+ }
c := &Cluster{
- Config: clusterConfig,
+ config: clusterConfig,
+ configID: configID,
+ runtimeState: &RuntimeState{},
acceptHealthEvents: true,
}
c.RefreshEndpointsFrom(previous)
// only handle one health checker
- if len(c.Config.HealthChecks) != 0 {
+ if clusterConfig != nil && len(clusterConfig.HealthChecks) != 0 {
c.HealthCheck = healthcheck.CreateHealthCheckWithCallback(
clusterConfig,
- c.Config.HealthChecks[0],
+ c.config.HealthChecks[0],
c.handleEndpointHealth,
)
c.HealthCheck.Start()
@@ -98,7 +117,7 @@ func (c *Cluster) RefreshEndpointsFrom(previous
*EndpointSnapshot) {
if current != nil && current != previous {
source = current
}
- next := newEndpointSnapshot(c.Config, source,
len(c.Config.HealthChecks) != 0)
+ next := newEndpointSnapshot(c.config, source,
len(c.config.HealthChecks) != 0)
if c.endpoints.CompareAndSwap(current, next) {
recordSnapshotPublish(c.clusterName(), next)
return
@@ -179,10 +198,62 @@ func (c *Cluster) handleEndpointHealth(event
healthcheck.EndpointHealthEvent) {
}
func (c *Cluster) clusterName() string {
- if c == nil || c.Config == nil {
+ if c == nil || c.config == nil {
return ""
}
- return c.Config.Name
+ return c.config.Name
+}
+
+// SyncConfigEndpoints updates the runtime's config endpoints to match the
+// store's authoritative copy. Required before RefreshEndpoints when the store
+// mutates endpoints through SetEndpoint or replaceEndpointAt.
+func (c *Cluster) SyncConfigEndpoints(endpoints []*model.Endpoint) {
+ if c == nil {
+ return
+ }
+ c.config.Endpoints = model.CloneEndpoints(endpoints)
+}
+
+// Config returns the cluster configuration this runtime was built from.
+// Callers must treat the returned pointer as read-only; the runtime owns the
+// config and never mutates it after construction.
+func (c *Cluster) Config() *model.ClusterConfig {
+ if c == nil {
+ return nil
+ }
+ return c.config
+}
+
+// ConfigIsIdenticalTo reports whether this runtime was built from cfg (by
+// config object identity, not by deep equality). Used after deep-cloning to
+// detect when a runtime needs replacement without relying on pointer equality.
+func (c *Cluster) ConfigIsIdenticalTo(cfg *model.ClusterConfig) bool {
+ if c == nil || cfg == nil {
+ return false
+ }
+ return c.configID == cfg.ConfigID
+}
+
+// RoundRobinCursor returns the runtime's atomic RoundRobin cursor. Snapshot
+// load balancers use this cursor instead of
ClusterConfig.PrePickEndpointIndex.
+func (c *Cluster) RoundRobinCursor() *atomic.Uint32 {
+ if c == nil || c.runtimeState == nil {
+ return nil
+ }
+ return &c.runtimeState.roundRobinCursor
+}
+
+// CarryOverCursorTo copies this runtime's RR cursor value into target, used
+// when replacing runtimes to preserve fairness state.
+func (c *Cluster) CarryOverCursorTo(target *Cluster) {
+ if c == nil || target == nil {
+ return
+ }
+ src := c.RoundRobinCursor()
+ dst := target.RoundRobinCursor()
+ if src != nil && dst != nil {
+ dst.Store(src.Load())
+ }
}
// EndpointSnapshot endpoint membership and health indexes are immutable after
diff --git a/pkg/cluster/loadbalancer/load_balancer.go
b/pkg/cluster/loadbalancer/load_balancer.go
index 4b577eab8..e3e87e7cb 100644
--- a/pkg/cluster/loadbalancer/load_balancer.go
+++ b/pkg/cluster/loadbalancer/load_balancer.go
@@ -52,6 +52,9 @@ type PickContext struct {
// Snapshot-aware balancers must treat endpoints as read-only and return
// the chosen endpoint without mutating or retaining it.
HealthyEndpoints []*model.Endpoint
+ // RoundRobinCursor is the runtime-owned atomic RR cursor.
Snapshot-aware
+ // RoundRobin balancers should use this instead of
Config.PrePickEndpointIndex.
+ RoundRobinCursor *atomic.Uint32
// HealthyByID resolves a healthy snapshot endpoint by ID in O(1) for
the
// post-pick identity recheck. It is set by the snapshot-published pick
path;
// when nil (e.g. a hand-built context in a test), the recheck falls
back to
@@ -231,12 +234,21 @@ func pickEndpoint(balancer LoadBalancer, context
PickContext, policy model.LbPol
}
config := *context.Config
config.Endpoints = model.CloneEndpoints(allEndpoints)
- cursorBefore := atomic.LoadUint32(&context.Config.PrePickEndpointIndex)
+ var cursorBefore, cursorAfter uint32
+ if context.RoundRobinCursor != nil {
+ cursorBefore = context.RoundRobinCursor.Load()
+ } else {
+ cursorBefore =
atomic.LoadUint32(&context.Config.PrePickEndpointIndex)
+ }
atomic.StoreUint32(&config.PrePickEndpointIndex, cursorBefore)
endpoint := balancer.Handler(&config, policy)
- cursorAfter := atomic.LoadUint32(&config.PrePickEndpointIndex)
+ cursorAfter = atomic.LoadUint32(&config.PrePickEndpointIndex)
if cursorAfter != cursorBefore {
- atomic.AddUint32(&context.Config.PrePickEndpointIndex,
cursorAfter-cursorBefore)
+ if context.RoundRobinCursor != nil {
+ context.RoundRobinCursor.Add(cursorAfter - cursorBefore)
+ } else {
+ atomic.AddUint32(&context.Config.PrePickEndpointIndex,
cursorAfter-cursorBefore)
+ }
}
return healthyEndpointFromSnapshot(endpoint, context)
}
diff --git a/pkg/cluster/loadbalancer/roundrobin/round_robin.go
b/pkg/cluster/loadbalancer/roundrobin/round_robin.go
index ecd787ddb..bc03a2404 100644
--- a/pkg/cluster/loadbalancer/roundrobin/round_robin.go
+++ b/pkg/cluster/loadbalancer/roundrobin/round_robin.go
@@ -49,7 +49,11 @@ func (RoundRobin) HandlerWithSnapshot(c
loadbalancer.PickContext, _ model.LbPoli
if len(endpoints) == 0 {
return nil
}
- // AddUint32 returns the incremented value, so subtract 1 for a
zero-based index.
- index := atomic.AddUint32(&c.Config.PrePickEndpointIndex, 1) - 1
+ var index uint32
+ if c.RoundRobinCursor != nil {
+ index = c.RoundRobinCursor.Add(1) - 1
+ } else {
+ index = atomic.AddUint32(&c.Config.PrePickEndpointIndex, 1) - 1
+ }
return endpoints[int(index%uint32(len(endpoints)))]
}
diff --git a/pkg/model/cluster.go b/pkg/model/cluster.go
index 6238f6001..41de7c4f9 100644
--- a/pkg/model/cluster.go
+++ b/pkg/model/cluster.go
@@ -57,7 +57,16 @@ var (
)
type (
- // ClusterConfig a single upstream cluster
+ // ClusterConfig represents the desired state of an upstream cluster.
+ // It is created from YAML, xDS, or ClusterManager APIs. After
publication
+ // to a runtime Cluster the runtime holds its own deep copy — mutations
to
+ // ClusterConfig are invisible to the running cluster until the next
+ // explicit update.
+ //
+ // Runtime state (cursors, health, snapshots) belongs on
cluster.Cluster,
+ // not here. PrePickEndpointIndex is retained only for the legacy
+ // LoadBalancer.Handler interface; new code should use
+ // PickContext.RoundRobinCursor.
ClusterConfig struct {
Name string `yaml:"name"
json:"name"` // Name the cluster unique name
TypeStr string `yaml:"type"
json:"type"` // Type the cluster discovery type string value
@@ -67,7 +76,8 @@ type (
ConsistentHash ConsistentHash `yaml:"consistent"
json:"consistent"` // Consistent hash config info
HealthChecks []HealthCheckConfig `yaml:"health_checks"
json:"health_checks"`
Endpoints []*Endpoint `yaml:"endpoints"
json:"endpoints"`
- PrePickEndpointIndex uint32 `yaml:"-" json:"-"` //
runtime-only round-robin cursor state
+ PrePickEndpointIndex uint32 `yaml:"-" json:"-"` //
runtime-only round-robin cursor state (legacy Handler compat)
+ ConfigID uint64 `yaml:"-" json:"-"` //
stable identity for config-runtime association
}
// EdsClusterConfig todo remove un-used EdsClusterConfig
@@ -305,3 +315,40 @@ func cloneLLMMeta(meta *LLMMeta) *LLMMeta {
cloned.RetryPolicy.Config =
copyutil.CloneStringAnyMap(meta.RetryPolicy.Config)
return &cloned
}
+
+// CloneClusterConfig returns a deep copy of c suitable for handing to a new
+// runtime Cluster. The clone owns its Endpoints and HealthChecks slices.
+// ConsistentHash.Hash (a mutable runtime object) is set to nil when the policy
+// has a registered factory, so the runtime can rebuild it from its endpoint
+// snapshot. For an unregistered/custom policy, a programmatically supplied
+// Hash is preserved because there is no factory available to reconstruct it;
+// custom implementations are responsible for their own concurrency safety.
+// configID is preserved so callers can detect config-object identity changes.
+// PrePickEndpointIndex is NOT copied — runtime cursor state belongs on the
+// runtime, not in the config clone.
+func CloneClusterConfig(c *ClusterConfig) *ClusterConfig {
+ if c == nil {
+ return nil
+ }
+ clone := *c
+ clone.Endpoints = CloneEndpoints(c.Endpoints)
+ clone.HealthChecks = cloneHealthChecks(c.HealthChecks)
+ clone.ConsistentHash = c.ConsistentHash
+ if c.HasConsistentHashFactory() {
+ clone.ConsistentHash.Hash = nil
+ }
+ return &clone
+}
+
+func cloneHealthChecks(checks []HealthCheckConfig) []HealthCheckConfig {
+ if checks == nil {
+ return nil
+ }
+ cloned := make([]HealthCheckConfig, len(checks))
+ for i, hc := range checks {
+ cloned[i] = hc
+ cloned[i].CommonCallbacks = append([]string(nil),
hc.CommonCallbacks...)
+ cloned[i].SessionConfig =
copyutil.CloneStringAnyMap(hc.SessionConfig)
+ }
+ return cloned
+}
diff --git a/pkg/model/cluster_test.go b/pkg/model/cluster_test.go
index b53a6fa6c..a1a71ad5f 100644
--- a/pkg/model/cluster_test.go
+++ b/pkg/model/cluster_test.go
@@ -108,6 +108,36 @@ func TestClusterConfig_EnsureConsistentHashBuildsOnce(t
*testing.T) {
assert.Same(t, first, cluster.ConsistentHash.Hash)
}
+func TestCloneClusterConfigHandlesConsistentHashByFactoryAvailability(t
*testing.T) {
+ customHash := &testConsistentHash{}
+
+ registered := &model.ClusterConfig{
+ LbStr: model.LoadBalancerRingHashing,
+ ConsistentHash: model.ConsistentHash{Hash: customHash},
+ }
+ registeredClone := model.CloneClusterConfig(registered)
+ assert.Nil(t, registeredClone.ConsistentHash.Hash,
+ "a registered factory can rebuild the hash for the runtime
snapshot")
+ assert.Same(t, customHash, registered.ConsistentHash.Hash,
+ "cloning must not mutate the source config")
+
+ unregistered := &model.ClusterConfig{
+ LbStr: model.LbPolicyType("ProgrammaticCustomHash"),
+ ConsistentHash: model.ConsistentHash{Hash: customHash},
+ }
+ unregisteredClone := model.CloneClusterConfig(unregistered)
+ assert.Same(t, customHash, unregisteredClone.ConsistentHash.Hash,
+ "a programmatic hash must survive when no factory can rebuild
it")
+}
+
+type testConsistentHash struct{}
+
+func (*testConsistentHash) Hash(string) uint32 { return 0 }
+func (*testConsistentHash) Get(string) (string, error) { return "", nil }
+func (*testConsistentHash) GetHash(uint32) (string, error) { return "", nil }
+func (*testConsistentHash) Add(string) {}
+func (*testConsistentHash) Remove(string) bool { return false }
+
func TestClusterConfig_PrePickEndpointIndexIsRuntimeOnly(t *testing.T) {
cluster := &model.ClusterConfig{
Name: "runtime-cursor",
diff --git a/pkg/server/cluster_manager.go b/pkg/server/cluster_manager.go
index 31c907872..c47304ef0 100644
--- a/pkg/server/cluster_manager.go
+++ b/pkg/server/cluster_manager.go
@@ -37,6 +37,7 @@ import (
// generate cluster name for unnamed cluster
var (
clusterIndex int32 = 1
+ configIDSeq uint64
)
type (
@@ -272,15 +273,6 @@ func (cm *ClusterManager)
GetHealthyEndpointByID(clusterName, endpointID string)
return runtimeCluster.EndpointSnapshot().HealthyEndpointByID(endpointID)
}
-// getCluster returns the cluster configuration by its name.
-func (cm *ClusterManager) getCluster(clusterName string) *model.ClusterConfig {
- runtimeCluster := cm.getRuntimeCluster(clusterName)
- if runtimeCluster == nil {
- return nil
- }
- return runtimeCluster.Config
-}
-
func (cm *ClusterManager) getRuntimeCluster(clusterName string)
*cluster.Cluster {
return cm.store.clustersMap[clusterName]
}
@@ -292,7 +284,7 @@ func (cm *ClusterManager) pickOneEndpoint(runtimeCluster
*cluster.Cluster, polic
}
healthyEndpoints := snapshot.HealthyEndpointsForPick()
- c := runtimeCluster.Config
+ c := runtimeCluster.Config()
loadBalancer, ok := loadbalancer.LoadBalancerStrategy[c.LbStr]
if !ok {
loadBalancer =
loadbalancer.LoadBalancerStrategy[model.LoadBalancerRand]
@@ -308,6 +300,7 @@ func (cm *ClusterManager) pickOneEndpoint(runtimeCluster
*cluster.Cluster, polic
HealthyConsistentHash: snapshot.HealthyConsistentHash(),
AllEndpoints: allEndpoints,
HealthyEndpoints: healthyEndpoints,
+ RoundRobinCursor: runtimeCluster.RoundRobinCursor(),
HealthyByID: snapshot,
}, policy)
}
@@ -384,6 +377,9 @@ func (s *ClusterStore) prepareClusterConfig(c
*model.ClusterConfig) {
// preserve any programmatically supplied hash because there is no factory
// available to rebuild it later.
func (s *ClusterStore) prepareOwnedClusterConfig(c *model.ClusterConfig) {
+ if c.ConfigID == 0 {
+ c.ConfigID = atomic.AddUint64(&configIDSeq, 1)
+ }
s.assembleClusterEndpoints(c)
if c.HasConsistentHashFactory() {
c.ConsistentHash.Hash = nil
@@ -448,7 +444,11 @@ func (s *ClusterStore) replaceClusterRuntimeWithSnapshot(
s.ensureRuntimeClusterMap()
oldRuntime := s.clustersMap[name]
- s.clustersMap[name] = cluster.NewClusterWithEndpointSnapshot(config,
previous)
+ // Deep-clone so the runtime owns its config copy. Mutations to
+ // store.Config[i] no longer affect the running cluster until the next
+ // explicit update.
+ cloned := model.CloneClusterConfig(config)
+ s.clustersMap[name] = cluster.NewClusterWithEndpointSnapshot(cloned,
previous)
return oldRuntime
}
@@ -475,7 +475,7 @@ func (s *ClusterStore) ensureRuntimeClusters()
[]*cluster.Cluster {
configsByName[clusterConfig.Name] = clusterConfig
runtimeCluster := s.clustersMap[clusterConfig.Name]
- if runtimeCluster == nil || runtimeCluster.Config !=
clusterConfig {
+ if runtimeCluster == nil ||
!runtimeCluster.ConfigIsIdenticalTo(clusterConfig) {
if oldRuntime :=
s.replaceClusterRuntime(clusterConfig.Name, clusterConfig); oldRuntime != nil {
replacedClusters = append(replacedClusters,
oldRuntime)
}
@@ -522,7 +522,7 @@ func (s *ClusterStore) repairRuntimeClusterFromPrevious(
s.prepareClusterConfig(clusterConfig)
previous := snapshotForRuntimeReplacement(old, clusterConfig.Name)
runtimeCluster := s.clustersMap[clusterConfig.Name]
- if runtimeCluster != nil && runtimeCluster.Config == clusterConfig &&
previous == nil {
+ if runtimeCluster != nil &&
runtimeCluster.ConfigIsIdenticalTo(clusterConfig) && previous == nil {
return clusterConfig.Name, nil
}
return clusterConfig.Name,
s.replaceClusterRuntimeWithSnapshot(clusterConfig.Name, clusterConfig, previous)
@@ -599,12 +599,14 @@ func (s *ClusterStore) UpdateCluster(new
*model.ClusterConfig) {
}
if c.Name == new.Name {
s.prepareClusterConfig(new)
- atomic.StoreUint32(
- &new.PrePickEndpointIndex,
- atomic.LoadUint32(&c.PrePickEndpointIndex),
- )
+ oldRuntime := s.clustersMap[new.Name]
s.Config[i] = new
-
stopClusters([]*cluster.Cluster{s.replaceClusterRuntime(new.Name, new)})
+ oldReplaced := s.replaceClusterRuntime(new.Name, new)
+ if oldRuntime != nil {
+ newRuntime := s.clustersMap[new.Name]
+ oldRuntime.CarryOverCursorTo(newRuntime)
+ }
+ stopClusters([]*cluster.Cluster{oldReplaced})
return
}
}
@@ -625,7 +627,7 @@ func (s *ClusterStore) SetEndpoint(clusterName string,
endpoint *model.Endpoint)
}
runtimeCluster := s.clustersMap[clusterName]
- if runtimeCluster == nil || runtimeCluster.Config != clusterConfig {
+ if runtimeCluster == nil ||
!runtimeCluster.ConfigIsIdenticalTo(clusterConfig) {
stopClusters([]*cluster.Cluster{s.replaceClusterRuntime(clusterName,
clusterConfig)})
runtimeCluster = s.clustersMap[clusterName]
}
@@ -646,6 +648,7 @@ func (s *ClusterStore) SetEndpoint(clusterName string,
endpoint *model.Endpoint)
case setEndpointAppend:
clusterConfig.Endpoints = append(clusterConfig.Endpoints,
endpoint)
s.prepareOwnedClusterConfig(clusterConfig)
+ runtimeCluster.SyncConfigEndpoints(clusterConfig.Endpoints)
runtimeCluster.RefreshEndpoints()
runtimeCluster.AddEndpoint(endpoint)
}
@@ -681,6 +684,7 @@ func (s *ClusterStore) replaceEndpointAt(
}
clusterConfig.Endpoints[idx] = endpoint
s.prepareOwnedClusterConfig(clusterConfig)
+ runtimeCluster.SyncConfigEndpoints(clusterConfig.Endpoints)
runtimeCluster.RefreshEndpoints()
if addressChanged {
runtimeCluster.AddEndpoint(endpoint)
@@ -936,7 +940,7 @@ func (s *ClusterStore) DeleteEndpoint(clusterName string,
endpointID string) {
}
runtimeCluster := s.clustersMap[clusterName]
- if runtimeCluster == nil || runtimeCluster.Config != clusterConfig {
+ if runtimeCluster == nil ||
!runtimeCluster.ConfigIsIdenticalTo(clusterConfig) {
stopClusters([]*cluster.Cluster{s.replaceClusterRuntime(clusterName,
clusterConfig)})
runtimeCluster = s.clustersMap[clusterName]
}
@@ -946,6 +950,7 @@ func (s *ClusterStore) DeleteEndpoint(clusterName string,
endpointID string) {
runtimeCluster.RemoveEndpoint(e)
clusterConfig.Endpoints =
slices.Delete(clusterConfig.Endpoints, i, i+1)
s.prepareOwnedClusterConfig(clusterConfig)
+
runtimeCluster.SyncConfigEndpoints(clusterConfig.Endpoints)
runtimeCluster.RefreshEndpoints()
return
}
@@ -980,23 +985,18 @@ func (s *ClusterStore) carryOverRuntimeStateFrom(old
*ClusterStore) {
return
}
- oldConfigsByName := make(map[string]*model.ClusterConfig,
len(old.Config))
- for _, clusterConfig := range old.Config {
- if clusterConfig != nil {
- oldConfigsByName[clusterConfig.Name] = clusterConfig
- }
- }
-
- // Preserve runtime-only load-balancer state when a rebuilt store is
swapped in.
+ // Preserve runtime-only RoundRobin cursor when a rebuilt store is
swapped in.
+ // The cursor lives on Cluster.runtimeState, not on ClusterConfig.
for _, clusterConfig := range s.Config {
if clusterConfig == nil {
continue
}
- if oldConfig := oldConfigsByName[clusterConfig.Name]; oldConfig
!= nil {
- atomic.StoreUint32(
- &clusterConfig.PrePickEndpointIndex,
-
atomic.LoadUint32(&oldConfig.PrePickEndpointIndex),
- )
+ newRuntime := s.clustersMap[clusterConfig.Name]
+ if newRuntime == nil {
+ continue
+ }
+ if oldRuntime := old.clustersMap[clusterConfig.Name];
oldRuntime != nil {
+ oldRuntime.CarryOverCursorTo(newRuntime)
}
}
}
diff --git a/pkg/server/cluster_manager_bench_test.go
b/pkg/server/cluster_manager_bench_test.go
index 523d14e7a..09724ab77 100644
--- a/pkg/server/cluster_manager_bench_test.go
+++ b/pkg/server/cluster_manager_bench_test.go
@@ -71,7 +71,10 @@ func BenchmarkClusterLookupSerial(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
cm.rw.RLock()
- benchmarkClusterSink =
cm.getCluster(names[i%len(names)])
+ runtimeCluster :=
cm.getRuntimeCluster(names[i%len(names)])
+ if runtimeCluster != nil {
+ benchmarkClusterSink =
runtimeCluster.Config()
+ }
cm.rw.RUnlock()
}
})
diff --git a/pkg/server/cluster_manager_test.go
b/pkg/server/cluster_manager_test.go
index c7834aeb0..264790350 100644
--- a/pkg/server/cluster_manager_test.go
+++ b/pkg/server/cluster_manager_test.go
@@ -38,6 +38,7 @@ import (
import (
"github.com/apache/dubbo-go-pixiu/pkg/cluster"
+ "github.com/apache/dubbo-go-pixiu/pkg/cluster/loadbalancer"
_ "github.com/apache/dubbo-go-pixiu/pkg/cluster/loadbalancer/maglev"
// Register Maglev for cluster-manager tests.
_ "github.com/apache/dubbo-go-pixiu/pkg/cluster/loadbalancer/rand"
// Register Rand for cluster-manager tests.
_ "github.com/apache/dubbo-go-pixiu/pkg/cluster/loadbalancer/ringhash"
// Register RingHash for cluster-manager tests.
@@ -180,7 +181,9 @@ func
TestClusterManager_CompareAndSetStorePreservesRoundRobinCursorAcrossRefresh
cm := testClusterManager(cluster)
const expectedCursor uint32 = 5
- atomic.StoreUint32(&cm.store.Config[0].PrePickEndpointIndex,
expectedCursor)
+ // Set cursor on the runtime, not the config.
+ oldRuntime := cm.store.clustersMap[cluster.Name]
+ oldRuntime.RoundRobinCursor().Store(expectedCursor)
oldStore, err := cm.CloneStore()
if !assert.NoError(t, err) {
@@ -194,7 +197,10 @@ func
TestClusterManager_CompareAndSetStorePreservesRoundRobinCursorAcrossRefresh
assert.True(t, cm.CompareAndSetStore(newStore))
if assert.Len(t, cm.store.Config, 1) {
- assert.Equal(t, expectedCursor,
atomic.LoadUint32(&cm.store.Config[0].PrePickEndpointIndex))
+ newRuntime := cm.store.clustersMap[cluster.Name]
+ if assert.NotNil(t, newRuntime) {
+ assert.Equal(t, expectedCursor,
newRuntime.RoundRobinCursor().Load())
+ }
}
endpoint := cm.PickEndpoint(cluster.Name, nil)
@@ -214,11 +220,11 @@ func
TestClusterManager_UpdateClusterRebuildsRuntimeCluster(t *testing.T) {
if !assert.NotNil(t, oldRuntime) {
return
}
- assert.Same(t, oldConfig, oldRuntime.Config)
+ assert.True(t, oldRuntime.ConfigIsIdenticalTo(oldConfig))
assert.Greater(t, healthCheckersLen(oldRuntime), 0)
const expectedCursor uint32 = 11
- atomic.StoreUint32(&oldConfig.PrePickEndpointIndex, expectedCursor)
+ oldRuntime.RoundRobinCursor().Store(expectedCursor)
newConfig := testCluster(oldConfig.Name, model.LoadBalancerRoundRobin,
[]*model.Endpoint{
testEndpoint("ep-2", "127.0.0.1", 19301),
@@ -231,9 +237,9 @@ func
TestClusterManager_UpdateClusterRebuildsRuntimeCluster(t *testing.T) {
return
}
assert.NotSame(t, oldRuntime, newRuntime)
- assert.Same(t, newConfig, newRuntime.Config)
+ assert.True(t, newRuntime.ConfigIsIdenticalTo(newConfig))
assert.Same(t, newConfig, cm.store.Config[0])
- assert.Equal(t, expectedCursor,
atomic.LoadUint32(&newConfig.PrePickEndpointIndex))
+ assert.Equal(t, expectedCursor, newRuntime.RoundRobinCursor().Load())
assert.Equal(t, 0, healthCheckersLen(oldRuntime))
assert.Greater(t, healthCheckersLen(newRuntime), 0)
}
@@ -279,7 +285,7 @@ func
TestClusterManager_CompareAndSetStoreEnsuresRuntimeAndStopsOld(t *testing.T
assert.Greater(t, healthCheckersLen(oldRuntime), 0)
const expectedCursor uint32 = 17
- atomic.StoreUint32(&oldConfig.PrePickEndpointIndex, expectedCursor)
+ oldRuntime.RoundRobinCursor().Store(expectedCursor)
newConfig := testCluster(oldConfig.Name, model.LoadBalancerRoundRobin,
[]*model.Endpoint{
testEndpoint("ep-2", "127.0.0.1", 19321),
@@ -298,8 +304,8 @@ func
TestClusterManager_CompareAndSetStoreEnsuresRuntimeAndStopsOld(t *testing.T
}
assert.Same(t, candidate, cm.store)
assert.NotSame(t, oldRuntime, newRuntime)
- assert.Same(t, newConfig, newRuntime.Config)
- assert.Equal(t, expectedCursor,
atomic.LoadUint32(&newConfig.PrePickEndpointIndex))
+ assert.True(t, newRuntime.ConfigIsIdenticalTo(newConfig))
+ assert.Equal(t, expectedCursor, newRuntime.RoundRobinCursor().Load())
assert.Equal(t, 0, healthCheckersLen(oldRuntime))
assert.Greater(t, healthCheckersLen(newRuntime), 0)
}
@@ -339,7 +345,7 @@ func
TestClusterStore_EnsureRuntimeClustersRepairsRuntimeMap(t *testing.T) {
assert.Empty(t, replaced)
if assert.NotNil(t, store.clustersMap[config.Name]) {
- assert.Same(t, config,
store.clustersMap[config.Name].Config)
+ assert.True(t,
store.clustersMap[config.Name].ConfigIsIdenticalTo(config))
}
})
@@ -380,13 +386,13 @@ func
TestClusterStore_EnsureRuntimeClustersRepairsRuntimeMap(t *testing.T) {
replaced := store.ensureRuntimeClusters()
stopClusters(replaced)
- assert.Same(t, correctRuntime,
store.clustersMap[correctConfig.Name])
+ assert.True(t,
store.clustersMap[correctConfig.Name].ConfigIsIdenticalTo(correctConfig))
if assert.NotNil(t, store.clustersMap[missingConfig.Name]) {
- assert.Same(t, missingConfig,
store.clustersMap[missingConfig.Name].Config)
+ assert.True(t,
store.clustersMap[missingConfig.Name].ConfigIsIdenticalTo(missingConfig))
}
if assert.NotNil(t,
store.clustersMap[newMismatchedConfig.Name]) {
assert.NotSame(t, mismatchedRuntime,
store.clustersMap[newMismatchedConfig.Name])
- assert.Same(t, newMismatchedConfig,
store.clustersMap[newMismatchedConfig.Name].Config)
+ assert.True(t,
store.clustersMap[newMismatchedConfig.Name].ConfigIsIdenticalTo(newMismatchedConfig))
}
assert.NotContains(t, store.clustersMap, staleConfig.Name)
assert.Contains(t, replaced, mismatchedRuntime)
@@ -478,7 +484,7 @@ func
TestClusterManager_DeleteEndpointRepairsRuntimeAndConsistentHash(t *testing
return
}
assert.NotSame(t, staleRuntime, runtime)
- assert.Same(t, config, runtime.Config)
+ assert.True(t, runtime.ConfigIsIdenticalTo(config))
if assert.Len(t, config.Endpoints, 1) {
assert.Equal(t, remainingEndpoint, config.Endpoints[0])
assert.NotSame(t, remainingEndpoint, config.Endpoints[0])
@@ -561,6 +567,70 @@ func
TestClusterManager_PrepareClusterConfigPreservesCustomHashWithoutFactory(t
assert.Same(t, customHash, cm.store.Config[0].ConsistentHash.Hash)
}
+func TestClusterManager_RuntimePreservesProgrammaticCustomHash(t *testing.T) {
+ const customPolicy model.LbPolicyType = "ProgrammaticCustomHashRuntime"
+ previousBalancer, hadPreviousBalancer :=
loadbalancer.LoadBalancerStrategy[customPolicy]
+ loadbalancer.LoadBalancerStrategy[customPolicy] =
programmaticHashSnapshotBalancer{}
+ defer func() {
+ if hadPreviousBalancer {
+ loadbalancer.LoadBalancerStrategy[customPolicy] =
previousBalancer
+ } else {
+ delete(loadbalancer.LoadBalancerStrategy, customPolicy)
+ }
+ }()
+
+ endpoint := testEndpoint("custom-hash-ep", "127.0.0.1", 19372)
+ config := testCluster("custom-hash-runtime", customPolicy,
[]*model.Endpoint{endpoint})
+ config.ConsistentHash.Hash = fixedEndpointHash{host: endpoint.GetHost()}
+
+ cm := testClusterManager(config)
+ defer stopStoreRuntimes(cm.store)
+
+ picked := cm.PickEndpoint(config.Name, nil)
+ require.NotNil(t, picked)
+ assert.Equal(t, endpoint.ID, picked.ID)
+}
+
+type fixedEndpointHash struct {
+ host string
+}
+
+func (fixedEndpointHash) Hash(string) uint32 { return 0 }
+func (h fixedEndpointHash) Get(string) (string, error) { return h.host,
nil }
+func (h fixedEndpointHash) GetHash(uint32) (string, error) { return h.host,
nil }
+func (fixedEndpointHash) Add(string) {}
+func (fixedEndpointHash) Remove(string) bool { return false }
+
+type programmaticHashSnapshotBalancer struct{}
+
+func (programmaticHashSnapshotBalancer) Handler(*model.ClusterConfig,
model.LbPolicy) *model.Endpoint {
+ return nil
+}
+
+func (programmaticHashSnapshotBalancer) HandlerWithSnapshot(
+ context loadbalancer.PickContext,
+ _ model.LbPolicy,
+) *model.Endpoint {
+ hash := loadbalancer.ConsistentHashForHealthyEndpoints(context)
+ if hash == nil {
+ return nil
+ }
+ host, err := hash.Get("reviewer-regression")
+ if err != nil {
+ return nil
+ }
+ for _, endpoint := range context.HealthyEndpoints {
+ if endpoint != nil && endpoint.GetHost() == host {
+ return endpoint
+ }
+ }
+ return nil
+}
+
+func (programmaticHashSnapshotBalancer) UseHealthyEndpointsOnly() bool {
+ return true
+}
+
func TestClusterManager_Race_RoundRobinPickEndpoint(t *testing.T) {
cluster := testCluster("race-round-robin",
model.LoadBalancerRoundRobin, []*model.Endpoint{
testEndpoint("ep-1", "127.0.0.1", 19100),
