You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by zh...@apache.org on 2026/05/14 04:45:28 UTC
(dubbo-kubernetes) branch master updated: Version 0.4.0 related functional logic (#907)
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 1127279a Version 0.4.0 related functional logic (#907)
1127279a is described below
commit 1127279ad4e036f36b2db7b14dda92d8c4bd1255
Author: mfordjody <11...@qq.com>
AuthorDate: Thu May 14 12:45:24 2026 +0800
Version 0.4.0 related functional logic (#907)
* Update kdubbo API module versions for MeshService
* Remove MeshService visibleTo usage
* Test proxyless mTLS xDS generation
* Wire proxyless runtime security config
* Update api and xds-api dependencies
* Add mtls gateway logic and necessary configuration
---
.asf.yaml | 2 +-
dubbod/discovery/cmd/app/cmd.go | 2 +
dubbod/discovery/cmd/app/cmd_test.go | 5 +-
dubbod/discovery/cmd/app/xds_client.go | 161 ++++++++++++++---
dubbod/discovery/cmd/app/xds_client_test.go | 43 +++++
dubbod/discovery/cmd/app/xds_server.go | 193 ++++++++++++++++++++
dubbod/discovery/cmd/app/xds_server_test.go | 172 ++++++++++++++++++
.../pkg/bootstrap/proxyless_grpc_controller.go | 130 +++++++++++++-
.../bootstrap/proxyless_grpc_controller_test.go | 199 +++++++++++++++++++++
.../kube/gateway/deployment_controller_test.go | 9 +
dubbod/discovery/pkg/model/virtualservice.go | 4 +-
.../discovery/pkg/networking/grpcgen/mtls_test.go | 192 ++++++++++++++++++++
dubbod/discovery/pkg/xds/eds.go | 1 -
.../pkg/xds/endpoints/endpoint_builder.go | 51 +++++-
.../pkg/xds/endpoints/endpoint_builder_test.go | 167 +++++++++++++++++
dubboinstaller/charts/base/Chart.yaml | 2 +-
dubboinstaller/charts/base/files/crd-all.gen.yaml | 5 -
dubboinstaller/charts/dubbod/Chart.yaml | 2 +-
.../charts/dubbod/files/grpc-engine.yaml | 21 +++
.../charts/dubbod/files/kube-gateway.yaml | 3 +
.../charts/dubbod/templates/deployment.yaml | 2 +-
.../templates/mutatingwebhookconfiguration.yaml | 38 +++-
go.mod | 10 +-
go.sum | 10 +-
pkg/kube/inject/inject.go | 2 +-
pkg/kube/inject/proxyless.go | 2 +
pkg/kube/inject/proxyless_test.go | 32 +++-
pkg/kube/inject/service_test.go | 188 +++++++++++++++++++
pkg/kube/inject/webhook.go | 97 ++++++++++
samples/addons/00-namespace.yaml | 19 --
samples/addons/README.md | 38 ----
samples/app/README.md | 20 ++-
samples/app/deployment.yaml | 16 ++
samples/app/meshservice.yaml | 41 -----
34 files changed, 1700 insertions(+), 179 deletions(-)
diff --git a/.asf.yaml b/.asf.yaml
index e7744c3e..a57176f4 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -21,7 +21,7 @@ notifications:
github:
homepage: "https://kdubbo.github.io"
- description: "Embed gRPC, use xDS, build inherent mesh."
+ description: "Build a native mesh using gRPC and xDS."
features:
# Enable wiki for documentation
wiki: false
diff --git a/dubbod/discovery/cmd/app/cmd.go b/dubbod/discovery/cmd/app/cmd.go
index ee16ef77..4ab203fd 100644
--- a/dubbod/discovery/cmd/app/cmd.go
+++ b/dubbod/discovery/cmd/app/cmd.go
@@ -37,6 +37,7 @@ const (
executeLogScope = "setup"
waitLogScope = "wait"
xclientLogScope = "xclient"
+ xserverLogScope = "xserver"
)
func NewRootCommand() *cobra.Command {
@@ -61,6 +62,7 @@ func NewRootCommand() *cobra.Command {
cmd.AddFlags(rootCmd)
rootCmd.AddCommand(waitCmd)
rootCmd.AddCommand(newXClientCommand())
+ rootCmd.AddCommand(newXServerCommand())
return rootCmd
}
diff --git a/dubbod/discovery/cmd/app/cmd_test.go b/dubbod/discovery/cmd/app/cmd_test.go
index 70d81fe7..dda5b982 100644
--- a/dubbod/discovery/cmd/app/cmd_test.go
+++ b/dubbod/discovery/cmd/app/cmd_test.go
@@ -32,7 +32,7 @@ func TestRootCommandRegistersRenamedCommands(t *testing.T) {
commands[command.Name()] = true
}
- for _, name := range []string{"execute", "xclient"} {
+ for _, name := range []string{"execute", "xclient", "xserver"} {
if !commands[name] {
t.Fatalf("expected command %q to be registered; commands=%v", name, commands)
}
@@ -47,7 +47,7 @@ func TestRootCommandRegistersRenamedCommands(t *testing.T) {
func TestCommandsSetLogScopes(t *testing.T) {
defer func() {
log.SetDefaultScope("log")
- for _, name := range []string{"log", executeLogScope, waitLogScope, xclientLogScope} {
+ for _, name := range []string{"log", executeLogScope, waitLogScope, xclientLogScope, xserverLogScope} {
if scope := log.FindScope(name); scope != nil {
scope.SetOutput(os.Stderr)
}
@@ -62,6 +62,7 @@ func TestCommandsSetLogScopes(t *testing.T) {
{command: "execute", scope: "setup"},
{command: "wait", scope: "wait"},
{command: "xclient", scope: "xclient"},
+ {command: "xserver", scope: "xserver"},
} {
t.Run(tt.command, func(t *testing.T) {
var out bytes.Buffer
diff --git a/dubbod/discovery/cmd/app/xds_client.go b/dubbod/discovery/cmd/app/xds_client.go
index f0be4e13..7153839d 100644
--- a/dubbod/discovery/cmd/app/xds_client.go
+++ b/dubbod/discovery/cmd/app/xds_client.go
@@ -39,6 +39,7 @@ import (
corev1 "github.com/kdubbo/xds-api/core/v1"
endpointv1 "github.com/kdubbo/xds-api/endpoint/v1"
hcmv1 "github.com/kdubbo/xds-api/extensions/filters/v1/network/http_connection_manager"
+ tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
xdsresolver "github.com/kdubbo/xds-api/grpc/resolver"
listenerv1 "github.com/kdubbo/xds-api/listener/v1"
routev1 "github.com/kdubbo/xds-api/route/v1"
@@ -85,7 +86,11 @@ type xdsDestination struct {
Host string `json:"host"`
Subset string `json:"subset,omitempty"`
Weight uint32 `json:"weight"`
+ TLSMode string `json:"tlsMode,omitempty"`
+ SNI string `json:"sni,omitempty"`
Endpoints []xdsEndpoint `json:"endpoints,omitempty"`
+
+ tlsContext *tlsv1.UpstreamTlsContext
}
type xdsRouteSnapshot struct {
@@ -102,12 +107,15 @@ type sampleADSClient struct {
host string
port int
- mu sync.RWMutex
- subs map[string][]string
- route map[string]uint32
- endpoints map[string][]xdsEndpoint
- updates chan struct{}
- errs chan error
+ mu sync.RWMutex
+ subs map[string][]string
+ route map[string]uint32
+ endpoints map[string][]xdsEndpoint
+ clusterTLS map[string]*tlsv1.UpstreamTlsContext
+ updates chan struct{}
+ errs chan error
+ bootstrapPath string
+ bootstrap *xdsresolver.BootstrapConfig
}
func newXClientCommand() *cobra.Command {
@@ -232,17 +240,19 @@ func newSampleADSClient(ctx context.Context, opts *xdsClientOptions) (*sampleADS
return nil, fmt.Errorf("open ADS stream: %w", err)
}
return &sampleADSClient{
- conn: conn,
- stream: stream,
- node: node,
- target: opts.target,
- host: opts.host,
- port: opts.port,
- subs: map[string][]string{},
- route: map[string]uint32{},
- endpoints: map[string][]xdsEndpoint{},
- updates: make(chan struct{}, 1),
- errs: make(chan error, 1),
+ conn: conn,
+ stream: stream,
+ node: node,
+ target: opts.target,
+ host: opts.host,
+ port: opts.port,
+ subs: map[string][]string{},
+ route: map[string]uint32{},
+ endpoints: map[string][]xdsEndpoint{},
+ clusterTLS: map[string]*tlsv1.UpstreamTlsContext{},
+ updates: make(chan struct{}, 1),
+ errs: make(chan error, 1),
+ bootstrapPath: opts.bootstrapPath,
}, nil
}
@@ -410,10 +420,16 @@ func (c *sampleADSClient) handleResponse(resp *discovery.DiscoveryResponse) erro
return c.subscribe(v1.ClusterType, clusters)
}
case v1.ClusterType:
- edsNames, err := edsNamesFromClusters(resp.Resources)
+ edsNames, clusterTLS, err := edsNamesAndTLSFromClusters(resp.Resources)
if err != nil {
return err
}
+ c.mu.Lock()
+ for clusterName, tlsContext := range clusterTLS {
+ c.clusterTLS[clusterName] = tlsContext
+ }
+ c.mu.Unlock()
+ c.notify()
if len(edsNames) > 0 {
return c.subscribe(v1.EndpointType, edsNames)
}
@@ -478,7 +494,7 @@ func (c *sampleADSClient) readySnapshot(expected map[string]uint32) (xdsRouteSna
continue
}
eps := append([]xdsEndpoint(nil), c.endpoints[clusterName]...)
- dest := destinationFromCluster(clusterName, weight, eps)
+ dest := destinationFromCluster(clusterName, weight, eps, c.clusterTLS[clusterName])
snapshot.Destinations = append(snapshot.Destinations, dest)
}
sort.Slice(snapshot.Destinations, func(i, j int) bool {
@@ -569,13 +585,15 @@ func routeWeightsFromRoutes(resources []*anypb.Any) (map[string]uint32, []string
return weights, sortedUnique(clusters), nil
}
-func edsNamesFromClusters(resources []*anypb.Any) ([]string, error) {
+func edsNamesAndTLSFromClusters(resources []*anypb.Any) ([]string, map[string]*tlsv1.UpstreamTlsContext, error) {
out := make([]string, 0, len(resources))
+ clusterTLS := map[string]*tlsv1.UpstreamTlsContext{}
for _, resource := range resources {
c := &clusterv1.Cluster{}
if err := proto.Unmarshal(resource.Value, c); err != nil {
- return nil, err
+ return nil, nil, err
}
+ clusterTLS[c.Name] = upstreamTLSContextFromCluster(c)
serviceName := c.GetName()
if eds := c.GetEdsClusterConfig(); eds != nil && eds.GetServiceName() != "" {
serviceName = eds.GetServiceName()
@@ -584,7 +602,22 @@ func edsNamesFromClusters(resources []*anypb.Any) ([]string, error) {
out = append(out, serviceName)
}
}
- return sortedUnique(out), nil
+ return sortedUnique(out), clusterTLS, nil
+}
+
+func upstreamTLSContextFromCluster(c *clusterv1.Cluster) *tlsv1.UpstreamTlsContext {
+ if c == nil || c.TransportSocket == nil {
+ return nil
+ }
+ typedConfig := c.TransportSocket.GetTypedConfig()
+ if typedConfig == nil {
+ return nil
+ }
+ upstreamTLS := &tlsv1.UpstreamTlsContext{}
+ if err := anypb.UnmarshalTo(typedConfig, upstreamTLS, proto.UnmarshalOptions{}); err != nil {
+ return nil
+ }
+ return upstreamTLS
}
func endpointsFromAssignments(resources []*anypb.Any) (map[string][]xdsEndpoint, error) {
@@ -622,13 +655,17 @@ func endpointsFromAssignments(resources []*anypb.Any) (map[string][]xdsEndpoint,
return out, nil
}
-func destinationFromCluster(clusterName string, weight uint32, endpoints []xdsEndpoint) xdsDestination {
+func destinationFromCluster(clusterName string, weight uint32, endpoints []xdsEndpoint, tlsContext *tlsv1.UpstreamTlsContext) xdsDestination {
parts := strings.Split(clusterName, "|")
- dest := xdsDestination{Cluster: clusterName, Weight: weight, Endpoints: endpoints}
+ dest := xdsDestination{Cluster: clusterName, Weight: weight, Endpoints: endpoints, tlsContext: tlsContext}
if len(parts) == 4 {
dest.Host = parts[3]
dest.Subset = parts[2]
}
+ if tlsContext != nil {
+ dest.TLSMode = "DUBBO_MUTUAL"
+ dest.SNI = tlsContext.Sni
+ }
return dest
}
@@ -638,7 +675,7 @@ func runSampleRequests(ctx context.Context, adsClient *sampleADSClient, snapshot
return err
}
currentSignature := snapshotSignature(snapshot)
- client := &http.Client{Timeout: requestTimeout}
+ clients := newSampleRequestClients(adsClient, requestTimeout)
for i := 0; i < count; i++ {
if updated, ok := adsClient.readySnapshot(nil); ok {
updatedSignature := snapshotSignature(updated)
@@ -651,17 +688,21 @@ func runSampleRequests(ctx context.Context, adsClient *sampleADSClient, snapshot
}
}
}
- _, endpoint, err := picker.Next()
+ destination, endpoint, err := picker.Next()
+ if err != nil {
+ return err
+ }
+ httpClient, scheme, err := clients.clientForDestination(destination)
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet,
- fmt.Sprintf("http://%s/", net.JoinHostPort(endpoint.Address, strconv.Itoa(int(endpoint.Port)))), nil)
+ fmt.Sprintf("%s://%s/", scheme, net.JoinHostPort(endpoint.Address, strconv.Itoa(int(endpoint.Port)))), nil)
if err != nil {
return err
}
req.Host = snapshot.Host
- resp, err := client.Do(req)
+ resp, err := httpClient.Do(req)
if err != nil {
return err
}
@@ -686,6 +727,70 @@ func runSampleRequests(ctx context.Context, adsClient *sampleADSClient, snapshot
return nil
}
+type sampleRequestClients struct {
+ adsClient *sampleADSClient
+ requestTimeout time.Duration
+ plaintext *http.Client
+ tlsClients map[string]*http.Client
+}
+
+func newSampleRequestClients(adsClient *sampleADSClient, requestTimeout time.Duration) *sampleRequestClients {
+ return &sampleRequestClients{
+ adsClient: adsClient,
+ requestTimeout: requestTimeout,
+ plaintext: &http.Client{Timeout: requestTimeout},
+ tlsClients: map[string]*http.Client{},
+ }
+}
+
+func (c *sampleRequestClients) clientForDestination(destination xdsDestination) (*http.Client, string, error) {
+ if destination.tlsContext == nil {
+ return c.plaintext, "http", nil
+ }
+ bootstrap, err := c.adsClient.bootstrapConfig()
+ if err != nil {
+ return nil, "", err
+ }
+ tlsConfig, err := xdsresolver.DataPlaneTLSConfigFromBootstrap(bootstrap, destination.tlsContext, destination.Host)
+ if err != nil {
+ return nil, "", err
+ }
+ key := destination.TLSMode + "|" + destination.SNI
+ if key == "|" {
+ key = destination.Cluster
+ }
+ if cached := c.tlsClients[key]; cached != nil {
+ return cached, "https", nil
+ }
+ client := &http.Client{
+ Timeout: c.requestTimeout,
+ Transport: &http.Transport{
+ TLSClientConfig: tlsConfig,
+ },
+ }
+ c.tlsClients[key] = client
+ return client, "https", nil
+}
+
+func (c *sampleADSClient) bootstrapConfig() (*xdsresolver.BootstrapConfig, error) {
+ if c.bootstrap != nil {
+ return c.bootstrap, nil
+ }
+ path := c.bootstrapPath
+ if path == "" {
+ path = os.Getenv("GRPC_XDS_BOOTSTRAP")
+ }
+ if path == "" {
+ return nil, fmt.Errorf("data-plane mTLS requires GRPC_XDS_BOOTSTRAP or --bootstrap")
+ }
+ bootstrap, err := xdsresolver.ParseBootstrap(path)
+ if err != nil {
+ return nil, err
+ }
+ c.bootstrap = bootstrap
+ return c.bootstrap, nil
+}
+
func activeDestinations(snapshot xdsRouteSnapshot) []xdsDestination {
out := make([]xdsDestination, 0, len(snapshot.Destinations))
for _, dest := range snapshot.Destinations {
diff --git a/dubbod/discovery/cmd/app/xds_client_test.go b/dubbod/discovery/cmd/app/xds_client_test.go
index 3beaa8d6..6a74ebfc 100644
--- a/dubbod/discovery/cmd/app/xds_client_test.go
+++ b/dubbod/discovery/cmd/app/xds_client_test.go
@@ -13,6 +13,8 @@ import (
"strings"
"testing"
"time"
+
+ tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
)
func TestAutoDiscoverServiceTargetSingleService(t *testing.T) {
@@ -203,6 +205,47 @@ func TestRunSampleRequestsUsesUpdatedSnapshotOnSameStream(t *testing.T) {
}
}
+func TestSampleRequestClientsDoNotBypassMTLS(t *testing.T) {
+ clients := newSampleRequestClients(&sampleADSClient{}, time.Second)
+ destination := xdsDestination{
+ Cluster: "outbound|80|v1|nginx.app.svc.cluster.local",
+ Host: "nginx.app.svc.cluster.local",
+ TLSMode: "DUBBO_MUTUAL",
+ SNI: "nginx.app.svc.cluster.local",
+ tlsContext: &tlsv1.UpstreamTlsContext{
+ Sni: "nginx.app.svc.cluster.local",
+ },
+ }
+
+ _, _, err := clients.clientForDestination(destination)
+ if err == nil {
+ t.Fatalf("clientForDestination() error = nil, want bootstrap error for mTLS route")
+ }
+ if !strings.Contains(err.Error(), "data-plane mTLS requires GRPC_XDS_BOOTSTRAP or --bootstrap") {
+ t.Fatalf("error = %q, want bootstrap requirement", err.Error())
+ }
+}
+
+func TestDestinationFromClusterExposesTLSContext(t *testing.T) {
+ tlsContext := &tlsv1.UpstreamTlsContext{Sni: "nginx.app.svc.cluster.local"}
+ destination := destinationFromCluster(
+ "outbound|80|v1|nginx.app.svc.cluster.local",
+ 50,
+ []xdsEndpoint{{Address: "10.0.0.1", Port: 80}},
+ tlsContext,
+ )
+
+ if destination.TLSMode != "DUBBO_MUTUAL" {
+ t.Fatalf("TLSMode = %q, want DUBBO_MUTUAL", destination.TLSMode)
+ }
+ if destination.SNI != "nginx.app.svc.cluster.local" {
+ t.Fatalf("SNI = %q, want nginx host", destination.SNI)
+ }
+ if destination.tlsContext != tlsContext {
+ t.Fatalf("tlsContext was not preserved")
+ }
+}
+
func endpointForServer(t *testing.T, server *httptest.Server) xdsEndpoint {
t.Helper()
parsed, err := neturl.Parse(server.URL)
diff --git a/dubbod/discovery/cmd/app/xds_server.go b/dubbod/discovery/cmd/app/xds_server.go
new file mode 100644
index 00000000..c53ea4f4
--- /dev/null
+++ b/dubbod/discovery/cmd/app/xds_server.go
@@ -0,0 +1,193 @@
+// 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 app
+
+import (
+ "context"
+ "crypto/tls"
+ "crypto/x509"
+ "fmt"
+ "io"
+ "net"
+ "os"
+ "sync"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/pkg/kube/inject"
+ "github.com/apache/dubbo-kubernetes/pkg/log"
+ xdsresolver "github.com/kdubbo/xds-api/grpc/resolver"
+ "github.com/spf13/cobra"
+)
+
+type xdsServerOptions struct {
+ listen string
+ upstream string
+ bootstrapPath string
+ acceptTimeout time.Duration
+ connectTimeout time.Duration
+}
+
+func newXServerCommand() *cobra.Command {
+ opts := &xdsServerOptions{
+ listen: firstNonEmpty(os.Getenv("DUBBO_XSERVER_LISTEN"), fmt.Sprintf(":%d", inject.ProxylessXServerPort)),
+ upstream: firstNonEmpty(os.Getenv("DUBBO_XSERVER_UPSTREAM"), "127.0.0.1:80"),
+ bootstrapPath: os.Getenv("GRPC_XDS_BOOTSTRAP"),
+ acceptTimeout: durationSecondsFromEnv("DUBBO_XSERVER_ACCEPT_TIMEOUT", 0),
+ connectTimeout: durationSecondsFromEnv("DUBBO_XSERVER_CONNECT_TIMEOUT", 5*time.Second),
+ }
+ c := &cobra.Command{
+ Use: "xserver",
+ Short: "run an inbound mTLS data-plane proxy for proxyless workloads",
+ Args: cobra.NoArgs,
+ PreRunE: func(cmd *cobra.Command, args []string) error {
+ log.SetDefaultScope(xserverLogScope)
+ return nil
+ },
+ RunE: func(cmd *cobra.Command, _ []string) error {
+ return opts.run(cmd.Context())
+ },
+ }
+ c.Flags().StringVar(&opts.listen, "listen", opts.listen, "mTLS listener address")
+ c.Flags().StringVar(&opts.upstream, "upstream", opts.upstream, "local plaintext upstream address")
+ c.Flags().StringVar(&opts.bootstrapPath, "bootstrap", opts.bootstrapPath, "gRPC xDS bootstrap file")
+ c.Flags().DurationVar(&opts.acceptTimeout, "accept-timeout", opts.acceptTimeout, "optional TLS handshake timeout")
+ c.Flags().DurationVar(&opts.connectTimeout, "connect-timeout", opts.connectTimeout, "timeout for connecting to the local upstream")
+ return c
+}
+
+func (o *xdsServerOptions) run(ctx context.Context) error {
+ if o.bootstrapPath == "" {
+ return fmt.Errorf("xserver requires GRPC_XDS_BOOTSTRAP or --bootstrap")
+ }
+ if o.listen == "" {
+ return fmt.Errorf("xserver listen address is required")
+ }
+ if o.upstream == "" {
+ return fmt.Errorf("xserver upstream address is required")
+ }
+ bootstrap, err := xdsresolver.ParseBootstrap(o.bootstrapPath)
+ if err != nil {
+ return err
+ }
+ tlsConfig, err := xserverTLSConfigFromBootstrap(bootstrap)
+ if err != nil {
+ return err
+ }
+ lis, err := net.Listen("tcp", o.listen)
+ if err != nil {
+ return fmt.Errorf("listen xserver %s: %w", o.listen, err)
+ }
+ defer lis.Close()
+ return serveXServer(ctx, lis, tlsConfig, o.upstream, o.acceptTimeout, o.connectTimeout)
+}
+
+func xserverTLSConfigFromBootstrap(bootstrap *xdsresolver.BootstrapConfig) (*tls.Config, error) {
+ if bootstrap == nil {
+ return nil, fmt.Errorf("bootstrap config is nil")
+ }
+ cfg, ok := bootstrap.CertProviders["default"]
+ if !ok {
+ return nil, fmt.Errorf("certificate_providers[default] not found")
+ }
+ if cfg.CertificateFile == "" || cfg.PrivateKeyFile == "" {
+ return nil, fmt.Errorf("xserver mTLS requires certificate_file and private_key_file")
+ }
+ if cfg.CACertificateFile == "" {
+ return nil, fmt.Errorf("xserver mTLS requires ca_certificate_file")
+ }
+ cert, err := tls.LoadX509KeyPair(cfg.CertificateFile, cfg.PrivateKeyFile)
+ if err != nil {
+ return nil, fmt.Errorf("load xserver certificate/key: %w", err)
+ }
+ rootPEM, err := os.ReadFile(cfg.CACertificateFile)
+ if err != nil {
+ return nil, fmt.Errorf("read xserver CA certificate %s: %w", cfg.CACertificateFile, err)
+ }
+ clientCAs := x509.NewCertPool()
+ if !clientCAs.AppendCertsFromPEM(rootPEM) {
+ return nil, fmt.Errorf("parse xserver CA certificate %s: no certificates found", cfg.CACertificateFile)
+ }
+ return &tls.Config{
+ MinVersion: tls.VersionTLS12,
+ Certificates: []tls.Certificate{cert},
+ ClientCAs: clientCAs,
+ ClientAuth: tls.RequireAndVerifyClientCert,
+ }, nil
+}
+
+func serveXServer(ctx context.Context, lis net.Listener, tlsConfig *tls.Config, upstream string, acceptTimeout, connectTimeout time.Duration) error {
+ tlsListener := tls.NewListener(lis, tlsConfig)
+ go func() {
+ <-ctx.Done()
+ _ = tlsListener.Close()
+ }()
+ for {
+ conn, err := tlsListener.Accept()
+ if err != nil {
+ select {
+ case <-ctx.Done():
+ return nil
+ default:
+ return err
+ }
+ }
+ go proxyXServerConnection(conn, upstream, acceptTimeout, connectTimeout)
+ }
+}
+
+func proxyXServerConnection(inbound net.Conn, upstream string, acceptTimeout, connectTimeout time.Duration) {
+ defer inbound.Close()
+ if acceptTimeout > 0 {
+ _ = inbound.SetDeadline(time.Now().Add(acceptTimeout))
+ }
+ if tlsConn, ok := inbound.(*tls.Conn); ok {
+ if err := tlsConn.Handshake(); err != nil {
+ return
+ }
+ }
+ _ = inbound.SetDeadline(time.Time{})
+ dialer := net.Dialer{Timeout: connectTimeout}
+ outbound, err := dialer.Dial("tcp", upstream)
+ if err != nil {
+ return
+ }
+ defer outbound.Close()
+ copyBothDirections(inbound, outbound)
+}
+
+func copyBothDirections(a, b net.Conn) {
+ var wg sync.WaitGroup
+ wg.Add(2)
+ go func() {
+ defer wg.Done()
+ _, _ = io.Copy(a, b)
+ closeWrite(a)
+ }()
+ go func() {
+ defer wg.Done()
+ _, _ = io.Copy(b, a)
+ closeWrite(b)
+ }()
+ wg.Wait()
+}
+
+func closeWrite(conn net.Conn) {
+ if c, ok := conn.(interface{ CloseWrite() error }); ok {
+ _ = c.CloseWrite()
+ return
+ }
+ _ = conn.Close()
+}
diff --git a/dubbod/discovery/cmd/app/xds_server_test.go b/dubbod/discovery/cmd/app/xds_server_test.go
new file mode 100644
index 00000000..3ce19265
--- /dev/null
+++ b/dubbod/discovery/cmd/app/xds_server_test.go
@@ -0,0 +1,172 @@
+package app
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/rsa"
+ "crypto/tls"
+ "crypto/x509"
+ "crypto/x509/pkix"
+ "encoding/pem"
+ "fmt"
+ "io"
+ "math/big"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ xdsresolver "github.com/kdubbo/xds-api/grpc/resolver"
+)
+
+func TestXServerRequiresClientCertificateAndProxiesHTTP(t *testing.T) {
+ caCert, caKey := newTestCA(t)
+ serverCert, serverKey := newSignedCert(t, caCert, caKey, "xserver")
+ clientCert, clientKey := newSignedCert(t, caCert, caKey, "xclient")
+ dir := t.TempDir()
+ writePEM(t, filepath.Join(dir, "root-cert.pem"), "CERTIFICATE", caCert.Raw)
+ writePEM(t, filepath.Join(dir, "cert-chain.pem"), "CERTIFICATE", serverCert.Raw)
+ writePEM(t, filepath.Join(dir, "key.pem"), "RSA PRIVATE KEY", x509.MarshalPKCS1PrivateKey(serverKey))
+ writePEM(t, filepath.Join(dir, "client-cert.pem"), "CERTIFICATE", clientCert.Raw)
+ writePEM(t, filepath.Join(dir, "client-key.pem"), "RSA PRIVATE KEY", x509.MarshalPKCS1PrivateKey(clientKey))
+
+ tlsConfig, err := xserverTLSConfigFromBootstrap(&xdsresolver.BootstrapConfig{
+ CertProviders: map[string]xdsresolver.FileWatcherCertConfig{
+ "default": {
+ CertificateFile: filepath.Join(dir, "cert-chain.pem"),
+ PrivateKeyFile: filepath.Join(dir, "key.pem"),
+ CACertificateFile: filepath.Join(dir, "root-cert.pem"),
+ },
+ },
+ })
+ if err != nil {
+ t.Fatalf("xserverTLSConfigFromBootstrap() failed: %v", err)
+ }
+
+ upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ _, _ = fmt.Fprintln(w, "nginx v1")
+ }))
+ defer upstream.Close()
+ upstreamAddr := upstream.Listener.Addr().String()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatalf("net.Listen() failed: %v", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ errCh := make(chan error, 1)
+ go func() {
+ errCh <- serveXServer(ctx, listener, tlsConfig, upstreamAddr, time.Second, time.Second)
+ }()
+ t.Cleanup(func() {
+ cancel()
+ select {
+ case err := <-errCh:
+ if err != nil {
+ t.Fatalf("serveXServer() error = %v", err)
+ }
+ case <-time.After(time.Second):
+ t.Fatalf("serveXServer() did not stop")
+ }
+ })
+
+ url := "https://" + listener.Addr().String() + "/"
+ _, err = (&http.Client{Timeout: time.Second, Transport: &http.Transport{TLSClientConfig: &tls.Config{
+ InsecureSkipVerify: true,
+ }}}).Get(url)
+ if err == nil {
+ t.Fatalf("plain TLS request without client certificate succeeded, want failure")
+ }
+
+ cert, err := tls.LoadX509KeyPair(filepath.Join(dir, "client-cert.pem"), filepath.Join(dir, "client-key.pem"))
+ if err != nil {
+ t.Fatalf("tls.LoadX509KeyPair() failed: %v", err)
+ }
+ resp, err := (&http.Client{Timeout: time.Second, Transport: &http.Transport{TLSClientConfig: &tls.Config{
+ InsecureSkipVerify: true,
+ Certificates: []tls.Certificate{cert},
+ }}}).Get(url)
+ if err != nil {
+ t.Fatalf("mTLS request failed: %v", err)
+ }
+ defer resp.Body.Close()
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ t.Fatalf("io.ReadAll() failed: %v", err)
+ }
+ if got, want := string(body), "nginx v1\n"; got != want {
+ t.Fatalf("body = %q, want %q", got, want)
+ }
+}
+
+func newTestCA(t *testing.T) (*x509.Certificate, *rsa.PrivateKey) {
+ t.Helper()
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatalf("rsa.GenerateKey() failed: %v", err)
+ }
+ tmpl := &x509.Certificate{
+ SerialNumber: big.NewInt(1),
+ Subject: pkix.Name{CommonName: "test-ca"},
+ NotBefore: time.Now().Add(-time.Minute),
+ NotAfter: time.Now().Add(time.Hour),
+ KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
+ BasicConstraintsValid: true,
+ IsCA: true,
+ }
+ der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
+ if err != nil {
+ t.Fatalf("x509.CreateCertificate(CA) failed: %v", err)
+ }
+ cert, err := x509.ParseCertificate(der)
+ if err != nil {
+ t.Fatalf("x509.ParseCertificate(CA) failed: %v", err)
+ }
+ return cert, key
+}
+
+func newSignedCert(t *testing.T, caCert *x509.Certificate, caKey *rsa.PrivateKey, commonName string) (*x509.Certificate, *rsa.PrivateKey) {
+ t.Helper()
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatalf("rsa.GenerateKey() failed: %v", err)
+ }
+ serial, err := rand.Int(rand.Reader, big.NewInt(1<<62))
+ if err != nil {
+ t.Fatalf("rand.Int() failed: %v", err)
+ }
+ tmpl := &x509.Certificate{
+ SerialNumber: serial,
+ Subject: pkix.Name{CommonName: commonName},
+ NotBefore: time.Now().Add(-time.Minute),
+ NotAfter: time.Now().Add(time.Hour),
+ KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+ ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+ DNSNames: []string{commonName},
+ }
+ der, err := x509.CreateCertificate(rand.Reader, tmpl, caCert, &key.PublicKey, caKey)
+ if err != nil {
+ t.Fatalf("x509.CreateCertificate(%s) failed: %v", commonName, err)
+ }
+ cert, err := x509.ParseCertificate(der)
+ if err != nil {
+ t.Fatalf("x509.ParseCertificate(%s) failed: %v", commonName, err)
+ }
+ return cert, key
+}
+
+func writePEM(t *testing.T, path, typ string, der []byte) {
+ t.Helper()
+ file, err := os.Create(path)
+ if err != nil {
+ t.Fatalf("os.Create(%s) failed: %v", path, err)
+ }
+ defer file.Close()
+ if err := pem.Encode(file, &pem.Block{Type: typ, Bytes: der}); err != nil {
+ t.Fatalf("pem.Encode(%s) failed: %v", path, err)
+ }
+}
diff --git a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
index 55c2749c..0c3b19c7 100644
--- a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
+++ b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller.go
@@ -32,6 +32,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/host"
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"
kubelib "github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
@@ -46,6 +47,7 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/security/pkg/pki/ca"
pkiutil "github.com/apache/dubbo-kubernetes/dubbod/security/pkg/pki/util"
caserver "github.com/apache/dubbo-kubernetes/dubbod/security/pkg/server/ca"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -88,6 +90,15 @@ func (s *Server) initProxylessGRPCWorkloads() error {
controller := newProxylessGRPCWorkloadController(s)
s.proxylessGRPCWorkloadController = controller
+ if s.XDSServer != nil {
+ previous := s.XDSServer.RuntimeConfigUpdate
+ s.XDSServer.RuntimeConfigUpdate = func(req *discoverymodel.PushRequest) {
+ if previous != nil {
+ previous(req)
+ }
+ controller.handleRuntimeConfigUpdate(req)
+ }
+ }
s.addStartFunc(proxylessGRPCControllerName, func(stop <-chan struct{}) error {
go controller.Run(stop)
return nil
@@ -164,6 +175,32 @@ func (c *proxylessGRPCWorkloadController) enqueueAllPods() {
}
}
+func (c *proxylessGRPCWorkloadController) handleRuntimeConfigUpdate(req *discoverymodel.PushRequest) {
+ if !proxylessGRPCRuntimeConfigNeedsUpdate(req) {
+ return
+ }
+ c.enqueueAllPods()
+}
+
+func proxylessGRPCRuntimeConfigNeedsUpdate(req *discoverymodel.PushRequest) bool {
+ if req == nil {
+ return false
+ }
+ if req.Full || req.Forced || len(req.AddressesUpdated) > 0 {
+ return true
+ }
+ for cfg := range req.ConfigsUpdated {
+ switch cfg.Kind {
+ case kind.MeshService, kind.PeerAuthentication, kind.Service, kind.EndpointSlice, kind.Endpoints, kind.Pod, kind.Namespace:
+ return true
+ }
+ }
+ if req.Reason.Has(discoverymodel.EndpointUpdate) || req.Reason.Has(discoverymodel.HeadlessEndpointUpdate) || req.Reason.Has(discoverymodel.GlobalUpdate) {
+ return true
+ }
+ return false
+}
+
func (c *proxylessGRPCWorkloadController) enqueueManagedPodsBatched(stop <-chan struct{}) {
keys := c.managedPodKeys()
for start := 0; start < len(keys); start += proxylessGRPCBundleRequeueBatch {
@@ -313,8 +350,9 @@ type proxylessGRPCServiceRuntimeConfig struct {
}
type proxylessGRPCPortRuntimeConfig struct {
- Name string `json:"name,omitempty"`
- Port int `json:"port"`
+ Name string `json:"name,omitempty"`
+ Port int `json:"port"`
+ MTLSMode string `json:"mtlsMode,omitempty"`
}
type proxylessGRPCRouteRuntimeConfig struct {
@@ -327,6 +365,7 @@ type proxylessGRPCDestinationRuntimeConfig struct {
Host string `json:"host"`
Subset string `json:"subset,omitempty"`
Weight int `json:"weight"`
+ TLSMode string `json:"tlsMode,omitempty"`
Endpoints []proxylessGRPCEndpointRuntimeConfig `json:"endpoints,omitempty"`
}
@@ -350,7 +389,8 @@ func (c *proxylessGRPCWorkloadController) buildSecret(pod *corev1.Pod, current *
return nil, time.Time{}, err
}
- runtimeConfigJSON, err := buildRuntimeConfigJSON(workload, nil, nil)
+ services, routes := c.buildRuntimeTrafficConfig()
+ runtimeConfigJSON, err := buildRuntimeConfigJSON(workload, services, routes)
if err != nil {
return nil, time.Time{}, err
}
@@ -555,7 +595,7 @@ func (c *proxylessGRPCWorkloadController) buildRuntimeTrafficConfig() ([]proxyle
if svc == nil {
continue
}
- serviceConfigs = append(serviceConfigs, buildRuntimeServiceConfig(endpoints, svc))
+ serviceConfigs = append(serviceConfigs, buildRuntimeServiceConfig(push, endpoints, svc))
for _, port := range svc.Ports {
if port == nil {
continue
@@ -566,7 +606,7 @@ func (c *proxylessGRPCWorkloadController) buildRuntimeTrafficConfig() ([]proxyle
return serviceConfigs, routeConfigs
}
-func buildRuntimeServiceConfig(endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service) proxylessGRPCServiceRuntimeConfig {
+func buildRuntimeServiceConfig(push *discoverymodel.PushContext, endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service) proxylessGRPCServiceRuntimeConfig {
cfg := proxylessGRPCServiceRuntimeConfig{
Host: string(svc.Hostname),
Namespace: svc.Attributes.Namespace,
@@ -576,7 +616,11 @@ func buildRuntimeServiceConfig(endpointIndex *discoverymodel.EndpointIndex, svc
if port == nil {
continue
}
- cfg.Ports = append(cfg.Ports, proxylessGRPCPortRuntimeConfig{Name: port.Name, Port: port.Port})
+ cfg.Ports = append(cfg.Ports, proxylessGRPCPortRuntimeConfig{
+ Name: port.Name,
+ Port: port.Port,
+ MTLSMode: runtimeInboundMTLSMode(push, svc.Attributes.Namespace, port.Port),
+ })
cfg.Endpoints = append(cfg.Endpoints, runtimeEndpointsForService(endpointIndex, svc, port.Port, nil)...)
}
sortRuntimeEndpoints(cfg.Endpoints)
@@ -590,10 +634,12 @@ func buildRuntimeRouteConfig(push *discoverymodel.PushContext, endpointIndex *di
}
vs := push.VirtualServiceForHost(svc.Hostname)
if vs == nil || len(vs.Http) == 0 {
+ tlsMode := runtimeDestinationTLSMode(push, svc.Attributes.Namespace, svc.Hostname, "")
cfg.Destinations = []proxylessGRPCDestinationRuntimeConfig{{
Host: string(svc.Hostname),
Weight: 100,
- Endpoints: runtimeEndpointsForService(endpointIndex, svc, port, nil),
+ TLSMode: tlsMode,
+ Endpoints: runtimeRouteEndpointsForService(endpointIndex, svc, port, nil, tlsMode),
}}
return cfg
}
@@ -624,25 +670,45 @@ func buildRuntimeRouteConfig(push *discoverymodel.PushContext, endpointIndex *di
if subsetFound {
endpoints = runtimeEndpointsForService(endpointIndex, targetSvc, port, selector)
}
+ tlsMode := runtimeDestinationTLSMode(push, targetSvc.Attributes.Namespace, targetSvc.Hostname, subset)
cfg.Destinations = append(cfg.Destinations, proxylessGRPCDestinationRuntimeConfig{
Host: targetHost,
Subset: subset,
Weight: int(weighted.Weight),
- Endpoints: endpoints,
+ TLSMode: tlsMode,
+ Endpoints: rewriteRuntimeEndpointPortsForTLS(endpoints, tlsMode),
})
}
}
if len(cfg.Destinations) == 0 {
+ tlsMode := runtimeDestinationTLSMode(push, svc.Attributes.Namespace, svc.Hostname, "")
cfg.Destinations = []proxylessGRPCDestinationRuntimeConfig{{
Host: string(svc.Hostname),
Weight: 100,
- Endpoints: runtimeEndpointsForService(endpointIndex, svc, port, nil),
+ TLSMode: tlsMode,
+ Endpoints: runtimeRouteEndpointsForService(endpointIndex, svc, port, nil, tlsMode),
}}
}
normalizeRuntimeRouteWeights(cfg.Destinations)
return cfg
}
+func runtimeRouteEndpointsForService(endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service, port int, selector configlabels.Instance, tlsMode string) []proxylessGRPCEndpointRuntimeConfig {
+ return rewriteRuntimeEndpointPortsForTLS(runtimeEndpointsForService(endpointIndex, svc, port, selector), tlsMode)
+}
+
+func rewriteRuntimeEndpointPortsForTLS(endpoints []proxylessGRPCEndpointRuntimeConfig, tlsMode string) []proxylessGRPCEndpointRuntimeConfig {
+ if tlsMode != "DUBBO_MUTUAL" {
+ return endpoints
+ }
+ out := make([]proxylessGRPCEndpointRuntimeConfig, len(endpoints))
+ copy(out, endpoints)
+ for i := range out {
+ out[i].Port = inject.ProxylessXServerPort
+ }
+ return out
+}
+
func runtimeSubsetSelector(push *discoverymodel.PushContext, namespace string, hostname host.Name, subset string) (configlabels.Instance, bool) {
if subset == "" {
return nil, true
@@ -659,6 +725,52 @@ func runtimeSubsetSelector(push *discoverymodel.PushContext, namespace string, h
return nil, false
}
+func runtimeDestinationTLSMode(push *discoverymodel.PushContext, namespace string, hostname host.Name, subset string) string {
+ if push == nil {
+ return ""
+ }
+ rule := push.DestinationRuleForService(namespace, hostname)
+ if rule == nil {
+ return ""
+ }
+ if subset != "" {
+ for _, ss := range rule.Subsets {
+ if ss.Name == subset {
+ if ss.TrafficPolicy != nil {
+ return runtimeTrafficPolicyTLSMode(ss.TrafficPolicy)
+ }
+ break
+ }
+ }
+ }
+ return runtimeTrafficPolicyTLSMode(rule.TrafficPolicy)
+}
+
+func runtimeTrafficPolicyTLSMode(policy *networking.TrafficPolicy) string {
+ if policy == nil || policy.Tls == nil {
+ return ""
+ }
+ return policy.Tls.Mode.String()
+}
+
+func runtimeInboundMTLSMode(push *discoverymodel.PushContext, namespace string, port int) string {
+ if push == nil || push.AuthenticationPolicies == nil || port <= 0 {
+ return ""
+ }
+ return runtimeMutualTLSModeString(push.AuthenticationPolicies.EffectiveMutualTLSMode(namespace, nil, uint32(port)))
+}
+
+func runtimeMutualTLSModeString(mode discoverymodel.MutualTLSMode) string {
+ switch mode {
+ case discoverymodel.MTLSDisable:
+ return "DISABLE"
+ case discoverymodel.MTLSStrict:
+ return "STRICT"
+ default:
+ return ""
+ }
+}
+
func normalizeRuntimeRouteWeights(destinations []proxylessGRPCDestinationRuntimeConfig) {
hasPositiveWeight := false
for i := range destinations {
diff --git a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
index d048bf87..f8ff7716 100644
--- a/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
+++ b/dubbod/discovery/pkg/bootstrap/proxyless_grpc_controller_test.go
@@ -20,9 +20,23 @@ import (
"testing"
"time"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/config/memory"
+ discoverymodel "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/constants"
+ "github.com/apache/dubbo-kubernetes/pkg/config/host"
+ "github.com/apache/dubbo-kubernetes/pkg/config/mesh"
+ "github.com/apache/dubbo-kubernetes/pkg/config/mesh/meshwatcher"
+ "github.com/apache/dubbo-kubernetes/pkg/config/protocol"
+ "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/kube/inject"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ security "github.com/kdubbo/api/security/v1alpha3"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
@@ -169,6 +183,80 @@ func TestBuildRuntimeConfigJSON(t *testing.T) {
}
}
+func TestBuildRuntimeTrafficConfigCapturesProxylessSecurity(t *testing.T) {
+ hostname := host.Name("provider.grpc-app.svc.cluster.local")
+ svc := newProxylessRuntimeTestService("provider", "grpc-app", string(hostname), 17070)
+ push := newProxylessRuntimeTestPushContext(t, []config.Config{
+ newProxylessMTLSMeshServiceConfig("provider-mtls", "grpc-app", hostname),
+ newProxylessStrictPeerAuthenticationConfig("grpc-app-strict-mtls", "grpc-app"),
+ }, []*discoverymodel.Service{svc})
+
+ serviceConfig := buildRuntimeServiceConfig(push, nil, svc)
+ if len(serviceConfig.Ports) != 1 {
+ t.Fatalf("ports = %d, want 1", len(serviceConfig.Ports))
+ }
+ if got := serviceConfig.Ports[0].MTLSMode; got != "STRICT" {
+ t.Fatalf("mtlsMode = %q, want STRICT", got)
+ }
+
+ routeConfig := buildRuntimeRouteConfig(push, nil, svc, 17070)
+ if len(routeConfig.Destinations) != 2 {
+ t.Fatalf("destinations = %d, want 2", len(routeConfig.Destinations))
+ }
+ wantWeights := map[string]int{"v1": 50, "v2": 50}
+ for _, destination := range routeConfig.Destinations {
+ if destination.TLSMode != "DUBBO_MUTUAL" {
+ t.Fatalf("destination %s tlsMode = %q, want DUBBO_MUTUAL", destination.Subset, destination.TLSMode)
+ }
+ if wantWeights[destination.Subset] != destination.Weight {
+ t.Fatalf("destination %s weight = %d, want %d", destination.Subset, destination.Weight, wantWeights[destination.Subset])
+ }
+ }
+}
+
+func TestProxylessGRPCRuntimeConfigNeedsUpdate(t *testing.T) {
+ tests := []struct {
+ name string
+ req *discoverymodel.PushRequest
+ want bool
+ }{
+ {
+ name: "full push",
+ req: &discoverymodel.PushRequest{Full: true},
+ want: true,
+ },
+ {
+ name: "meshservice",
+ req: &discoverymodel.PushRequest{
+ ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.MeshService, Name: "provider-mtls", Namespace: "grpc-app"}),
+ },
+ want: true,
+ },
+ {
+ name: "peerauthentication",
+ req: &discoverymodel.PushRequest{
+ ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.PeerAuthentication, Name: "strict", Namespace: "grpc-app"}),
+ },
+ want: true,
+ },
+ {
+ name: "unrelated configmap",
+ req: &discoverymodel.PushRequest{
+ ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.ConfigMap, Name: "ui", Namespace: "dubbo-system"}),
+ },
+ want: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if got := proxylessGRPCRuntimeConfigNeedsUpdate(tt.req); got != tt.want {
+ t.Fatalf("proxylessGRPCRuntimeConfigNeedsUpdate() = %v, want %v", got, tt.want)
+ }
+ })
+ }
+}
+
func TestNextRotationTime(t *testing.T) {
now := time.Now()
rotations := map[types.NamespacedName]time.Time{
@@ -189,6 +277,117 @@ func TestNextRotationTime(t *testing.T) {
}
}
+func newProxylessRuntimeTestPushContext(t *testing.T, configs []config.Config, services []*discoverymodel.Service) *discoverymodel.PushContext {
+ t.Helper()
+
+ store := memory.Make(collections.DubboGatewayAPI())
+ for _, cfg := range configs {
+ if _, err := store.Create(cfg); err != nil {
+ t.Fatalf("create config %s/%s: %v", cfg.Namespace, cfg.Name, err)
+ }
+ }
+
+ env := discoverymodel.NewEnvironment()
+ env.ConfigStore = store
+ env.ServiceDiscovery = proxylessRuntimeStaticServiceDiscovery{services: services}
+ env.Watcher = meshwatcher.ConfigAdapter(krt.NewStatic(&meshwatcher.MeshConfigResource{
+ MeshConfig: mesh.DefaultMeshConfig(),
+ }, true))
+ env.Init()
+
+ push := discoverymodel.NewPushContext()
+ push.InitContext(env, nil, nil)
+ return push
+}
+
+func newProxylessRuntimeTestService(name, namespace, hostname string, port int) *discoverymodel.Service {
+ return &discoverymodel.Service{
+ Hostname: host.Name(hostname),
+ Ports: discoverymodel.PortList{
+ {
+ Name: "grpc",
+ Port: port,
+ Protocol: protocol.HTTP2,
+ },
+ },
+ Attributes: discoverymodel.ServiceAttributes{
+ Name: name,
+ Namespace: namespace,
+ },
+ }
+}
+
+func newProxylessMTLSMeshServiceConfig(name, namespace string, hostname host.Name) config.Config {
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.MeshService,
+ Name: name,
+ Namespace: namespace,
+ Domain: "cluster.local",
+ },
+ Spec: &networking.MeshService{
+ Hosts: []string{string(hostname)},
+ TrafficPolicy: &networking.TrafficPolicy{
+ Tls: &networking.ClientTLSSettings{
+ Mode: networking.ClientTLSSettings_DUBBO_MUTUAL,
+ },
+ },
+ Routes: []*networking.MeshServiceRoute{{
+ Service: []*networking.ServiceDestination{{
+ Name: "v1",
+ Host: string(hostname),
+ Labels: map[string]string{"version": "v1"},
+ Port: &networking.ServicePort{Number: 17070},
+ Weight: 50,
+ }, {
+ Name: "v2",
+ Host: string(hostname),
+ Labels: map[string]string{"version": "v2"},
+ Port: &networking.ServicePort{Number: 17070},
+ Weight: 50,
+ }},
+ }},
+ },
+ }
+}
+
+func newProxylessStrictPeerAuthenticationConfig(name, namespace string) config.Config {
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.PeerAuthentication,
+ Name: name,
+ Namespace: namespace,
+ Domain: "cluster.local",
+ },
+ Spec: &security.PeerAuthentication{
+ Mtls: &security.PeerAuthentication_MutualTLS{
+ Mode: security.PeerAuthentication_MutualTLS_STRICT,
+ },
+ },
+ }
+}
+
+type proxylessRuntimeStaticServiceDiscovery struct {
+ services []*discoverymodel.Service
+}
+
+func (s proxylessRuntimeStaticServiceDiscovery) Services() []*discoverymodel.Service {
+ return s.services
+}
+
+func (s proxylessRuntimeStaticServiceDiscovery) GetService(hostname host.Name) *discoverymodel.Service {
+ for _, svc := range s.services {
+ if svc.Hostname == hostname {
+ return svc
+ }
+ }
+ return nil
+}
+
+func (s proxylessRuntimeStaticServiceDiscovery) GetProxyServiceTargets(*discoverymodel.Proxy) []discoverymodel.ServiceTarget {
+ return nil
+}
+
func TestScheduleRotationTracksEarliestTimer(t *testing.T) {
now := time.Now()
controller := &proxylessGRPCWorkloadController{
diff --git a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
index 25de3817..3cc8b9a8 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
@@ -318,12 +318,21 @@ func TestKubeGatewayTemplateRendersDxgateResources(t *testing.T) {
if !strings.Contains(rendered[2], "DXGATE_BOOTSTRAP") || strings.Contains(rendered[2], "DXGATE_STATIC_CONFIG") {
t.Fatalf("deployment did not switch from static config to bootstrap:\n%s", rendered[2])
}
+ if !strings.Contains(rendered[2], `proxyless.dubbo.apache.org/inject: "true"`) {
+ t.Fatalf("deployment pod template did not enable proxyless injection for dxgate mTLS certs:\n%s", rendered[2])
+ }
+ if !strings.Contains(rendered[2], "inject.dubbo.apache.org/templates: grpc-engine") {
+ t.Fatalf("deployment pod template did not request grpc-engine injection:\n%s", rendered[2])
+ }
if !strings.Contains(rendered[2], "app.kubernetes.io/instance: public-dubbo") {
t.Fatalf("deployment did not render stable dxgate instance label:\n%s", rendered[2])
}
if !strings.Contains(rendered[3], "targetPort: http") {
t.Fatalf("service did not target dxgate http port:\n%s", rendered[3])
}
+ if !strings.Contains(rendered[3], `proxyless.dubbo.apache.org/inject: "false"`) {
+ t.Fatalf("service did not opt out of proxyless targetPort rewriting:\n%s", rendered[3])
+ }
if !strings.Contains(rendered[3], "app.kubernetes.io/instance: public-dubbo") {
t.Fatalf("service did not render stable dxgate instance selector:\n%s", rendered[3])
}
diff --git a/dubbod/discovery/pkg/model/virtualservice.go b/dubbod/discovery/pkg/model/virtualservice.go
index a4a31607..5ff17981 100644
--- a/dubbod/discovery/pkg/model/virtualservice.go
+++ b/dubbod/discovery/pkg/model/virtualservice.go
@@ -28,8 +28,7 @@ func meshServiceToVirtualServiceConfig(msConfig config.Config) config.Config {
return r
}
vs := &networking.VirtualService{
- Hosts: append([]string(nil), ms.Hosts...),
- ExportTo: append([]string(nil), ms.VisibleTo...),
+ Hosts: append([]string(nil), ms.Hosts...),
}
for _, route := range ms.Routes {
if route == nil || len(route.Service) == 0 {
@@ -69,7 +68,6 @@ func meshServiceToDestinationRuleConfig(msConfig config.Config) config.Config {
dr := &networking.DestinationRule{
Host: firstMeshServiceHost(ms, r.Meta),
TrafficPolicy: ms.TrafficPolicy,
- ExportTo: append([]string(nil), ms.VisibleTo...),
}
seen := setsString{}
for _, route := range ms.Routes {
diff --git a/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go b/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go
new file mode 100644
index 00000000..55bd2225
--- /dev/null
+++ b/dubbod/discovery/pkg/networking/grpcgen/mtls_test.go
@@ -0,0 +1,192 @@
+package grpcgen
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/host"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/dubboagency/grpcxds"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ 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"
+ listener "github.com/kdubbo/xds-api/listener/v1"
+)
+
+func TestMeshServiceDUBBOMutualBuildsUpstreamTLS(t *testing.T) {
+ hostname := host.Name("nginx.app.svc.cluster.local")
+ clusterName := model.BuildSubsetKey(model.TrafficDirectionOutbound, "", hostname, 80)
+ push := newRDSTestPushContext(t, []config.Config{
+ newMTLSMeshServiceConfig("nginx-mtls", "app", hostname),
+ }, []*model.Service{
+ newRDSTestService("nginx", "app", string(hostname), 80),
+ })
+
+ resources := (&GrpcConfigGenerator{}).BuildClusters(newMTLSTestProxy(), push, []string{clusterName})
+ c := findCluster(t, resources, clusterName)
+ if c.GetTransportSocket() == nil {
+ t.Fatalf("cluster %s has no transport socket", clusterName)
+ }
+
+ tlsContext := &tlsv1.UpstreamTlsContext{}
+ if err := c.GetTransportSocket().GetTypedConfig().UnmarshalTo(tlsContext); err != nil {
+ t.Fatalf("unmarshal upstream tls context: %v", err)
+ }
+ if tlsContext.GetSni() != string(hostname) {
+ t.Fatalf("SNI = %q, want %q", tlsContext.GetSni(), hostname)
+ }
+ assertCommonTLSContext(t, tlsContext.GetCommonTlsContext())
+}
+
+func TestPeerAuthenticationStrictBuildsDownstreamMTLSListener(t *testing.T) {
+ svc := newRDSTestService("nginx", "app", "nginx.app.svc.cluster.local", 80)
+ proxy := newMTLSTestProxy()
+ proxy.ServiceTargets = []model.ServiceTarget{{
+ Service: svc,
+ Port: model.ServiceInstancePort{
+ ServicePort: svc.Ports[0],
+ TargetPort: 80,
+ },
+ }}
+ push := newRDSTestPushContext(t, []config.Config{
+ newStrictPeerAuthenticationConfig("app-strict-mtls", "app"),
+ }, []*model.Service{svc})
+
+ listenerName := grpcxds.ServerListenerNamePrefix + "0.0.0.0:80"
+ resources := (&GrpcConfigGenerator{}).BuildListeners(proxy, push, []string{listenerName})
+ l := findListener(t, resources, listenerName)
+ if len(l.GetFilterChains()) != 1 {
+ t.Fatalf("filter chains = %d, want 1", len(l.GetFilterChains()))
+ }
+ transportSocket := l.GetFilterChains()[0].GetTransportSocket()
+ if transportSocket == nil {
+ t.Fatalf("listener %s has no transport socket", listenerName)
+ }
+
+ tlsContext := &tlsv1.DownstreamTlsContext{}
+ if err := transportSocket.GetTypedConfig().UnmarshalTo(tlsContext); err != nil {
+ t.Fatalf("unmarshal downstream tls context: %v", err)
+ }
+ if tlsContext.GetRequireClientCertificate() == nil || !tlsContext.GetRequireClientCertificate().GetValue() {
+ t.Fatalf("RequireClientCertificate = %v, want true", tlsContext.GetRequireClientCertificate())
+ }
+ assertCommonTLSContext(t, tlsContext.GetCommonTlsContext())
+}
+
+func newMTLSTestProxy() *model.Proxy {
+ return &model.Proxy{
+ ID: "proxyless~10.0.0.1~nginx-consumer.app~app.svc.cluster.local",
+ Type: model.Proxyless,
+ DNSDomain: "app.svc.cluster.local",
+ ConfigNamespace: "app",
+ Metadata: &model.NodeMetadata{
+ Namespace: "app",
+ },
+ }
+}
+
+func newMTLSMeshServiceConfig(name, namespace string, hostname host.Name) config.Config {
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.MeshService,
+ Name: name,
+ Namespace: namespace,
+ Domain: "cluster.local",
+ },
+ Spec: &networking.MeshService{
+ Hosts: []string{string(hostname)},
+ TrafficPolicy: &networking.TrafficPolicy{
+ Tls: &networking.ClientTLSSettings{
+ Mode: networking.ClientTLSSettings_DUBBO_MUTUAL,
+ },
+ },
+ Routes: []*networking.MeshServiceRoute{{
+ Service: []*networking.ServiceDestination{{
+ Name: "v1",
+ Host: string(hostname),
+ Labels: map[string]string{"version": "v1"},
+ Port: &networking.ServicePort{Number: 80},
+ Weight: 50,
+ }, {
+ Name: "v2",
+ Host: string(hostname),
+ Labels: map[string]string{"version": "v2"},
+ Port: &networking.ServicePort{Number: 80},
+ Weight: 50,
+ }},
+ }},
+ },
+ }
+}
+
+func newStrictPeerAuthenticationConfig(name, namespace string) config.Config {
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.PeerAuthentication,
+ Name: name,
+ Namespace: namespace,
+ Domain: "cluster.local",
+ },
+ Spec: &security.PeerAuthentication{
+ Mtls: &security.PeerAuthentication_MutualTLS{
+ Mode: security.PeerAuthentication_MutualTLS_STRICT,
+ },
+ },
+ }
+}
+
+func findCluster(t *testing.T, resources model.Resources, name string) *cluster.Cluster {
+ t.Helper()
+ for _, resource := range resources {
+ if resource.Name != name {
+ continue
+ }
+ c := &cluster.Cluster{}
+ if err := resource.Resource.UnmarshalTo(c); err != nil {
+ t.Fatalf("unmarshal cluster %s: %v", name, err)
+ }
+ return c
+ }
+ t.Fatalf("cluster %s not found in %d resources", name, len(resources))
+ return nil
+}
+
+func findListener(t *testing.T, resources model.Resources, name string) *listener.Listener {
+ t.Helper()
+ for _, resource := range resources {
+ if resource.Name != name {
+ continue
+ }
+ l := &listener.Listener{}
+ if err := resource.Resource.UnmarshalTo(l); err != nil {
+ t.Fatalf("unmarshal listener %s: %v", name, err)
+ }
+ return l
+ }
+ t.Fatalf("listener %s not found in %d resources", name, len(resources))
+ return nil
+}
+
+func assertCommonTLSContext(t *testing.T, ctx *tlsv1.CommonTlsContext) {
+ t.Helper()
+ if ctx == nil {
+ t.Fatal("common TLS context is nil")
+ }
+ if got := ctx.GetAlpnProtocols(); len(got) != 1 || got[0] != "h2" {
+ t.Fatalf("ALPN protocols = %v, want [h2]", got)
+ }
+ cert := ctx.GetTlsCertificateCertificateProviderInstance()
+ if cert == nil || cert.GetInstanceName() != "default" || cert.GetCertificateName() != "default" {
+ t.Fatalf("workload cert provider = %v, want default/default", cert)
+ }
+ combined := ctx.GetCombinedValidationContext()
+ if combined == nil {
+ t.Fatal("combined validation context is nil")
+ }
+ root := combined.GetValidationContextCertificateProviderInstance()
+ if root == nil || root.GetInstanceName() != "default" || root.GetCertificateName() != "ROOTCA" {
+ t.Fatalf("root cert provider = %v, want default/ROOTCA", root)
+ }
+}
diff --git a/dubbod/discovery/pkg/xds/eds.go b/dubbod/discovery/pkg/xds/eds.go
index d49ec190..bdd484e1 100644
--- a/dubbod/discovery/pkg/xds/eds.go
+++ b/dubbod/discovery/pkg/xds/eds.go
@@ -36,7 +36,6 @@ type EdsGenerator struct {
var _ model.XdsDeltaResourceGenerator = &EdsGenerator{}
var skippedEdsConfigs = sets.New(
- kind.MeshService,
kind.PeerAuthentication,
kind.Secret,
)
diff --git a/dubbod/discovery/pkg/xds/endpoints/endpoint_builder.go b/dubbod/discovery/pkg/xds/endpoints/endpoint_builder.go
index d99fd205..fea15978 100644
--- a/dubbod/discovery/pkg/xds/endpoints/endpoint_builder.go
+++ b/dubbod/discovery/pkg/xds/endpoints/endpoint_builder.go
@@ -23,8 +23,10 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/util"
"github.com/apache/dubbo-kubernetes/pkg/config/host"
"github.com/apache/dubbo-kubernetes/pkg/config/labels"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/inject"
dubbolog "github.com/apache/dubbo-kubernetes/pkg/log"
"github.com/cespare/xxhash/v2"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
// core "github.com/kdubbo/xds-api/core/v1"
core "github.com/kdubbo/xds-api/core/v1"
// endpoint "github.com/kdubbo/xds-api/endpoint/v1"
@@ -84,14 +86,6 @@ func (b *EndpointBuilder) BuildClusterLoadAssignment(endpointIndex *model.Endpoi
return buildEmptyClusterLoadAssignment(b.clusterName)
}
- // Get endpoints from the endpoint index
- allServices := endpointIndex.AllServices()
- if len(allServices) > 0 {
- log.Infof("EndpointIndex contains %d services: %v", len(allServices), allServices)
- } else {
- log.Warnf("EndpointIndex is empty - no services registered")
- }
-
shards, ok := endpointIndex.ShardsForService(string(b.hostname), b.service.Attributes.Namespace)
if !ok {
// Also log what services ARE available in the namespace
@@ -267,7 +261,7 @@ func (b *EndpointBuilder) buildLbEndpoint(ep *model.DubboEndpoint) *endpoint.LbE
return nil
}
- address := util.BuildAddress(ep.Addresses[0], ep.EndpointPort)
+ address := util.BuildAddress(ep.Addresses[0], b.endpointPort(ep))
if address == nil {
return nil
}
@@ -298,6 +292,45 @@ func (b *EndpointBuilder) buildLbEndpoint(ep *model.DubboEndpoint) *endpoint.LbE
}
}
+func (b *EndpointBuilder) endpointPort(ep *model.DubboEndpoint) uint32 {
+ if b.useXServerEndpointPort() {
+ return inject.ProxylessXServerPort
+ }
+ return ep.EndpointPort
+}
+
+func (b *EndpointBuilder) useXServerEndpointPort() bool {
+ if b == nil || b.proxy == nil || !b.proxy.IsProxylessGrpc() || b.push == nil || b.service == nil {
+ return false
+ }
+ return b.destinationUsesDUBBOMutual()
+}
+
+func (b *EndpointBuilder) destinationUsesDUBBOMutual() bool {
+ rule := b.push.DestinationRuleForService(b.service.Attributes.Namespace, b.hostname)
+ if rule == nil && b.service.Hostname != "" && b.service.Hostname != b.hostname {
+ rule = b.push.DestinationRuleForService(b.service.Attributes.Namespace, b.service.Hostname)
+ }
+ if rule == nil {
+ return false
+ }
+ if b.subsetName != "" {
+ for _, subset := range rule.Subsets {
+ if subset.Name == b.subsetName && subset.TrafficPolicy != nil {
+ return trafficPolicyUsesDUBBOMutual(subset.TrafficPolicy)
+ }
+ }
+ }
+ return trafficPolicyUsesDUBBOMutual(rule.TrafficPolicy)
+}
+
+func trafficPolicyUsesDUBBOMutual(policy *networking.TrafficPolicy) bool {
+ if policy == nil || policy.Tls == nil {
+ return false
+ }
+ return policy.Tls.Mode == networking.ClientTLSSettings_DUBBO_MUTUAL || policy.Tls.Mode.String() == "DUBBO_MUTUAL"
+}
+
func buildEmptyClusterLoadAssignment(clusterName string) *endpoint.ClusterLoadAssignment {
// preventing "weighted-target: no targets to pick from" errors
return &endpoint.ClusterLoadAssignment{
diff --git a/dubbod/discovery/pkg/xds/endpoints/endpoint_builder_test.go b/dubbod/discovery/pkg/xds/endpoints/endpoint_builder_test.go
new file mode 100644
index 00000000..21714332
--- /dev/null
+++ b/dubbod/discovery/pkg/xds/endpoints/endpoint_builder_test.go
@@ -0,0 +1,167 @@
+package endpoints
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/config/memory"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/host"
+ "github.com/apache/dubbo-kubernetes/pkg/config/mesh"
+ "github.com/apache/dubbo-kubernetes/pkg/config/mesh/meshwatcher"
+ "github.com/apache/dubbo-kubernetes/pkg/config/protocol"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/collections"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/inject"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ networking "github.com/kdubbo/api/networking/v1alpha3"
+ endpoint "github.com/kdubbo/xds-api/endpoint/v1"
+)
+
+func TestBuildClusterLoadAssignmentUsesXServerPortForDUBBOMutual(t *testing.T) {
+ hostname := host.Name("nginx.app.svc.cluster.local")
+ svc := newEndpointTestService("nginx", "app", string(hostname), 80)
+ push := newEndpointTestPushContext(t, []config.Config{
+ newEndpointTestMTLSMeshService("nginx-routing", "app", hostname),
+ }, []*model.Service{svc})
+ index := model.NewEndpointIndex(model.DisabledCache{})
+ index.UpdateServiceEndpoints(model.ShardKey{}, string(hostname), "app", []*model.DubboEndpoint{{
+ Addresses: []string{"10.0.0.1"},
+ EndpointPort: 80,
+ ServicePortName: "http",
+ Labels: map[string]string{"version": "v1"},
+ HealthStatus: model.Healthy,
+ }}, false)
+
+ clusterName := model.BuildSubsetKey(model.TrafficDirectionOutbound, "v1", hostname, 80)
+ builder := NewEndpointBuilder(clusterName, newEndpointTestProxy(), push)
+ cla := builder.BuildClusterLoadAssignment(index)
+
+ if got := firstEndpointPort(t, cla); got != inject.ProxylessXServerPort {
+ t.Fatalf("endpoint port = %d, want xserver port %d", got, inject.ProxylessXServerPort)
+ }
+}
+
+func TestBuildClusterLoadAssignmentKeepsAppPortWithoutDUBBOMutual(t *testing.T) {
+ hostname := host.Name("nginx.app.svc.cluster.local")
+ svc := newEndpointTestService("nginx", "app", string(hostname), 80)
+ push := newEndpointTestPushContext(t, nil, []*model.Service{svc})
+ index := model.NewEndpointIndex(model.DisabledCache{})
+ index.UpdateServiceEndpoints(model.ShardKey{}, string(hostname), "app", []*model.DubboEndpoint{{
+ Addresses: []string{"10.0.0.1"},
+ EndpointPort: 80,
+ ServicePortName: "http",
+ HealthStatus: model.Healthy,
+ }}, false)
+
+ clusterName := model.BuildSubsetKey(model.TrafficDirectionOutbound, "", hostname, 80)
+ builder := NewEndpointBuilder(clusterName, newEndpointTestProxy(), push)
+ cla := builder.BuildClusterLoadAssignment(index)
+
+ if got := firstEndpointPort(t, cla); got != 80 {
+ t.Fatalf("endpoint port = %d, want app port 80", got)
+ }
+}
+
+func firstEndpointPort(t *testing.T, cla *endpoint.ClusterLoadAssignment) uint32 {
+ t.Helper()
+ localities := cla.GetEndpoints()
+ if len(localities) == 0 || len(localities[0].GetLbEndpoints()) == 0 {
+ t.Fatalf("CLA has no endpoints")
+ }
+ return localities[0].GetLbEndpoints()[0].GetEndpoint().GetAddress().GetSocketAddress().GetPortValue()
+}
+
+func newEndpointTestPushContext(t *testing.T, configs []config.Config, services []*model.Service) *model.PushContext {
+ t.Helper()
+ store := memory.Make(collections.DubboGatewayAPI())
+ for _, cfg := range configs {
+ if _, err := store.Create(cfg); err != nil {
+ t.Fatalf("create config %s/%s: %v", cfg.Namespace, cfg.Name, err)
+ }
+ }
+ env := model.NewEnvironment()
+ env.ConfigStore = store
+ env.ServiceDiscovery = endpointTestServiceDiscovery{services: services}
+ env.Watcher = meshwatcher.ConfigAdapter(krt.NewStatic(&meshwatcher.MeshConfigResource{
+ MeshConfig: mesh.DefaultMeshConfig(),
+ }, true))
+ env.Init()
+ push := model.NewPushContext()
+ push.InitContext(env, nil, nil)
+ return push
+}
+
+func newEndpointTestProxy() *model.Proxy {
+ return &model.Proxy{
+ ID: "proxyless~10.0.0.2~nginx-consumer.app~app.svc.cluster.local",
+ Type: model.Proxyless,
+ ConfigNamespace: "app",
+ Metadata: &model.NodeMetadata{
+ Generator: "grpc",
+ Namespace: "app",
+ },
+ }
+}
+
+func newEndpointTestService(name, namespace, hostname string, port int) *model.Service {
+ return &model.Service{
+ Hostname: host.Name(hostname),
+ Ports: model.PortList{{
+ Name: "http",
+ Port: port,
+ Protocol: protocol.HTTP2,
+ }},
+ Attributes: model.ServiceAttributes{
+ Name: name,
+ Namespace: namespace,
+ },
+ }
+}
+
+func newEndpointTestMTLSMeshService(name, namespace string, hostname host.Name) config.Config {
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.MeshService,
+ Name: name,
+ Namespace: namespace,
+ Domain: "cluster.local",
+ },
+ Spec: &networking.MeshService{
+ Hosts: []string{string(hostname)},
+ TrafficPolicy: &networking.TrafficPolicy{
+ Tls: &networking.ClientTLSSettings{Mode: networking.ClientTLSSettings_DUBBO_MUTUAL},
+ },
+ Routes: []*networking.MeshServiceRoute{{
+ Service: []*networking.ServiceDestination{{
+ Name: "v1",
+ Host: string(hostname),
+ Labels: map[string]string{"version": "v1"},
+ Port: &networking.ServicePort{Number: 80},
+ Weight: 100,
+ }},
+ }},
+ },
+ }
+}
+
+type endpointTestServiceDiscovery struct {
+ services []*model.Service
+}
+
+func (s endpointTestServiceDiscovery) Services() []*model.Service {
+ return s.services
+}
+
+func (s endpointTestServiceDiscovery) GetService(hostname host.Name) *model.Service {
+ for _, svc := range s.services {
+ if svc.Hostname == hostname {
+ return svc
+ }
+ }
+ return nil
+}
+
+func (s endpointTestServiceDiscovery) GetProxyServiceTargets(*model.Proxy) []model.ServiceTarget {
+ return nil
+}
diff --git a/dubboinstaller/charts/base/Chart.yaml b/dubboinstaller/charts/base/Chart.yaml
index 0827c2ca..0fb64454 100644
--- a/dubboinstaller/charts/base/Chart.yaml
+++ b/dubboinstaller/charts/base/Chart.yaml
@@ -20,7 +20,7 @@ name: base
home: https://github.com/apache/dubbo-kubernetes
icon: https://avatars.githubusercontent.com/u/11751992?s=200&v=4
description: Helm chart for deploying dubbo cluster resources and CRDs.
-version: 0.3.7
+version: 0.3.9
appVersion: "latest"
maintainers:
- name: mfordjody
diff --git a/dubboinstaller/charts/base/files/crd-all.gen.yaml b/dubboinstaller/charts/base/files/crd-all.gen.yaml
index ea4cd3fc..575eccd1 100644
--- a/dubboinstaller/charts/base/files/crd-all.gen.yaml
+++ b/dubboinstaller/charts/base/files/crd-all.gen.yaml
@@ -232,11 +232,6 @@ spec:
x-kubernetes-int-or-string: true
type: object
type: object
- visibleTo:
- description: Namespaces that can see this mesh service.
- items:
- type: string
- type: array
type: object
status:
properties:
diff --git a/dubboinstaller/charts/dubbod/Chart.yaml b/dubboinstaller/charts/dubbod/Chart.yaml
index cae1aa51..64091abc 100644
--- a/dubboinstaller/charts/dubbod/Chart.yaml
+++ b/dubboinstaller/charts/dubbod/Chart.yaml
@@ -25,7 +25,7 @@ keywords:
- dubbod
- dubbo-discovery
- dubbo-control-plane
-version: 0.3.7
+version: 0.3.9
appVersion: "latest"
maintainers:
- name: mfordjody
diff --git a/dubboinstaller/charts/dubbod/files/grpc-engine.yaml b/dubboinstaller/charts/dubbod/files/grpc-engine.yaml
index 61a84dc6..7f3a2d9e 100644
--- a/dubboinstaller/charts/dubbod/files/grpc-engine.yaml
+++ b/dubboinstaller/charts/dubbod/files/grpc-engine.yaml
@@ -74,6 +74,27 @@ spec:
- name: dubbo-xds
mountPath: /etc/dubbo/proxy
readOnly: true
+{{- $appPort := 80 }}
+{{- if and (gt (len .Spec.Containers) 0) (gt (len ((index .Spec.Containers 0).Ports)) 0) }}
+{{- $appPort = int ((index ((index .Spec.Containers 0).Ports) 0).ContainerPort) }}
+{{- end }}
+ - name: dubbo-xserver
+ image: {{ .ProxyImage | quote }}
+ imagePullPolicy: IfNotPresent
+ args:
+ - xserver
+ - --listen
+ - :25080
+ - --upstream
+ - 127.0.0.1:{{ $appPort }}
+ ports:
+ - name: xserver
+ containerPort: 25080
+ protocol: TCP
+ volumeMounts:
+ - name: dubbo-xds
+ mountPath: /etc/dubbo/proxy
+ readOnly: true
volumes:
- name: dubbo-xds
secret:
diff --git a/dubboinstaller/charts/dubbod/files/kube-gateway.yaml b/dubboinstaller/charts/dubbod/files/kube-gateway.yaml
index 138247f9..e880d51f 100644
--- a/dubboinstaller/charts/dubbod/files/kube-gateway.yaml
+++ b/dubboinstaller/charts/dubbod/files/kube-gateway.yaml
@@ -63,9 +63,11 @@ spec:
app.kubernetes.io/managed-by: dubbod
gateway.dubbo.apache.org/managed: dubbo.apache.org-gateway-controller
gateway.networking.k8s.io/gateway-name: {{ .Gateway.Name }}
+ proxyless.dubbo.apache.org/inject: "true"
annotations:
dubbo.apache.org/rev: {{ .Revision | quote }}
gateway.dubbo.apache.org/bootstrap-hash: {{ .BootstrapConfigHash | quote }}
+ inject.dubbo.apache.org/templates: grpc-engine
spec:
serviceAccountName: {{ .ServiceAccount }}
containers:
@@ -149,6 +151,7 @@ metadata:
app.kubernetes.io/instance: {{ .DeploymentName }}
app.kubernetes.io/managed-by: dubbod
gateway.networking.k8s.io/gateway-name: {{ .Gateway.Name }}
+ proxyless.dubbo.apache.org/inject: "false"
spec:
type: {{ .ServiceType }}
selector:
diff --git a/dubboinstaller/charts/dubbod/templates/deployment.yaml b/dubboinstaller/charts/dubbod/templates/deployment.yaml
index cb89138b..075c58cd 100644
--- a/dubboinstaller/charts/dubbod/templates/deployment.yaml
+++ b/dubboinstaller/charts/dubbod/templates/deployment.yaml
@@ -40,7 +40,7 @@ spec:
containers:
- name: execute
image: "kdubbo/dubbod:debug"
- imagePullPolicy: Always
+ imagePullPolicy: IfNotPresent
args:
- "execute"
- --domain
diff --git a/dubboinstaller/charts/dubbod/templates/mutatingwebhookconfiguration.yaml b/dubboinstaller/charts/dubbod/templates/mutatingwebhookconfiguration.yaml
index 4da36af4..7d70ffd7 100644
--- a/dubboinstaller/charts/dubbod/templates/mutatingwebhookconfiguration.yaml
+++ b/dubboinstaller/charts/dubbod/templates/mutatingwebhookconfiguration.yaml
@@ -55,6 +55,15 @@ webhooks:
- CREATE
resources:
- pods
+ - apiGroups:
+ - ""
+ apiVersions:
+ - v1
+ operations:
+ - CREATE
+ - UPDATE
+ resources:
+ - services
sideEffects: None
- admissionReviewVersions:
- v1
@@ -92,6 +101,15 @@ webhooks:
- CREATE
resources:
- pods
+ - apiGroups:
+ - ""
+ apiVersions:
+ - v1
+ operations:
+ - CREATE
+ - UPDATE
+ resources:
+ - services
sideEffects: None
- admissionReviewVersions:
- v1
@@ -124,6 +142,15 @@ webhooks:
- CREATE
resources:
- pods
+ - apiGroups:
+ - ""
+ apiVersions:
+ - v1
+ operations:
+ - CREATE
+ - UPDATE
+ resources:
+ - services
sideEffects: None
- admissionReviewVersions:
- v1
@@ -154,4 +181,13 @@ webhooks:
- CREATE
resources:
- pods
- sideEffects: None
\ No newline at end of file
+ - apiGroups:
+ - ""
+ apiVersions:
+ - v1
+ operations:
+ - CREATE
+ - UPDATE
+ resources:
+ - services
+ sideEffects: None
diff --git a/go.mod b/go.mod
index a8db9367..5c97c5f6 100644
--- a/go.mod
+++ b/go.mod
@@ -53,9 +53,9 @@ require (
github.com/hashicorp/go-multierror v1.1.1
github.com/hashicorp/golang-lru/v2 v2.0.5
github.com/heroku/color v0.0.6
- github.com/kdubbo/api v0.0.0-20260430152956-c0ec9648793f
- github.com/kdubbo/client-go v0.0.0-20260322200550-82e70d319918
- github.com/kdubbo/xds-api v0.0.0-20260423094330-4e01ed7e8f82
+ github.com/kdubbo/api v0.0.0-20260513025923-86f2ad63c4ab
+ github.com/kdubbo/client-go v0.0.0-20260506035847-a9460155f65d
+ github.com/kdubbo/xds-api v0.0.0-20260513100730-a07294e322cc
github.com/moby/term v0.5.2
github.com/ory/viper v1.7.5
github.com/pkg/errors v0.9.1
@@ -257,7 +257,3 @@ replace (
github.com/moby/buildkit => github.com/moby/buildkit v0.10.6
github.com/moby/dockerfile => github.com/moby/dockerfile v1.4.1
)
-
-replace github.com/kdubbo/api => ../api
-
-replace github.com/kdubbo/client-go => ../client-go
diff --git a/go.sum b/go.sum
index 0be007d5..e59f6c4d 100644
--- a/go.sum
+++ b/go.sum
@@ -335,8 +335,14 @@ github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm
github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
-github.com/kdubbo/xds-api v0.0.0-20260423094330-4e01ed7e8f82 h1:pZBmQ8vP6gW1502ANOxx00rUtSsZNdijldFNwFgJSYo=
-github.com/kdubbo/xds-api v0.0.0-20260423094330-4e01ed7e8f82/go.mod h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
+github.com/kdubbo/api v0.0.0-20260513025923-86f2ad63c4ab h1:HJdyq0CIsepvNZlGW/mWg78FreoVF4E9HXi8dlpXfAQ=
+github.com/kdubbo/api v0.0.0-20260513025923-86f2ad63c4ab/go.mod h1:8BtJiIovg7QCPsCxXcw3gDf922VcvYq5ihOSvj49Rq8=
+github.com/kdubbo/client-go v0.0.0-20260506035847-a9460155f65d h1:YYS3OfwT5mfqIgf17yFqCjio9NGHNIURNh1OvjLCujs=
+github.com/kdubbo/client-go v0.0.0-20260506035847-a9460155f65d/go.mod h1:VoiR0j6MxAcabmLL3dyz5vC3ShxCc3gKUjEcju3QeiE=
+github.com/kdubbo/xds-api v0.0.0-20260513025923-effbe30b748a h1:Quuv4Y8zq6ttf0qEFhwPbDBPTH/jfqKjkHEvkd2/oEE=
+github.com/kdubbo/xds-api v0.0.0-20260513025923-effbe30b748a/go.mod h1:o2HDUgL1ntaDbWomZ4cD2tt8jBamuG2qRtjXOa1zZ0Q=
+github.com/kdubbo/xds-api v0.0.0-20260513100730-a07294e322cc h1:dhfVaVHjt9RmNjeq8qcOWZPLSvRT3BHEbDyUeFiAkJg=
+github.com/kdubbo/xds-api v0.0.0-20260513100730-a07294e322cc/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/kube/inject/inject.go b/pkg/kube/inject/inject.go
index 95b9fd59..f77488e5 100644
--- a/pkg/kube/inject/inject.go
+++ b/pkg/kube/inject/inject.go
@@ -115,7 +115,7 @@ func RunTemplate(params InjectionParameters) (mergedPod *corev1.Pod, templatePod
MeshConfig: meshConfig,
Values: params.valuesConfig.asMap,
Revision: params.revision,
- ProxyImage: getProxyImage(params.valuesConfig.asMap, "dubboregistry/dubbo-proxy:0.3.6"),
+ ProxyImage: getProxyImage(params.valuesConfig.asMap, "kdubbo/dubbod:debug"),
CompliancePolicy: common_features.CompliancePolicy,
}
diff --git a/pkg/kube/inject/proxyless.go b/pkg/kube/inject/proxyless.go
index 4add5856..567c83ed 100644
--- a/pkg/kube/inject/proxyless.go
+++ b/pkg/kube/inject/proxyless.go
@@ -35,6 +35,8 @@ const (
ProxylessGRPCConfigEnvName = "DUBBO_GRPC_XDS_CONFIG"
ProxylessGRPCConfigFileName = "dubbo-grpc-xds.json"
ProxylessGRPCConfigPath = ProxylessXDSMountPath + "/" + ProxylessGRPCConfigFileName
+ ProxylessXServerContainerName = "dubbo-xserver"
+ ProxylessXServerPort = 25080
)
var ProxylessInjectTemplatesAnnoName = annotation.OrgApacheDubboInjectTemplates.Name
diff --git a/pkg/kube/inject/proxyless_test.go b/pkg/kube/inject/proxyless_test.go
index 41554a07..3300cce3 100644
--- a/pkg/kube/inject/proxyless_test.go
+++ b/pkg/kube/inject/proxyless_test.go
@@ -83,17 +83,18 @@ func TestInstallerGRPCEngineTemplateInjectsDirectXDSConnection(t *testing.T) {
t.Fatalf("RunTemplate() failed: %v", err)
}
- if len(injectedPod.Spec.Containers) != 1 {
- t.Fatalf("template containers = %d, want 1 hard-coded application container overlay", len(injectedPod.Spec.Containers))
+ if len(injectedPod.Spec.Containers) != 2 {
+ t.Fatalf("template containers = %d, want app overlay plus xserver", len(injectedPod.Spec.Containers))
}
if err := postProcessPod(mergedPod, *injectedPod, req); err != nil {
t.Fatalf("postProcessPod() failed: %v", err)
}
- if len(mergedPod.Spec.Containers) != 1 {
- t.Fatalf("containers = %d, want 1 application container", len(mergedPod.Spec.Containers))
+ if len(mergedPod.Spec.Containers) != 2 {
+ t.Fatalf("containers = %d, want application container plus xserver", len(mergedPod.Spec.Containers))
}
assertDirectXDSConnection(t, mergedPod, "app", ProxylessGRPCSecretNameForMeta(pod.ObjectMeta))
+ assertXServerContainer(t, mergedPod)
}
func TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t *testing.T) {
@@ -152,10 +153,11 @@ func TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t *testing
if err := postProcessPod(mergedPod, *injectedPod, req); err != nil {
t.Fatalf("postProcessPod() failed: %v", err)
}
- if len(mergedPod.Spec.Containers) != 1 {
- t.Fatalf("containers = %d, want original nginx container only", len(mergedPod.Spec.Containers))
+ if len(mergedPod.Spec.Containers) != 2 {
+ t.Fatalf("containers = %d, want original nginx container plus xserver", len(mergedPod.Spec.Containers))
}
assertDirectXDSConnection(t, mergedPod, "nginx", ProxylessGRPCSecretNameForMeta(pod.ObjectMeta))
+ assertXServerContainer(t, mergedPod)
if got := mergedPod.Spec.Volumes[0].Secret.SecretName; got == ProxylessGRPCSecretName("") {
t.Fatalf("secret name = %q, want generateName-based secret", got)
}
@@ -228,6 +230,24 @@ func assertNoArgs(t *testing.T, pod *corev1.Pod) {
}
}
+func assertXServerContainer(t *testing.T, pod *corev1.Pod) {
+ t.Helper()
+ container := FindContainer(ProxylessXServerContainerName, pod.Spec.Containers)
+ if container == nil {
+ t.Fatalf("%s container missing", ProxylessXServerContainerName)
+ }
+ if container.Image != "kdubbo/dubbod:debug" {
+ t.Fatalf("xserver image = %q, want kdubbo/dubbod:debug", container.Image)
+ }
+ wantArgs := []string{"xserver", "--listen", ":25080", "--upstream", "127.0.0.1:80"}
+ if strings.Join(container.Args, ",") != strings.Join(wantArgs, ",") {
+ t.Fatalf("xserver args = %v, want %v", container.Args, wantArgs)
+ }
+ if !hasMount(container.VolumeMounts, ProxylessXDSVolumeName, ProxylessXDSMountPath, true) {
+ t.Fatalf("xserver proxyless xds mount missing")
+ }
+}
+
func TestAddApplicationContainerConfigInjectsProxylessGRPCContract(t *testing.T) {
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
diff --git a/pkg/kube/inject/service_test.go b/pkg/kube/inject/service_test.go
new file mode 100644
index 00000000..52f82f22
--- /dev/null
+++ b/pkg/kube/inject/service_test.go
@@ -0,0 +1,188 @@
+// 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 inject
+
+import (
+ "encoding/json"
+ "strings"
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/pkg/kube"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+ "k8s.io/apimachinery/pkg/types"
+ "k8s.io/apimachinery/pkg/util/intstr"
+)
+
+func TestRewriteProxylessServiceTargetPortsRoutesTCPServiceToXServer(t *testing.T) {
+ svc := &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "nginx", Namespace: "app"},
+ Spec: corev1.ServiceSpec{
+ Selector: map[string]string{"app": "nginx"},
+ Ports: []corev1.ServicePort{
+ {Name: "http", Port: 80, TargetPort: intstr.FromInt(80), Protocol: corev1.ProtocolTCP},
+ {Name: "metrics", Port: 9090, TargetPort: intstr.FromString("metrics"), Protocol: corev1.ProtocolTCP},
+ },
+ },
+ }
+
+ if !rewriteProxylessServiceTargetPorts(svc) {
+ t.Fatalf("rewriteProxylessServiceTargetPorts() = false, want true")
+ }
+ for _, port := range svc.Spec.Ports {
+ if got := port.TargetPort.IntVal; got != ProxylessXServerPort {
+ t.Fatalf("port %s targetPort = %d, want %d", port.Name, got, ProxylessXServerPort)
+ }
+ }
+}
+
+func TestRewriteProxylessServiceTargetPortsSkipsUnsafeServices(t *testing.T) {
+ cases := []struct {
+ name string
+ svc corev1.Service
+ }{
+ {
+ name: "selectorless",
+ svc: corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "manual", Namespace: "app"},
+ Spec: corev1.ServiceSpec{
+ Ports: []corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromInt(80)}},
+ },
+ },
+ },
+ {
+ name: "external-name",
+ svc: corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "external", Namespace: "app"},
+ Spec: corev1.ServiceSpec{
+ Type: corev1.ServiceTypeExternalName,
+ ExternalName: "example.com",
+ Selector: map[string]string{"app": "external"},
+ Ports: []corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromInt(80)}},
+ },
+ },
+ },
+ }
+
+ for _, tt := range cases {
+ t.Run(tt.name, func(t *testing.T) {
+ svc := tt.svc
+ if rewriteProxylessServiceTargetPorts(&svc) {
+ t.Fatalf("rewriteProxylessServiceTargetPorts() = true, want false")
+ }
+ if got := svc.Spec.Ports[0].TargetPort.IntVal; got != 80 {
+ t.Fatalf("targetPort = %d, want 80", got)
+ }
+ })
+ }
+}
+
+func TestRewriteProxylessServiceTargetPortsSkipsNonTCPPorts(t *testing.T) {
+ svc := &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "dns", Namespace: "app"},
+ Spec: corev1.ServiceSpec{
+ Selector: map[string]string{"app": "dns"},
+ Ports: []corev1.ServicePort{
+ {Name: "dns", Port: 53, TargetPort: intstr.FromInt(53), Protocol: corev1.ProtocolUDP},
+ },
+ },
+ }
+
+ if rewriteProxylessServiceTargetPorts(svc) {
+ t.Fatalf("rewriteProxylessServiceTargetPorts() = true, want false")
+ }
+ if got := svc.Spec.Ports[0].TargetPort.IntVal; got != 53 {
+ t.Fatalf("targetPort = %d, want 53", got)
+ }
+}
+
+func TestInjectServiceCreatesXServerTargetPortPatch(t *testing.T) {
+ svc := corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{Name: "nginx", Namespace: "app"},
+ Spec: corev1.ServiceSpec{
+ Selector: map[string]string{"app": "nginx"},
+ Ports: []corev1.ServicePort{
+ {Name: "http", Port: 80, TargetPort: intstr.FromInt(80), Protocol: corev1.ProtocolTCP},
+ },
+ },
+ }
+ raw, err := json.Marshal(svc)
+ if err != nil {
+ t.Fatalf("json.Marshal() failed: %v", err)
+ }
+
+ wh := &Webhook{Config: &Config{Policy: InjectionPolicyEnabled}}
+ resp := wh.injectService(&kube.AdmissionReview{
+ Request: &kube.AdmissionRequest{
+ UID: types.UID("test"),
+ Kind: metav1.GroupVersionKind{Version: "v1", Kind: "Service"},
+ Namespace: "app",
+ Operation: kube.Create,
+ Object: runtime.RawExtension{Raw: raw},
+ },
+ }, "/inject")
+
+ if !resp.Allowed {
+ t.Fatalf("Allowed = false, want true")
+ }
+ if resp.PatchType == nil || *resp.PatchType != "JSONPatch" {
+ t.Fatalf("PatchType = %v, want JSONPatch", resp.PatchType)
+ }
+ if !strings.Contains(string(resp.Patch), `"value":25080`) {
+ t.Fatalf("patch = %s, want targetPort 25080", string(resp.Patch))
+ }
+}
+
+func TestInjectServiceSkipsProxylessOptOutService(t *testing.T) {
+ svc := corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "dxgate-gateway",
+ Namespace: "app",
+ Labels: map[string]string{
+ "proxyless.dubbo.apache.org/inject": "false",
+ },
+ },
+ Spec: corev1.ServiceSpec{
+ Selector: map[string]string{"app.kubernetes.io/name": "dxgate"},
+ Ports: []corev1.ServicePort{
+ {Name: "http", Port: 80, TargetPort: intstr.FromString("http"), Protocol: corev1.ProtocolTCP},
+ },
+ },
+ }
+ raw, err := json.Marshal(svc)
+ if err != nil {
+ t.Fatalf("json.Marshal() failed: %v", err)
+ }
+
+ wh := &Webhook{Config: &Config{Policy: InjectionPolicyEnabled}}
+ resp := wh.injectService(&kube.AdmissionReview{
+ Request: &kube.AdmissionRequest{
+ UID: types.UID("test"),
+ Kind: metav1.GroupVersionKind{Version: "v1", Kind: "Service"},
+ Namespace: "app",
+ Operation: kube.Create,
+ Object: runtime.RawExtension{Raw: raw},
+ },
+ }, "/inject")
+
+ if !resp.Allowed {
+ t.Fatalf("Allowed = false, want true")
+ }
+ if len(resp.Patch) != 0 {
+ t.Fatalf("patch = %s, want no patch", string(resp.Patch))
+ }
+}
diff --git a/pkg/kube/inject/webhook.go b/pkg/kube/inject/webhook.go
index 4df7f767..760b932b 100644
--- a/pkg/kube/inject/webhook.go
+++ b/pkg/kube/inject/webhook.go
@@ -45,6 +45,7 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/types"
+ "k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/mergepatch"
"k8s.io/apimachinery/pkg/util/strategicpatch"
"sigs.k8s.io/yaml"
@@ -230,6 +231,20 @@ func (wh *Webhook) serveInject(w http.ResponseWriter, r *http.Request) {
}
func (wh *Webhook) inject(ar *kube.AdmissionReview, path string) *kube.AdmissionResponse {
+ if ar == nil || ar.Request == nil {
+ return toAdmissionResponse(errors.New("admission request is nil"))
+ }
+ switch ar.Request.Kind.Kind {
+ case "Pod":
+ return wh.injectPod(ar, path)
+ case "Service":
+ return wh.injectService(ar, path)
+ default:
+ return &kube.AdmissionResponse{Allowed: true}
+ }
+}
+
+func (wh *Webhook) injectPod(ar *kube.AdmissionReview, path string) *kube.AdmissionResponse {
log := webhookLog.WithLabels("path", path)
req := ar.Request
var pod corev1.Pod
@@ -295,6 +310,55 @@ func (wh *Webhook) inject(ar *kube.AdmissionReview, path string) *kube.Admission
return &reviewResponse
}
+func (wh *Webhook) injectService(ar *kube.AdmissionReview, path string) *kube.AdmissionResponse {
+ log := webhookLog.WithLabels("path", path)
+ req := ar.Request
+ var svc corev1.Service
+ if err := json.Unmarshal(req.Object.Raw, &svc); err != nil {
+ log.Errorf("Could not unmarshal raw service object: %v %s", err, string(req.Object.Raw))
+ return toAdmissionResponse(err)
+ }
+ if svc.Namespace == "" {
+ svc.Namespace = req.Namespace
+ }
+
+ log = log.WithLabels("service", svc.Namespace+"/"+svc.Name)
+ log.Infof("Process proxyless service request")
+
+ wh.mu.RLock()
+ required := injectRequired(IgnoredNamespaces.UnsortedList(), wh.Config, &corev1.PodSpec{}, svc.ObjectMeta)
+ wh.mu.RUnlock()
+ if !required {
+ log.Infof("Skipping service due to policy check")
+ return &kube.AdmissionResponse{Allowed: true}
+ }
+
+ originalService, err := json.Marshal(svc)
+ if err != nil {
+ return toAdmissionResponse(err)
+ }
+ if !rewriteProxylessServiceTargetPorts(&svc) {
+ return &kube.AdmissionResponse{Allowed: true}
+ }
+
+ patchBytes, err := createServicePatch(&svc, originalService)
+ if err != nil {
+ log.Errorf("Service injection failed: %v", err)
+ return toAdmissionResponse(err)
+ }
+
+ log.Infof("Service injection successfully, patch size: %d bytes", len(patchBytes))
+ reviewResponse := kube.AdmissionResponse{
+ Allowed: true,
+ Patch: patchBytes,
+ PatchType: func() *string {
+ pt := "JSONPatch"
+ return &pt
+ }(),
+ }
+ return &reviewResponse
+}
+
func (wh *Webhook) Run(stop <-chan struct{}) {
go wh.watcher.Run(stop)
}
@@ -609,6 +673,7 @@ func reorderPod(pod *corev1.Pod, req InjectionParameters) error {
// Proxy container should be last to ensure `kubectl exec` and similar commands
// continue to default to the user's container
pod.Spec.Containers = modifyContainers(pod.Spec.Containers, ProxyContainerName, MoveLast)
+ pod.Spec.Containers = modifyContainers(pod.Spec.Containers, ProxylessXServerContainerName, MoveLast)
return nil
}
@@ -624,6 +689,38 @@ func createPatch(pod *corev1.Pod, original []byte) ([]byte, error) {
return json.Marshal(p)
}
+func createServicePatch(svc *corev1.Service, original []byte) ([]byte, error) {
+ reinjected, err := json.Marshal(svc)
+ if err != nil {
+ return nil, err
+ }
+ p, err := jsonpatch.CreatePatch(original, reinjected)
+ if err != nil {
+ return nil, err
+ }
+ return json.Marshal(p)
+}
+
+func rewriteProxylessServiceTargetPorts(svc *corev1.Service) bool {
+ if svc.Spec.Type == corev1.ServiceTypeExternalName || len(svc.Spec.Selector) == 0 {
+ return false
+ }
+
+ changed := false
+ for i := range svc.Spec.Ports {
+ port := &svc.Spec.Ports[i]
+ if port.Protocol != "" && port.Protocol != corev1.ProtocolTCP {
+ continue
+ }
+ if port.TargetPort.Type == intstr.Int && port.TargetPort.IntVal == ProxylessXServerPort {
+ continue
+ }
+ port.TargetPort = intstr.FromInt(ProxylessXServerPort)
+ changed = true
+ }
+ return changed
+}
+
func applyOverlayYAML(target *corev1.Pod, overlayYAML []byte) (*corev1.Pod, error) {
currentJSON, err := json.Marshal(target)
if err != nil {
diff --git a/samples/addons/00-namespace.yaml b/samples/addons/00-namespace.yaml
deleted file mode 100644
index 42e674c6..00000000
--- a/samples/addons/00-namespace.yaml
+++ /dev/null
@@ -1,19 +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.
-
-apiVersion: v1
-kind: Namespace
-metadata:
- name: dubbo-system
diff --git a/samples/addons/README.md b/samples/addons/README.md
deleted file mode 100644
index 0fed25f2..00000000
--- a/samples/addons/README.md
+++ /dev/null
@@ -1,38 +0,0 @@
-# Dubbo 可观测性 Addons
-
-这个目录提供开箱即用的 Prometheus 和 Grafana 部署，用于观察 `dubbo-system/dubbod` 的日志信号、监控指标和控制面追踪耗时。
-
-## 安装
-
-```bash
-kubectl apply -f samples/addons
-kubectl -n dubbo-system rollout status deploy/prometheus --timeout=180s
-kubectl -n dubbo-system rollout status deploy/grafana --timeout=180s
-```
-
-## 打开 Grafana
-
-```bash
-kubectl -n dubbo-system port-forward svc/grafana 3000:3000
-```
-
-浏览器打开 `http://127.0.0.1:3000`，进入 `Dubbo / Dubbo control plane Observability` 仪表盘。
-
-## 验证 Prometheus
-
-```bash
-kubectl -n dubbo-system port-forward svc/prometheus 9090:9090
-curl 'http://127.0.0.1:9090/api/v1/query?query=dubbod_uptime_seconds'
-```
-
-## 覆盖内容
-
-- 日志：`dubbod_log_messages_total` 按 `level` 和 `scope` 统计日志速率，原始日志仍通过 `kubectl -n dubbo-system logs deploy/dubbod` 查看。
-- 监控：Prometheus 抓取 `dubbod.dubbo-system.svc:8080/metrics`，Grafana 内置控制面总览面板。
-- 追踪：通过 `dubbod_xds_push_time`、`dubbod_xds_send_time`、`dubbod_proxy_queue_time` 等直方图展示 xDS 推送链路耗时。
-
-## 清理
-
-```bash
-kubectl delete -f samples/addons --ignore-not-found=true
-```
diff --git a/samples/app/README.md b/samples/app/README.md
index 13d3dbc5..8ee4663d 100644
--- a/samples/app/README.md
+++ b/samples/app/README.md
@@ -15,7 +15,7 @@ kubectl -n app rollout status deploy/nginx-v2 --timeout=180s
kubectl -n app rollout status deploy/nginx-consumer --timeout=180s
```
-`dubbo-injection=enabled` 开启后，会自动注入 `grpc-engine`。`nginx-consumer` 使用 `kdubbo/dubbod` 镜像，会自动启动 `xclient --watch`。
+`dubbo-injection=enabled` 开启后，会自动注入 `grpc-engine`。`xclient` 只主动发起请求，不监听端口；服务端 Pod 会注入 `xserver` 接收 mTLS 并转发到本地 nginx，Service 会被准入层改到 `targetPort: 25080`，样例 NetworkPolicy 会禁止从其他 Pod 直连 nginx `80`。
## 配置流量规则
@@ -46,10 +46,24 @@ nginx.app.svc.cluster.local:80 v1=23 endpoints=1,v2=77 endpoints=1
## 验证流量结果
```bash
-kubectl -n app exec deploy/nginx-consumer -- dubbod xclient 100 | sort | uniq -c
+kubectl -n app exec deploy/nginx-consumer -- \
+ dubbod xclient --print-route --expect v1=50,v2=50
+kubectl -n app exec deploy/nginx-consumer -- \
+ dubbod xclient --expect v1=50,v2=50 100 | sort | uniq -c
```
-`xclient` 现在是逐请求选路，不再先算好固定数量再批量发送。
+`--print-route` 里应看到 `tlsMode: DUBBO_MUTUAL`，endpoint 端口应为 `25080`。
+
+确认 Service 也指向 `xserver`：
+
+```bash
+kubectl -n app get svc nginx -o jsonpath='{.spec.ports[0].targetPort}{"\n"}'
+kubectl -n app run plain-curl --rm -i --restart=Never \
+ --image=curlimages/curl:8.5.0 -- \
+ curl -sv --max-time 5 http://nginx.app.svc.cluster.local/
+```
+
+第一条应输出 `25080`，第二条不能返回 `nginx v1` 或 `nginx v2`。
## 在线更新验证
diff --git a/samples/app/deployment.yaml b/samples/app/deployment.yaml
index 778f8e0e..7ee87728 100644
--- a/samples/app/deployment.yaml
+++ b/samples/app/deployment.yaml
@@ -37,6 +37,22 @@ spec:
targetPort: 80
protocol: TCP
---
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+ name: nginx-proxyless-only
+ namespace: app
+spec:
+ podSelector:
+ matchLabels:
+ app: nginx
+ policyTypes:
+ - Ingress
+ ingress:
+ - ports:
+ - protocol: TCP
+ port: 25080
+---
apiVersion: v1
kind: ConfigMap
metadata:
diff --git a/samples/app/meshservice.yaml b/samples/app/meshservice.yaml
deleted file mode 100644
index 2a24ae88..00000000
--- a/samples/app/meshservice.yaml
+++ /dev/null
@@ -1,41 +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.
-
-apiVersion: networking.dubbo.apache.org/v1alpha3
-kind: MeshService
-metadata:
- name: nginx-routing
- namespace: app
-spec:
- hosts:
- - nginx.app.svc.cluster.local
- visibleTo:
- - "."
- routes:
- - service:
- - name: v1
- host: nginx.app.svc.cluster.local
- labels:
- version: v1
- port:
- number: 80
- weight: 20
- - name: v2
- host: nginx.app.svc.cluster.local
- labels:
- version: v2
- port:
- number: 80
- weight: 80