This is an automated email from the ASF dual-hosted git repository.
mfordjody pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new 58913c3f Update the delta xDS framework (#923)
58913c3f is described below
commit 58913c3fa01cf47fb9a5a1c3d40a56d667de19d4
Author: mfordjody <[email protected]>
AuthorDate: Thu Jun 25 19:18:02 2026 +0800
Update the delta xDS framework (#923)
---
.../discovery/pkg/networking/grpcgen/delta_test.go | 94 ++++++++++
dubbod/discovery/pkg/networking/grpcgen/grpcgen.go | 61 +++++++
dubbod/discovery/pkg/xds/delta.go | 15 +-
dubbod/discovery/pkg/xds/delta_test.go | 199 +++++++++++++++++++++
dubbod/discovery/pkg/xds/xdsgen.go | 8 +
pkg/cni/config.go | 2 +-
pkg/cni/install.go | 16 +-
7 files changed, 386 insertions(+), 9 deletions(-)
diff --git a/dubbod/discovery/pkg/networking/grpcgen/delta_test.go
b/dubbod/discovery/pkg/networking/grpcgen/delta_test.go
new file mode 100644
index 00000000..f14db62c
--- /dev/null
+++ b/dubbod/discovery/pkg/networking/grpcgen/delta_test.go
@@ -0,0 +1,94 @@
+//
+// 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 grpcgen
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
+ "github.com/apache/dubbo-kubernetes/pkg/config/host"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+)
+
+func TestGenerateDeltasBuildsOnlyChangedServiceClusters(t *testing.T) {
+ nginxHost := "nginx.app.svc.cluster.local"
+ apiHost := "api.app.svc.cluster.local"
+ nginxCluster := model.BuildSubsetKey(model.TrafficDirectionOutbound,
"", host.Name(nginxHost), 80)
+ apiCluster := model.BuildSubsetKey(model.TrafficDirectionOutbound, "",
host.Name(apiHost), 80)
+ push := newRDSTestPushContext(t, nil, []*model.Service{
+ newRDSTestService("nginx", "app", nginxHost, 80),
+ newRDSTestService("api", "app", apiHost, 80),
+ })
+
+ resources, removed, details, usedDelta, err :=
(&GrpcConfigGenerator{}).GenerateDeltas(newMTLSTestProxy(), &model.PushRequest{
+ Push: push,
+ Full: true,
+ ConfigsUpdated: sets.New(model.ConfigKey{Kind: kind.Service,
Name: nginxHost, Namespace: "app"}),
+ }, &model.WatchedResource{
+ TypeUrl: v1.ClusterType,
+ ResourceNames: sets.New(nginxCluster, apiCluster),
+ })
+ if err != nil {
+ t.Fatalf("GenerateDeltas() failed: %v", err)
+ }
+ if !usedDelta || !details.Incremental {
+ t.Fatalf("usedDelta/details.Incremental = %v/%v, want
true/true", usedDelta, details.Incremental)
+ }
+ if got := resourceNamesForTest(resources); len(got) != 1 || got[0] !=
nginxCluster {
+ t.Fatalf("resources = %v, want [%s]", got, nginxCluster)
+ }
+ if len(removed) != 0 {
+ t.Fatalf("removed = %v, want empty", removed)
+ }
+}
+
+func TestGenerateDeltasRemovesClustersForDeletedService(t *testing.T) {
+ nginxHost := "nginx.app.svc.cluster.local"
+ nginxCluster := model.BuildSubsetKey(model.TrafficDirectionOutbound,
"", host.Name(nginxHost), 80)
+ push := newRDSTestPushContext(t, nil, nil)
+
+ resources, removed, _, usedDelta, err :=
(&GrpcConfigGenerator{}).GenerateDeltas(newMTLSTestProxy(), &model.PushRequest{
+ Push: push,
+ Full: true,
+ ConfigsUpdated: sets.New(model.ConfigKey{Kind: kind.Service,
Name: nginxHost, Namespace: "app"}),
+ }, &model.WatchedResource{
+ TypeUrl: v1.ClusterType,
+ ResourceNames: sets.New(nginxCluster),
+ })
+ if err != nil {
+ t.Fatalf("GenerateDeltas() failed: %v", err)
+ }
+ if !usedDelta {
+ t.Fatal("usedDelta = false, want true")
+ }
+ if len(resources) != 0 {
+ t.Fatalf("resources = %v, want empty",
resourceNamesForTest(resources))
+ }
+ if len(removed) != 1 || removed[0] != nginxCluster {
+ t.Fatalf("removed = %v, want [%s]", removed, nginxCluster)
+ }
+}
+
+func resourceNamesForTest(resources model.Resources) []string {
+ out := make([]string, 0, len(resources))
+ for _, resource := range resources {
+ out = append(out, resource.Name)
+ }
+ return out
+}
diff --git a/dubbod/discovery/pkg/networking/grpcgen/grpcgen.go
b/dubbod/discovery/pkg/networking/grpcgen/grpcgen.go
index 9b35f745..bbd44ad7 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/grpcgen.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/grpcgen.go
@@ -19,7 +19,9 @@ package grpcgen
import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
dubbolog "github.com/apache/dubbo-kubernetes/pkg/log"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
)
@@ -27,6 +29,8 @@ var log = dubbolog.RegisterScope("grpcgen", "xDS Generator
for Proxyless gRPC")
type GrpcConfigGenerator struct{}
+var _ model.XdsDeltaResourceGenerator = &GrpcConfigGenerator{}
+
func (g *GrpcConfigGenerator) Generate(proxy *model.Proxy, w
*model.WatchedResource, req *model.PushRequest) (model.Resources,
model.XdsLogDetails, error) {
// Extract requested resource names from WatchedResource
// If ResourceNames is empty (wildcard request), pass empty slice
@@ -52,6 +56,63 @@ func (g *GrpcConfigGenerator) Generate(proxy *model.Proxy, w
*model.WatchedResou
return nil, model.DefaultXdsLogDetails, nil
}
+func (g *GrpcConfigGenerator) GenerateDeltas(proxy *model.Proxy, req
*model.PushRequest, w *model.WatchedResource) (model.Resources,
model.DeletedResources, model.XdsLogDetails, bool, error) {
+ if w == nil {
+ return nil, nil, model.DefaultXdsLogDetails, false, nil
+ }
+ if w.TypeUrl != v1.ClusterType || !grpcCanUseDeltaClusters(req) {
+ resources, logDetails, err := g.Generate(proxy, w, req)
+ return resources, nil, logDetails, false, err
+ }
+
+ changedNames := grpcDeltaClusterNames(w, req)
+ if len(changedNames) == 0 {
+ return nil, nil, model.DefaultXdsLogDetails, true, nil
+ }
+ resources := g.BuildClusters(proxy, req.Push,
sets.SortedList(changedNames))
+
+ built := sets.NewWithLength[string](len(resources))
+ for _, resource := range resources {
+ if resource != nil {
+ built.Insert(resource.Name)
+ }
+ }
+ removed := changedNames.Copy()
+ for name := range built {
+ removed.Delete(name)
+ }
+
+ logDetails := model.DefaultXdsLogDetails
+ logDetails.Incremental = true
+ return resources, sets.SortedList(removed), logDetails, true, nil
+}
+
+func grpcCanUseDeltaClusters(req *model.PushRequest) bool {
+ if req == nil || req.Forced || len(req.ConfigsUpdated) == 0 {
+ return false
+ }
+ for cfg := range req.ConfigsUpdated {
+ if cfg.Kind != kind.Service || cfg.Name == "" {
+ return false
+ }
+ }
+ return true
+}
+
+func grpcDeltaClusterNames(w *model.WatchedResource, req *model.PushRequest)
sets.String {
+ updatedServices := sets.New[string]()
+ for cfg := range req.ConfigsUpdated {
+ updatedServices.Insert(cfg.Name)
+ }
+ changed := sets.New[string]()
+ for clusterName := range w.ResourceNames {
+ if
updatedServices.Contains(model.ParseSubsetKeyHostname(clusterName)) {
+ changed.Insert(clusterName)
+ }
+ }
+ return changed
+}
+
// buildCommonTLSContext creates a TLS context that matches gRPC xDS
expectations.
// - Uses certificate provider "default" for workload certs and root CA
// - Does not configure explicit SAN matches (left to future hardening)
diff --git a/dubbod/discovery/pkg/xds/delta.go
b/dubbod/discovery/pkg/xds/delta.go
index 8647c770..bb305e6b 100644
--- a/dubbod/discovery/pkg/xds/delta.go
+++ b/dubbod/discovery/pkg/xds/delta.go
@@ -20,6 +20,7 @@ import (
"errors"
dubbolog "github.com/apache/dubbo-kubernetes/pkg/log"
"strings"
+ "time"
dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/grpc"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
@@ -335,7 +336,19 @@ func (conn *Connection) sendDelta(res
*discovery.DeltaDiscoveryResponse, newReso
return conn.deltaStream.Send(res)
}
err := sendResonse()
- if status.Convert(err).Code() == codes.DeadlineExceeded {
+ if err == nil {
+ conn.proxy.UpdateWatchedResource(res.TypeUrl, func(wr
*model.WatchedResource) *model.WatchedResource {
+ if wr == nil {
+ wr = &model.WatchedResource{TypeUrl:
res.TypeUrl}
+ }
+ if newResourceNames != nil {
+ wr.ResourceNames = newResourceNames
+ }
+ wr.NonceSent = res.Nonce
+ wr.LastSendTime = time.Now()
+ return wr
+ })
+ } else if status.Convert(err).Code() == codes.DeadlineExceeded {
deltaLog.Infof("Timeout writing %s: %v", conn.ID(),
v1.GetShortType(res.TypeUrl))
}
return err
diff --git a/dubbod/discovery/pkg/xds/delta_test.go
b/dubbod/discovery/pkg/xds/delta_test.go
new file mode 100644
index 00000000..992060a0
--- /dev/null
+++ b/dubbod/discovery/pkg/xds/delta_test.go
@@ -0,0 +1,199 @@
+//
+// 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 (
+ "context"
+ "io"
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/util/protoconv"
+ v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ cluster "github.com/kdubbo/xds-api/cluster/v1"
+ discovery "github.com/kdubbo/xds-api/service/discovery/v1"
+ "google.golang.org/grpc/metadata"
+)
+
+func TestPushDeltaXdsFiltersSubscribedResources(t *testing.T) {
+ oldName := "outbound|80||old.app.svc.cluster.local"
+ newName := "outbound|80||new.app.svc.cluster.local"
+ server, con, stream := newDeltaXDSTestServer(filteredDeltaGenerator{})
+ con.proxy.NewWatchedResource(v1.ClusterType, []string{oldName, newName})
+
+ err := server.pushDeltaXds(con,
con.proxy.GetWatchedResource(v1.ClusterType), &model.PushRequest{
+ Push: con.proxy.LastPushContext,
+ Delta: model.ResourceDelta{
+ Subscribed: sets.New(newName),
+ },
+ })
+ if err != nil {
+ t.Fatalf("pushDeltaXds() failed: %v", err)
+ }
+
+ resp := stream.takeDeltaResponse(t)
+ if got := resourceNames(resp.Resources); len(got) != 1 || got[0] !=
newName {
+ t.Fatalf("resources = %v, want [%s]", got, newName)
+ }
+ assertWatchedNames(t, con.proxy.GetWatchedResource(v1.ClusterType),
newName)
+ if con.proxy.GetWatchedResource(v1.ClusterType).NonceSent == "" {
+ t.Fatal("NonceSent was not recorded")
+ }
+}
+
+func TestPushDeltaXdsAppliesRemovedResourcesToWatchedState(t *testing.T) {
+ removedName := "outbound|80||removed.app.svc.cluster.local"
+ keptName := "outbound|80||kept.app.svc.cluster.local"
+ addedName := "outbound|80||added.app.svc.cluster.local"
+ server, con, stream := newDeltaXDSTestServer(staticDeltaGenerator{
+ resources: model.Resources{clusterResource(addedName)},
+ removed: []string{removedName},
+ })
+ con.proxy.NewWatchedResource(v1.ClusterType, []string{removedName,
keptName})
+
+ err := server.pushDeltaXds(con,
con.proxy.GetWatchedResource(v1.ClusterType), &model.PushRequest{
+ Push: con.proxy.LastPushContext,
+ })
+ if err != nil {
+ t.Fatalf("pushDeltaXds() failed: %v", err)
+ }
+
+ resp := stream.takeDeltaResponse(t)
+ if got := resp.RemovedResources; len(got) != 1 || got[0] != removedName
{
+ t.Fatalf("removed = %v, want [%s]", got, removedName)
+ }
+ assertWatchedNames(t, con.proxy.GetWatchedResource(v1.ClusterType),
keptName, addedName)
+}
+
+func newDeltaXDSTestServer(generator model.XdsResourceGenerator)
(*DiscoveryServer, *Connection, *fakeDeltaADSStream) {
+ push := model.NewPushContext()
+ push.PushVersion = "test-version"
+ server := NewDiscoveryServer(model.NewEnvironment(), nil, nil)
+ server.Generators = map[string]model.XdsResourceGenerator{
+ v1.ClusterType: generator,
+ }
+ stream := newFakeDeltaADSStream()
+ con := newDeltaConnection("127.0.0.1:26010", stream)
+ con.s = server
+ con.proxy = &model.Proxy{
+ ID: "pod-1",
+ Type: model.Proxyless,
+ Metadata: &model.NodeMetadata{Generator: "grpc",
Namespace: "app"},
+ WatchedResources: map[string]*model.WatchedResource{},
+ LastPushContext: push,
+ }
+ return server, con, stream
+}
+
+type filteredDeltaGenerator struct{}
+
+func (filteredDeltaGenerator) Generate(_ *model.Proxy, w
*model.WatchedResource, _ *model.PushRequest) (model.Resources,
model.XdsLogDetails, error) {
+ resources := make(model.Resources, 0, len(w.ResourceNames))
+ for name := range w.ResourceNames {
+ resources = append(resources, clusterResource(name))
+ }
+ return resources, model.DefaultXdsLogDetails, nil
+}
+
+type staticDeltaGenerator struct {
+ resources model.Resources
+ removed []string
+}
+
+func (g staticDeltaGenerator) Generate(*model.Proxy, *model.WatchedResource,
*model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
+ return g.resources, model.DefaultXdsLogDetails, nil
+}
+
+func (g staticDeltaGenerator) GenerateDeltas(*model.Proxy, *model.PushRequest,
*model.WatchedResource) (model.Resources, model.DeletedResources,
model.XdsLogDetails, bool, error) {
+ return g.resources, g.removed, model.DefaultXdsLogDetails, true, nil
+}
+
+func clusterResource(name string) *discovery.Resource {
+ return &discovery.Resource{
+ Name: name,
+ Resource: protoconv.MessageToAny(&cluster.Cluster{Name: name}),
+ }
+}
+
+func resourceNames(resources model.Resources) []string {
+ out := make([]string, 0, len(resources))
+ for _, resource := range resources {
+ out = append(out, resource.Name)
+ }
+ return out
+}
+
+type fakeDeltaADSStream struct {
+ ctx context.Context
+ recvCh chan *discovery.DeltaDiscoveryRequest
+ sendCh chan *discovery.DeltaDiscoveryResponse
+}
+
+func newFakeDeltaADSStream() *fakeDeltaADSStream {
+ return &fakeDeltaADSStream{
+ ctx: context.Background(),
+ recvCh: make(chan *discovery.DeltaDiscoveryRequest, 8),
+ sendCh: make(chan *discovery.DeltaDiscoveryResponse, 8),
+ }
+}
+
+func (f *fakeDeltaADSStream) Send(resp *discovery.DeltaDiscoveryResponse)
error {
+ f.sendCh <- resp
+ return nil
+}
+
+func (f *fakeDeltaADSStream) Recv() (*discovery.DeltaDiscoveryRequest, error) {
+ req, ok := <-f.recvCh
+ if !ok {
+ return nil, io.EOF
+ }
+ return req, nil
+}
+
+func (f *fakeDeltaADSStream) SetHeader(metadata.MD) error {
+ return nil
+}
+
+func (f *fakeDeltaADSStream) SendHeader(metadata.MD) error {
+ return nil
+}
+
+func (f *fakeDeltaADSStream) SetTrailer(metadata.MD) {}
+
+func (f *fakeDeltaADSStream) Context() context.Context {
+ return f.ctx
+}
+
+func (f *fakeDeltaADSStream) SendMsg(any) error {
+ return nil
+}
+
+func (f *fakeDeltaADSStream) RecvMsg(any) error {
+ return nil
+}
+
+func (f *fakeDeltaADSStream) takeDeltaResponse(t *testing.T)
*discovery.DeltaDiscoveryResponse {
+ t.Helper()
+ select {
+ case resp := <-f.sendCh:
+ return resp
+ default:
+ t.Fatal("expected delta response")
+ return nil
+ }
+}
diff --git a/dubbod/discovery/pkg/xds/xdsgen.go
b/dubbod/discovery/pkg/xds/xdsgen.go
index 93262b60..ab9588cc 100644
--- a/dubbod/discovery/pkg/xds/xdsgen.go
+++ b/dubbod/discovery/pkg/xds/xdsgen.go
@@ -535,6 +535,14 @@ func (s *DiscoveryServer) pushDeltaXds(con *Connection, w
*model.WatchedResource
}
var logFiltered string
+ if !req.Delta.IsEmpty() {
+ logFiltered = " filtered:" +
strconv.Itoa(len(w.ResourceNames)-len(req.Delta.Subscribed))
+ w = &model.WatchedResource{
+ TypeUrl: w.TypeUrl,
+ ResourceNames: req.Delta.Subscribed,
+ }
+ }
+
var res model.Resources
var deletedRes model.DeletedResources
var logdata model.XdsLogDetails
diff --git a/pkg/cni/config.go b/pkg/cni/config.go
index eda915fd..82620bdb 100644
--- a/pkg/cni/config.go
+++ b/pkg/cni/config.go
@@ -26,7 +26,7 @@ import (
)
const (
- defaultCNIVersion = "1.0.0"
+ defaultCNIVersion = "1.0.0" // 无用版本
defaultStateDir = "/var/run/dubbo-cni"
defaultIPTablesPath = "iptables"
defaultIPSetPath = "ipset"
diff --git a/pkg/cni/install.go b/pkg/cni/install.go
index 86ae9829..df8bdb93 100644
--- a/pkg/cni/install.go
+++ b/pkg/cni/install.go
@@ -30,11 +30,13 @@ import (
)
const (
- defaultCNIBinDir = "/opt/cni/bin"
- defaultCNIConfDir = "/etc/cni/net.d"
- defaultInstallerInterval = time.Minute
- defaultKubeConfigFileName = "dubbo-cni-kubeconfig"
- installBackupSuffix = ".dubbo-cni.bak"
+ defaultCNIBinDir = "/opt/cni/bin"
+ defaultCNIConfDir = "/etc/cni/net.d"
+ defaultServiceAccountTokenPath =
"/var/run/secrets/kubernetes.io/serviceaccount/token"
+ defaultServiceAccountCAPath =
"/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+ defaultInstallerInterval = time.Minute
+ defaultKubeConfigFileName = "dubbo-cni-kubeconfig"
+ installBackupSuffix = ".dubbo-cni.bak"
)
type InstallerOptions struct {
@@ -60,8 +62,8 @@ func DefaultInstallerOptions() InstallerOptions {
BinDir: defaultCNIBinDir,
ConfDir: defaultCNIConfDir,
StateDir: defaultStateDir,
- ServiceAccountTokenPath:
"/var/run/secrets/kubernetes.io/serviceaccount/token",
- ServiceAccountCAPath:
"/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+ ServiceAccountTokenPath: defaultServiceAccountTokenPath,
+ ServiceAccountCAPath: defaultServiceAccountCAPath,
ManagedLabel: inject.ProxylessManagedLabel,
ManagedLabelValue: inject.ProxylessManagedLabelValue,
IPTablesPath: defaultIPTablesPath,
