This is an automated email from the ASF dual-hosted git repository.
AlexStocks 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 9c0cd206a Feat(cluster): reduce endpoint cloning (#961)
9c0cd206a is described below
commit 9c0cd206aa7774208978276e4daff52e99a5500d
Author: wm_03 <[email protected]>
AuthorDate: Mon Jul 6 20:16:52 2026 +0800
Feat(cluster): reduce endpoint cloning (#961)
* test(cluster): adds endpoint membership churn benchmark.
* feat(cluster): Avoid recloning store-owned endpoints
* feat(cluster): drop redundant nil guards
* feat(cluster): use slices.Delete for endpoint removal
---
pkg/server/cluster_manager.go | 33 +++++++++++++++++++++-----------
pkg/server/cluster_manager_bench_test.go | 33 ++++++++++++++++++++++++++++++++
2 files changed, 55 insertions(+), 11 deletions(-)
diff --git a/pkg/server/cluster_manager.go b/pkg/server/cluster_manager.go
index c708e60b4..09a4dd7a5 100644
--- a/pkg/server/cluster_manager.go
+++ b/pkg/server/cluster_manager.go
@@ -20,6 +20,7 @@ package server
import (
"fmt"
"reflect"
+ "slices"
"sync"
"sync/atomic"
)
@@ -356,23 +357,33 @@ func (s *ClusterStore) AddCluster(c *model.ClusterConfig)
{
stopClusters([]*cluster.Cluster{s.replaceClusterRuntime(c.Name, c)})
}
-// prepareClusterConfig rebuilds endpoint defaults and hash from current
endpoints.
+// prepareClusterConfig clones operator-supplied endpoints, then rebuilds
+// endpoint defaults and hash from current endpoints.
func (s *ClusterStore) prepareClusterConfig(c *model.ClusterConfig) {
+ c.Endpoints = model.CloneEndpoints(c.Endpoints)
+ s.prepareOwnedClusterConfig(c)
+}
+
+// prepareOwnedClusterConfig rebuilds endpoint defaults and hash for endpoints
+// already owned by ClusterStore. Callers must not pass operator-owned endpoint
+// pointers here; use prepareClusterConfig at external input boundaries.
+func (s *ClusterStore) prepareOwnedClusterConfig(c *model.ClusterConfig) {
s.assembleClusterEndpoints(c)
c.CreateConsistentHash()
}
-// assembleClusterEndpoints assembles the cluster endpoints by formatting the
-// ID, name and domains for each endpoint. If endpoint.LLMMeta is not nil, the
-// assimilation of name and domain is based on the LLM provider denoted in the
-// endpoint LLMMeta. The store first deep-clones c.Endpoints, so ID/name
-// defaulting never mutates operator-supplied *model.Endpoint values.
+// assembleClusterEndpoints assembles the cluster endpoints by assigning stable
+// unique IDs and default names for each endpoint. If endpoint.LLMMeta is not
nil,
+// the default endpoint name is based on the LLM provider denoted in the
endpoint
+// LLMMeta. Callers choose the ownership boundary before invoking this helper:
+// external input paths clone endpoints in prepareClusterConfig, while
store-owned
+// mutation paths call prepareOwnedClusterConfig to avoid a second full
endpoint
+// clone before snapshot publication.
func (s *ClusterStore) assembleClusterEndpoints(c *model.ClusterConfig) {
if c == nil {
return
}
- c.Endpoints = model.CloneEndpoints(c.Endpoints)
endpointIDs := make(map[string]struct{}, len(c.Endpoints))
for i, endpoint := range c.Endpoints {
if endpoint == nil {
@@ -642,7 +653,7 @@ func (s *ClusterStore) SetEndpoint(clusterName string,
endpoint *model.Endpoint)
s.replaceEndpointAt(clusterConfig, runtimeCluster,
outcome.replaceIdx, endpoint)
case setEndpointAppend:
clusterConfig.Endpoints = append(clusterConfig.Endpoints,
endpoint)
- s.prepareClusterConfig(clusterConfig)
+ s.prepareOwnedClusterConfig(clusterConfig)
runtimeCluster.RefreshEndpoints()
runtimeCluster.AddEndpoint(endpoint)
}
@@ -677,7 +688,7 @@ func (s *ClusterStore) replaceEndpointAt(
logSetEndpointOverwrite(clusterConfig.Name, endpoint.ID, old,
endpoint)
}
clusterConfig.Endpoints[idx] = endpoint
- s.prepareClusterConfig(clusterConfig)
+ s.prepareOwnedClusterConfig(clusterConfig)
runtimeCluster.RefreshEndpoints()
if addressChanged {
runtimeCluster.AddEndpoint(endpoint)
@@ -941,8 +952,8 @@ func (s *ClusterStore) DeleteEndpoint(clusterName string,
endpointID string) {
for i, e := range clusterConfig.Endpoints {
if e.ID == endpointID {
runtimeCluster.RemoveEndpoint(e)
- clusterConfig.Endpoints =
append(clusterConfig.Endpoints[:i], clusterConfig.Endpoints[i+1:]...)
- s.prepareClusterConfig(clusterConfig)
+ clusterConfig.Endpoints =
slices.Delete(clusterConfig.Endpoints, i, i+1)
+ s.prepareOwnedClusterConfig(clusterConfig)
runtimeCluster.RefreshEndpoints()
return
}
diff --git a/pkg/server/cluster_manager_bench_test.go
b/pkg/server/cluster_manager_bench_test.go
index d7b7a2563..63faf3db0 100644
--- a/pkg/server/cluster_manager_bench_test.go
+++ b/pkg/server/cluster_manager_bench_test.go
@@ -165,6 +165,39 @@ func benchmarkHealthySnapshotAccessor(
}
}
+func BenchmarkClusterSetEndpointMembershipChurn(b *testing.B) {
+ endpointCount := 1000
+ clusterName := "endpoint-churn"
+
+ clusterConfig := benchmarkClusterConfig(clusterName,
model.LoadBalancerRoundRobin, endpointCount, 0)
+ for _, endpoint := range clusterConfig.Endpoints {
+ endpoint.Metadata = map[string]string{
+ "first": "0",
+ "second": "0",
+ "third": "0",
+ }
+ }
+
+ cm := testClusterManager(clusterConfig)
+ endpoints := cm.store.Config[0].Endpoints
+
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ endpoint := endpoints[i%len(endpoints)]
+
+ cm.SetEndpoint(clusterName, &model.Endpoint{
+ ID: endpoint.ID,
+ Address: endpoint.Address,
+ Metadata: map[string]string{
+ "first": fmt.Sprintf("%d", i),
+ "second": fmt.Sprintf("%d", i),
+ "third": fmt.Sprintf("%d", i),
+ },
+ })
+ }
+}
+
func BenchmarkClusterCompareAndSetStoreMixed(b *testing.B) {
cm, names := benchmarkClusterManager(128, 4,
model.LoadBalancerRoundRobin)
readIndex := 0
