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 81793ce9 Optimize project defects (#936)
81793ce9 is described below
commit 81793ce9ee7ea12c6adac0eabfb7b5aed846031b
Author: mfordjody <[email protected]>
AuthorDate: Thu Jul 9 10:10:24 2026 +0800
Optimize project defects (#936)
---
Makefile | 5 +
cni/main.go | 2 +-
{pkg => cni/pkg}/cni/config.go | 0
{pkg => cni/pkg}/cni/config_test.go | 0
{pkg => cni/pkg}/cni/install.go | 0
{pkg => cni/pkg}/cni/install_test.go | 0
{pkg => cni/pkg}/cni/iptables.go | 0
{pkg => cni/pkg}/cni/iptables_test.go | 0
{pkg => cni/pkg}/cni/kubernetes.go | 0
{pkg => cni/pkg}/cni/plugin.go | 0
{pkg => cni/pkg}/cni/plugin_test.go | 0
{pkg => cni/pkg}/cni/state.go | 0
dubbod/agency/config/config.go | 32 -
.../pkg/bootstrap/proxyless_grpc_controller.go | 2 +-
.../bootstrap/proxyless_grpc_controller_test.go | 2 +-
dubbod/discovery/pkg/networking/grpcgen/lds.go | 2 +-
.../discovery/pkg/networking/grpcgen/mtls_test.go | 2 +-
go.mod | 2 +-
pkg/dubboagency/agent.go | 629 --------------
pkg/dubboagency/config/config.go | 98 ---
pkg/dubboagency/plugins.go | 74 --
pkg/dubboagency/xds_proxy.go | 907 ---------------------
pkg/dubboagency/xds_proxy_delta.go | 323 --------
pkg/{dubboagency => }/grpcxds/grpc_bootstrap.go | 0
.../grpcxds/grpc_bootstrap_test.go | 0
25 files changed, 11 insertions(+), 2069 deletions(-)
diff --git a/Makefile b/Makefile
index 67dbbea9..f5189ed6 100644
--- a/Makefile
+++ b/Makefile
@@ -19,6 +19,11 @@ build-dubboctl:
-ldflags "-X
github.com/apache/dubbo-kubernetes/pkg/version.gitTag=$(GIT_VERSION)" \
-o bin/dubboctl cli/main.go
+.PHONY: build-dubbod
+build-dubbod:
+ CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH) go build \
+ -o bin/dubbod dubbod/discovery/cmd/main.go
+
.PHONY: build-dubbo-cni
build-dubbo-cni:
CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH) go build \
diff --git a/cni/main.go b/cni/main.go
index 1f23cee7..7b46b68b 100644
--- a/cni/main.go
+++ b/cni/main.go
@@ -25,7 +25,7 @@ import (
"syscall"
"time"
- "github.com/apache/dubbo-kubernetes/pkg/cni"
+ "github.com/apache/dubbo-kubernetes/cni/pkg/cni"
)
func main() {
diff --git a/pkg/cni/config.go b/cni/pkg/cni/config.go
similarity index 100%
rename from pkg/cni/config.go
rename to cni/pkg/cni/config.go
diff --git a/pkg/cni/config_test.go b/cni/pkg/cni/config_test.go
similarity index 100%
rename from pkg/cni/config_test.go
rename to cni/pkg/cni/config_test.go
diff --git a/pkg/cni/install.go b/cni/pkg/cni/install.go
similarity index 100%
rename from pkg/cni/install.go
rename to cni/pkg/cni/install.go
diff --git a/pkg/cni/install_test.go b/cni/pkg/cni/install_test.go
similarity index 100%
rename from pkg/cni/install_test.go
rename to cni/pkg/cni/install_test.go
diff --git a/pkg/cni/iptables.go b/cni/pkg/cni/iptables.go
similarity index 100%
rename from pkg/cni/iptables.go
rename to cni/pkg/cni/iptables.go
diff --git a/pkg/cni/iptables_test.go b/cni/pkg/cni/iptables_test.go
similarity index 100%
rename from pkg/cni/iptables_test.go
rename to cni/pkg/cni/iptables_test.go
diff --git a/pkg/cni/kubernetes.go b/cni/pkg/cni/kubernetes.go
similarity index 100%
rename from pkg/cni/kubernetes.go
rename to cni/pkg/cni/kubernetes.go
diff --git a/pkg/cni/plugin.go b/cni/pkg/cni/plugin.go
similarity index 100%
rename from pkg/cni/plugin.go
rename to cni/pkg/cni/plugin.go
diff --git a/pkg/cni/plugin_test.go b/cni/pkg/cni/plugin_test.go
similarity index 100%
rename from pkg/cni/plugin_test.go
rename to cni/pkg/cni/plugin_test.go
diff --git a/pkg/cni/state.go b/cni/pkg/cni/state.go
similarity index 100%
rename from pkg/cni/state.go
rename to cni/pkg/cni/state.go
diff --git a/dubbod/agency/config/config.go b/dubbod/agency/config/config.go
deleted file mode 100644
index a008b88c..00000000
--- a/dubbod/agency/config/config.go
+++ /dev/null
@@ -1,32 +0,0 @@
-//
-// 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 config
-
-import "strings"
-
-const (
- local = "localhost"
-)
-
-func GetDubboSan(discoveryAddress string) string {
- discHost := strings.Split(discoveryAddress, ":")[0]
- // For local debugging - the discoveryAddress is set to localhost, but
the cert issued for normal PA.
- if discHost == local {
- discHost = "dubbod.dubbo-system.svc"
- }
- return discHost
-}
diff --git a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
index bd2f512b..3c76dae4 100644
--- a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
+++ b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
@@ -33,7 +33,7 @@ import (
configlabels "github.com/apache/dubbo-kubernetes/pkg/config/labels"
meshconfig "github.com/apache/dubbo-kubernetes/pkg/config/mesh"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
- "github.com/apache/dubbo-kubernetes/pkg/dubboagency/grpcxds"
+ "github.com/apache/dubbo-kubernetes/pkg/grpcxds"
kubelib "github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/inject"
diff --git a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
index 8f903f6b..b4ad480d 100644
--- a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
+++ b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
@@ -31,7 +31,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collections"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
- "github.com/apache/dubbo-kubernetes/pkg/dubboagency/grpcxds"
+ "github.com/apache/dubbo-kubernetes/pkg/grpcxds"
"github.com/apache/dubbo-kubernetes/pkg/kube/inject"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
diff --git a/dubbod/discovery/pkg/networking/grpcgen/lds.go
b/dubbod/discovery/pkg/networking/grpcgen/lds.go
index c93db5eb..ee46f3ee 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/lds.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/lds.go
@@ -26,7 +26,7 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/util"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/util/protoconv"
"github.com/apache/dubbo-kubernetes/pkg/config/host"
- "github.com/apache/dubbo-kubernetes/pkg/dubboagency/grpcxds"
+ "github.com/apache/dubbo-kubernetes/pkg/grpcxds"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/pkg/wellknown"
core "github.com/kdubbo/xds-api/core/v1"
diff --git a/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go
b/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go
index 7f5a637d..68218140 100644
--- a/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go
+++ b/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go
@@ -6,7 +6,7 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
"github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
- "github.com/apache/dubbo-kubernetes/pkg/dubboagency/grpcxds"
+ "github.com/apache/dubbo-kubernetes/pkg/grpcxds"
security "github.com/kdubbo/api/security/v1alpha3"
cluster "github.com/kdubbo/xds-api/cluster/v1"
tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
diff --git a/go.mod b/go.mod
index 91cb6a8a..9efff7c5 100644
--- a/go.mod
+++ b/go.mod
@@ -72,7 +72,6 @@ require (
golang.org/x/term v0.38.0
golang.org/x/time v0.12.0
gomodules.xyz/jsonpatch/v2 v2.5.0
- google.golang.org/genproto/googleapis/rpc
v0.0.0-20260319201613-d00831a3d3e7
google.golang.org/grpc v1.79.3
google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v3 v3.0.1
@@ -234,6 +233,7 @@ require (
golang.org/x/sync v0.19.0 // indirect
golang.org/x/text v0.32.0 // indirect
google.golang.org/genproto/googleapis/api
v0.0.0-20251202230838-ff82c1b0f217 // indirect
+ google.golang.org/genproto/googleapis/rpc
v0.0.0-20260319201613-d00831a3d3e7 // indirect
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
diff --git a/pkg/dubboagency/agent.go b/pkg/dubboagency/agent.go
deleted file mode 100644
index a0847d88..00000000
--- a/pkg/dubboagency/agent.go
+++ /dev/null
@@ -1,629 +0,0 @@
-//
-// 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 dubboagency
-
-import (
- "context"
- "encoding/json"
- "fmt"
- "github.com/apache/dubbo-kubernetes/dubbod/agency/config"
- "net/http"
- "os"
- "path"
- "path/filepath"
- "strings"
- "sync"
- "time"
-
- "github.com/apache/dubbo-kubernetes/pkg/log"
-
- "github.com/apache/dubbo-kubernetes/pkg/model"
-
- "github.com/apache/dubbo-kubernetes/dubbod/security/pkg/nodeagent/cache"
- "github.com/apache/dubbo-kubernetes/pkg/backoff"
- "github.com/apache/dubbo-kubernetes/pkg/bootstrap"
- "github.com/apache/dubbo-kubernetes/pkg/config/constants"
- "github.com/apache/dubbo-kubernetes/pkg/dubboagency/grpcxds"
- "github.com/apache/dubbo-kubernetes/pkg/filewatcher"
- "github.com/apache/dubbo-kubernetes/pkg/security"
- mesh "github.com/kdubbo/api/mesh/v1alpha1"
- core "github.com/kdubbo/xds-api/core/v1"
- "google.golang.org/grpc"
- "google.golang.org/grpc/credentials/insecure"
- "google.golang.org/protobuf/types/known/structpb"
-)
-
-const (
- serviceNodeSeparator = "~"
-)
-
-const (
- DubboCACertPath = "./var/run/secrets/dubbo"
-)
-
-const (
- MetadataClientCertKey = "DUBBO_META_TLS_CLIENT_KEY"
- MetadataClientCertChain = "DUBBO_META_TLS_CLIENT_CERT_CHAIN"
- MetadataClientRootCert = "DUBBO_META_TLS_CLIENT_ROOT_CERT"
-)
-
-type SDSServiceFactory = func(_ *security.Options, _ security.SecretManager)
SDSService
-
-type SDSService interface {
- OnSecretUpdate(resourceName string)
- Stop()
-}
-
-type Proxy struct {
- ID string
- DNSDomain string
- IPAddresses []string
- Type model.NodeType
- ipMode model.IPMode
-}
-
-type Agent struct {
- proxyConfig *mesh.ProxyConfig
- cfg *AgentOptions
- EnableDynamicProxyConfig bool
- secOpts *security.Options
- sdsServer SDSService
- secretCache *cache.SecretManagerClient
- sdsMu sync.Mutex
-
- xdsProxy *XdsProxy
- fileWatcher filewatcher.FileWatcher
- statusSrv *http.Server
-
- wg sync.WaitGroup
-}
-
-type AgentOptions struct {
- WorkloadIdentitySocketFile string
- GRPCBootstrapPath string
- XDSHeaders map[string]string
- XdsUdsPath string
- XDSRootCerts string
- ProxyIPAddresses []string
- ProxyDomain string
- EnableDynamicProxyConfig bool
- ServiceNode string
- MetadataDiscovery *bool
- CARootCerts string
- DubbodSAN string
- DownstreamGrpcOptions []grpc.ServerOption
- ProxyType model.NodeType
- SDSFactory func(options *security.Options,
workloadSecretCache security.SecretManager) SDSService
-}
-
-func NewAgent(proxyConfig *mesh.ProxyConfig, agentOpts *AgentOptions, sopts
*security.Options) *Agent {
- return &Agent{
- proxyConfig: proxyConfig,
- cfg: agentOpts,
- secOpts: sopts,
- fileWatcher: filewatcher.NewWatcher(),
- }
-}
-
-func (a *Agent) Run(ctx context.Context) (func(), error) {
- if a.cfg.WorkloadIdentitySocketFile !=
filepath.Base(a.cfg.WorkloadIdentitySocketFile) {
- return nil, fmt.Errorf("workload identity socket file override
must be a filename, not a path: %s", a.cfg.WorkloadIdentitySocketFile)
- }
-
- configuredAgentSocketPath :=
security.GetWorkloadSDSSocketListenPath(a.cfg.WorkloadIdentitySocketFile)
-
- isDubboSDS := configuredAgentSocketPath ==
security.GetDubboSDSServerSocketPath()
-
- socketExists, err := checkSocket(ctx, configuredAgentSocketPath)
- if err != nil {
- return nil, fmt.Errorf("failed to check SDS socket: %v", err)
- }
- if socketExists {
- log.Infof("Existing workload SDS socket found at %s. Default
Dubbo SDS Server will only serve files", configuredAgentSocketPath)
- a.secOpts.ServeOnlyFiles = true
- } else if !isDubboSDS {
- return nil, fmt.Errorf("agent configured for non-default SDS
socket path: %s but no socket found", configuredAgentSocketPath)
- }
-
- log.Info("Starting default Dubbo SDS Server")
- err = a.initSdsServer(ctx)
- if err != nil {
- return nil, fmt.Errorf("failed to start default Dubbo SDS
server: %v", err)
- }
- a.xdsProxy, err = initXdsProxy(a)
- if err != nil {
- return nil, fmt.Errorf("failed to start xds proxy: %v", err)
- }
-
- var bootstrapNode *core.Node
- if a.cfg.GRPCBootstrapPath != "" {
- node, err := a.generateGRPCBootstrapWithNode()
- if err != nil {
- return nil, fmt.Errorf("failed generating gRPC XDS
bootstrap: %v", err)
- }
- // Prepare Node for upstream connection, but don't set it yet
- // We'll set it after status port starts and certificates are
generated
- if node != nil && a.xdsProxy != nil {
- bootstrapNode = &core.Node{
- Id: node.ID,
- Locality: node.Locality,
- }
- if node.Metadata != nil {
- bytes, _ := json.Marshal(node.Metadata)
- rawMeta := map[string]any{}
- if err := json.Unmarshal(bytes, &rawMeta); err
== nil {
- if metaStruct, err :=
structpb.NewStruct(rawMeta); err == nil {
- bootstrapNode.Metadata =
metaStruct
- }
- }
- }
- }
- } else {
- log.Warn("GRPC_XDS_BOOTSTRAP not set, bootstrap file will not
be generated")
- }
-
- if a.proxyConfig.ControlPlaneAuthPolicy !=
mesh.AuthenticationPolicy_NONE {
- rootCAForXDS, err := a.FindRootCAForXDS()
- if err != nil {
- return nil, fmt.Errorf("failed to find root XDS CA:
%v", err)
- }
- go a.startFileWatcher(ctx, rootCAForXDS, func() {
- if err := a.xdsProxy.initDubbodDialOptions(a); err !=
nil {
- log.Warnf("Failed to update xds proxy dial
options: %v", err)
- } else {
- log.Info("Updated xds proxy dial options after
certificate change")
- }
- })
-
- // also watch CA root for CA client; rebuild SDS/CA client when
it changes
- if caRoot, err := a.FindRootCAForCA(); err == nil && caRoot !=
"" {
- go a.startFileWatcher(ctx, caRoot, func() {
- log.Info("CA root changed, rebuilding CA client
and SDS server")
- a.rebuildSDSWithNewCAClient()
- })
- }
- }
-
- // start status HTTP server for health checks
- mux := http.NewServeMux()
- mux.HandleFunc("/healthz/ready", func(w http.ResponseWriter, r
*http.Request) {
- w.WriteHeader(http.StatusOK)
- _, _ = w.Write([]byte("ok"))
- })
- a.statusSrv = &http.Server{
- Addr: fmt.Sprintf(":%d", a.proxyConfig.StatusPort),
- Handler: mux,
- }
- a.wg.Add(1)
- go func() {
- defer a.wg.Done()
- if err := a.statusSrv.ListenAndServe(); err != nil && err !=
http.ErrServerClosed {
- log.Errorf("status server error: %v", err)
- }
- }()
-
- // Now set bootstrap node to trigger upstream connection.
- // This ensures upstream connection logs appear after certificate logs.
- if bootstrapNode != nil && a.xdsProxy != nil {
- a.xdsProxy.SetBootstrapNode(bootstrapNode)
- }
-
- if a.WaitForSigterm() {
- // wait for SIGTERM and perform graceful shutdown
- a.wg.Add(1)
- go func() {
- defer a.wg.Done()
- <-ctx.Done()
- }()
- }
-
- return a.wg.Wait, nil
-}
-
-func (a *Agent) WaitForSigterm() bool {
- return a.cfg.ProxyType == model.Router
-}
-
-func (a *Agent) Close() {
- if a.xdsProxy != nil {
- a.xdsProxy.close()
- }
- if a.sdsServer != nil {
- a.sdsServer.Stop()
- }
- if a.secretCache != nil {
- a.secretCache.Close()
- }
- if a.fileWatcher != nil {
- _ = a.fileWatcher.Close()
- }
- if a.statusSrv != nil {
- _ = a.statusSrv.Shutdown(context.Background())
- }
-}
-
-func (a *Agent) FindRootCAForXDS() (string, error) {
- var rootCAPath string
-
- if a.cfg.XDSRootCerts == security.SystemRootCerts {
- // Special case input for root cert configuration to use system
root certificates
- return "", nil
- } else if a.cfg.XDSRootCerts != "" {
- // Using specific platform certs or custom roots
- rootCAPath = a.cfg.XDSRootCerts
- } else if fileExists(security.DefaultRootCertFilePath) {
- // Old style - mounted cert. This is used for XDS auth only,
- // not connecting to CA_ADDRESS because this mode uses external
- // agent (Secret refresh, etc)
- return security.DefaultRootCertFilePath, nil
- } else if a.secOpts.ProvCert != "" {
- // This was never completely correct - PROV_CERT are only
intended for auth with CA_ADDRESS,
- // and should not be involved in determining the root CA.
- // For VMs, the root cert file used to auth may be populated
afterwards.
- // Thus, return directly here and skip checking for existence.
- return a.secOpts.ProvCert + "/root-cert.pem", nil
- } else if a.secOpts.FileMountedCerts {
- // FileMountedCerts - Load it from Proxy Metadata.
- rootCAPath = a.proxyConfig.ProxyMetadata[MetadataClientRootCert]
- } else if a.secOpts.DubboCertProvider == constants.CertProviderNone {
- return "", fmt.Errorf("root CA file for XDS required but
configured provider as none")
- } else {
- rootCAPath = path.Join(DubboCACertPath,
constants.CACertNamespaceConfigMapDataName)
- }
-
- if fileExists(rootCAPath) {
- return rootCAPath, nil
- }
-
- return "", fmt.Errorf("root CA file for XDS does not exist %s",
rootCAPath)
-}
-
-func (a *Agent) GetKeyCertsForXDS() (string, string) {
- var key, cert string
- if a.secOpts.ProvCert != "" {
- key, cert = getKeyCertInner(a.secOpts.ProvCert)
- } else if a.secOpts.FileMountedCerts {
- key = a.proxyConfig.ProxyMetadata[MetadataClientCertKey]
- cert = a.proxyConfig.ProxyMetadata[MetadataClientCertChain]
- }
- return key, cert
-}
-
-func (a *Agent) GetKeyCertsForCA() (string, string) {
- var key, cert string
- if a.secOpts.ProvCert != "" {
- key, cert = getKeyCertInner(a.secOpts.ProvCert)
- }
- return key, cert
-}
-
-func (a *Agent) FindRootCAForCA() (string, error) {
- var rootCAPath string
-
- if a.cfg.CARootCerts == security.SystemRootCerts {
- return "", nil
- } else if a.cfg.CARootCerts != "" {
- rootCAPath = a.cfg.CARootCerts
- } else if a.secOpts.DubboCertProvider == constants.CertProviderCustom {
- rootCAPath = security.DefaultRootCertFilePath //
./etc/certs/root-cert.pem
- } else if a.secOpts.ProvCert != "" {
- // This was never completely correct - PROV_CERT are only
intended for auth with CA_ADDRESS,
- // and should not be involved in determining the root CA.
- // For VMs, the root cert file used to auth may be populated
afterwards.
- // Thus, return directly here and skip checking for existence.
- return a.secOpts.ProvCert + "/root-cert.pem", nil
- } else if a.secOpts.DubboCertProvider == constants.CertProviderNone {
- return "", fmt.Errorf("root CA file for CA required but
configured provider as none")
- } else {
- rootCAPath = path.Join(DubboCACertPath,
constants.CACertNamespaceConfigMapDataName)
- }
-
- if fileExists(rootCAPath) {
- return rootCAPath, nil
- }
-
- return "", fmt.Errorf("root CA file for CA does not exist %s",
rootCAPath)
-}
-
-func (a *Agent) startFileWatcher(ctx context.Context, filePath string, handler
func()) {
- absPath, err := filepath.Abs(filePath)
- if err != nil {
- log.Warnf("Failed to get absolute path for %s: %v", filePath,
err)
- return
- }
- // Ensure parent directory exists for filewatcher
- parentDir := filepath.Dir(absPath)
- if _, err := os.Stat(parentDir); os.IsNotExist(err) {
- if err := os.MkdirAll(parentDir, 0755); err != nil {
- log.Warnf("Failed to create parent directory %s for
file watcher: %v", parentDir, err)
- return
- }
- }
- // Filewatcher watches the parent directory, so the file doesn't need
to exist yet
- if err := a.fileWatcher.Add(absPath); err != nil {
- // If the file is already being watched, this is expected and
should be silently skipped
- // Only log as warning if it's a different error
- if strings.Contains(err.Error(), "is already being watched") {
- log.Debugf("File watcher already exists for %s,
skipping", absPath)
- return
- }
- log.Warnf("Failed to add file watcher %s: %v", absPath, err)
- return
- }
-
- log.Debugf("Add file %s watcher", absPath)
- for {
- select {
- case gotEvent := <-a.fileWatcher.Events(absPath):
- log.Debugf("Receive file %s event %v", absPath,
gotEvent)
- handler()
- case err := <-a.fileWatcher.Errors(absPath):
- log.Warnf("Watch file %s error: %v", absPath, err)
- case <-ctx.Done():
- return
- }
- }
-}
-
-func (a *Agent) initSdsServer(ctx context.Context) error {
- var err error
- if
security.CheckWorkloadCertificate(security.WorkloadIdentityCertChainPath,
security.WorkloadIdentityKeyPath, security.WorkloadIdentityRootCertPath) {
- log.Info("workload certificate files detected, creating secret
manager without caClient")
- a.secOpts.RootCertFilePath =
security.WorkloadIdentityRootCertPath
- a.secOpts.CertChainFilePath =
security.WorkloadIdentityCertChainPath
- a.secOpts.KeyFilePath = security.WorkloadIdentityKeyPath
- a.secOpts.FileMountedCerts = true
- }
-
- createCaClient := !a.secOpts.FileMountedCerts &&
!a.secOpts.ServeOnlyFiles
- a.secretCache, err = a.newSecretManager(createCaClient)
- if err != nil {
- return fmt.Errorf("failed to start workload secret manager %v",
err)
- }
-
- a.sdsServer = a.cfg.SDSFactory(a.secOpts, a.secretCache)
- return a.registerSecretHandler(ctx)
-}
-
-func (a *Agent) rebuildSDSWithNewCAClient() {
- a.sdsMu.Lock()
- defer a.sdsMu.Unlock()
- if a.sdsServer != nil {
- log.Info("Stopping existing SDS server for CA client rebuild")
- a.sdsServer.Stop()
- }
- if a.secretCache != nil {
- log.Info("Closing existing SecretManagerClient")
- a.secretCache.Close()
- }
- // recreate secret manager with CA client enabled
- sc, err := a.newSecretManager(true)
- if err != nil {
- log.Errorf("failed to recreate secret manager with new CA
client: %v", err)
- return
- }
- a.secretCache = sc
- a.sdsServer = a.cfg.SDSFactory(a.secOpts, a.secretCache)
- if err := a.registerSecretHandler(context.Background()); err != nil {
- log.Errorf("failed to refresh workload certificates after CA
rebuild: %v", err)
- } else {
- log.Info("SDS server and CA client rebuilt successfully")
- }
-}
-
-func (a *Agent) registerSecretHandler(ctx context.Context) error {
- if a.secretCache == nil {
- return nil
- }
- handler := func(resourceName string) {
- if a.sdsServer != nil {
- a.sdsServer.OnSecretUpdate(resourceName)
- }
- if resourceName == security.WorkloadKeyCertResourceName ||
resourceName == security.RootCertReqResourceName {
- go func() {
- if err :=
a.ensureWorkloadCertificates(context.Background()); err != nil {
- log.Warnf("failed to refresh workload
certificates after %s update: %v", resourceName, err)
- }
- }()
- }
- }
- a.secretCache.RegisterSecretHandler(handler)
- return a.ensureWorkloadCertificates(ctx)
-}
-
-func (a *Agent) ensureWorkloadCertificates(ctx context.Context) error {
- if a.secretCache == nil || a.secOpts == nil ||
a.secOpts.OutputKeyCertToDir == "" {
- // Nothing to write
- return nil
- }
- generate := func(resource string) error {
- b := backoff.NewExponentialBackOff(backoff.DefaultOption())
- return b.RetryWithContext(ctx, func() error {
- if ctx.Err() != nil {
- return ctx.Err()
- }
- _, err := a.secretCache.GenerateSecret(resource)
- if err != nil {
- log.Warnf("failed to generate %s: %v",
resource, err)
- }
- return err
- })
- }
- if err := generate(security.WorkloadKeyCertResourceName); err != nil {
- return err
- }
- return generate(security.RootCertReqResourceName)
-}
-
-func (a *Agent) generateGRPCBootstrapWithNode() (*model.Node, error) {
- // Convert relative path to absolute path for bootstrap file
- bootstrapPath := a.cfg.GRPCBootstrapPath
- absBootstrapPath, err := filepath.Abs(bootstrapPath)
- if err != nil {
- return nil, fmt.Errorf("failed to resolve absolute path for
bootstrap file: %v", err)
- }
- log.Debugf("Generating gRPC bootstrap file at: %s (absolute: %s)",
bootstrapPath, absBootstrapPath)
-
- // generate metadata
- node, err := a.generateNodeMetadata()
- if err != nil {
- return nil, fmt.Errorf("failed generating node metadata: %v",
err)
- }
-
- log.Infof("Dubbo SAN: %v", node.Metadata.DubboSubjectAltName)
-
- node.Metadata.Generator = "grpc"
-
- if err := os.MkdirAll(filepath.Dir(absBootstrapPath), 0o700); err !=
nil {
- return nil, err
- }
-
- // Use absolute XdsUdsPath from xdsProxy (already converted in
initXdsProxy)
- absUdsPath := a.xdsProxy.xdsUdsPath
-
- _, err = grpcxds.GenerateBootstrapFile(grpcxds.GenerateBootstrapOptions{
- Node: node,
- XdsUdsPath: absUdsPath,
- DiscoveryAddress: a.proxyConfig.DiscoveryAddress,
- CertDir: a.secOpts.OutputKeyCertToDir,
- }, absBootstrapPath)
- if err != nil {
- return nil, err
- }
- log.Debugf("gRPC bootstrap file generated successfully at: %s",
absBootstrapPath)
- return node, nil
-}
-
-func (a *Agent) generateGRPCBootstrap() error {
- _, err := a.generateGRPCBootstrapWithNode()
- return err
-}
-
-func (a *Agent) newSecretManager(createCaClient bool)
(*cache.SecretManagerClient, error) {
- if !createCaClient {
- log.Info("Workload is using file mounted certificates. Skipping
connecting to CA")
- return cache.NewSecretManagerClient(nil, a.secOpts)
- }
- log.Infof("CA Endpoint %s, provider %s", a.secOpts.CAEndpoint,
a.secOpts.CAProviderName)
-
- caClient, err := createCAClient(a.secOpts, a)
- if err != nil {
- return nil, err
- }
- return cache.NewSecretManagerClient(caClient, a.secOpts)
-}
-
-func (a *Agent) generateNodeMetadata() (*model.Node, error) {
- var dubboSAN []string
- if a.proxyConfig.ControlPlaneAuthPolicy ==
mesh.AuthenticationPolicy_MUTUAL_TLS {
- dubboSAN =
[]string{config.GetDubboSan(a.proxyConfig.DiscoveryAddress)}
- }
-
- credentialSocketExists, err := checkSocket(context.TODO(),
security.CredentialNameSocketPath)
- if err != nil {
- return nil, fmt.Errorf("failed to check credential SDS socket:
%v", err)
- }
- if credentialSocketExists {
- log.Info("Credential SDS socket found")
- }
-
- return bootstrap.GetNodeMetaData(bootstrap.MetadataOptions{
- ID: a.cfg.ServiceNode,
- Envs: os.Environ(),
- InstanceIPs: a.cfg.ProxyIPAddresses,
- StsPort: a.secOpts.STSPort,
- ProxyConfig: a.proxyConfig,
- DubboSubjectAltName: dubboSAN,
- CredentialSocketExists: credentialSocketExists,
- XDSRootCert: a.cfg.XDSRootCerts,
- MetadataDiscovery: a.cfg.MetadataDiscovery,
- })
-}
-
-func (node *Proxy) DiscoverIPMode() {
- node.ipMode = model.DiscoverIPMode(node.IPAddresses)
-}
-
-func (node *Proxy) ServiceNode() string {
- ip := ""
- if len(node.IPAddresses) > 0 {
- ip = node.IPAddresses[0]
- }
- return strings.Join([]string{
- string(node.Type), ip, node.ID, node.DNSDomain,
- }, serviceNodeSeparator)
-}
-
-func getKeyCertInner(certPath string) (string, string) {
- key := path.Join(certPath, constants.KeyFilename)
- cert := path.Join(certPath, constants.CertChainFilename)
- return key, cert
-}
-
-func fileExists(path string) bool {
- if fi, err := os.Stat(path); err == nil && fi.Mode().IsRegular() {
- return true
- }
- return false
-}
-
-func socketHealthCheck(ctx context.Context, socketPath string) error {
- ctx, cancel := context.WithDeadline(ctx, time.Now().Add(time.Second))
- defer cancel()
-
- conn, err := grpc.DialContext(ctx, fmt.Sprintf("unix:%s", socketPath),
- grpc.WithTransportCredentials(insecure.NewCredentials()),
- grpc.FailOnNonTempDialError(true),
- grpc.WithReturnConnectionError(),
- grpc.WithBlock(),
- )
- if err != nil {
- return err
- }
- err = conn.Close()
- if err != nil {
- log.Infof("connection is not closed: %v", err)
- }
-
- return nil
-}
-
-func socketFileExists(path string) bool {
- if fi, err := os.Stat(path); err == nil && !fi.Mode().IsRegular() {
- return true
- }
- return false
-}
-
-func checkSocket(ctx context.Context, socketPath string) (bool, error) {
- socketExists := socketFileExists(socketPath)
- if !socketExists {
- return false, nil
- }
-
- err := socketHealthCheck(ctx, socketPath)
- if err != nil {
- log.Debugf("SDS socket detected but not healthy: %v", err)
- err = os.Remove(socketPath)
- if err != nil {
- return false, fmt.Errorf("existing SDS socket could not
be removed: %v", err)
- }
- return false, nil
- }
-
- return true, nil
-}
diff --git a/pkg/dubboagency/config/config.go b/pkg/dubboagency/config/config.go
deleted file mode 100644
index fd98d7d3..00000000
--- a/pkg/dubboagency/config/config.go
+++ /dev/null
@@ -1,98 +0,0 @@
-//
-// 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 config
-
-import (
- "fmt"
- "os"
-
- "github.com/apache/dubbo-kubernetes/pkg/bootstrap"
- "github.com/apache/dubbo-kubernetes/pkg/config/mesh"
- "github.com/apache/dubbo-kubernetes/pkg/log"
- meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
-)
-
-// ConstructProxyConfig returns proxyConfig
-func ConstructProxyConfig(meshConfigFile, proxyConfigEnv string)
(*meshv1alpha1.ProxyConfig, error) {
- annotations, err := bootstrap.ReadPodAnnotations("")
- if err != nil {
- if os.IsNotExist(err) {
- log.Debugf("failed to read pod annotations: %v", err)
- } else {
- log.Warnf("failed to read pod annotations: %v", err)
- }
- }
- var fileMeshContents string
- if fileExists(meshConfigFile) {
- contents, err := os.ReadFile(meshConfigFile)
- if err != nil {
- return nil, fmt.Errorf("failed to read mesh config file
%v: %v", meshConfigFile, err)
- }
- fileMeshContents = string(contents)
- }
- meshConfig, err := getMeshConfig(fileMeshContents,
annotations["proxy.dubbo.apache.org/config"], proxyConfigEnv)
- if err != nil {
- return nil, err
- }
- proxyConfig := mesh.DefaultProxyConfig()
- if meshConfig.DefaultConfig != nil {
- proxyConfig = meshConfig.DefaultConfig
- }
- // TODO ResolveAddr
- // TODO ValidateMeshConfigProxyConfig
- return proxyConfig, nil
-}
-
-func getMeshConfig(fileOverride, annotationOverride, proxyConfigEnv string)
(*meshv1alpha1.MeshConfig, error) {
- mc := mesh.DefaultMeshConfig()
- if fileOverride != "" {
- log.Infof("Apply mesh config from file %v", fileOverride)
- fileMesh, err := mesh.ApplyMeshConfig(fileOverride, mc)
- if err != nil || fileMesh == nil {
- return nil, fmt.Errorf("failed to unmarshal mesh config
from file [%v]: %v", fileOverride, err)
- }
- mc = fileMesh
- }
-
- // Original order: env first, then annotation
- if proxyConfigEnv != "" {
- log.Infof("Apply proxy config from env %v", proxyConfigEnv)
- envMesh, err := mesh.ApplyProxyConfig(proxyConfigEnv, mc)
- if err != nil || envMesh == nil {
- return nil, fmt.Errorf("failed to unmarshal mesh config
from environment [%v]: %v", proxyConfigEnv, err)
- }
- mc = envMesh
- }
-
- if annotationOverride != "" {
- log.Infof("Apply proxy config from annotation %v",
annotationOverride)
- annotationMesh, err :=
mesh.ApplyProxyConfig(annotationOverride, mc)
- if err != nil || annotationMesh == nil {
- return nil, fmt.Errorf("failed to unmarshal mesh config
from annotation [%v]: %v", annotationOverride, err)
- }
- mc = annotationMesh
- }
-
- return mc, nil
-}
-
-func fileExists(path string) bool {
- if _, err := os.Stat(path); os.IsNotExist(err) {
- return false
- }
- return true
-}
diff --git a/pkg/dubboagency/plugins.go b/pkg/dubboagency/plugins.go
deleted file mode 100644
index 3a7f6ec3..00000000
--- a/pkg/dubboagency/plugins.go
+++ /dev/null
@@ -1,74 +0,0 @@
-//
-// 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 dubboagency
-
-import (
- "fmt"
- "github.com/apache/dubbo-kubernetes/pkg/log"
- "strings"
-
-
"github.com/apache/dubbo-kubernetes/dubbod/security/pkg/nodeagent/caclient/providers/aegis"
- "github.com/apache/dubbo-kubernetes/pkg/security"
-)
-
-var providers = make(map[string]func(*security.Options, RootCertProvider)
(security.Client, error))
-
-type RootCertProvider interface {
- GetKeyCertsForCA() (string, string)
- FindRootCAForCA() (string, error)
-}
-
-func createAegis(opts *security.Options, a RootCertProvider) (security.Client,
error) {
- var tlsOpts *aegis.TLSOptions
- var err error
-
- if strings.HasSuffix(opts.CAEndpoint, ":26010") {
- log.Warn("Debug mode or IP-secure network")
- } else {
- tlsOpts = &aegis.TLSOptions{}
- tlsOpts.RootCert, err = a.FindRootCAForCA()
- if err != nil {
- return nil, fmt.Errorf("failed to find root CA cert for
CA: %v", err)
- }
-
- if tlsOpts.RootCert == "" {
- log.Infof("Using CA %s cert with system certs",
opts.CAEndpoint)
- } else if !fileExists(tlsOpts.RootCert) {
- log.Fatalf("invalid config - %s missing a root
certificate %s", opts.CAEndpoint, tlsOpts.RootCert)
- } else {
- log.Infof("Using CA %s cert with certs: %s",
opts.CAEndpoint, tlsOpts.RootCert)
- }
-
- tlsOpts.Key, tlsOpts.Cert = a.GetKeyCertsForCA()
- }
-
- tlsOpts.Key, tlsOpts.Cert = a.GetKeyCertsForCA()
-
- return aegis.NewAegisClient(opts, tlsOpts)
-}
-
-func createCAClient(opts *security.Options, a RootCertProvider)
(security.Client, error) {
- provider, ok := providers[opts.CAProviderName]
- if !ok {
- return nil, fmt.Errorf("CA provider %q not registered",
opts.CAProviderName)
- }
- return provider(opts, a)
-}
-
-func init() {
- providers["Aegis"] = createAegis
-}
diff --git a/pkg/dubboagency/xds_proxy.go b/pkg/dubboagency/xds_proxy.go
deleted file mode 100644
index 1c777cda..00000000
--- a/pkg/dubboagency/xds_proxy.go
+++ /dev/null
@@ -1,907 +0,0 @@
-//
-// 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 dubboagency
-
-import (
- "context"
- "fmt"
- "math"
- "net"
- "path/filepath"
- "sync"
- "time"
-
- "github.com/apache/dubbo-kubernetes/pkg/log"
-
- dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/grpc"
-
"github.com/apache/dubbo-kubernetes/dubbod/security/pkg/nodeagent/caclient"
- "github.com/apache/dubbo-kubernetes/pkg/channels"
- dubbokeepalive "github.com/apache/dubbo-kubernetes/pkg/keepalive"
- "github.com/apache/dubbo-kubernetes/pkg/model"
- "github.com/apache/dubbo-kubernetes/pkg/uds"
- meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
- core "github.com/kdubbo/xds-api/core/v1"
- discovery "github.com/kdubbo/xds-api/service/discovery/v1"
- "go.uber.org/atomic"
- google_rpc "google.golang.org/genproto/googleapis/rpc/status"
- "google.golang.org/grpc"
- "google.golang.org/grpc/codes"
- "google.golang.org/grpc/credentials/insecure"
- "google.golang.org/grpc/metadata"
- "google.golang.org/grpc/reflection"
- "google.golang.org/protobuf/types/known/anypb"
-)
-
-const (
- defaultClientMaxReceiveMessageSize = math.MaxInt32
-)
-
-var proxyLog = log.RegisterScope("xdsproxy", "xDS Proxy in Dubbo Agent")
-
-var connectionNumber = atomic.NewUint32(0)
-
-type adsStream interface {
- Send(*discovery.DiscoveryResponse) error
- Recv() (*discovery.DiscoveryRequest, error)
- Context() context.Context
-}
-
-type (
- DiscoveryStream =
discovery.AggregatedDiscoveryService_StreamAggregatedResourcesServer
- DeltaDiscoveryStream =
discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer
- DiscoveryClient =
discovery.AggregatedDiscoveryService_StreamAggregatedResourcesClient
- DeltaDiscoveryClient =
discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesClient
-)
-
-type ResponseHandler func(resp *anypb.Any) error
-
-type ProxyConnection struct {
- conID uint32
- upstreamError chan error
- downstreamError chan error
- requestsChan *channels.Unbounded[*discovery.DiscoveryRequest]
- responsesChan chan *discovery.DiscoveryResponse
- deltaRequestsChan *channels.Unbounded[*discovery.DeltaDiscoveryRequest]
- deltaResponsesChan chan *discovery.DeltaDiscoveryResponse
- stopChan chan struct{}
- downstream adsStream
- upstream DiscoveryClient
- downstreamDeltas DeltaDiscoveryStream
- upstreamDeltas DeltaDiscoveryClient
- node *core.Node // Preserve Node from first request
- nodeMutex sync.RWMutex
-}
-
-type XdsProxy struct {
- discovery.UnimplementedAggregatedDiscoveryServiceServer
-
- stopChan chan struct{}
- downstreamGrpcServer *grpc.Server
- downstreamListener net.Listener
- optsMutex sync.RWMutex
- dialOptions []grpc.DialOption
- dubbodSAN string
- dubbodAddress string
- xdsHeaders map[string]string
- xdsUdsPath string
- proxyAddresses []string
- ia *Agent
- clusterID string
- handlers map[string]ResponseHandler
- downstreamGrpcOptions []grpc.ServerOption
- connectedMutex sync.RWMutex
- connected *ProxyConnection
- ecdsLastAckVersion atomic.String
- ecdsLastNonce atomic.String
- initialHealthRequest *discovery.DiscoveryRequest
- initialDeltaHealthRequest *discovery.DeltaDiscoveryRequest
- // Upstream connection for proxyless mode
- bootstrapNode *core.Node
- ConnMutex sync.RWMutex
- Conn *ProxyConnection
-}
-
-func initXdsProxy(ia *Agent) (*XdsProxy, error) {
- var err error
-
- proxy := &XdsProxy{
- dubbodAddress: ia.proxyConfig.DiscoveryAddress,
- dubbodSAN: ia.cfg.DubbodSAN,
- clusterID: ia.secOpts.ClusterID,
- stopChan: make(chan struct{}),
- xdsHeaders: ia.cfg.XDSHeaders,
- xdsUdsPath: ia.cfg.XdsUdsPath,
- proxyAddresses: ia.cfg.ProxyIPAddresses,
- ia: ia,
- downstreamGrpcOptions: ia.cfg.DownstreamGrpcOptions,
- handlers: make(map[string]ResponseHandler),
- }
-
- // Initialize dial options immediately, required for connecting to
upstream
- if err = proxy.initDubbodDialOptions(ia); err != nil {
- return nil, fmt.Errorf("failed to init dubbod dial options:
%v", err)
- }
-
- proxyLog.Infof("Initializing with upstream address %q and cluster %q",
proxy.dubbodAddress, proxy.clusterID)
-
- if err = proxy.initDownstreamServer(); err != nil {
- return nil, err
- }
-
- ia.wg.Add(1)
- go func() {
- defer ia.wg.Done()
- proxyLog.Infof("Starting XDS proxy server listening on UDS:
%s", proxy.xdsUdsPath)
- proxyLog.Infof("XDS proxy server ready to accept connections on
UDS: %s", proxy.xdsUdsPath)
- if err :=
proxy.downstreamGrpcServer.Serve(proxy.downstreamListener); err != nil {
- proxyLog.Errorf("XDS proxy server stopped accepting
connections: %v", err)
- }
- proxyLog.Warnf("XDS proxy server stopped serving on UDS: %s",
proxy.xdsUdsPath)
- }()
-
- // For proxyless mode, establish an upstream connection using bootstrap
Node
- // This ensures the proxy is ready even before downstream clients
connect
- // Use a retry loop with exponential backoff to automatically reconnect
on failures
- ia.wg.Add(1)
- go func() {
- defer ia.wg.Done()
- // Wait for bootstrap Node to be set (with timeout)
- // The Node is set synchronously after bootstrap file
generation in agent.Run()
- for i := 0; i < 50; i++ {
- proxy.ConnMutex.RLock()
- nodeReady := proxy.bootstrapNode != nil
- proxy.ConnMutex.RUnlock()
- if nodeReady {
- break
- }
- select {
- case <-proxy.stopChan:
- return
- case <-time.After(100 * time.Millisecond):
- }
- }
- proxy.ConnMutex.RLock()
- nodeReady := proxy.bootstrapNode != nil
- proxy.ConnMutex.RUnlock()
- if !nodeReady {
- proxyLog.Warnf("Bootstrap Node not set after 5 seconds,
proceeding anyway")
- }
-
- maxBackoff := 30 * time.Second
- backoff := time.Second
-
- for {
- select {
- case <-proxy.stopChan:
- return
- default:
- }
-
- // Establish connection
- connDone, err := proxy.establishConnection(ia)
- if err != nil {
- // Connection failed, log and retry with
exponential backoff
- proxyLog.Warnf("Failed to establish upstream
connection: %v, retrying in %v", err, backoff)
-
- select {
- case <-proxy.stopChan:
- return
- case <-time.After(backoff):
- backoff *= 2
- if backoff > maxBackoff {
- backoff = maxBackoff
- }
- }
- continue
- }
-
- // Connection successful, reset backoff
- backoff = time.Second
- proxyLog.Infof("Upstream Connected Successfully")
- // Wait for connection to terminate (connDone will be
closed when connection dies)
- select {
- case <-proxy.stopChan:
- return
- case <-connDone:
- proxyLog.Warnf("connection terminated, will
retry")
- }
- }
- }()
-
- return proxy, nil
-}
-
-func (p *XdsProxy) StreamAggregatedResources(downstream DiscoveryStream) error
{
- return p.handleStream(downstream)
-}
-
-func (p *XdsProxy) handleStream(downstream adsStream) error {
- conID := connectionNumber.Inc()
- proxyLog.Infof("new downstream connection #%d", conID)
- con := &ProxyConnection{
- conID: conID,
- upstreamError: make(chan error), // can be produced by recv
and send
- downstreamError: make(chan error), // can be produced by recv
and send
- requestsChan:
channels.NewUnbounded[*discovery.DiscoveryRequest](),
- responsesChan: make(chan *discovery.DiscoveryResponse, 1),
- stopChan: make(chan struct{}),
- downstream: downstream,
- }
-
- p.registerStream(con)
- defer p.unregisterStream(con)
- defer proxyLog.Infof("downstream connection #%d closed", conID)
-
- ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
- defer cancel()
-
- proxyLog.Infof("connection #%d building upstream connection to %s",
conID, p.dubbodAddress)
- upstreamConn, err := p.buildUpstreamConn(ctx)
- if err != nil {
- proxyLog.Errorf("connection #%d failed to connect to upstream
%s: %v", conID, p.dubbodAddress, err)
- return err
- }
- proxyLog.Infof("connection #%d successfully built upstream connection",
conID)
- defer upstreamConn.Close()
-
- xds := discovery.NewAggregatedDiscoveryServiceClient(upstreamConn)
- ctx = metadata.AppendToOutgoingContext(context.Background(),
"ClusterID", p.clusterID)
- for k, v := range p.xdsHeaders {
- ctx = metadata.AppendToOutgoingContext(ctx, k, v)
- }
- return p.handleUpstream(ctx, con, xds)
-}
-
-func (p *XdsProxy) handleUpstream(ctx context.Context, con *ProxyConnection,
xds discovery.AggregatedDiscoveryServiceClient) error {
- proxyLog.Infof("connection #%d connecting to upstream: %s", con.conID,
p.dubbodAddress)
- upstream, err := xds.StreamAggregatedResources(ctx,
- grpc.MaxCallRecvMsgSize(defaultClientMaxReceiveMessageSize))
- if err != nil {
- proxyLog.Errorf("connection #%d failed to create stream to
upstream %s: %v", con.conID, p.dubbodAddress, err)
- return err
- }
- proxyLog.Infof("Connected to upstream XDS server: %s id=%d",
p.dubbodAddress, con.conID)
-
- // Log when we start receiving responses
- go func() {
- firstResponse := true
- for {
- resp, err := upstream.Recv()
- if err != nil {
- if firstResponse {
- proxyLog.Warnf("connection #%d upstream
Recv failed before first response: %v", con.conID, err)
- }
- upstreamErr(con, err)
- return
- }
- if firstResponse {
- proxyLog.Infof("connection #%d received first
response from upstream: TypeUrl=%s, Resources=%d",
- con.conID,
model.GetShortType(resp.TypeUrl), len(resp.Resources))
- firstResponse = false
- } else {
- proxyLog.Debugf("connection #%d received
response from upstream: TypeUrl=%s, Resources=%d, VersionInfo=%s",
- con.conID,
model.GetShortType(resp.TypeUrl), len(resp.Resources), resp.VersionInfo)
- }
- select {
- case con.responsesChan <- resp:
- case <-con.stopChan:
- return
- }
- }
- }()
-
- con.upstream = upstream
-
- go p.handleUpstreamRequest(con)
- go p.handleUpstreamResponse(con)
-
- for {
- select {
- case err := <-con.upstreamError:
- return err
- case err := <-con.downstreamError:
- return err
- case <-con.stopChan:
- return nil
- }
- }
-}
-
-func (p *XdsProxy) handleUpstreamRequest(con *ProxyConnection) {
- initialRequestsSent := atomic.NewBool(false)
- nodeReceived := atomic.NewBool(false)
- go func() {
- for {
- req, err := con.downstream.Recv()
- if err != nil {
- proxyLog.Warnf("connection #%d downstream Recv
error: %v", con.conID, err)
- downstreamErr(con, err)
- return
- }
-
- proxyLog.Debugf("connection #%d received request:
TypeUrl=%s, Node=%v, ResourceNames=%d",
- con.conID, model.GetShortType(req.TypeUrl),
req.Node != nil, len(req.ResourceNames))
-
- // Save Node from first request that contains it
- if req.Node != nil && req.Node.Id != "" {
- con.nodeMutex.Lock()
- if con.node == nil {
- // Deep copy to preserve the Node
information
- con.node = &core.Node{
- Id: req.Node.Id,
- Cluster: req.Node.Cluster,
- Locality: req.Node.Locality,
- Metadata: req.Node.Metadata,
- }
- proxyLog.Infof("connection #%d saved
Node: %s", con.conID, req.Node.Id)
- nodeReceived.Store(true)
- }
- con.nodeMutex.Unlock()
- }
-
- // Skip health check probes that don't have Node
information (before we've received any Node)
- if req.TypeUrl == model.HealthInfoType &&
!nodeReceived.Load() {
- proxyLog.Debugf("connection #%d skipping health
check probe without Node", con.conID)
- continue
- }
-
- // For LDS requests (typically the first request), we
must have Node for connection initialization
- // For other requests, if we already have Node saved,
we can inject it
- con.nodeMutex.RLock()
- hasNode := con.node != nil
- con.nodeMutex.RUnlock()
-
- // For XDS, any first request (not just LDS) without
Node should wait
- // because dubbo-discovery needs Node in the first
request to initialize connection
- if !nodeReceived.Load() && req.Node == nil &&
req.TypeUrl != model.HealthInfoType {
- proxyLog.Debugf("connection #%d received first
request without Node (TypeUrl=%s), waiting for Node information", con.conID,
model.GetShortType(req.TypeUrl))
- continue
- }
-
- // Ensure Node is set in request before forwarding
- // For proxyless gRPC, we must ensure Node is always
set in every request
- con.nodeMutex.RLock()
- if con.node != nil {
- if req.Node == nil {
- // Deep copy Node to avoid race
conditions
- req.Node = &core.Node{
- Id: con.node.Id,
- Cluster: con.node.Cluster,
- Locality: con.node.Locality,
- Metadata: con.node.Metadata,
- }
- proxyLog.Debugf("connection #%d added
saved Node to request", con.conID)
- } else if req.Node.Id == "" {
- // If Node exists but Id is empty, copy
from saved node
- req.Node.Id = con.node.Id
- if req.Node.Metadata == nil {
- req.Node.Metadata =
con.node.Metadata
- }
- proxyLog.Debugf("connection #%d filled
empty Node.Id in request", con.conID)
- }
- }
- con.nodeMutex.RUnlock()
-
- // Final check: for initialization, we must have Node
- if !hasNode && req.Node == nil && req.TypeUrl !=
model.HealthInfoType {
- proxyLog.Warnf("connection #%d cannot forward
request without Node: TypeUrl=%s", con.conID, model.GetShortType(req.TypeUrl))
- continue
- }
-
- // forward to dubbod
- con.sendRequest(req)
- if !initialRequestsSent.Load() && req.TypeUrl ==
model.ListenerType {
- // fire off an initial PCDS request
- if _, f := p.handlers[model.ProxyConfigType]; f
{
- pcdsReq := &discovery.DiscoveryRequest{
- TypeUrl: model.ProxyConfigType,
- }
- // Include Node in internal requests
- con.nodeMutex.RLock()
- if con.node != nil {
- pcdsReq.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendRequest(pcdsReq)
- }
- // set flag before sending the initial request
to prevent race.
- initialRequestsSent.Store(true)
- // Fire of a configured initial request, if
there is one
- p.connectedMutex.RLock()
- initialRequest := p.initialHealthRequest
- if initialRequest != nil {
- // Ensure Node is set in initial health
request
- con.nodeMutex.RLock()
- if con.node != nil &&
initialRequest.Node == nil {
- initialRequest.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendRequest(initialRequest)
- }
- p.connectedMutex.RUnlock()
- }
- }
- }()
-
- defer con.upstream.CloseSend() // nolint
- for {
- select {
- case req := <-con.requestsChan.Get():
- con.requestsChan.Load()
- if req.TypeUrl == model.HealthInfoType &&
!initialRequestsSent.Load() {
- // only send healthcheck probe after LDS
request has been sent
- continue
- }
-
- // Ensure Node is set before sending to upstream
- // For proxyless gRPC, we must ensure Node is always
set in every request
- con.nodeMutex.RLock()
- if con.node != nil {
- if req.Node == nil {
- // Deep copy Node to avoid race
conditions
- req.Node = &core.Node{
- Id: con.node.Id,
- Cluster: con.node.Cluster,
- Locality: con.node.Locality,
- Metadata: con.node.Metadata,
- }
- } else if req.Node.Id == "" {
- // If Node exists but Id is empty, copy
from saved node
- req.Node.Id = con.node.Id
- if req.Node.Metadata == nil {
- req.Node.Metadata =
con.node.Metadata
- }
- }
- }
- con.nodeMutex.RUnlock()
-
- // Final safety check: don't send if still no Node
- if req.Node == nil || req.Node.Id == "" {
- proxyLog.Warnf("connection #%d cannot send
request without Node: TypeUrl=%s", con.conID, model.GetShortType(req.TypeUrl))
- continue
- }
-
- // Only log at debug level to reduce noise - these are
normal operations
- proxyLog.Debugf("connection #%d forwarding request:
TypeUrl=%s, Node=%v",
- con.conID, model.GetShortType(req.TypeUrl),
req.Node != nil && req.Node.Id != "")
- if err := con.upstream.Send(req); err != nil {
- proxyLog.Errorf("connection #%d failed to send
request upstream: TypeUrl=%s, error=%v",
- con.conID,
model.GetShortType(req.TypeUrl), err)
- err = fmt.Errorf("send error for type url %s:
%v", req.TypeUrl, err)
- upstreamErr(con, err)
- return
- }
- // Only log at debug level to reduce noise - successful
sends are normal
- proxyLog.Debugf("connection #%d successfully sent
request upstream: TypeUrl=%s",
- con.conID, model.GetShortType(req.TypeUrl))
- case <-con.stopChan:
- return
- }
- }
-}
-
-func (p *XdsProxy) handleUpstreamResponse(con *ProxyConnection) {
- for {
- select {
- case resp := <-con.responsesChan:
- // TODO: separate upstream response handling from
requests sending, which are both time costly
- proxyLog.Debugf("Upstream response id=%d type=%s
resources=%d",
- con.conID,
- model.GetShortType(resp.TypeUrl),
- len(resp.Resources))
- // Handle internal types (e.g., ProxyConfig) that need
special processing
- hasHandler := false
- if h, f := p.handlers[resp.TypeUrl]; f {
- hasHandler = true
- if len(resp.Resources) > 0 {
- // Process the resource with the handler
- err := h(resp.Resources[0])
- var errorResp *google_rpc.Status
- if err != nil {
- errorResp = &google_rpc.Status{
- Code:
int32(codes.Internal),
- Message: err.Error(),
- }
- }
- // Send ACK/NACK for internal types
- ackReq := &discovery.DiscoveryRequest{
- VersionInfo: resp.VersionInfo,
- TypeUrl: resp.TypeUrl,
- ResponseNonce: resp.Nonce,
- ErrorDetail: errorResp,
- }
- // Ensure Node is set in ACK requests
- con.nodeMutex.RLock()
- if con.node != nil {
- ackReq.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendRequest(ackReq)
- // For ProxyConfig type, we don't
forward to downstream as it's only for agent
- if resp.TypeUrl ==
model.ProxyConfigType {
- continue
- }
- }
- }
-
- // Forward all non-internal responses to downstream
(gRPC client)
- proxyLog.Debugf("connection #%d forwarding response to
downstream: TypeUrl=%s, Resources=%d, VersionInfo=%s",
- con.conID, model.GetShortType(resp.TypeUrl),
len(resp.Resources), resp.VersionInfo)
- if err := con.downstream.Send(resp); err != nil {
- proxyLog.Errorf("connection #%d failed to send
response to downstream: %v", con.conID, err)
- downstreamErr(con, err)
- return
- }
- proxyLog.Debugf("connection #%d successfully forwarded
response to downstream: TypeUrl=%s, Resources=%d",
- con.conID, model.GetShortType(resp.TypeUrl),
len(resp.Resources))
-
- // Send ACK for normal XDS responses (if not already
sent by handler)
- if !hasHandler {
- ackReq := &discovery.DiscoveryRequest{
- VersionInfo: resp.VersionInfo,
- TypeUrl: resp.TypeUrl,
- ResponseNonce: resp.Nonce,
- }
- // Ensure Node is set in ACK requests
- con.nodeMutex.RLock()
- if con.node != nil {
- ackReq.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendRequest(ackReq)
- }
- case <-con.stopChan:
- return
- }
- }
-}
-
-func (p *XdsProxy) buildUpstreamConn(ctx context.Context) (*grpc.ClientConn,
error) {
- p.optsMutex.RLock()
- opts := p.dialOptions
- p.optsMutex.RUnlock()
-
- if len(opts) == 0 {
- proxyLog.Warnf("dialOptions is empty, connection may fail.
Address: %s", p.dubbodAddress)
- }
-
- proxyLog.Debugf("dialing %s with %d options", p.dubbodAddress,
len(opts))
- conn, err := grpc.DialContext(ctx, p.dubbodAddress, opts...)
- if err != nil {
- return nil, fmt.Errorf("grpc.DialContext failed for %s: %w",
p.dubbodAddress, err)
- }
- return conn, nil
-}
-
-func (p *XdsProxy) unregisterStream(c *ProxyConnection) {
- p.connectedMutex.Lock()
- defer p.connectedMutex.Unlock()
- if p.connected != nil && p.connected == c {
- close(p.connected.stopChan)
- p.connected = nil
- }
-}
-
-func (p *XdsProxy) registerStream(c *ProxyConnection) {
- p.connectedMutex.Lock()
- defer p.connectedMutex.Unlock()
- if p.connected != nil {
- close(p.connected.stopChan)
- }
- p.connected = c
-}
-
-func (p *XdsProxy) initDownstreamServer() error {
- // Convert relative path to absolute path for UDS socket
- absPath, err := filepath.Abs(p.xdsUdsPath)
- if err != nil {
- return fmt.Errorf("failed to resolve absolute path for UDS:
%v", err)
- }
- p.xdsUdsPath = absPath
-
- l, err := uds.NewListener(p.xdsUdsPath)
- if err != nil {
- return err
- }
- // TODO: Expose keepalive options to agent cmd line flags.
- opts := p.downstreamGrpcOptions
- opts = append(opts,
dubbogrpc.ServerOptions(dubbokeepalive.DefaultOption())...)
- grpcs := grpc.NewServer(opts...)
- discovery.RegisterAggregatedDiscoveryServiceServer(grpcs, p)
- reflection.Register(grpcs)
- p.downstreamGrpcServer = grpcs
- p.downstreamListener = l
- proxyLog.Infof("XDS proxy server initialized, listening on UDS socket:
%s", p.xdsUdsPath)
- return nil
-}
-
-func (p *XdsProxy) initDubbodDialOptions(agent *Agent) error {
- opts, err := p.buildUpstreamClientDialOpts(agent)
- if err != nil {
- return err
- }
-
- p.optsMutex.Lock()
- p.dialOptions = opts
- p.optsMutex.Unlock()
- return nil
-}
-
-func (p *XdsProxy) buildUpstreamClientDialOpts(sa *Agent) ([]grpc.DialOption,
error) {
- // For NONE auth policy, use insecure credentials
- if sa.proxyConfig.ControlPlaneAuthPolicy ==
meshv1alpha1.AuthenticationPolicy_NONE {
- options := []grpc.DialOption{
-
grpc.WithTransportCredentials(insecure.NewCredentials()),
- }
- if sa.secOpts.CredFetcher != nil {
- options = append(options,
grpc.WithPerRPCCredentials(caclient.NewDefaultTokenProvider(sa.secOpts)))
- }
- return options, nil
- }
-
- tlsOpts, err := p.getTLSOptions(sa)
- if err != nil {
- return nil, fmt.Errorf("failed to get TLS options to talk to
upstream: %v", err)
- }
- options, err := dubbogrpc.ClientOptions(nil, tlsOpts)
- if err != nil {
- return nil, err
- }
- if sa.secOpts.CredFetcher != nil {
- options = append(options,
grpc.WithPerRPCCredentials(caclient.NewDefaultTokenProvider(sa.secOpts)))
- }
- return options, nil
-}
-
-func (p *XdsProxy) getTLSOptions(agent *Agent) (*dubbogrpc.TLSOptions, error) {
- if agent.proxyConfig.ControlPlaneAuthPolicy ==
meshv1alpha1.AuthenticationPolicy_NONE {
- return nil, nil
- }
- xdsCACertPath, err := agent.FindRootCAForXDS()
- if err != nil {
- return nil, fmt.Errorf("failed to find root CA cert for XDS:
%v", err)
- }
- key, cert := agent.GetKeyCertsForXDS()
- return &dubbogrpc.TLSOptions{
- RootCert: xdsCACertPath,
- Key: key,
- Cert: cert,
- ServerAddress: agent.proxyConfig.DiscoveryAddress,
- SAN: p.dubbodSAN,
- }, nil
-}
-
-func (p *XdsProxy) SetBootstrapNode(node *core.Node) {
- p.ConnMutex.Lock()
- p.bootstrapNode = node
- p.ConnMutex.Unlock()
-}
-
-func (p *XdsProxy) establishConnection(ia *Agent) (<-chan struct{}, error) {
- p.ConnMutex.Lock()
- node := p.bootstrapNode
- // Clean up old connection if it exists
- if p.Conn != nil {
- close(p.Conn.stopChan)
- p.Conn = nil
- }
- p.ConnMutex.Unlock()
-
- if node == nil {
- return nil, fmt.Errorf("bootstrap node not available")
- }
-
- if ia.cfg.ProxyType == model.Proxyless {
- proxyLog.Infof("Connecting proxyless upstream connection with
Node: %s", node.Id)
- } else if ia.cfg.ProxyType == model.Router {
- proxyLog.Infof("Connecting router upstream connection with
Node: %s", node.Id)
- }
-
- ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
- defer cancel()
-
- upstreamConn, err := p.buildUpstreamConn(ctx)
- if err != nil {
- return nil, fmt.Errorf("failed to build upstream connection:
%w", err)
- }
- // Don't defer close here - connection will be closed when connection
terminates
-
- // Create a channel that will be closed when connection terminates
- connDone := make(chan struct{})
-
- xds := discovery.NewAggregatedDiscoveryServiceClient(upstreamConn)
- ctx = metadata.AppendToOutgoingContext(context.Background(),
"ClusterID", p.clusterID)
- for k, v := range p.xdsHeaders {
- ctx = metadata.AppendToOutgoingContext(ctx, k, v)
- }
-
- upstream, err := xds.StreamAggregatedResources(ctx,
- grpc.MaxCallRecvMsgSize(defaultClientMaxReceiveMessageSize))
- if err != nil {
- _ = upstreamConn.Close()
- return nil, fmt.Errorf("failed to create upstream stream: %w",
err)
- }
- proxyLog.Infof("Connected to upstream XDS server: %s", p.dubbodAddress)
-
- conID := connectionNumber.Inc()
- con := &ProxyConnection{
- conID: conID,
- upstreamError: make(chan error, 1),
- requestsChan:
channels.NewUnbounded[*discovery.DiscoveryRequest](),
- responsesChan: make(chan *discovery.DiscoveryResponse, 1),
- stopChan: make(chan struct{}),
- upstream: upstream,
- node: node,
- }
-
- p.ConnMutex.Lock()
- p.Conn = con
- p.ConnMutex.Unlock()
-
- // Close upstream connection when connection terminates and signal done
channel
- go func() {
- select {
- case <-con.stopChan:
- case <-con.upstreamError:
- case <-p.stopChan:
- }
- _ = upstreamConn.Close()
- p.ConnMutex.Lock()
- if p.Conn == con {
- p.Conn = nil
- }
- p.ConnMutex.Unlock()
- close(connDone)
- }()
-
- // Send initial LDS request with bootstrap Node to initialize connection
- ldsReq := &discovery.DiscoveryRequest{
- TypeUrl: model.ListenerType,
- Node: node,
- }
- proxyLog.Debugf("connection sending initial LDS request with Node: %s",
node.Id)
- if err := upstream.Send(ldsReq); err != nil {
- _ = upstreamConn.Close()
- p.ConnMutex.Lock()
- if p.Conn == con {
- p.Conn = nil
- }
- p.ConnMutex.Unlock()
- close(connDone)
- return nil, fmt.Errorf("failed to send initial LDS request:
%w", err)
- }
-
- // Handle responses (discard or queue for future downstream clients)
- go func() {
- defer func() {
- // Signal connection termination
- select {
- case con.upstreamError <- fmt.Errorf("response handler
terminated"):
- case <-con.stopChan:
- }
- }()
- for {
- select {
- case <-con.stopChan:
- return
- case <-p.stopChan:
- return
- default:
- }
-
- resp, err := upstream.Recv()
- if err != nil {
- upstreamErr(con, err)
- return
- }
- proxyLog.Debugf("connection received response:
TypeUrl=%s, Resources=%d",
- model.GetShortType(resp.TypeUrl),
len(resp.Resources))
- // Send ACK
- ackReq := &discovery.DiscoveryRequest{
- VersionInfo: resp.VersionInfo,
- TypeUrl: resp.TypeUrl,
- ResponseNonce: resp.Nonce,
- Node: node,
- }
- con.sendRequest(ackReq)
- }
- }()
-
- // Send requests
- go func() {
- defer func() {
- // Signal connection termination if send handler exits
- select {
- case con.upstreamError <- fmt.Errorf("send handler
terminated"):
- case <-con.stopChan:
- }
- }()
- for {
- select {
- case req := <-con.requestsChan.Get():
- con.requestsChan.Load()
- if req.Node == nil {
- req.Node = node
- }
- if err := upstream.Send(req); err != nil {
- upstreamErr(con, err)
- return
- }
- case <-con.stopChan:
- return
- case <-p.stopChan:
- return
- }
- }
- }()
-
- // Return immediately - connection is connected and running in
background
- // The connDone channel will be closed when connection terminates
- return connDone, nil
-}
-
-func (p *XdsProxy) close() {
- close(p.stopChan)
- p.ConnMutex.Lock()
- if p.Conn != nil {
- close(p.Conn.stopChan)
- p.Conn = nil
- }
- p.ConnMutex.Unlock()
- if p.downstreamGrpcServer != nil {
- p.downstreamGrpcServer.Stop()
- }
- if p.downstreamListener != nil {
- _ = p.downstreamListener.Close()
- }
-}
-
-func (con *ProxyConnection) sendRequest(req *discovery.DiscoveryRequest) {
- con.requestsChan.Put(req)
-}
-
-func upstreamErr(con *ProxyConnection, err error) {
- switch dubbogrpc.GRPCErrorType(err) {
- case dubbogrpc.GracefulTermination:
- err = nil
- fallthrough
- case dubbogrpc.ExpectedError:
- proxyLog.Errorf("upstream terminated with status %v", err)
- default:
- proxyLog.Errorf("upstream terminated with unexpected error %v",
err)
- }
- select {
- case con.upstreamError <- err:
- case <-con.stopChan:
- }
-}
-
-func downstreamErr(con *ProxyConnection, err error) {
- switch dubbogrpc.GRPCErrorType(err) {
- case dubbogrpc.GracefulTermination:
- err = nil
- fallthrough
- case dubbogrpc.ExpectedError:
- proxyLog.Errorf("downstream terminated with status %v", err)
- default:
- proxyLog.Errorf("downstream terminated with unexpected error
%v", err)
- }
- select {
- case con.downstreamError <- err:
- case <-con.stopChan:
- }
-}
diff --git a/pkg/dubboagency/xds_proxy_delta.go
b/pkg/dubboagency/xds_proxy_delta.go
deleted file mode 100644
index 35e94e73..00000000
--- a/pkg/dubboagency/xds_proxy_delta.go
+++ /dev/null
@@ -1,323 +0,0 @@
-//
-// 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 dubboagency
-
-import (
- "context"
- "fmt"
- "time"
-
- "github.com/apache/dubbo-kubernetes/pkg/channels"
- "github.com/apache/dubbo-kubernetes/pkg/model"
- core "github.com/kdubbo/xds-api/core/v1"
- discovery "github.com/kdubbo/xds-api/service/discovery/v1"
- "go.uber.org/atomic"
- google_rpc "google.golang.org/genproto/googleapis/rpc/status"
- "google.golang.org/grpc"
- "google.golang.org/grpc/codes"
- "google.golang.org/grpc/metadata"
-)
-
-func (p *XdsProxy) DeltaAggregatedResources(downstream DeltaDiscoveryStream)
error {
- conID := connectionNumber.Inc()
- proxyLog.Infof("new delta downstream connection #%d", conID)
- con := &ProxyConnection{
- conID: conID,
- upstreamError: make(chan error), // can be produced by recv
and send
- downstreamError: make(chan error), // can be produced by recv
and send
- deltaRequestsChan:
channels.NewUnbounded[*discovery.DeltaDiscoveryRequest](),
- // Allow a buffer of 1. This ensures we queue up at most 2 (one
in process, 1 pending) responses before forwarding.
- deltaResponsesChan: make(chan
*discovery.DeltaDiscoveryResponse, 1),
- stopChan: make(chan struct{}),
- downstreamDeltas: downstream,
- }
- p.registerStream(con)
- defer p.unregisterStream(con)
- defer proxyLog.Infof("delta downstream connection #%d closed", conID)
-
- ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
- defer cancel()
-
- proxyLog.Infof("delta connection #%d building upstream connection to
%s", conID, p.dubbodAddress)
- upstreamConn, err := p.buildUpstreamConn(ctx)
- if err != nil {
- proxyLog.Errorf("delta connection #%d failed to connect to
upstream %s: %v", conID, p.dubbodAddress, err)
- return err
- }
- proxyLog.Infof("delta connection #%d successfully built upstream
connection", conID)
- defer upstreamConn.Close()
-
- xds := discovery.NewAggregatedDiscoveryServiceClient(upstreamConn)
- ctx = metadata.AppendToOutgoingContext(context.Background(),
"ClusterID", p.clusterID)
- for k, v := range p.xdsHeaders {
- ctx = metadata.AppendToOutgoingContext(ctx, k, v)
- }
- return p.handleDeltaUpstream(ctx, con, xds)
-}
-
-func (p *XdsProxy) handleDeltaUpstream(ctx context.Context, con
*ProxyConnection, xds discovery.AggregatedDiscoveryServiceClient) error {
- proxyLog.Infof("delta connection #%d connecting to upstream: %s",
con.conID, p.dubbodAddress)
- deltaUpstream, err := xds.DeltaAggregatedResources(ctx,
- grpc.MaxCallRecvMsgSize(defaultClientMaxReceiveMessageSize))
- if err != nil {
- proxyLog.Errorf("delta connection #%d failed to create stream
to upstream %s: %v", con.conID, p.dubbodAddress, err)
- return err
- }
- proxyLog.Infof("xdsproxy connected to delta upstream XDS server: %s
id=%d", p.dubbodAddress, con.conID)
- defer proxyLog.Infof("xdsproxy disconnected from delta XDS server: %s
id=%d", p.dubbodAddress, con.conID)
-
- con.upstreamDeltas = deltaUpstream
-
- go func() {
- for {
- resp, err := con.upstreamDeltas.Recv()
- if err != nil {
- upstreamErr(con, err)
- return
- }
- select {
- case con.deltaResponsesChan <- resp:
- case <-con.stopChan:
- }
- }
- }()
-
- go p.handleUpstreamDeltaRequest(con)
- go p.handleUpstreamDeltaResponse(con)
-
- for {
- select {
- case err := <-con.upstreamError:
- return err
- case err := <-con.downstreamError:
- return err
- case <-con.stopChan:
- return nil
- }
- }
-}
-
-func (p *XdsProxy) handleUpstreamDeltaRequest(con *ProxyConnection) {
- initialRequestsSent := atomic.NewBool(false)
- nodeReceived := atomic.NewBool(false)
- go func() {
- for {
- req, err := con.downstreamDeltas.Recv()
- if err != nil {
- downstreamErr(con, err)
- return
- }
-
- // Save Node from first request that contains it
- if req.Node != nil && req.Node.Id != "" {
- con.nodeMutex.Lock()
- if con.node == nil {
- // Deep copy to preserve the Node
information
- con.node = &core.Node{
- Id: req.Node.Id,
- Cluster: req.Node.Cluster,
- Locality: req.Node.Locality,
- Metadata: req.Node.Metadata,
- }
- proxyLog.Debugf("delta connection #%d
saved Node: %s", con.conID, req.Node.Id)
- nodeReceived.Store(true)
- }
- con.nodeMutex.Unlock()
- }
-
- // Skip health check probes that don't have Node
information (before we've received any Node)
- if req.TypeUrl == model.HealthInfoType &&
!nodeReceived.Load() {
- proxyLog.Debugf("delta connection #%d skipping
health check probe without Node", con.conID)
- continue
- }
-
- // For LDS requests (typically the first request), we
must have Node for connection initialization
- // For other requests, if we already have Node saved,
we can inject it
- con.nodeMutex.RLock()
- hasNode := con.node != nil
- con.nodeMutex.RUnlock()
-
- // For Delta XDS, any first request (not just LDS)
without Node should wait
- // because dubbo-discovery needs Node in the first
request to initialize connection
- if !nodeReceived.Load() && req.Node == nil &&
req.TypeUrl != model.HealthInfoType {
- proxyLog.Debugf("delta connection #%d received
first request without Node (TypeUrl=%s), waiting for Node information",
con.conID, model.GetShortType(req.TypeUrl))
- continue
- }
-
- // Ensure Node is set in request before forwarding
- con.nodeMutex.RLock()
- if con.node != nil && req.Node == nil {
- req.Node = con.node
- }
- con.nodeMutex.RUnlock()
-
- // Final check: for initialization, we must have Node
- if !hasNode && req.Node == nil && req.TypeUrl !=
model.HealthInfoType {
- proxyLog.Debugf("delta connection #%d cannot
forward request without Node: TypeUrl=%s", con.conID,
model.GetShortType(req.TypeUrl))
- continue
- }
-
- // forward to dubbod
- con.sendDeltaRequest(req)
- if !initialRequestsSent.Load() && req.TypeUrl ==
model.ListenerType {
- // fire off an initial PCDS request
- if _, f := p.handlers[model.ProxyConfigType]; f
{
- pcdsReq :=
&discovery.DeltaDiscoveryRequest{
- TypeUrl: model.ProxyConfigType,
- }
- // Include Node in internal requests
- con.nodeMutex.RLock()
- if con.node != nil {
- pcdsReq.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendDeltaRequest(pcdsReq)
- }
- // set flag before sending the initial request
to prevent race.
- initialRequestsSent.Store(true)
- // Fire of a configured initial request, if
there is one
- p.connectedMutex.RLock()
- initialRequest := p.initialDeltaHealthRequest
- if initialRequest != nil {
- // Ensure Node is set in initial health
request
- con.nodeMutex.RLock()
- if con.node != nil &&
initialRequest.Node == nil {
- initialRequest.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendDeltaRequest(initialRequest)
- }
- p.connectedMutex.RUnlock()
- }
- }
- }()
-
- defer func() {
- _ = con.upstreamDeltas.CloseSend()
- }()
- for {
- select {
- case req := <-con.deltaRequestsChan.Get():
- con.deltaRequestsChan.Load()
- if req.TypeUrl == model.HealthInfoType &&
!initialRequestsSent.Load() {
- // only send healthcheck probe after LDS
request has been sent
- continue
- }
-
- // Ensure Node is set before sending to upstream
- con.nodeMutex.RLock()
- if con.node != nil && req.Node == nil {
- req.Node = con.node
- }
- con.nodeMutex.RUnlock()
-
- // Final safety check: don't send if still no Node
- if req.Node == nil || req.Node.Id == "" {
- proxyLog.Warnf("delta connection #%d cannot
send request without Node: TypeUrl=%s", con.conID,
model.GetShortType(req.TypeUrl))
- continue
- }
-
- proxyLog.Debugf("Delta request type=%s sub=%d unsub=%d
nonce=%s initial=%d",
- model.GetShortType(req.TypeUrl),
- len(req.ResourceNamesSubscribe),
- len(req.ResourceNamesUnsubscribe),
- req.ResponseNonce,
- len(req.InitialResourceVersions))
- if err := con.upstreamDeltas.Send(req); err != nil {
- err = fmt.Errorf("send error for type url %s:
%v", req.TypeUrl, err)
- upstreamErr(con, err)
- return
- }
- case <-con.stopChan:
- return
- }
- }
-}
-
-func (p *XdsProxy) handleUpstreamDeltaResponse(con *ProxyConnection) {
- for {
- select {
- case resp := <-con.deltaResponsesChan:
- // TODO: separate upstream response handling from
requests sending, which are both time costly
- proxyLog.Debugf(
- "Upstream delta response id=%d type=%s nonce=%s
resources=%d removes=%d",
- con.conID,
- model.GetShortType(resp.TypeUrl),
- resp.Nonce,
- len(resp.Resources),
- len(resp.RemovedResources),
- )
- // Handle internal types (e.g., ProxyConfig) that need
special processing
- if h, f := p.handlers[resp.TypeUrl]; f {
- if len(resp.Resources) > 0 {
- // Process the resource with the handler
- err := h(resp.Resources[0].Resource)
- var errorResp *google_rpc.Status
- if err != nil {
- errorResp = &google_rpc.Status{
- Code:
int32(codes.Internal),
- Message: err.Error(),
- }
- }
- // Send ACK/NACK
- ackReq :=
&discovery.DeltaDiscoveryRequest{
- TypeUrl: resp.TypeUrl,
- ResponseNonce: resp.Nonce,
- ErrorDetail: errorResp,
- }
- // Ensure Node is set in ACK requests
- con.nodeMutex.RLock()
- if con.node != nil {
- ackReq.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendDeltaRequest(ackReq)
- }
- // Continue to forward to downstream for
transparency
- }
-
- // Forward all delta responses to downstream (gRPC
client)
- if err := con.downstreamDeltas.Send(resp); err != nil {
- proxyLog.Errorf("delta connection #%d failed to
send response to downstream: %v", con.conID, err)
- downstreamErr(con, err)
- return
- }
-
- // If there's no handler, we still need to send ACK for
normal XDS types
- if _, hasHandler := p.handlers[resp.TypeUrl];
!hasHandler {
- // Send ACK for normal XDS delta responses
- ackReq := &discovery.DeltaDiscoveryRequest{
- TypeUrl: resp.TypeUrl,
- ResponseNonce: resp.Nonce,
- }
- // Ensure Node is set in ACK requests
- con.nodeMutex.RLock()
- if con.node != nil {
- ackReq.Node = con.node
- }
- con.nodeMutex.RUnlock()
- con.sendDeltaRequest(ackReq)
- }
- case <-con.stopChan:
- return
- }
- }
-}
-
-func (con *ProxyConnection) sendDeltaRequest(req
*discovery.DeltaDiscoveryRequest) {
- con.deltaRequestsChan.Put(req)
-}
diff --git a/pkg/dubboagency/grpcxds/grpc_bootstrap.go
b/pkg/grpcxds/grpc_bootstrap.go
similarity index 100%
rename from pkg/dubboagency/grpcxds/grpc_bootstrap.go
rename to pkg/grpcxds/grpc_bootstrap.go
diff --git a/pkg/dubboagency/grpcxds/grpc_bootstrap_test.go
b/pkg/grpcxds/grpc_bootstrap_test.go
similarity index 100%
rename from pkg/dubboagency/grpcxds/grpc_bootstrap_test.go
rename to pkg/grpcxds/grpc_bootstrap_test.go
