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 c723c6be Complete SDS delta xDS support (#924)
c723c6be is described below
commit c723c6be93e0e6f6ccb184f1d71fc2159ab2f98d
Author: mfordjody <[email protected]>
AuthorDate: Sat Jun 27 22:08:38 2026 +0800
Complete SDS delta xDS support (#924)
* Complete SDS delta xDS support
* Use released xds-api secret resources
* Update moviereview samples
---
dubbod/security/pkg/nodeagent/sds/sdsservice.go | 278 +++++++++++++++++++-
.../security/pkg/nodeagent/sds/sdsservice_test.go | 291 +++++++++++++++++++++
go.mod | 2 +-
go.sum | 2 +
pkg/xds/server.go | 3 +
pkg/xds/server_test.go | 18 ++
samples/moviereview/deployment.yaml | 2 +-
7 files changed, 589 insertions(+), 7 deletions(-)
diff --git a/dubbod/security/pkg/nodeagent/sds/sdsservice.go
b/dubbod/security/pkg/nodeagent/sds/sdsservice.go
index c909b219..1ffe8abf 100644
--- a/dubbod/security/pkg/nodeagent/sds/sdsservice.go
+++ b/dubbod/security/pkg/nodeagent/sds/sdsservice.go
@@ -18,26 +18,34 @@ package sds
import (
"context"
- "github.com/apache/dubbo-kubernetes/pkg/log"
+ "fmt"
+ "io"
"strconv"
"sync"
"sync/atomic"
+ "time"
"github.com/apache/dubbo-kubernetes/pkg/backoff"
+ "github.com/apache/dubbo-kubernetes/pkg/log"
+ xdsmodel "github.com/apache/dubbo-kubernetes/pkg/model"
"github.com/apache/dubbo-kubernetes/pkg/security"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/pkg/xds"
core "github.com/kdubbo/xds-api/core/v1"
+ tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
discovery "github.com/kdubbo/xds-api/service/discovery/v1"
sds "github.com/kdubbo/xds-api/service/secret/v1"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
+ "google.golang.org/protobuf/types/known/anypb"
)
var sdsServiceLog = log.RegisterScope("sds", "SDS service debugging")
var connectionNumber = int64(0)
+var deltaNonceNumber = int64(0)
+var versionNumber = int64(0)
type sdsservice struct {
sds.UnimplementedSecretDiscoveryServiceServer
@@ -117,11 +125,81 @@ func (s *sdsservice) StreamSecrets(stream
sds.SecretDiscoveryService_StreamSecre
}
func (s *sdsservice) DeltaSecrets(stream
sds.SecretDiscoveryService_DeltaSecretsServer) error {
- return status.Error(codes.Unimplemented, "DeltaSecrets not implemented")
+ ctx := &Context{
+ BaseConnection: xds.NewConnection("", nil),
+ s: s,
+ w: &Watch{},
+ }
+ reqCh := make(chan *discovery.DeltaDiscoveryRequest, 1)
+ errCh := make(chan error, 1)
+ go func() {
+ defer close(reqCh)
+ for {
+ req, err := stream.Recv()
+ if err != nil {
+ errCh <- err
+ return
+ }
+ select {
+ case reqCh <- req:
+ case <-stream.Context().Done():
+ errCh <- stream.Context().Err()
+ return
+ }
+ }
+ }()
+
+ initialized := false
+ for {
+ select {
+ case req, ok := <-reqCh:
+ if !ok {
+ err := <-errCh
+ if err == io.EOF {
+ return nil
+ }
+ return err
+ }
+ if !initialized {
+ if req.Node == nil || req.Node.Id == "" {
+ return
status.New(codes.InvalidArgument, "missing node information").Err()
+ }
+ if err := ctx.Initialize(req.Node); err != nil {
+ return err
+ }
+ defer ctx.Close()
+ initialized = true
+ }
+ shouldRespond, delta := ctx.shouldRespondDelta(req)
+ if !shouldRespond {
+ continue
+ }
+ resourceNames := ctx.deltaResourceNames(req.TypeUrl,
delta)
+ if err := ctx.sendDelta(stream, req.TypeUrl,
resourceNames, delta.Unsubscribed.UnsortedList()); err != nil {
+ return err
+ }
+ case ev := <-ctx.XdsConnection().PushCh():
+ secretName := ev.(string)
+ if !ctx.w.requested(secretName) {
+ continue
+ }
+ if err := ctx.sendDelta(stream, xdsmodel.SecretType,
[]string{secretName}, nil); err != nil {
+ return err
+ }
+ case <-stream.Context().Done():
+ if initialized {
+ ctx.Close()
+ }
+ return stream.Context().Err()
+ }
+ }
}
-func (s *sdsservice) FetchSecrets(ctx context.Context, discReq
*discovery.DiscoveryRequest) (*discovery.DiscoveryResponse, error) {
- return nil, status.Error(codes.Unimplemented, "FetchSecrets not
implemented")
+func (s *sdsservice) FetchSecrets(_ context.Context, discReq
*discovery.DiscoveryRequest) (*discovery.DiscoveryResponse, error) {
+ if discReq == nil {
+ return nil, status.New(codes.InvalidArgument, "missing
discovery request").Err()
+ }
+ return s.generate(discReq.ResourceNames)
}
// register adds the SDS handle to the grpc server
@@ -130,7 +208,197 @@ func (s *sdsservice) register(rpcs *grpc.Server) {
}
func (s *sdsservice) generate(resourceNames []string)
(*discovery.DiscoveryResponse, error) {
- return &discovery.DiscoveryResponse{}, nil
+ resources := make([]*anypb.Any, 0, len(resourceNames))
+ for _, resourceName := range resourceNames {
+ secret, err := s.st.GenerateSecret(resourceName)
+ if err != nil {
+ return nil, fmt.Errorf("failed to generate secret for
%s: %v", resourceName, err)
+ }
+ if secret == nil {
+ return nil, fmt.Errorf("failed to generate secret for
%s: secret is nil", resourceName)
+ }
+ res, err := anypb.New(toXdsSecret(secret, resourceName))
+ if err != nil {
+ return nil, fmt.Errorf("failed to marshal secret for
%s: %v", resourceName, err)
+ }
+ resources = append(resources, res)
+ }
+ version := strconv.FormatInt(atomic.AddInt64(&versionNumber, 1), 10)
+ return &discovery.DiscoveryResponse{
+ TypeUrl: xdsmodel.SecretType,
+ VersionInfo: version,
+ Nonce: version,
+ Resources: resources,
+ }, nil
+}
+
+func toXdsSecret(item *security.SecretItem, requestedName string)
*tlsv1.Secret {
+ name := item.ResourceName
+ if name == "" {
+ name = requestedName
+ }
+ secret := &tlsv1.Secret{Name: name}
+ if name == security.RootCertReqResourceName {
+ secret.Type = &tlsv1.Secret_ValidationContext{
+ ValidationContext: &tlsv1.CertificateValidationContext{
+ TrustedCa: inlineBytes(item.RootCert),
+ },
+ }
+ return secret
+ }
+ if cfg, ok := security.SdsCertificateConfigFromResourceName(name); ok
&& cfg.IsRootCertificate() {
+ secret.Type = &tlsv1.Secret_ValidationContext{
+ ValidationContext: &tlsv1.CertificateValidationContext{
+ TrustedCa: inlineBytes(item.RootCert),
+ },
+ }
+ return secret
+ }
+ secret.Type = &tlsv1.Secret_TlsCertificate{
+ TlsCertificate: &tlsv1.TlsCertificate{
+ CertificateChain: inlineBytes(item.CertificateChain),
+ PrivateKey: inlineBytes(item.PrivateKey),
+ },
+ }
+ return secret
+}
+
+func inlineBytes(value []byte) *core.DataSource {
+ return &core.DataSource{
+ Specifier: &core.DataSource_InlineBytes{
+ InlineBytes: value,
+ },
+ }
+}
+
+func (c *Context) shouldRespondDelta(req *discovery.DeltaDiscoveryRequest)
(bool, xds.ResourceDelta) {
+ if req.ErrorDetail != nil {
+ errCode := codes.Code(req.ErrorDetail.Code)
+ sdsServiceLog.Warnf("%s: ACK ERROR %s %s:%s",
xdsmodel.GetShortType(req.TypeUrl), c.XdsConnection().ID(), errCode.String(),
req.ErrorDetail.GetMessage())
+ c.w.UpdateWatchedResource(req.TypeUrl, func(wr
*xds.WatchedResource) *xds.WatchedResource {
+ if wr == nil {
+ wr = &xds.WatchedResource{TypeUrl: req.TypeUrl}
+ }
+ wr.LastError = req.ErrorDetail.GetMessage()
+ return wr
+ })
+ return false, xds.ResourceDelta{}
+ }
+
+ previous := c.w.GetWatchedResource(req.TypeUrl)
+ if previous == nil {
+ names, _ := deltaWatchedSecrets(nil, req)
+ c.w.NewWatchedResource(req.TypeUrl, names.UnsortedList())
+ return true, xds.ResourceDelta{Subscribed: names, Unsubscribed:
sets.New(req.ResourceNamesUnsubscribe...)}
+ }
+
+ if req.ResponseNonce != "" && req.ResponseNonce != previous.NonceSent {
+ sdsServiceLog.Debugf("%s: REQ %s expired nonce received %s,
sent %s",
+ xdsmodel.GetShortType(req.TypeUrl),
c.XdsConnection().ID(), req.ResponseNonce, previous.NonceSent)
+ return false, xds.ResourceDelta{}
+ }
+
+ spontaneousReq := req.ResponseNonce == ""
+ var subChanged bool
+ var alwaysRespond bool
+ var subscribed sets.String
+ c.w.UpdateWatchedResource(req.TypeUrl, func(wr *xds.WatchedResource)
*xds.WatchedResource {
+ if wr == nil {
+ wr = &xds.WatchedResource{TypeUrl: req.TypeUrl}
+ }
+ before := wr.ResourceNames
+ wr.ResourceNames, subChanged =
deltaWatchedSecrets(wr.ResourceNames, req)
+ subscribed = wr.ResourceNames.Difference(before)
+ if !spontaneousReq {
+ wr.LastError = ""
+ wr.NonceAcked = req.ResponseNonce
+ }
+ alwaysRespond = wr.AlwaysRespond
+ wr.AlwaysRespond = false
+ return wr
+ })
+
+ if !subChanged {
+ if alwaysRespond {
+ return true, xds.ResourceDelta{}
+ }
+ return false, xds.ResourceDelta{}
+ }
+ return true, xds.ResourceDelta{
+ Subscribed: subscribed,
+ Unsubscribed: sets.New(req.ResourceNamesUnsubscribe...),
+ }
+}
+
+func deltaWatchedSecrets(existing sets.String, req
*discovery.DeltaDiscoveryRequest) (sets.String, bool) {
+ out := sets.New[string]()
+ if existing != nil {
+ out = existing.Copy()
+ }
+ before := out.Copy()
+ out.InsertAll(req.ResourceNamesSubscribe...)
+ for name := range req.InitialResourceVersions {
+ out.Insert(name)
+ }
+ out.DeleteAll(req.ResourceNamesUnsubscribe...)
+ return out, !out.Equals(before)
+}
+
+func (c *Context) deltaResourceNames(typeURL string, delta xds.ResourceDelta)
[]string {
+ if len(delta.Subscribed) > 0 {
+ return delta.Subscribed.UnsortedList()
+ }
+ if len(delta.Unsubscribed) > 0 {
+ return nil
+ }
+ wr := c.w.GetWatchedResource(typeURL)
+ if wr == nil || wr.ResourceNames == nil {
+ return nil
+ }
+ return wr.ResourceNames.UnsortedList()
+}
+
+func (c *Context) sendDelta(stream
sds.SecretDiscoveryService_DeltaSecretsServer, typeURL string, resourceNames
[]string, removed []string) error {
+ res, err := c.s.generate(resourceNames)
+ if err != nil {
+ return err
+ }
+ nonce := strconv.FormatInt(atomic.AddInt64(&deltaNonceNumber, 1), 10)
+ resp := &discovery.DeltaDiscoveryResponse{
+ SystemVersionInfo: res.VersionInfo,
+ TypeUrl: typeURL,
+ RemovedResources: removed,
+ Nonce: nonce,
+ ControlPlane: res.ControlPlane,
+ }
+ if resp.SystemVersionInfo == "" {
+ resp.SystemVersionInfo = nonce
+ }
+ if res.TypeUrl != "" {
+ resp.TypeUrl = res.TypeUrl
+ }
+ for i, resource := range res.Resources {
+ name := ""
+ if i < len(resourceNames) {
+ name = resourceNames[i]
+ }
+ resp.Resources = append(resp.Resources, &discovery.Resource{
+ Name: name,
+ Resource: resource,
+ })
+ }
+ if err := stream.Send(resp); err != nil {
+ return err
+ }
+ c.w.UpdateWatchedResource(resp.TypeUrl, func(wr *xds.WatchedResource)
*xds.WatchedResource {
+ if wr == nil {
+ wr = &xds.WatchedResource{TypeUrl: resp.TypeUrl}
+ }
+ wr.NonceSent = resp.Nonce
+ wr.LastSendTime = time.Now()
+ return wr
+ })
+ return nil
}
func (s *sdsservice) push(secretName string) {
diff --git a/dubbod/security/pkg/nodeagent/sds/sdsservice_test.go
b/dubbod/security/pkg/nodeagent/sds/sdsservice_test.go
new file mode 100644
index 00000000..bf9712c8
--- /dev/null
+++ b/dubbod/security/pkg/nodeagent/sds/sdsservice_test.go
@@ -0,0 +1,291 @@
+//
+// 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 sds
+
+import (
+ "bytes"
+ "context"
+ "io"
+ "testing"
+ "time"
+
+ xdsmodel "github.com/apache/dubbo-kubernetes/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/security"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ "github.com/apache/dubbo-kubernetes/pkg/xds"
+ core "github.com/kdubbo/xds-api/core/v1"
+ tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
+ discovery "github.com/kdubbo/xds-api/service/discovery/v1"
+ "google.golang.org/grpc/metadata"
+)
+
+func TestDeltaSecretsRespondsToSubscribeAndPush(t *testing.T) {
+ service := &sdsservice{
+ st: &fakeSecretManager{
+ secrets: map[string]*security.SecretItem{
+ "default": {
+ ResourceName: "default",
+ CertificateChain: []byte("cert-v1"),
+ PrivateKey: []byte("key-v1"),
+ },
+ },
+ },
+ stop: make(chan struct{}),
+ clients: map[string]*Context{},
+ }
+ stream := newFakeDeltaSecretsStream()
+ errCh := make(chan error, 1)
+ go func() {
+ errCh <- service.DeltaSecrets(stream)
+ }()
+
+ stream.recvCh <- &discovery.DeltaDiscoveryRequest{
+ Node: &core.Node{Id:
"proxyless~10.0.0.1~pod-1~app.svc.cluster.local"},
+ TypeUrl: xdsmodel.SecretType,
+ ResourceNamesSubscribe: []string{"default"},
+ }
+ resp := stream.takeResponse(t)
+ if resp.TypeUrl != xdsmodel.SecretType {
+ t.Fatalf("TypeUrl = %s, want %s", resp.TypeUrl,
xdsmodel.SecretType)
+ }
+ if resp.Nonce == "" || resp.SystemVersionInfo == "" {
+ t.Fatalf("nonce/version = %q/%q, want both set", resp.Nonce,
resp.SystemVersionInfo)
+ }
+ assertTLSSecret(t, resp, "default", []byte("cert-v1"), []byte("key-v1"))
+
+ service.st.(*fakeSecretManager).secrets["default"] =
&security.SecretItem{
+ ResourceName: "default",
+ CertificateChain: []byte("cert-v2"),
+ PrivateKey: []byte("key-v2"),
+ }
+ service.push("default")
+ pushResp := stream.takeResponse(t)
+ if pushResp.TypeUrl != xdsmodel.SecretType {
+ t.Fatalf("push TypeUrl = %s, want %s", pushResp.TypeUrl,
xdsmodel.SecretType)
+ }
+ if pushResp.Nonce == resp.Nonce {
+ t.Fatalf("push nonce = %s, want a new nonce", pushResp.Nonce)
+ }
+ assertTLSSecret(t, pushResp, "default", []byte("cert-v2"),
[]byte("key-v2"))
+
+ close(stream.recvCh)
+ select {
+ case err := <-errCh:
+ if err != nil {
+ t.Fatalf("DeltaSecrets() error = %v", err)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("DeltaSecrets() did not exit after client close")
+ }
+}
+
+func TestGenerateBuildsRootValidationContext(t *testing.T) {
+ service := &sdsservice{
+ st: &fakeSecretManager{
+ secrets: map[string]*security.SecretItem{
+ security.RootCertReqResourceName: {
+ ResourceName:
security.RootCertReqResourceName,
+ RootCert: []byte("root"),
+ },
+ },
+ },
+ }
+
+ resp, err :=
service.generate([]string{security.RootCertReqResourceName})
+ if err != nil {
+ t.Fatalf("generate() error = %v", err)
+ }
+ if resp.TypeUrl != xdsmodel.SecretType {
+ t.Fatalf("TypeUrl = %s, want %s", resp.TypeUrl,
xdsmodel.SecretType)
+ }
+ if len(resp.Resources) != 1 {
+ t.Fatalf("resources = %d, want 1", len(resp.Resources))
+ }
+ secret := &tlsv1.Secret{}
+ if err := resp.Resources[0].UnmarshalTo(secret); err != nil {
+ t.Fatalf("unmarshal secret: %v", err)
+ }
+ if secret.GetName() != security.RootCertReqResourceName {
+ t.Fatalf("secret name = %s, want %s", secret.GetName(),
security.RootCertReqResourceName)
+ }
+ if got :=
secret.GetValidationContext().GetTrustedCa().GetInlineBytes();
!bytes.Equal(got, []byte("root")) {
+ t.Fatalf("trusted CA = %q, want root", got)
+ }
+}
+
+func TestFetchSecretsReturnsGeneratedResources(t *testing.T) {
+ service := &sdsservice{
+ st: &fakeSecretManager{
+ secrets: map[string]*security.SecretItem{
+ "default": {
+ ResourceName: "default",
+ CertificateChain: []byte("cert"),
+ PrivateKey: []byte("key"),
+ },
+ },
+ },
+ }
+
+ resp, err := service.FetchSecrets(context.Background(),
&discovery.DiscoveryRequest{
+ ResourceNames: []string{"default"},
+ })
+ if err != nil {
+ t.Fatalf("FetchSecrets() error = %v", err)
+ }
+ if len(resp.Resources) != 1 {
+ t.Fatalf("resources = %d, want 1", len(resp.Resources))
+ }
+ secret := &tlsv1.Secret{}
+ if err := resp.Resources[0].UnmarshalTo(secret); err != nil {
+ t.Fatalf("unmarshal secret: %v", err)
+ }
+ if got :=
secret.GetTlsCertificate().GetCertificateChain().GetInlineBytes();
!bytes.Equal(got, []byte("cert")) {
+ t.Fatalf("cert = %q, want cert", got)
+ }
+}
+
+func TestShouldRespondDeltaReportsNewSubscriptions(t *testing.T) {
+ ctx := &Context{
+ BaseConnection: xds.NewConnection("", nil),
+ w: &Watch{},
+ }
+
+ shouldRespond, _ :=
ctx.shouldRespondDelta(&discovery.DeltaDiscoveryRequest{
+ TypeUrl: xdsmodel.SecretType,
+ ResourceNamesSubscribe: []string{"default"},
+ })
+ if !shouldRespond {
+ t.Fatalf("initial subscribe should respond")
+ }
+
+ shouldRespond, delta :=
ctx.shouldRespondDelta(&discovery.DeltaDiscoveryRequest{
+ TypeUrl: xdsmodel.SecretType,
+ ResourceNamesSubscribe: []string{"next"},
+ })
+ if !shouldRespond {
+ t.Fatalf("new subscribe should respond")
+ }
+ if !delta.Subscribed.Contains("next") {
+ t.Fatalf("Subscribed = %v, want next",
delta.Subscribed.UnsortedList())
+ }
+ if delta.Subscribed.Contains("default") {
+ t.Fatalf("Subscribed = %v, want only newly subscribed
resources", delta.Subscribed.UnsortedList())
+ }
+}
+
+func TestDeltaResourceNamesSkipsUnsubscribeOnlyResources(t *testing.T) {
+ ctx := &Context{w: &Watch{}}
+ ctx.w.NewWatchedResource(xdsmodel.SecretType, []string{"kept"})
+
+ names := ctx.deltaResourceNames(xdsmodel.SecretType, xds.ResourceDelta{
+ Unsubscribed: sets.New("removed"),
+ })
+ if len(names) != 0 {
+ t.Fatalf("resource names = %v, want none for unsubscribe-only
delta", names)
+ }
+}
+
+type fakeSecretManager struct {
+ secrets map[string]*security.SecretItem
+}
+
+func (f *fakeSecretManager) GenerateSecret(resourceName string)
(*security.SecretItem, error) {
+ return f.secrets[resourceName], nil
+}
+
+func assertTLSSecret(t *testing.T, resp *discovery.DeltaDiscoveryResponse,
name string, cert, key []byte) {
+ t.Helper()
+ if len(resp.Resources) != 1 {
+ t.Fatalf("resources = %d, want 1", len(resp.Resources))
+ }
+ if resp.Resources[0].GetName() != name {
+ t.Fatalf("resource name = %s, want %s",
resp.Resources[0].GetName(), name)
+ }
+ secret := &tlsv1.Secret{}
+ if err := resp.Resources[0].GetResource().UnmarshalTo(secret); err !=
nil {
+ t.Fatalf("unmarshal secret: %v", err)
+ }
+ if secret.GetName() != name {
+ t.Fatalf("secret name = %s, want %s", secret.GetName(), name)
+ }
+ if got :=
secret.GetTlsCertificate().GetCertificateChain().GetInlineBytes();
!bytes.Equal(got, cert) {
+ t.Fatalf("cert = %q, want %q", got, cert)
+ }
+ if got := secret.GetTlsCertificate().GetPrivateKey().GetInlineBytes();
!bytes.Equal(got, key) {
+ t.Fatalf("key = %q, want %q", got, key)
+ }
+}
+
+type fakeDeltaSecretsStream struct {
+ ctx context.Context
+ recvCh chan *discovery.DeltaDiscoveryRequest
+ sendCh chan *discovery.DeltaDiscoveryResponse
+}
+
+func newFakeDeltaSecretsStream() *fakeDeltaSecretsStream {
+ return &fakeDeltaSecretsStream{
+ ctx: context.Background(),
+ recvCh: make(chan *discovery.DeltaDiscoveryRequest, 8),
+ sendCh: make(chan *discovery.DeltaDiscoveryResponse, 8),
+ }
+}
+
+func (f *fakeDeltaSecretsStream) Send(resp *discovery.DeltaDiscoveryResponse)
error {
+ f.sendCh <- resp
+ return nil
+}
+
+func (f *fakeDeltaSecretsStream) Recv() (*discovery.DeltaDiscoveryRequest,
error) {
+ req, ok := <-f.recvCh
+ if !ok {
+ return nil, io.EOF
+ }
+ return req, nil
+}
+
+func (f *fakeDeltaSecretsStream) SetHeader(metadata.MD) error {
+ return nil
+}
+
+func (f *fakeDeltaSecretsStream) SendHeader(metadata.MD) error {
+ return nil
+}
+
+func (f *fakeDeltaSecretsStream) SetTrailer(metadata.MD) {}
+
+func (f *fakeDeltaSecretsStream) Context() context.Context {
+ return f.ctx
+}
+
+func (f *fakeDeltaSecretsStream) SendMsg(any) error {
+ return nil
+}
+
+func (f *fakeDeltaSecretsStream) RecvMsg(any) error {
+ return nil
+}
+
+func (f *fakeDeltaSecretsStream) takeResponse(t *testing.T)
*discovery.DeltaDiscoveryResponse {
+ t.Helper()
+ select {
+ case resp := <-f.sendCh:
+ return resp
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for delta SDS response")
+ return nil
+ }
+}
diff --git a/go.mod b/go.mod
index 5fe70718..14e0205f 100644
--- a/go.mod
+++ b/go.mod
@@ -55,7 +55,7 @@ require (
github.com/heroku/color v0.0.6
github.com/kdubbo/api v0.0.0-20260514125132-e21515234c56
github.com/kdubbo/client-go v0.0.0-20260514125254-aa566bd2f6d5
- github.com/kdubbo/xds-api v0.0.0-20260618170142-573d9453aff2
+ github.com/kdubbo/xds-api v0.0.0-20260627135022-d265dbcbd12f
github.com/moby/term v0.5.2
github.com/ory/viper v1.7.5
github.com/pkg/errors v0.9.1
diff --git a/go.sum b/go.sum
index c19e77a7..52a98f00 100644
--- a/go.sum
+++ b/go.sum
@@ -341,6 +341,8 @@ github.com/kdubbo/client-go
v0.0.0-20260514125254-aa566bd2f6d5 h1:ynMV7FtdzoEMBG
github.com/kdubbo/client-go v0.0.0-20260514125254-aa566bd2f6d5/go.mod
h1:NYuRXnoNWCR5uZLjLmWi1ACz+GYFHjckMtLJHWRI4VU=
github.com/kdubbo/xds-api v0.0.0-20260618170142-573d9453aff2
h1:hOAD6w8dNGZL/C2LyWVgLQ/CZ3CqatWlyEb49nRcPWE=
github.com/kdubbo/xds-api v0.0.0-20260618170142-573d9453aff2/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
+github.com/kdubbo/xds-api v0.0.0-20260627135022-d265dbcbd12f
h1:uO3+UyFwH5nyycY8VxHMt0tL4029SNrVknSxkyNf47g=
+github.com/kdubbo/xds-api v0.0.0-20260627135022-d265dbcbd12f/go.mod
h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
github.com/kevinburke/ssh_config v1.2.0
h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4=
github.com/kevinburke/ssh_config v1.2.0/go.mod
h1:CT57kijsi8u/K/BOFA39wgDQJ9CxiF4nAY/ojJ6r6mM=
github.com/kisielk/errcheck v1.5.0/go.mod
h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
diff --git a/pkg/xds/server.go b/pkg/xds/server.go
index ba999ab2..fd3cfa31 100644
--- a/pkg/xds/server.go
+++ b/pkg/xds/server.go
@@ -267,6 +267,9 @@ func (conn *Connection) StopCh() chan struct{} {
}
func (conn *Connection) StreamDone() <-chan struct{} {
+ if conn.stream == nil {
+ return conn.stop
+ }
return conn.stream.Context().Done()
}
diff --git a/pkg/xds/server_test.go b/pkg/xds/server_test.go
index 812abba1..2132844d 100644
--- a/pkg/xds/server_test.go
+++ b/pkg/xds/server_test.go
@@ -17,6 +17,7 @@ package xds
import (
"testing"
+ "time"
"github.com/apache/dubbo-kubernetes/pkg/model"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
@@ -56,6 +57,23 @@ func (w *testWatcher) GetID() string {
return "test"
}
+func TestStreamDoneHandlesNilStream(t *testing.T) {
+ conn := NewConnection("", nil)
+
+ select {
+ case <-conn.StreamDone():
+ t.Fatalf("StreamDone closed before StopCh")
+ default:
+ }
+
+ close(conn.StopCh())
+ select {
+ case <-conn.StreamDone():
+ case <-time.After(time.Second):
+ t.Fatalf("StreamDone did not close after StopCh")
+ }
+}
+
func TestShouldRespondPreservesNonceAckWithoutResourceNames(t *testing.T) {
watcher := newTestWatcher()
routeName := "outbound|80||nginx.app.svc.cluster.local"
diff --git a/samples/moviereview/deployment.yaml
b/samples/moviereview/deployment.yaml
index 92777d57..cd649908 100644
--- a/samples/moviereview/deployment.yaml
+++ b/samples/moviereview/deployment.yaml
@@ -34,7 +34,7 @@ spec:
- name: http
port: 80
targetPort: 9080
- nodePort: 30080
+ nodePort: 19080
protocol: TCP
---
apiVersion: v1
