Alanxtl commented on code in PR #1042:
URL: https://github.com/apache/dubbo-go-pixiu/pull/1042#discussion_r4043781508
##########
pkg/config/xds/apiclient/grpc.go:
##########
@@ -145,101 +145,141 @@ func (g *GrpcExtensionApiClient) Delta() (chan
*DeltaResources, error) {
}
func (g *GrpcExtensionApiClient) runDelta(output chan<- *DeltaResources) error
{
- var delta
extensionpb.ExtensionConfigDiscoveryService_DeltaExtensionConfigsClient
- var cancel context.CancelFunc
- var xState xdsState
- backoff := func() {
+ ctx, cancel := context.WithCancel(context.Background())
+ if g.exitCh != nil {
+ go func() {
+ select {
+ case <-g.exitCh:
+ cancel()
+ case <-ctx.Done():
+ }
+ }()
+ }
+
+ xState := &xdsState{deltaVersion: make(map[string]string)}
+ delta, err := g.connectDelta(ctx, xState)
+ if err != nil {
+ cancel()
+ close(output)
+ if stderr.Is(err, context.Canceled) {
+ return nil
+ }
+ return err
+ }
+
+ go func() {
+ defer cancel()
+ defer close(output)
for {
- //back off
+ if err := g.consumeDeltaStream(ctx, delta, xState,
output); err != nil && !stderr.Is(err, context.Canceled) {
+ logger.Errorf("delta extension config stream
closed: %v", err)
+ }
+ if ctx.Err() != nil {
+ return
+ }
var err error
- var ctx context.Context // context to sync exitCh
- ctx, cancel = context.WithCancel(context.TODO())
- delta, err = g.sendInitDeltaRequest(ctx, &xState)
+ delta, err = g.connectDelta(ctx, xState)
if err != nil {
- logger.Error("can not receive delta discovery
request, will back off 1 sec later", err)
- select {
- case <-time.After(1 * time.Second):
- case <-g.exitCh:
- logger.Infof("get close single.")
- return
- }
-
- continue //backoff
+ return
}
- return //success
}
- }
+ }()
+ return nil
+}
- backoff()
- if delta == nil { // delta instance not created because exitCh
- return nil
+func (g *GrpcExtensionApiClient) connectDelta(ctx context.Context, xState
*xdsState)
(extensionpb.ExtensionConfigDiscoveryService_DeltaExtensionConfigsClient,
error) {
+ for {
+ delta, err := g.sendInitDeltaRequest(ctx, xState)
+ if err == nil {
+ return delta, nil
+ }
+ logger.Errorf("can not start delta extension config stream;
retrying in 1s: %v", err)
+ select {
+ case <-time.After(time.Second):
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ }
}
- go func() {
- //waiting exitCh close
- for range g.exitCh {
+}
+
+func (g *GrpcExtensionApiClient) consumeDeltaStream(ctx context.Context, delta
extensionpb.ExtensionConfigDiscoveryService_DeltaExtensionConfigsClient, xState
*xdsState, output chan<- *DeltaResources) error {
+ for {
+ resp, err := delta.Recv()
+ if err != nil {
+ return err
}
- cancel()
- }()
- //get message
- go func() {
- for { // delta response backoff.
- for { //loop consume recv data form xds
server(sendInitDeltaRequest)
- resp, err := delta.Recv()
- if err != nil { //todo backoff retry
- logger.Error("can not receive delta
discovery request", err)
- break
- }
- g.handleDeltaResponse(resp, &xState, output)
-
- err = g.subscribeOnGoingChang(delta, &xState)
- if err != nil {
- logger.Error("can not recv delta
discovery request", err)
- break
- }
+
+ resources, decodeErr := g.handleDeltaResponse(resp)
+ if decodeErr != nil {
+ if err := g.respondToDelta(delta, resp.Nonce,
decodeErr); err != nil {
+ return err
}
- backoff()
+ continue
}
- }()
- return nil
-}
+ select {
+ case output <- resources:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
-func (g *GrpcExtensionApiClient) handleDeltaResponse(resp
*discoverypb.DeltaDiscoveryResponse, xState *xdsState, output chan<-
*DeltaResources) {
- // save the xds state
- xState.deltaVersion = make(map[string]string, 1)
- xState.nonce = resp.Nonce
+ var applyErr error
+ select {
+ case applyErr = <-resources.applyResult:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ if applyErr == nil {
+ acceptDeltaResources(xState, resources)
+ }
+ if err := g.respondToDelta(delta, resp.Nonce, applyErr); err !=
nil {
+ return err
+ }
+ }
+}
- resources := &DeltaResources{
- NewResources: make([]*ProtoAny, 0, 1),
- RemovedResource: make([]string, 0, 1),
+func (g *GrpcExtensionApiClient) handleDeltaResponse(resp
*discoverypb.DeltaDiscoveryResponse) (*DeltaResources, error) {
+ if resp == nil {
+ return nil, errors.New("delta discovery response is nil")
}
+ resources := newDeltaResources()
logger.Infof("get xDS message nonce, %s", resp.Nonce)
for _, res := range resp.RemovedResources {
- logger.Infof("remove resource found ", res)
- resources.RemovedResource = append(resources.RemovedResource,
res)
+ logger.Infof("remove resource found %s", res)
+ resources.RemovedResources = append(resources.RemovedResources,
res)
}
for _, res := range resp.Resources {
logger.Infof("new resource found %s version=%s", res.Name,
res.Version)
- xState.deltaVersion[res.Name] = res.Version
elems, err := g.decodeSource(res.Resource)
if err != nil {
- logger.Infof("can not decode source %s version=%s",
res.Name, res.Version, err)
+ return nil, errors.Wrapf(err, "decode resource %s
version=%s", res.Name, res.Version)
}
resources.NewResources = append(resources.NewResources, elems)
+ resources.versions[res.Name] = res.Version
}
Review Comment:
handleDeltaResponse 虽然检查了 resp == nil，但调用方随后仍会访问 resp.Nonce；同时循环直接访问
res.Name，没有判断 res == nil。此外也没有拒绝空名或重复的 resource name，可能导致版本覆盖或重复资源被应用。Envoy
协议要求重复资源名应
NACK。[官方协议说明](https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol.html#duplicate-resource-names)
##########
admin/xds/snapshot.go:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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 xds
+
+import (
+ "fmt"
+ "strconv"
+ "strings"
+)
+
+import (
+ corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+ "github.com/envoyproxy/go-control-plane/pkg/cache/types"
+ cache "github.com/envoyproxy/go-control-plane/pkg/cache/v3"
+ "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+
+ "google.golang.org/protobuf/types/known/anypb"
+ "google.golang.org/protobuf/types/known/structpb"
+)
+
+import (
+ adminconfig "github.com/apache/dubbo-go-pixiu/admin/config"
+ "github.com/apache/dubbo-go-pixiu/admin/logic"
+ "github.com/apache/dubbo-go-pixiu/pkg/common/constant"
+ "github.com/apache/dubbo-go-pixiu/pkg/config"
+ xdsmodel "github.com/apache/dubbo-go-pixiu/pkg/config/xds/model"
+ "github.com/apache/dubbo-go-pixiu/pkg/model"
+)
+
+// ResourceLoader reads one logical Admin resource view for snapshot building.
+// A successful empty slice means that the resource type is intentionally
+// empty; an error means no candidate snapshot may be published.
+type ResourceLoader interface {
+ LoadListeners() ([]config.Listener, error)
+ LoadClusters() ([]config.Cluster, error)
+}
+
+// LogicResourceLoader reads the published listener and cluster configuration
+// through the existing Admin logic layer.
+type LogicResourceLoader struct{}
+
+func (LogicResourceLoader) LoadListeners() ([]config.Listener, error) {
+ if adminconfig.Client == nil {
+ return nil, fmt.Errorf("admin etcd client is not initialized")
+ }
+ return logic.BizGetListeners()
+}
+
+func (LogicResourceLoader) LoadClusters() ([]config.Cluster, error) {
+ if adminconfig.Client == nil {
+ return nil, fmt.Errorf("admin etcd client is not initialized")
+ }
+ return logic.BizGetClusters()
+}
+
+// SnapshotBuildResult carries the validated snapshot and the metadata needed
+// to update Admin diagnostics after successful cache publication.
+type SnapshotBuildResult struct {
+ Snapshot *cache.Snapshot
+ Version string
+ ListenerCount int
+ ClusterCount int
+}
+
+type SnapshotBuilder struct {
+ loader ResourceLoader
+}
+
+func NewSnapshotBuilder(loader ResourceLoader) *SnapshotBuilder {
+ return &SnapshotBuilder{loader: loader}
+}
+
+// Build loads, converts, packs, and validates a candidate snapshot. It does
+// not mutate the SnapshotCache, version state, or last-good diagnostics.
+func (b *SnapshotBuilder) Build(version string) (*SnapshotBuildResult, error) {
+ if b == nil || b.loader == nil {
+ return nil, fmt.Errorf("xDS snapshot resource loader is not
configured")
+ }
+ if version == "" {
+ return nil, fmt.Errorf("xDS snapshot version is empty")
+ }
+
+ listeners, err := b.loader.LoadListeners()
+ if err != nil {
+ return nil, fmt.Errorf("load listeners: %w", err)
+ }
+ clusters, err := b.loader.LoadClusters()
+ if err != nil {
+ return nil, fmt.Errorf("load clusters: %w", err)
+ }
+ if err := validateListeners(listeners); err != nil {
+ return nil, fmt.Errorf("validate listeners: %w", err)
+ }
+ if err := validateClusters(clusters); err != nil {
+ return nil, fmt.Errorf("validate clusters: %w", err)
+ }
+
+ listenerConfig, err := makeListeners(listeners)
+ if err != nil {
+ return nil, fmt.Errorf("convert listeners: %w", err)
+ }
+ clusterConfig := makeClusters(clusters)
+
+ listenerResource, err := anypb.New(listenerConfig)
+ if err != nil {
+ return nil, fmt.Errorf("pack listener ExtensionConfig: %w", err)
+ }
+ clusterResource, err := anypb.New(clusterConfig)
+ if err != nil {
+ return nil, fmt.Errorf("pack cluster ExtensionConfig: %w", err)
+ }
+
+ snapshot, err := cache.NewSnapshot(version,
map[resource.Type][]types.Resource{
+ resource.ExtensionConfigType: {
+ &corev3.TypedExtensionConfig{
+ Name: constant.ClusterType,
+ TypedConfig: clusterResource,
+ },
+ &corev3.TypedExtensionConfig{
+ Name: constant.ListenerType,
+ TypedConfig: listenerResource,
+ },
+ },
+ })
+ if err != nil {
+ return nil, fmt.Errorf("construct xDS snapshot: %w", err)
+ }
+ if err := snapshot.Consistent(); err != nil {
+ return nil, fmt.Errorf("validate xDS snapshot: %w", err)
+ }
+
+ return &SnapshotBuildResult{
+ Snapshot: snapshot,
+ Version: version,
+ ListenerCount: len(listenerConfig.Listeners),
+ ClusterCount: len(clusterConfig.Clusters),
+ }, nil
+}
+
+func validateListeners(listeners []config.Listener) error {
+ names := make(map[string]struct{}, len(listeners))
+ addresses := make(map[string]struct{}, len(listeners))
+ for index, listener := range listeners {
+ name := strings.TrimSpace(listener.Name)
+ if name == "" {
+ return fmt.Errorf("listener %d has an empty name",
index)
+ }
+ if _, duplicate := names[name]; duplicate {
+ return fmt.Errorf("duplicate listener name %q", name)
+ }
+ names[name] = struct{}{}
+
+ address :=
strings.TrimSpace(listener.Address.SocketAddress.Address)
+ port := listener.Address.SocketAddress.Port
+ if address == "" || port <= 0 || port > 65535 {
+ return fmt.Errorf("listener %q has invalid socket
address %q:%d", name, address, port)
+ }
+ addressKey := address + ":" + strconv.Itoa(port)
+ if _, duplicate := addresses[addressKey]; duplicate {
+ return fmt.Errorf("duplicate listener socket address
%q", addressKey)
+ }
+ addresses[addressKey] = struct{}{}
+
+ for routeIndex, route := range listener.RouteConfig.Routes {
+ if strings.TrimSpace(route.Route.Cluster) == "" {
+ return fmt.Errorf("listener %q route %d has an
empty cluster", name, routeIndex)
+ }
+ }
+ }
+ return nil
+}
Review Comment:
例如 type: " Static " 会通过校验，但发布时仍携带空格，后续 CdsManager 的精确匹配会失败并 NACK。address 和
route 的 cluster 也有同样问题，可能导致监听绑定或路由失败。
建议要么在构建快照前规范化字段，要么发现原值与 trim 后值不同就直接拒绝。
--
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]
