Similarityoung commented on code in PR #1042:
URL: https://github.com/apache/dubbo-go-pixiu/pull/1042#discussion_r4015896190
##########
pkg/config/xds/apiclient/grpc_envoy.go:
##########
@@ -133,246 +143,258 @@ func (g *AggGrpcApiClient) Delta() (chan
*DeltaResources, error) {
}
type refEndpoint struct {
- IsPending bool
- RawProto proto.Message
+ Clusters []*clusterpb.Cluster
}
func (g *AggGrpcApiClient) pipeline(output chan *DeltaResources) error {
- // all endpoint refer cluster or listener. map[type]map[resource
name]endpoint info
- edsResources := make(map[resource.Type]map[string]refEndpoint)
- req := g.makeDiscoveryRequest(g.resourceNames, g.typeUrl)
- var handler discoveryResponseHandler
- switch g.typeUrl {
- case resource.ListenerType:
- handler = func(any2 []*anypb.Any) {
- for _, res := range any2 {
- l := listenerpb.Listener{}
- if err := res.UnmarshalTo(&l); err != nil {
- logger.Warnf("can not decode source %s
, %v", res.TypeUrl, res)
- continue
- }
+ if g.typeUrl != resource.ClusterType {
+ close(output)
+ return errors.Errorf("standard ADS currently supports CDS/EDS,
got %s", g.typeUrl)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ if g.exitCh != nil {
+ go func() {
+ select {
+ case <-g.exitCh:
+ cancel()
+ case <-ctx.Done():
}
+ }()
+ }
+ states := map[string]*adsResourceState{
+ resource.ClusterType: {resourceNames: append([]string(nil),
g.resourceNames...)},
+ resource.EndpointType: {},
+ }
+ references := make(map[string]refEndpoint)
+ stream, err := g.connectADS(ctx, states, references)
+ if err != nil {
+ cancel()
+ close(output)
+ if errors.Is(err, context.Canceled) {
+ return nil
}
- case resource.ClusterType:
- handler = func(any2 []*anypb.Any) {
- // only one goroutine handle response, no need to lock
for local var.
- for _, res := range any2 {
- logger.Infof("new resource found %s",
res.TypeUrl)
- c := clusterpb.Cluster{}
- if err := res.UnmarshalTo(&c); err != nil {
- logger.Warnf("can not decode source %s
, %v", res.TypeUrl, res)
- continue
- }
- //needn't lock edsResources
- g.getClusterResourceReference(&c, edsResources)
- }
+ return err
+ }
- pendingResourceNames := make([]string, 0)
- clusterRefEndpoints :=
edsResources[resource.ClusterType]
- // list all pending pendingResourceNames
- for name, b := range clusterRefEndpoints {
- if b.IsPending {
- pendingResourceNames =
append(pendingResourceNames, name)
- }
+ go func() {
+ defer cancel()
+ defer close(output)
+ for {
+ if err := g.consumeADSStream(ctx, stream, states,
references, output); err != nil && !errors.Is(err, context.Canceled) {
+ logger.Errorf("ADS stream closed: %v", err)
}
+ if ctx.Err() != nil {
+ return
+ }
+ stream, err = g.connectADS(ctx, states, references)
+ if err != nil {
+ return
+ }
+ }
+ }()
+ return nil
+}
- // do not block, watch new resource at another goroutine
- err := g.runEndpointReferences(pendingResourceNames,
func(any2 []*anypb.Any) {
- // run on another goroutine
- extCluster := xdsmodel.PixiuExtensionClusters{
- Clusters: []*xdsmodel.Cluster{
- {
- Name: "",
- TypeStr:
constant.ClusterType,
- Type: 0,
- EdsClusterConfig: nil,
- LbStr: "",
- Lb: 0,
- HealthChecks: nil,
- Endpoints:
make([]*xdsmodel.Endpoint, 0, len(any2)),
- },
- },
- }
-
- for _, one := range any2 {
- l := endpointpb.ClusterLoadAssignment{}
- if err := one.UnmarshalTo(&l); err !=
nil {
- logger.Warnf("unmarshal error",
err)
- continue
- }
- c :=
clusterRefEndpoints[l.ClusterName].RawProto.(*clusterpb.Cluster)
- extCluster.Clusters[0].Name =
g.readServiceNameOfCluster(c)
-
- for _, ep := range l.Endpoints {
- address :=
ep.LbEndpoints[0].GetEndpoint().GetAddress().GetSocketAddress()
-
extCluster.Clusters[0].Endpoints = append(extCluster.Clusters[0].Endpoints,
&xdsmodel.Endpoint{
- Id: "",
- Name: "",
- Address:
&xdsmodel.SocketAddress{
- Address:
address.Address,
- Port:
int64(address.GetPortValue()),
- },
- Metadata: nil,
- })
- }
+func (g *AggGrpcApiClient) connectADS(ctx context.Context, states
map[string]*adsResourceState, references map[string]refEndpoint)
(discoverypb.AggregatedDiscoveryService_StreamAggregatedResourcesClient, error)
{
+ for {
+ stream, err := g.xDSAggClient.StreamAggregatedResources(ctx)
+ if err == nil {
+ if err =
stream.Send(g.makeADSRequest(resource.ClusterType,
states[resource.ClusterType], "", nil)); err == nil {
+ if len(references) == 0 {
+ return stream, nil
}
-
- //make output
- output <- &DeltaResources{
- NewResources: []*ProtoAny{
- {
- typeConfig:
&envoyconfigcorev3.TypedExtensionConfig{
- Name:
"cluster", //todo cluster name
- TypedConfig:
func() *anypb.Any { //make any.Any from extCluster
- a, err
:= anypb.New(&extCluster)
- if err
!= nil {
-
logger.Warnf("can not make anypb.Any %v", err)
-
return nil
- }
- return a
- }(),
- },
- },
- },
- RemovedResource: nil,
+ states[resource.EndpointType].resourceNames =
referenceNames(references)
+ if err =
stream.Send(g.makeADSRequest(resource.EndpointType,
states[resource.EndpointType], "", nil)); err == nil {
+ return stream, nil
}
- })
- if err != nil { //todo retry
- logger.Errorf("can not run reference request
%v", err)
}
+ _ = stream.CloseSend()
+ }
+ logger.Errorf("can not start ADS stream; retrying in 1s: %v",
err)
+ select {
+ case <-time.After(time.Second):
+ case <-ctx.Done():
+ return nil, ctx.Err()
}
-
- default:
- return errors.Errorf("nedd listenerType of clusterType but get
%s", g.typeUrl)
- }
-
- if err := g.runDelta(req, handler); err != nil {
- return errors.WithMessagef(err, "start run %s failed",
req.TypeUrl)
}
-
- return nil
}
-// readServiceNameOfCluster get service name of k8s
-func (g *AggGrpcApiClient) readServiceNameOfCluster(c *clusterpb.Cluster)
string {
- if c.Metadata == nil {
- return ""
- }
- return c.
- Metadata.
- FilterMetadata["istio"].
- Fields["services"].
- GetListValue().
- GetValues()[0].
- GetStructValue().
- Fields["name"].
- GetStringValue()
-}
+func (g *AggGrpcApiClient) consumeADSStream(ctx context.Context, stream
discoverypb.AggregatedDiscoveryService_StreamAggregatedResourcesClient, states
map[string]*adsResourceState, references map[string]refEndpoint, output chan<-
*DeltaResources) error {
+ for {
+ resp, err := stream.Recv()
+ if err != nil {
+ return err
+ }
+ if resp == nil {
+ return errors.New("ADS response is nil")
+ }
+ switch resp.TypeUrl {
+ case resource.ClusterType:
+ newReferences, applyErr :=
g.decodeCDSReferences(resp.Resources)
+ if applyErr == nil && len(newReferences) == 0 {
+ applyErr = publishADSClusters(ctx, output,
&xdsmodel.PixiuExtensionClusters{})
+ }
+ if applyErr == nil {
+ clear(references)
+ for name, ref := range newReferences {
+ references[name] = ref
+ }
+ states[resource.ClusterType].versionInfo =
resp.VersionInfo
Review Comment:
**在非空 CDS 更新中同步删除已移除的运行时集群**
只有 newReferences 完全为空时 CDS 分支才向运行时发布变更。若已有 A/B，控制面下发只保留 B 的 CDS，代码虽然删除 A
的本地引用并 ACK 新版本，但 ClusterManager 仍保留 A。B 的 EDS 未变化时，控制面无需对这次退订重发 EDS，A
因此可一直继续承接流量。应在接收 CDS 时对已接受的运行时集合执行删除，不能把删除生效依赖于未来 EDS 响应。
##########
pkg/server/listener_manager.go:
##########
@@ -164,6 +183,268 @@ func (lm *ListenerManager) UpdateListener(m
*model.Listener) error {
return nil
}
+// UpsertXDSListener adds or updates a listener owned by xDS. An xDS resource
+// is not allowed to replace a listener created from static configuration.
+func (lm *ListenerManager) UpsertXDSListener(m *model.Listener) error {
+ if m == nil {
+ return errors.New("xDS listener config is nil")
+ }
+
+ listenerKey := resolveListenerName(m)
+ lm.rwLock.Lock()
+ if active, exists := lm.activeListenerService[listenerKey]; exists {
+ if _, owned := lm.xdsManaged[listenerKey]; !owned {
+ lm.rwLock.Unlock()
+ return errors.Errorf("xDS listener %q conflicts with a
non-xDS listener", listenerKey)
+ }
+ logger.Infof("Update xDS Listener %s (key: %s)", m.Name,
listenerKey)
+ if err := active.Refresh(*m); err != nil {
+ lm.rwLock.Unlock()
+ return errors.Wrapf(err, "refresh xDS listener %q",
listenerKey)
+ }
+ active.config = m
+ lm.rwLock.Unlock()
+ return nil
+ }
+
+ gate := lm.listenerUpdateGateLocked()
+ ls, err := listener.CreateListenerServiceWithUpdateGate(m,
lm.bootstrap, gate)
+ if err != nil {
+ lm.rwLock.Unlock()
+ return err
+ }
+ setListenerActive(ls, false)
+ gate.Lock()
+ if err := startListenerServiceSafely(ls); err != nil {
+ gate.Unlock()
+ lm.rwLock.Unlock()
+ _ = closeListenerServiceSafely(ls)
+ return errors.Wrapf(err, "start xDS listener %q", listenerKey)
+ }
+ lm.activeListenerService[listenerKey] = &wrapListenerService{
+ config: m,
+ ListenerService: ls,
+ }
+ lm.xdsManaged[listenerKey] = struct{}{}
+ setListenerActive(ls, true)
+ gate.Unlock()
+ lm.rwLock.Unlock()
+
+ logger.Infof("Add xDS Listener %s (key: %s)", m.Name, listenerKey)
+ return nil
+}
+
+// ReplaceXDSListeners replaces the complete xDS-owned listener set as one
+// transaction. Every filter chain is fully prepared first; new sockets must
+// bind successfully before the shared request gate publishes all changes.
+// Rejected responses therefore leave the last-good listener set running.
+func (lm *ListenerManager) ReplaceXDSListeners(listeners []*model.Listener)
error {
+ lm.rwLock.Lock()
+ gate := lm.listenerUpdateGateLocked()
+
+ newManaged := make(map[string]struct{}, len(listeners))
+ staged := make(map[string]*wrapListenerService)
+ type refreshTarget struct {
+ key string
+ active *wrapListenerService
+ transactional transactionalListenerService
+ next *model.Listener
+ prepared *listener.PreparedUpdate
+ }
+ refreshes := make([]refreshTarget, 0, len(listeners))
+ closePrepared := func() {
+ for _, target := range refreshes {
+ if err :=
listener.ClosePreparedUpdate(target.prepared); err != nil {
+ logger.Warnf("close rejected xDS listener %s
filter chain: %v", target.key, err)
+ }
+ }
+ }
+
+ for _, listenerConfig := range listeners {
+ if listenerConfig == nil {
+ lm.rwLock.Unlock()
+ return errors.New("xDS listener config is nil")
+ }
+ key := resolveListenerName(listenerConfig)
+ if _, duplicate := newManaged[key]; duplicate {
+ lm.rwLock.Unlock()
+ return errors.Errorf("duplicate xDS listener %q", key)
+ }
+ newManaged[key] = struct{}{}
+
+ if active := lm.activeListenerService[key]; active != nil {
+ if _, owned := lm.xdsManaged[key]; !owned {
+ lm.rwLock.Unlock()
+ return errors.Errorf("xDS listener %q conflicts
with a non-xDS listener", key)
+ }
+ transactional, prepared, err :=
prepareListenerRefreshSafely(active.ListenerService, *listenerConfig)
+ if err != nil {
+ closePrepared()
+ lm.rwLock.Unlock()
+ return errors.Wrapf(err, "prepare xDS listener
%q", key)
Review Comment:
**刷新失败时关闭此前暂存的新监听器**
批次中先为新增监听器成功创建 service，随后已有监听器 PrepareRefresh 失败时，此分支只关闭 refreshes，遗漏
staged。此前创建的网络过滤链及其 HTTP 过滤器工厂永远没有 Close；例如 grpcproxy
工厂构造时注册的端点回调也不会注销。重复提交失败配置会累积这些资源。静态冲突、重复 key 等早退也存在同类遗漏。应让所有提交前错误出口统一释放 staged
和 prepared。
##########
pkg/config/xds/apiclient/grpc_envoy.go:
##########
@@ -133,246 +143,258 @@ func (g *AggGrpcApiClient) Delta() (chan
*DeltaResources, error) {
}
type refEndpoint struct {
- IsPending bool
- RawProto proto.Message
+ Clusters []*clusterpb.Cluster
}
func (g *AggGrpcApiClient) pipeline(output chan *DeltaResources) error {
- // all endpoint refer cluster or listener. map[type]map[resource
name]endpoint info
- edsResources := make(map[resource.Type]map[string]refEndpoint)
- req := g.makeDiscoveryRequest(g.resourceNames, g.typeUrl)
- var handler discoveryResponseHandler
- switch g.typeUrl {
- case resource.ListenerType:
- handler = func(any2 []*anypb.Any) {
- for _, res := range any2 {
- l := listenerpb.Listener{}
- if err := res.UnmarshalTo(&l); err != nil {
- logger.Warnf("can not decode source %s
, %v", res.TypeUrl, res)
- continue
- }
+ if g.typeUrl != resource.ClusterType {
+ close(output)
+ return errors.Errorf("standard ADS currently supports CDS/EDS,
got %s", g.typeUrl)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ if g.exitCh != nil {
+ go func() {
+ select {
+ case <-g.exitCh:
+ cancel()
+ case <-ctx.Done():
}
+ }()
+ }
+ states := map[string]*adsResourceState{
+ resource.ClusterType: {resourceNames: append([]string(nil),
g.resourceNames...)},
+ resource.EndpointType: {},
+ }
+ references := make(map[string]refEndpoint)
+ stream, err := g.connectADS(ctx, states, references)
+ if err != nil {
+ cancel()
+ close(output)
+ if errors.Is(err, context.Canceled) {
+ return nil
}
- case resource.ClusterType:
- handler = func(any2 []*anypb.Any) {
- // only one goroutine handle response, no need to lock
for local var.
- for _, res := range any2 {
- logger.Infof("new resource found %s",
res.TypeUrl)
- c := clusterpb.Cluster{}
- if err := res.UnmarshalTo(&c); err != nil {
- logger.Warnf("can not decode source %s
, %v", res.TypeUrl, res)
- continue
- }
- //needn't lock edsResources
- g.getClusterResourceReference(&c, edsResources)
- }
+ return err
+ }
- pendingResourceNames := make([]string, 0)
- clusterRefEndpoints :=
edsResources[resource.ClusterType]
- // list all pending pendingResourceNames
- for name, b := range clusterRefEndpoints {
- if b.IsPending {
- pendingResourceNames =
append(pendingResourceNames, name)
- }
+ go func() {
+ defer cancel()
+ defer close(output)
+ for {
+ if err := g.consumeADSStream(ctx, stream, states,
references, output); err != nil && !errors.Is(err, context.Canceled) {
+ logger.Errorf("ADS stream closed: %v", err)
}
+ if ctx.Err() != nil {
+ return
+ }
+ stream, err = g.connectADS(ctx, states, references)
+ if err != nil {
+ return
+ }
+ }
+ }()
+ return nil
+}
- // do not block, watch new resource at another goroutine
- err := g.runEndpointReferences(pendingResourceNames,
func(any2 []*anypb.Any) {
- // run on another goroutine
- extCluster := xdsmodel.PixiuExtensionClusters{
- Clusters: []*xdsmodel.Cluster{
- {
- Name: "",
- TypeStr:
constant.ClusterType,
- Type: 0,
- EdsClusterConfig: nil,
- LbStr: "",
- Lb: 0,
- HealthChecks: nil,
- Endpoints:
make([]*xdsmodel.Endpoint, 0, len(any2)),
- },
- },
- }
-
- for _, one := range any2 {
- l := endpointpb.ClusterLoadAssignment{}
- if err := one.UnmarshalTo(&l); err !=
nil {
- logger.Warnf("unmarshal error",
err)
- continue
- }
- c :=
clusterRefEndpoints[l.ClusterName].RawProto.(*clusterpb.Cluster)
- extCluster.Clusters[0].Name =
g.readServiceNameOfCluster(c)
-
- for _, ep := range l.Endpoints {
- address :=
ep.LbEndpoints[0].GetEndpoint().GetAddress().GetSocketAddress()
-
extCluster.Clusters[0].Endpoints = append(extCluster.Clusters[0].Endpoints,
&xdsmodel.Endpoint{
- Id: "",
- Name: "",
- Address:
&xdsmodel.SocketAddress{
- Address:
address.Address,
- Port:
int64(address.GetPortValue()),
- },
- Metadata: nil,
- })
- }
+func (g *AggGrpcApiClient) connectADS(ctx context.Context, states
map[string]*adsResourceState, references map[string]refEndpoint)
(discoverypb.AggregatedDiscoveryService_StreamAggregatedResourcesClient, error)
{
+ for {
+ stream, err := g.xDSAggClient.StreamAggregatedResources(ctx)
+ if err == nil {
+ if err =
stream.Send(g.makeADSRequest(resource.ClusterType,
states[resource.ClusterType], "", nil)); err == nil {
+ if len(references) == 0 {
+ return stream, nil
}
-
- //make output
- output <- &DeltaResources{
- NewResources: []*ProtoAny{
- {
- typeConfig:
&envoyconfigcorev3.TypedExtensionConfig{
- Name:
"cluster", //todo cluster name
- TypedConfig:
func() *anypb.Any { //make any.Any from extCluster
- a, err
:= anypb.New(&extCluster)
- if err
!= nil {
-
logger.Warnf("can not make anypb.Any %v", err)
-
return nil
- }
- return a
- }(),
- },
- },
- },
- RemovedResource: nil,
+ states[resource.EndpointType].resourceNames =
referenceNames(references)
+ if err =
stream.Send(g.makeADSRequest(resource.EndpointType,
states[resource.EndpointType], "", nil)); err == nil {
+ return stream, nil
}
- })
- if err != nil { //todo retry
- logger.Errorf("can not run reference request
%v", err)
}
+ _ = stream.CloseSend()
+ }
+ logger.Errorf("can not start ADS stream; retrying in 1s: %v",
err)
+ select {
+ case <-time.After(time.Second):
+ case <-ctx.Done():
+ return nil, ctx.Err()
}
-
- default:
- return errors.Errorf("nedd listenerType of clusterType but get
%s", g.typeUrl)
- }
-
- if err := g.runDelta(req, handler); err != nil {
- return errors.WithMessagef(err, "start run %s failed",
req.TypeUrl)
}
-
- return nil
}
-// readServiceNameOfCluster get service name of k8s
-func (g *AggGrpcApiClient) readServiceNameOfCluster(c *clusterpb.Cluster)
string {
- if c.Metadata == nil {
- return ""
- }
- return c.
- Metadata.
- FilterMetadata["istio"].
- Fields["services"].
- GetListValue().
- GetValues()[0].
- GetStructValue().
- Fields["name"].
- GetStringValue()
-}
+func (g *AggGrpcApiClient) consumeADSStream(ctx context.Context, stream
discoverypb.AggregatedDiscoveryService_StreamAggregatedResourcesClient, states
map[string]*adsResourceState, references map[string]refEndpoint, output chan<-
*DeltaResources) error {
+ for {
+ resp, err := stream.Recv()
+ if err != nil {
+ return err
+ }
+ if resp == nil {
+ return errors.New("ADS response is nil")
+ }
+ switch resp.TypeUrl {
+ case resource.ClusterType:
+ newReferences, applyErr :=
g.decodeCDSReferences(resp.Resources)
+ if applyErr == nil && len(newReferences) == 0 {
+ applyErr = publishADSClusters(ctx, output,
&xdsmodel.PixiuExtensionClusters{})
+ }
+ if applyErr == nil {
+ clear(references)
+ for name, ref := range newReferences {
+ references[name] = ref
+ }
+ states[resource.ClusterType].versionInfo =
resp.VersionInfo
+ }
+ if err :=
stream.Send(g.makeADSRequest(resource.ClusterType,
states[resource.ClusterType], resp.Nonce, applyErr)); err != nil {
+ return err
+ }
+ if applyErr != nil || len(references) == 0 {
+ continue
+ }
+ states[resource.EndpointType].resourceNames =
referenceNames(references)
+ if err :=
stream.Send(g.makeADSRequest(resource.EndpointType,
states[resource.EndpointType], "", nil)); err != nil {
+ return err
+ }
-// request EDS for the allResourceNames
-func (g *AggGrpcApiClient) runEndpointReferences(allResourceNames []string,
- output discoveryResponseHandler) (err error) {
+ case resource.EndpointType:
+ clusters, applyErr :=
decodeAndConvertEDS(resp.Resources, references)
+ if applyErr == nil {
+ applyErr = publishADSClusters(ctx, output,
clusters)
Review Comment:
**保留 EDS 响应中未出现的集群**
标准 SotW EDS 允许只返回发生变化的 ClusterLoadAssignment，空响应也是无操作。此处只转换当前响应，然后经
CdsManager.setupCluster 调用 ReplaceXDSClusters 替换全部 xDS 集群。先收到 A/B，再收到只含 A 的合法
EDS 更新时，B 会被删除，其路由随后无法选择端点；空 EDS 响应会清空全部动态集群。应按 EDS 资源名缓存已接受的
assignment，合并本次更新后构造完整集合；由 CDS 引用变化决定删除。
协议依据：[xDS — Grouping Resources into Responses / Deleting
Resources](https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol.html#grouping-resources-into-responses)。
##########
pkg/server/cluster_manager.go:
##########
@@ -158,6 +164,121 @@ func (cm *ClusterManager) UpdateCluster(new
*model.ClusterConfig) {
cm.notifyEndpointChanges(endpointStateChanges(old, newSnapshot,
version))
}
+// UpsertXDSCluster adds or updates a cluster owned by xDS. An xDS resource is
+// not allowed to overwrite a cluster created from static configuration.
+func (cm *ClusterManager) UpsertXDSCluster(c *model.ClusterConfig) error {
+ if c == nil || c.Name == "" {
+ return fmt.Errorf("xDS cluster must have a name")
+ }
+
+ cm.rw.Lock()
+ defer cm.rw.Unlock()
+
+ _, owned := cm.xdsManaged[c.Name]
+ if cm.store.HasCluster(c.Name) && !owned {
+ return fmt.Errorf("xDS cluster %q conflicts with a non-xDS
cluster", c.Name)
+ }
+
+ cm.store.IncreaseVersion()
+ if owned {
+ cm.store.UpdateCluster(c)
+ } else {
+ cm.store.AddCluster(c)
+ cm.xdsManaged[c.Name] = struct{}{}
+ }
+ return nil
+}
+
+// ReplaceXDSClusters atomically replaces the complete xDS-owned cluster set.
+// Validation and runtime construction finish before the new store is
+// published, so a rejected xDS response cannot partially mutate live state.
+func (cm *ClusterManager) ReplaceXDSClusters(clusters []*model.ClusterConfig)
error {
+ cm.rw.Lock()
+ candidate := &ClusterStore{
+ Version: cm.store.Version,
+ clustersMap: make(map[string]*cluster.Cluster,
len(cm.store.clustersMap)+len(clusters)),
+ }
+
+ staticNames := make(map[string]struct{}, len(cm.store.Config))
+ for _, clusterConfig := range cm.store.Config {
+ if clusterConfig == nil {
+ continue
+ }
+ if _, owned := cm.xdsManaged[clusterConfig.Name]; !owned {
+ staticNames[clusterConfig.Name] = struct{}{}
+ candidate.Config = append(candidate.Config,
clusterConfig)
+ if runtimeCluster :=
cm.store.clustersMap[clusterConfig.Name]; runtimeCluster != nil {
+ candidate.clustersMap[clusterConfig.Name] =
runtimeCluster
+ }
+ }
+ }
+
+ newManaged := make(map[string]struct{}, len(clusters))
+ for _, clusterConfig := range clusters {
+ if clusterConfig == nil || clusterConfig.Name == "" {
+ cm.rw.Unlock()
+ return fmt.Errorf("xDS cluster must have a name")
+ }
+ if _, duplicate := newManaged[clusterConfig.Name]; duplicate {
+ cm.rw.Unlock()
+ return fmt.Errorf("duplicate xDS cluster %q",
clusterConfig.Name)
+ }
+ if _, conflict := staticNames[clusterConfig.Name]; conflict {
+ cm.rw.Unlock()
+ return fmt.Errorf("xDS cluster %q conflicts with a
non-xDS cluster", clusterConfig.Name)
+ }
+ newManaged[clusterConfig.Name] = struct{}{}
+ }
+
+ for _, clusterConfig := range clusters {
+ candidate.prepareClusterConfig(clusterConfig)
+ candidate.Config = append(candidate.Config, clusterConfig)
+ candidate.replaceClusterRuntimeWithSnapshot(
+ clusterConfig.Name,
+ clusterConfig,
+ snapshotForRuntimeReplacement(cm.store,
clusterConfig.Name),
+ )
+ }
+ candidate.IncreaseVersion()
+ replacedClusters := make([]*cluster.Cluster, 0, len(cm.xdsManaged))
+ for name := range cm.xdsManaged {
+ if runtimeCluster := cm.store.clustersMap[name]; runtimeCluster
!= nil {
+ replacedClusters = append(replacedClusters,
runtimeCluster)
+ }
+ }
+ cm.store = candidate
+ cm.xdsManaged = newManaged
+ cm.rw.Unlock()
+
+ stopClusters(replacedClusters)
+ return nil
Review Comment:
**替换 xDS 集群后发送端点生命周期通知**
ReplaceXDSClusters 是 CDS/EDS 应用的生产路径，但发布新 store 后只调用 stopClusters，未像原有
UpdateCluster/RemoveCluster 路径一样发送 notifyEndpointChanges。HTTP grpcproxy 依赖
AddEndpointStateHandler 关闭删除端点的缓存 ClientConn 并清理 descriptor。仅更新/删除 xDS cluster
且监听器不变时，已移除地址的连接和描述符会留在长期存活的过滤器工厂内。应在锁内取得旧/新端点差异及事件版本，在解锁后发送通知。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
