You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by gi...@apache.org on 2026/09/11 03:04:09 UTC
(dubbo-kubernetes) branch master updated: Architecture replace 0.4.7 pre release (#1026)
This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] 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 93992e9d Architecture replace 0.4.7 pre release (#1026)
93992e9d is described below
commit 93992e9d8440cd8edd3eab71d14521abcba5928b
Author: mfordjody <11...@qq.com>
AuthorDate: Fri Sep 11 11:04:03 2026 +0800
Architecture replace 0.4.7 pre release (#1026)
* Upgrade the core architecture
* 0.4.7 pre-release revision 1
---
.asf.yaml | 2 +-
README.md | 11 +-
architecture/inherent/application.md | 6 -
dubboctl/cmd/multicluster.go | 4 +
dubboctl/cmd/multicluster_test.go | 4 +-
dubbod/discovery/cmd/app/grpc_outbound.go | 13 +-
dubbod/discovery/cmd/app/grpc_outbound_test.go | 9 +
dubbod/discovery/pkg/bootstrap/discovery.go | 25 +-
.../pkg/bootstrap/inherent_grpc_controller.go | 115 ++-
.../pkg/bootstrap/inherent_grpc_controller_test.go | 113 ++-
dubbod/discovery/pkg/bootstrap/server.go | 9 +-
dubbod/discovery/pkg/bootstrap/xds_auth.go | 110 +++
dubbod/discovery/pkg/bootstrap/xds_auth_test.go | 77 ++
dubbod/discovery/pkg/bootstrap/xds_secrets.go | 136 +++
dubbod/discovery/pkg/bootstrap/xds_secrets_test.go | 150 ++++
.../discovery/pkg/config/kube/gateway/context.go | 4 +-
.../pkg/config/kube/gateway/controller.go | 2 +-
.../config/kube/gateway/deployment_controller.go | 7 +-
.../kube/gateway/deployment_controller_test.go | 23 +-
dubbod/discovery/pkg/features/dubbo.go | 7 +-
dubbod/discovery/pkg/features/tuning.go | 13 -
.../pkg/leaderelection/leaderelection_test.go | 116 +++
.../model/{push_context.go => config_snapshot.go} | 412 ++-------
dubbod/discovery/pkg/model/config_state.go | 65 ++
dubbod/discovery/pkg/model/context.go | 41 +-
dubbod/discovery/pkg/model/push_context.go | 935 +--------------------
dubbod/discovery/pkg/model/push_context_test.go | 4 +-
dubbod/discovery/pkg/networking/apigen/apigen.go | 36 -
dubbod/discovery/pkg/networking/core/configgen.go | 39 -
.../kube/controller/multicluster.go | 18 +-
.../kube/controller/namespace_controller.go | 3 +-
dubbod/discovery/pkg/util/network/ip.go | 30 -
dubbod/discovery/pkg/xds/ads.go | 59 +-
dubbod/discovery/pkg/xds/ads_benchmark_test.go | 3 +-
dubbod/discovery/pkg/xds/ads_test.go | 12 +-
dubbod/discovery/pkg/xds/auth.go | 59 +-
dubbod/discovery/pkg/xds/auth_test.go | 160 ++++
dubbod/discovery/pkg/xds/cds.go | 60 --
dubbod/discovery/pkg/xds/delta.go | 7 +-
dubbod/discovery/pkg/xds/delta_test.go | 3 +-
dubbod/discovery/pkg/xds/discovery.go | 41 +-
dubbod/discovery/pkg/xds/lds.go | 47 --
dubbod/discovery/pkg/xds/rds.go | 65 --
dubbod/discovery/pkg/xds/statusgen.go | 45 -
dubbod/discovery/pkg/xds/stream_test.go | 40 +-
dubbod/discovery/pkg/xds/tls_stream_test.go | 155 ++++
dubbod/discovery/pkg/xds/v1/model.go | 11 +-
dubbod/discovery/pkg/xds/xdsgen.go | 103 +--
.../nodeagent/caclient/providers/aegis/client.go | 2 +-
.../server/ca/authenticate/cert_authenticator.go | 12 +-
dubbod/security/pkg/server/ca/server.go | 4 +-
pkg/features/transport.go | 40 +
{dubbod/discovery/pkg => pkg}/grpc/grpc.go | 2 +-
{dubbod/discovery/pkg => pkg}/grpc/tls.go | 0
pkg/introspection/options_test.go | 20 +-
pkg/kube/inject/inherent.go | 14 +-
pkg/kube/inject/inherent_test.go | 30 +-
pkg/kube/inject/webhook.go | 14 +-
pkg/kube/kclient/client.go | 2 +-
pkg/kube/util.go | 2 +-
pkg/model/proxy.go | 2 +-
pkg/model/xds.go | 7 -
pkg/spiffe/spiffe.go | 1 +
pkg/util/net/ip.go | 30 +
pkg/xds/server.go | 6 +-
samples/{ai-mesh => ai}/README.md | 0
samples/{ai-mesh => ai}/backends.yaml | 0
samples/{ai-mesh => ai}/gateway.yaml | 0
samples/{ai-mesh => ai}/routes.yaml | 0
samples/{ai-mesh => ai}/services.yaml | 0
tests/e2e/xdscheck/main.go | 158 ++++
71 files changed, 1812 insertions(+), 1943 deletions(-)
diff --git a/.asf.yaml b/.asf.yaml
index 0896c407..6e72f3a4 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -19,7 +19,7 @@ notifications:
pullrequests: notifications@dubbo.apache.org
github:
- homepage: "https://dsmlx.github.io"
+ homepage: "https://dubml.github.io"
description: "Build a native mesh using gRPC and xDS"
pull_requests:
allow_auto_merge: true
diff --git a/README.md b/README.md
index f8592be4..c84ddd13 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,4 @@
-# Apache Kdubbo - dubbo-mesh
+# Apache Kdubbo - Dubbo Mesh
[](https://pkg.go.dev/github.com/apache/dubbo-kubernetes)
[](https://goreportcard.com/report/github.com/apache/dubbo-kubernetes)
@@ -14,13 +14,14 @@ Dubbo service mesh enables workloads to join natively, receive policies from the
> Releases `0.4.6–0.4.9` will be in the **Beta** phase.
>
> Release `0.5.0` will be the first **RC** version.
+>
+> Release `0.6.0` will be the first **GA** version.
Dubbo’s control plane provides an abstraction layer over the underlying cluster management platform.
Dubbo component composition:
- **dubbod** — Dubbo xDS control plane. It provides service discovery, configuration and certificate issuance.
-- **Application runtime** — The original workload consumes the injected xDS bootstrap and policies natively; no proxy container is added.
- **dxgate** — Dubbo delegated gateway for Gateway API. It consumes routing configuration from dubbod over xDS and proxies north-south traffic into the mesh.
## Repositories
@@ -33,11 +34,11 @@ Projects are distributed across the code directory repositories:
- [client-go](https://github.com/kdubbo/client-go). — Defines the Kubernetes clients automatically generated for Dubbo control plane resources.
-- [dubboctl](./cli). — Provides dubboctl command line tools for control plane management and other operations.
+- [dubboctl](https://github.com/kdubbo/dubboctl). — Provides dubboctl command line tools for control plane management and other operations.
-- [dubbod](./dubbod) — The main code directory for the Dubbo control plane.
+- [dubbod](https://github.com/kdubbo/dubbod) — The main code directory for the Dubbo control plane.
-- [operator](./operator). — Provides user friendly options for operating the service mesh.
+- [operator](https://github.com/kdubbo/operator). — Provides user friendly options for operating the service mesh.
- [dxgate](https://github.com/kdubbo/dxgate) — Provides the delegated gateway that serves Gateway API traffic at the mesh edge.
diff --git a/architecture/inherent/application.md b/architecture/inherent/application.md
deleted file mode 100644
index 911c2470..00000000
--- a/architecture/inherent/application.md
+++ /dev/null
@@ -1,6 +0,0 @@
-# Inherent Application
-
-Inherent is proxyless. The injector adds the xDS bootstrap, runtime configuration,
-certificate volume and telemetry settings to the original application container.
-The application consumes them natively; no proxy image, sidecar container, Service
-port rewrite or node-level traffic interception is added.
diff --git a/dubboctl/cmd/multicluster.go b/dubboctl/cmd/multicluster.go
index 60c372b2..48c9e327 100644
--- a/dubboctl/cmd/multicluster.go
+++ b/dubboctl/cmd/multicluster.go
@@ -234,6 +234,10 @@ func buildEastWestGatewayManifest(args eastWestGatewayArgs) (*gatewayv1.Gateway,
if strings.TrimSpace(args.xdsAddress) == "" {
return nil, fmt.Errorf("xds-address is required")
}
+ endpoint, err := url.Parse(args.xdsAddress)
+ if err != nil || endpoint.Scheme != "https" || endpoint.Hostname() == "" {
+ return nil, fmt.Errorf("xds-address must be an https URL for authenticated ADS")
+ }
annotations := map[string]string{
"gateway.dubbo.apache.org/eastwest": "true",
"gateway.dubbo.apache.org/service-type": args.serviceType,
diff --git a/dubboctl/cmd/multicluster_test.go b/dubboctl/cmd/multicluster_test.go
index 560f653f..4e4ba624 100644
--- a/dubboctl/cmd/multicluster_test.go
+++ b/dubboctl/cmd/multicluster_test.go
@@ -86,7 +86,7 @@ func TestBuildEastWestGatewayManifest(t *testing.T) {
port: 15443,
targetPort: 15080,
nodePort: 32443,
- xdsAddress: "http://192.168.15.164:32010",
+ xdsAddress: "https://dubbod.remote.example:32012",
})
if err != nil {
t.Fatalf("buildEastWestGatewayManifest() error = %v", err)
@@ -97,7 +97,7 @@ func TestBuildEastWestGatewayManifest(t *testing.T) {
if gateway.Annotations["gateway.dubbo.apache.org/service-type"] != "NodePort" {
t.Fatalf("service type annotation = %q", gateway.Annotations["gateway.dubbo.apache.org/service-type"])
}
- if gateway.Annotations["gateway.dubbo.apache.org/xds-address"] != "http://192.168.15.164:32010" {
+ if gateway.Annotations["gateway.dubbo.apache.org/xds-address"] != "https://dubbod.remote.example:32012" {
t.Fatalf("xds address annotation = %q", gateway.Annotations["gateway.dubbo.apache.org/xds-address"])
}
if len(gateway.Spec.Listeners) != 1 || gateway.Spec.Listeners[0].Port != 15443 {
diff --git a/dubbod/discovery/cmd/app/grpc_outbound.go b/dubbod/discovery/cmd/app/grpc_outbound.go
index 2810b367..7a7e6944 100644
--- a/dubbod/discovery/cmd/app/grpc_outbound.go
+++ b/dubbod/discovery/cmd/app/grpc_outbound.go
@@ -17,7 +17,6 @@ package app
import (
"context"
- "crypto/tls"
"encoding/json"
"fmt"
"io"
@@ -48,7 +47,6 @@ import (
discovery "github.com/kdubbo/xds-api/service/discovery/v1"
"github.com/spf13/cobra"
"google.golang.org/grpc"
- "google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/anypb"
@@ -309,20 +307,19 @@ func adsDialConfig(opts *grpcOutboundOptions) (*corev1.Node, string, []grpc.Dial
creds = insecure.NewCredentials()
}
return bootstrap.Node, xdsresolver.DialAddress(bootstrap.ServerURI), []grpc.DialOption{grpc.WithTransportCredentials(creds)}, nil
- } else if !os.IsNotExist(err) {
+ } else {
return nil, "", nil, err
}
}
+ if !opts.insecure {
+ return nil, "", nil, fmt.Errorf("authenticated ADS requires --bootstrap or GRPC_XDS_BOOTSTRAP with workload credentials")
+ }
node, err := buildADSNode(opts)
if err != nil {
return nil, "", nil, err
}
- creds := grpc.WithTransportCredentials(insecure.NewCredentials())
- if !opts.insecure && strings.HasSuffix(opts.xdsAddress, ":26012") {
- creds = grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12}))
- }
- return node, opts.xdsAddress, []grpc.DialOption{creds}, nil
+ return node, opts.xdsAddress, []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}, nil
}
func buildADSNode(opts *grpcOutboundOptions) (*corev1.Node, error) {
diff --git a/dubbod/discovery/cmd/app/grpc_outbound_test.go b/dubbod/discovery/cmd/app/grpc_outbound_test.go
index a871cdf9..6abbd928 100644
--- a/dubbod/discovery/cmd/app/grpc_outbound_test.go
+++ b/dubbod/discovery/cmd/app/grpc_outbound_test.go
@@ -56,6 +56,15 @@ func TestAutoDiscoverServiceTargetSingleService(t *testing.T) {
}
}
+func TestADSDialRequiresWorkloadBootstrap(t *testing.T) {
+ if _, _, _, err := adsDialConfig(&grpcOutboundOptions{}); err == nil || !strings.Contains(err.Error(), "authenticated ADS requires") {
+ t.Fatalf("expected missing credentials error, got %v", err)
+ }
+ if _, _, _, err := adsDialConfig(&grpcOutboundOptions{bootstrapPath: t.TempDir() + "/bootstrap.json", insecure: true}); err == nil {
+ t.Fatalf("missing explicit bootstrap must not silently downgrade: %v", err)
+ }
+}
+
func TestAutoDiscoverServiceTargetMultipleServices(t *testing.T) {
_, _, err := autoDiscoverServiceTarget([]string{
"NGINX_SERVICE_HOST=10.96.12.34",
diff --git a/dubbod/discovery/pkg/bootstrap/discovery.go b/dubbod/discovery/pkg/bootstrap/discovery.go
index 2e52cc48..1b797a32 100644
--- a/dubbod/discovery/pkg/bootstrap/discovery.go
+++ b/dubbod/discovery/pkg/bootstrap/discovery.go
@@ -18,34 +18,19 @@ package bootstrap
import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/apigen"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/core"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/grpcgen"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds"
v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
)
-func InitGenerators(
- s *xds.DiscoveryServer,
- cg core.ConfigGenerator,
-) {
+func InitGenerators(s *xds.DiscoveryServer) {
env := s.Env
generators := map[string]model.XdsResourceGenerator{}
edsGen := &xds.EdsGenerator{Cache: s.Cache, EndpointIndex: env.EndpointIndex}
- generators[v1.ClusterType] = &xds.CdsGenerator{ConfigGenerator: cg}
- generators[v1.ListenerType] = &xds.LdsGenerator{ConfigGenerator: cg}
- generators[v1.RouteType] = &xds.RdsGenerator{ConfigGenerator: cg}
- generators[v1.EndpointType] = edsGen
-
- generators["grpc"] = &grpcgen.GrpcConfigGenerator{}
+ grpcGenerator := &grpcgen.GrpcConfigGenerator{}
generators["grpc/"+v1.EndpointType] = edsGen
- generators["grpc/"+v1.ListenerType] = generators["grpc"]
- generators["grpc/"+v1.RouteType] = generators["grpc"]
- generators["grpc/"+v1.ClusterType] = generators["grpc"]
-
- generators["api"] = apigen.NewGenerator(env.ConfigStore)
- generators["api/"+v1.EndpointType] = edsGen
-
- generators["event"] = xds.NewStatusGen(s)
+ generators["grpc/"+v1.ListenerType] = grpcGenerator
+ generators["grpc/"+v1.RouteType] = grpcGenerator
+ generators["grpc/"+v1.ClusterType] = grpcGenerator
s.Generators = generators
}
diff --git a/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller.go b/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller.go
index 231c3393..29fa045b 100644
--- a/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller.go
+++ b/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller.go
@@ -72,9 +72,10 @@ const (
var inherentGRPCLog = log.RegisterScope("inherentgrpc", "Inherent gRPC workload controller")
type inherentGRPCWorkloadController struct {
- server *Server
- client kubelib.Client
- pods kclient.Client[*corev1.Pod]
+ server *Server
+ client kubelib.Client
+ pods kclient.Client[*corev1.Pod]
+ secrets kclient.Client[*corev1.Secret]
// managedPods keeps bundle/cert updates from scanning every Pod in the cluster.
managedPods kclient.RawIndexer
queue controllers.Queue
@@ -125,22 +126,17 @@ func (s *Server) initInherentGRPCWorkloads() error {
return wrapped
})
}
- if s.XDSServer != nil {
- previous := s.XDSServer.RuntimeConfigUpdate
- s.XDSServer.RuntimeConfigUpdate = func(req *discoverymodel.PushRequest) {
- if previous != nil {
- previous(req)
- }
- controller.handleRuntimeConfigUpdate(req)
- if s.inherentGRPCRemoteControllers != nil {
- for _, remote := range s.inherentGRPCRemoteControllers.All() {
- if remote.controller != nil {
- remote.controller.handleRuntimeConfigUpdate(req)
- }
+ s.environment.AddConfigHandler(func(change *discoverymodel.ConfigChange) {
+ controller.handleRuntimeConfigUpdate(change)
+ if s.inherentGRPCRemoteControllers != nil {
+ for _, remote := range s.inherentGRPCRemoteControllers.All() {
+ if remote.controller != nil {
+ remote.controller.handleRuntimeConfigUpdate(change)
}
}
}
- }
+ })
+
s.addStartFunc(inherentGRPCControllerName, func(stop <-chan struct{}) error {
go controller.Run(stop)
return nil
@@ -162,6 +158,9 @@ func newInherentGRPCWorkloadController(s *Server, client kubelib.Client) *inhere
pods: kclient.NewFiltered[*corev1.Pod](client, kclient.Filter{
ObjectFilter: client.ObjectFilter(),
}),
+ secrets: kclient.NewFiltered[*corev1.Secret](client, kclient.Filter{
+ ObjectFilter: client.ObjectFilter(),
+ }),
rotations: make(map[types.NamespacedName]time.Time),
}
c.managedPods = c.pods.Index(inherentGRPCManagedPodIndex, func(pod *corev1.Pod) []string {
@@ -181,13 +180,32 @@ func newInherentGRPCWorkloadController(s *Server, client kubelib.Client) *inhere
pod := controllers.Extract[*corev1.Pod](o)
return shouldManageInherentGRPCPod(pod)
}))
+ c.secrets.AddEventHandler(controllers.ObjectHandler(func(o controllers.Object) {
+ secret := controllers.Extract[*corev1.Secret](o)
+ if secret == nil {
+ return
+ }
+ for _, owner := range secret.OwnerReferences {
+ if owner.Kind != "Pod" || owner.APIVersion != "v1" {
+ continue
+ }
+ pod := c.pods.Get(owner.Name, secret.Namespace)
+ if pod != nil && pod.UID == owner.UID && secret.Name == inject.InherentGRPCSecretNameForMeta(pod.ObjectMeta) {
+ c.queue.AddObject(pod)
+ if s.XDSServer != nil {
+ s.XDSServer.ProxyUpdate(client.ClusterID(), pod.Status.PodIP)
+ }
+ }
+ }
+ }))
return c
}
func (c *inherentGRPCWorkloadController) Run(stop <-chan struct{}) {
c.pods.Start(stop)
- if !kubelib.WaitForCacheSync(inherentGRPCControllerName, stop, c.pods.HasSynced) {
+ c.secrets.Start(stop)
+ if !kubelib.WaitForCacheSync(inherentGRPCControllerName, stop, c.pods.HasSynced, c.secrets.HasSynced) {
c.queue.ShutDownEarly()
return
}
@@ -202,7 +220,7 @@ func (c *inherentGRPCWorkloadController) Run(stop <-chan struct{}) {
}
func (c *inherentGRPCWorkloadController) HasSynced() bool {
- return c.pods.HasSynced() && c.queue.HasSynced()
+ return c.pods.HasSynced() && c.secrets.HasSynced() && c.queue.HasSynced()
}
func (c *inherentGRPCWorkloadController) watchBundleChanges(stop <-chan struct{}) {
@@ -225,14 +243,14 @@ func (c *inherentGRPCWorkloadController) enqueueAllPods() {
}
}
-func (c *inherentGRPCWorkloadController) handleRuntimeConfigUpdate(req *discoverymodel.PushRequest) {
+func (c *inherentGRPCWorkloadController) handleRuntimeConfigUpdate(req *discoverymodel.ConfigChange) {
if !inherentGRPCRuntimeConfigNeedsUpdate(req) {
return
}
c.enqueueAllPods()
}
-func inherentGRPCRuntimeConfigNeedsUpdate(req *discoverymodel.PushRequest) bool {
+func inherentGRPCRuntimeConfigNeedsUpdate(req *discoverymodel.ConfigChange) bool {
if req == nil {
return false
}
@@ -245,7 +263,7 @@ func inherentGRPCRuntimeConfigNeedsUpdate(req *discoverymodel.PushRequest) bool
return true
}
}
- if req.Reason.Has(discoverymodel.EndpointUpdate) || req.Reason.Has(discoverymodel.HeadlessEndpointUpdate) || req.Reason.Has(discoverymodel.GlobalUpdate) {
+ if req.EndpointsChanged || req.Global {
return true
}
return false
@@ -306,7 +324,9 @@ func (c *inherentGRPCWorkloadController) reconcile(key types.NamespacedName) err
pod := c.pods.Get(key.Name, key.Namespace)
if pod == nil {
c.clearRotation(key)
- return c.deleteSecret(key)
+ // Kubernetes garbage collection uses the Pod UID; deleting by name here
+ // could remove credentials of a replacement Pod before our cache catches up.
+ return nil
}
if !shouldManageInherentGRPCPod(pod) || pod.DeletionTimestamp != nil {
c.clearRotation(key)
@@ -314,6 +334,11 @@ func (c *inherentGRPCWorkloadController) reconcile(key types.NamespacedName) err
}
secrets := c.client.Kube().CoreV1().Secrets(pod.Namespace)
secretName := inject.InherentGRPCSecretNameForMeta(pod.ObjectMeta)
+ for _, volume := range pod.Spec.Volumes {
+ if volume.Name == inject.InherentXDSVolumeName && volume.Secret != nil && volume.Secret.SecretName != secretName {
+ return fmt.Errorf("pod %s/%s mounts legacy shared credentials; recreate the Pod to obtain its own Secret", pod.Namespace, pod.Name)
+ }
+ }
current, err := secrets.Get(context.Background(), secretName, metav1.GetOptions{})
if err != nil {
if !apierrors.IsNotFound(err) {
@@ -321,6 +346,9 @@ func (c *inherentGRPCWorkloadController) reconcile(key types.NamespacedName) err
}
current = nil
}
+ if current != nil && !workloadSecretOwnedByPod(current, pod) {
+ return fmt.Errorf("refusing to replace credentials Secret %s/%s owned by another workload", pod.Namespace, secretName)
+ }
desired, expireAt, err := c.buildSecret(pod, current)
if err != nil {
@@ -343,6 +371,18 @@ func (c *inherentGRPCWorkloadController) reconcile(key types.NamespacedName) err
return nil
}
+func workloadSecretOwnedByPod(secret *corev1.Secret, pod *corev1.Pod) bool {
+ if secret.Namespace != pod.Namespace || pod.UID == "" {
+ return false
+ }
+ for _, owner := range secret.OwnerReferences {
+ if owner.APIVersion == "v1" && owner.Kind == "Pod" && owner.Name == pod.Name && owner.UID == pod.UID {
+ return true
+ }
+ }
+ return false
+}
+
const inherentGRPCRuntimeConfigVersion = "dubbo.apache.org/inherent-grpc/v1"
type inherentGRPCWorkloadContext struct {
@@ -833,7 +873,7 @@ func (c *inherentGRPCWorkloadController) buildRuntimeTrafficConfig() ([]inherent
if c.server == nil || c.server.environment == nil {
return nil, nil
}
- push := c.server.environment.PushContext()
+ push := c.server.environment.ConfigSnapshot()
if push == nil || !push.InitDone.Load() {
return nil, nil
}
@@ -860,7 +900,7 @@ func (c *inherentGRPCWorkloadController) buildRuntimeTrafficConfig() ([]inherent
return serviceConfigs, routeConfigs
}
-func buildRuntimeServiceConfig(push *discoverymodel.PushContext, endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service) inherentGRPCServiceRuntimeConfig {
+func buildRuntimeServiceConfig(push *discoverymodel.ConfigSnapshot, endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service) inherentGRPCServiceRuntimeConfig {
cfg := inherentGRPCServiceRuntimeConfig{
Host: string(svc.Hostname),
Namespace: svc.Attributes.Namespace,
@@ -884,7 +924,7 @@ func buildRuntimeServiceConfig(push *discoverymodel.PushContext, endpointIndex *
}
func runtimeWorkloadAuthorizationPolicies(
- push *discoverymodel.PushContext,
+ push *discoverymodel.ConfigSnapshot,
svc *discoverymodel.Service,
) []inherentGRPCAuthorizationPolicyRuntimeConfig {
if push == nil || svc == nil {
@@ -947,7 +987,7 @@ func runtimeWorkloadAuthorizationRule(
return projected, true
}
-func runtimeFaultInjection(push *discoverymodel.PushContext, namespace, name, portName string) *inherentGRPCFaultRuntimeConfig {
+func runtimeFaultInjection(push *discoverymodel.ConfigSnapshot, namespace, name, portName string) *inherentGRPCFaultRuntimeConfig {
settings, found := push.FaultInjectionForService(namespace, name, portName)
if !found {
return nil
@@ -971,7 +1011,7 @@ func runtimeFaultInjection(push *discoverymodel.PushContext, namespace, name, po
return fault
}
-func buildRuntimeRouteConfig(_ *discoverymodel.PushContext, endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service, port int) inherentGRPCRouteRuntimeConfig {
+func buildRuntimeRouteConfig(_ *discoverymodel.ConfigSnapshot, endpointIndex *discoverymodel.EndpointIndex, svc *discoverymodel.Service, port int) inherentGRPCRouteRuntimeConfig {
return inherentGRPCRouteRuntimeConfig{
Host: string(svc.Hostname),
Port: port,
@@ -983,7 +1023,7 @@ func buildRuntimeRouteConfig(_ *discoverymodel.PushContext, endpointIndex *disco
}
}
-func runtimeInboundMTLSMode(push *discoverymodel.PushContext, namespace string, port int) string {
+func runtimeInboundMTLSMode(push *discoverymodel.ConfigSnapshot, namespace string, port int) string {
if push == nil || push.AuthenticationPolicies == nil || port <= 0 {
return ""
}
@@ -1209,15 +1249,18 @@ func shouldManageInherentGRPCPod(pod *corev1.Pod) bool {
return false
}
-func (c *inherentGRPCWorkloadController) deleteSecret(key types.NamespacedName) error {
- err := c.client.Kube().CoreV1().Secrets(key.Namespace).Delete(context.Background(),
- inject.InherentGRPCSecretName(key.Name), metav1.DeleteOptions{})
- return controllers.IgnoreNotFound(err)
-}
-
func (c *inherentGRPCWorkloadController) deleteSecretForPod(pod *corev1.Pod) error {
- err := c.client.Kube().CoreV1().Secrets(pod.Namespace).Delete(context.Background(),
- inject.InherentGRPCSecretNameForMeta(pod.ObjectMeta), metav1.DeleteOptions{})
+ secrets := c.client.Kube().CoreV1().Secrets(pod.Namespace)
+ secret, err := secrets.Get(context.Background(), inject.InherentGRPCSecretNameForMeta(pod.ObjectMeta), metav1.GetOptions{})
+ if err != nil {
+ return controllers.IgnoreNotFound(err)
+ }
+ if !workloadSecretOwnedByPod(secret, pod) {
+ return nil
+ }
+ err = secrets.Delete(context.Background(), secret.Name, metav1.DeleteOptions{
+ Preconditions: &metav1.Preconditions{UID: &secret.UID, ResourceVersion: &secret.ResourceVersion},
+ })
return controllers.IgnoreNotFound(err)
}
diff --git a/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go b/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
index 073c3c28..d0347ea3 100644
--- a/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
+++ b/dubbod/discovery/pkg/bootstrap/inherent_grpc_controller_test.go
@@ -16,7 +16,9 @@
package bootstrap
import (
+ "context"
"encoding/json"
+ "strings"
"testing"
"time"
@@ -33,8 +35,10 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
telemetryconfig "github.com/apache/dubbo-kubernetes/pkg/config/telemetry"
"github.com/apache/dubbo-kubernetes/pkg/grpcxds"
+ "github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/inject"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/kclient"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
networking "github.com/kdubbo/api/networking/v1alpha3"
@@ -45,8 +49,47 @@ import (
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/kubernetes/fake"
)
+func TestWorkloadControllerCannotOverwriteOrDeleteForeignCredentials(t *testing.T) {
+ pod, secret := testWorkloadSecret(t, "spiffe://cluster.local/ns/app/sa/default")
+ pod.Annotations = map[string]string{inject.InherentInjectTemplatesAnnoName: inject.InherentGRPCTemplateName}
+ secret.OwnerReferences[0].UID = "another-pod-uid"
+ client := credentialsTestClient{core: fake.NewClientset(pod, secret)}
+ c := &inherentGRPCWorkloadController{client: client, pods: credentialsTestPods{pod: pod}}
+ if err := c.reconcile(types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}); err == nil || !strings.Contains(err.Error(), "owned by another workload") {
+ t.Fatalf("expected ownership conflict, got %v", err)
+ }
+ if err := c.deleteSecretForPod(pod); err != nil {
+ t.Fatal(err)
+ }
+ got, err := client.Kube().CoreV1().Secrets(secret.Namespace).Get(context.Background(), secret.Name, metav1.GetOptions{})
+ if err != nil || got.OwnerReferences[0].UID != "another-pod-uid" {
+ t.Fatalf("foreign credentials changed or deleted: error=%v", err)
+ }
+}
+
+type credentialsTestClient struct {
+ kube.Client
+ core kubernetes.Interface
+}
+
+func (c credentialsTestClient) Kube() kubernetes.Interface { return c.core }
+
+type credentialsTestPods struct {
+ kclient.Client[*corev1.Pod]
+ pod *corev1.Pod
+}
+
+func (p credentialsTestPods) Get(name, namespace string) *corev1.Pod {
+ if p.pod.Name == name && p.pod.Namespace == namespace {
+ return p.pod
+ }
+ return nil
+}
+
func TestShouldManageInherentGRPCPod(t *testing.T) {
tests := []struct {
name string
@@ -380,7 +423,7 @@ func TestBuildWorkloadContextUsesInjectedRemoteValues(t *testing.T) {
func TestBuildRuntimeTrafficConfigCapturesInherentSecurity(t *testing.T) {
hostname := host.Name("provider.grpc-app.svc.cluster.local")
svc := newInherentRuntimeTestService("provider", "grpc-app", string(hostname), 17070)
- push := newInherentRuntimeTestPushContext(t, []config.Config{
+ push := newInherentRuntimeTestSnapshot(t, []config.Config{
newInherentStrictPeerAuthenticationConfig("grpc-app-strict-mtls", "grpc-app"),
}, []*discoverymodel.Service{svc})
@@ -402,6 +445,40 @@ func TestBuildRuntimeTrafficConfigCapturesInherentSecurity(t *testing.T) {
}
}
+func TestInherentReconcilesAndRevokesPolicyWithoutXDSServer(t *testing.T) {
+ svc := newInherentRuntimeTestService("provider", "grpc-app", "provider.grpc-app.svc.cluster.local", 17070)
+ policy := newInherentStrictPeerAuthenticationConfig("strict", "grpc-app")
+ env, previous := newInherentRuntimeTestEnvironment(t, []config.Config{policy}, []*discoverymodel.Service{svc})
+ controller := &inherentGRPCWorkloadController{server: &Server{environment: env}}
+ if env.PushContext().InitDone.Load() {
+ t.Fatal("test initialized an xDS push context")
+ }
+ services, _ := controller.buildRuntimeTrafficConfig()
+ if len(services) != 1 || services[0].Ports[0].MTLSMode != "STRICT" {
+ t.Fatalf("initial application policy = %+v", services)
+ }
+ if err := env.Delete(gvk.PeerAuthentication, policy.Name, policy.Namespace, nil); err != nil {
+ t.Fatal(err)
+ }
+ notified := false
+ env.AddConfigHandler(func(change *discoverymodel.ConfigChange) {
+ notified = true
+ services, _ := controller.buildRuntimeTrafficConfig()
+ if len(services) != 1 || services[0].Ports[0].MTLSMode == "STRICT" {
+ t.Fatalf("consumer saw stale policy after publication: %+v", services)
+ }
+ })
+ current := env.ReconcileConfig(&discoverymodel.ConfigChange{
+ ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.PeerAuthentication, Name: policy.Name, Namespace: policy.Namespace}),
+ })
+ if !notified || current == previous || env.ConfigSnapshot() != current {
+ t.Fatal("configuration change was not independently published")
+ }
+ if got := buildRuntimeServiceConfig(previous, nil, svc).Ports[0].MTLSMode; got != "STRICT" {
+ t.Fatalf("previous snapshot was mutated: %s", got)
+ }
+}
+
func TestBuildRuntimeTrafficConfigProjectsOnlyWorkloadPrincipalAuthorization(t *testing.T) {
svc := newInherentRuntimeTestService("provider", "grpc-app", "provider.grpc-app.svc.cluster.local", 17070)
workloadPolicy := config.Config{
@@ -430,7 +507,7 @@ func TestBuildRuntimeTrafficConfigProjectsOnlyWorkloadPrincipalAuthorization(t *
}}}}},
},
}
- push := newInherentRuntimeTestPushContext(t, []config.Config{workloadPolicy, jwtPolicy}, []*discoverymodel.Service{svc})
+ push := newInherentRuntimeTestSnapshot(t, []config.Config{workloadPolicy, jwtPolicy}, []*discoverymodel.Service{svc})
serviceConfig := buildRuntimeServiceConfig(push, nil, svc)
got := serviceConfig.Ports[0].AuthorizationPolicies
@@ -446,7 +523,7 @@ func TestBuildRuntimeTrafficConfigProjectsOnlyWorkloadPrincipalAuthorization(t *
func TestBuildRuntimeTrafficConfigCapturesPermissivePeerAuthentication(t *testing.T) {
hostname := host.Name("provider.grpc-app.svc.cluster.local")
svc := newInherentRuntimeTestService("provider", "grpc-app", string(hostname), 17070)
- push := newInherentRuntimeTestPushContext(t, []config.Config{
+ push := newInherentRuntimeTestSnapshot(t, []config.Config{
newInherentPeerAuthenticationConfig("grpc-app-permissive-mtls", "grpc-app", security.PeerAuthentication_MutualTLS_PERMISSIVE),
}, []*discoverymodel.Service{svc})
@@ -462,7 +539,7 @@ func TestBuildRuntimeTrafficConfigCapturesPermissivePeerAuthentication(t *testin
func TestBuildRuntimeTrafficConfigCapturesFaultInjection(t *testing.T) {
hostname := host.Name("provider.grpc-app.svc.cluster.local")
svc := newInherentRuntimeTestService("provider", "grpc-app", string(hostname), 17070)
- push := newInherentRuntimeTestPushContext(t, []config.Config{{
+ push := newInherentRuntimeTestSnapshot(t, []config.Config{{
Meta: config.Meta{
GroupVersionKind: gvk.FaultInjectionPolicy,
Name: "provider-fault",
@@ -501,52 +578,52 @@ func TestBuildRuntimeTrafficConfigCapturesFaultInjection(t *testing.T) {
func TestInherentGRPCRuntimeConfigNeedsUpdate(t *testing.T) {
tests := []struct {
name string
- req *discoverymodel.PushRequest
+ req *discoverymodel.ConfigChange
want bool
}{
{
name: "full push",
- req: &discoverymodel.PushRequest{Full: true},
+ req: &discoverymodel.ConfigChange{Full: true},
want: true,
},
{
name: "httproute",
- req: &discoverymodel.PushRequest{
+ req: &discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.HTTPRoute, Name: "provider-routing", Namespace: "grpc-app"}),
},
want: true,
},
{
name: "fault injection policy",
- req: &discoverymodel.PushRequest{
+ req: &discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.FaultInjectionPolicy, Name: "provider-fault", Namespace: "grpc-app"}),
},
want: true,
},
{
name: "peerauthentication",
- req: &discoverymodel.PushRequest{
+ req: &discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.PeerAuthentication, Name: "strict", Namespace: "grpc-app"}),
},
want: true,
},
{
name: "requestauthentication",
- req: &discoverymodel.PushRequest{
+ req: &discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.RequestAuthentication, Name: "jwt", Namespace: "grpc-app"}),
},
want: true,
},
{
name: "authorizationpolicy",
- req: &discoverymodel.PushRequest{
+ req: &discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.AuthorizationPolicy, Name: "require-jwt", Namespace: "grpc-app"}),
},
want: true,
},
{
name: "unrelated configmap",
- req: &discoverymodel.PushRequest{
+ req: &discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.ConfigMap, Name: "ui", Namespace: "dubbo-system"}),
},
want: false,
@@ -591,10 +668,10 @@ func TestHandleRuntimeConfigUpdateEnqueuesManagedPods(t *testing.T) {
controllers.WithMaxAttempts(1),
)
- controller.handleRuntimeConfigUpdate(&discoverymodel.PushRequest{
+ controller.handleRuntimeConfigUpdate(&discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.ConfigMap, Name: "unrelated"}),
})
- controller.handleRuntimeConfigUpdate(&discoverymodel.PushRequest{
+ controller.handleRuntimeConfigUpdate(&discoverymodel.ConfigChange{
ConfigsUpdated: sets.New(discoverymodel.ConfigKey{Kind: kind.Service, Name: "provider", Namespace: "team-a"}),
})
@@ -650,12 +727,12 @@ func TestNextRotationTime(t *testing.T) {
}
}
-func newInherentRuntimeTestPushContext(t *testing.T, configs []config.Config, services []*discoverymodel.Service) *discoverymodel.PushContext {
+func newInherentRuntimeTestSnapshot(t *testing.T, configs []config.Config, services []*discoverymodel.Service) *discoverymodel.ConfigSnapshot {
_, push := newInherentRuntimeTestEnvironment(t, configs, services)
return push
}
-func newInherentRuntimeTestEnvironment(t *testing.T, configs []config.Config, services []*discoverymodel.Service) (*discoverymodel.Environment, *discoverymodel.PushContext) {
+func newInherentRuntimeTestEnvironment(t *testing.T, configs []config.Config, services []*discoverymodel.Service) (*discoverymodel.Environment, *discoverymodel.ConfigSnapshot) {
t.Helper()
store := memory.Make(collections.DubboGatewayAPI())
@@ -673,9 +750,7 @@ func newInherentRuntimeTestEnvironment(t *testing.T, configs []config.Config, se
}, true))
env.Init()
- push := discoverymodel.NewPushContext()
- push.InitContext(env, nil, nil)
- env.SetPushContext(push)
+ push := env.ReconcileConfig(nil)
return env, push
}
diff --git a/dubbod/discovery/pkg/bootstrap/server.go b/dubbod/discovery/pkg/bootstrap/server.go
index db7ec9e8..359c0ce8 100644
--- a/dubbod/discovery/pkg/bootstrap/server.go
+++ b/dubbod/discovery/pkg/bootstrap/server.go
@@ -32,10 +32,8 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/activation"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
- dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/grpc"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/keycertbundle"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/core"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/server"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/aggregate"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/provider"
@@ -50,6 +48,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collections"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
"github.com/apache/dubbo-kubernetes/pkg/filewatcher"
+ dubbogrpc "github.com/apache/dubbo-kubernetes/pkg/grpc"
"github.com/apache/dubbo-kubernetes/pkg/introspection"
dubbokeepalive "github.com/apache/dubbo-kubernetes/pkg/keepalive"
kubelib "github.com/apache/dubbo-kubernetes/pkg/kube"
@@ -199,7 +198,7 @@ func NewServer(args *DubboArgs, initFuncs ...func(*Server)) (*Server, error) {
}
s.XDSServer = xds.NewDiscoveryServer(e, args.RegistryOptions.KubeOptions.ClusterAliases, args.KrtDebugger)
- configGen := core.NewConfigGenerator(s.XDSServer.Cache)
+ s.initXDSAuthentication()
s.initReadinessProbes()
@@ -223,6 +222,7 @@ func NewServer(args *DubboArgs, initFuncs ...func(*Server)) (*Server, error) {
s.environment.Init()
caOpts := &caOptions{
+ Authenticators: s.XDSServer.Authenticators,
TrustDomain: s.environment.Mesh().TrustDomain,
Namespace: args.Namespace,
ExternalCAType: ra.CaExternalType(externalCaType),
@@ -241,7 +241,8 @@ func NewServer(args *DubboArgs, initFuncs ...func(*Server)) (*Server, error) {
return nil, err
}
- InitGenerators(s.XDSServer, configGen)
+ InitGenerators(s.XDSServer)
+ s.XDSServer.Generators["grpc/"+sec_model.SecretType] = workloadSecretGenerator{lookup: s.lookupXDSWorkloadSecret}
if err := s.initManagement(); err != nil {
return nil, fmt.Errorf("error initializing management API: %v", err)
diff --git a/dubbod/discovery/pkg/bootstrap/xds_auth.go b/dubbod/discovery/pkg/bootstrap/xds_auth.go
new file mode 100644
index 00000000..72fa0d50
--- /dev/null
+++ b/dubbod/discovery/pkg/bootstrap/xds_auth.go
@@ -0,0 +1,110 @@
+//
+// 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 bootstrap
+
+import (
+ "fmt"
+ "strings"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/dubbod/security/pkg/server/ca/authenticate"
+ "github.com/apache/dubbo-kubernetes/pkg/security"
+ "github.com/apache/dubbo-kubernetes/pkg/spiffe"
+ corev1 "k8s.io/api/core/v1"
+)
+
+func (s *Server) initXDSAuthentication() {
+ s.XDSServer.Authenticators = []security.Authenticator{&authenticate.ClientCertAuthenticator{}}
+ s.XDSServer.Authorize = s.authorizeXDSWorkload
+}
+
+func (s *Server) authorizeXDSWorkload(proxy *model.Proxy, identities []string) error {
+ if proxy == nil || proxy.Metadata == nil || proxy.Metadata.Namespace == "" || proxy.Metadata.ClusterID == "" {
+ return fmt.Errorf("xDS requires a workload namespace and cluster")
+ }
+ controller := s.xdsWorkloadController(proxy)
+ if controller == nil {
+ return fmt.Errorf("xDS workload cluster is not managed")
+ }
+ suffix := "." + proxy.Metadata.Namespace
+ if !strings.HasSuffix(proxy.ID, suffix) {
+ return fmt.Errorf("xDS node namespace does not match metadata")
+ }
+ pod := controller.pods.Get(strings.TrimSuffix(proxy.ID, suffix), proxy.Metadata.Namespace)
+ mesh := s.environment.Mesh()
+ trustDomains := append([]string{mesh.GetTrustDomain()}, mesh.GetTrustDomainAliases()...)
+ return authorizeXDSPod(proxy, identities, pod, trustDomains)
+}
+
+func (s *Server) xdsWorkloadController(proxy *model.Proxy) *inherentGRPCWorkloadController {
+ controller := s.inherentGRPCWorkloadController
+ if proxy.Metadata.ClusterID != s.clusterID {
+ controller = nil
+ if s.inherentGRPCRemoteControllers != nil {
+ for _, remote := range s.inherentGRPCRemoteControllers.All() {
+ if remote.controller != nil && remote.controller.client.ClusterID() == proxy.Metadata.ClusterID {
+ controller = remote.controller
+ break
+ }
+ }
+ }
+ }
+ return controller
+}
+
+func authorizeXDSPod(proxy *model.Proxy, identities []string, pod *corev1.Pod, trustDomains []string) error {
+ if pod == nil || pod.DeletionTimestamp != nil || !shouldManageInherentGRPCPod(pod) {
+ return fmt.Errorf("xDS node is not an active managed workload")
+ }
+ if proxy.Metadata.Namespace != pod.Namespace || proxy.ID != pod.Name+"."+pod.Namespace {
+ return fmt.Errorf("xDS node does not match workload")
+ }
+ if len(proxy.IPAddresses) != 1 || !podHasIP(pod, proxy.IPAddresses[0]) {
+ return fmt.Errorf("xDS node IP does not match workload")
+ }
+ if proxy.IsRouter() && pod.Labels["gateway.networking.k8s.io/gateway-name"] == "" {
+ return fmt.Errorf("xDS router is not a managed gateway")
+ }
+ serviceAccount := pod.Spec.ServiceAccountName
+ if serviceAccount == "" {
+ serviceAccount = "default"
+ }
+ for _, raw := range identities {
+ identity, err := spiffe.ParseIdentity(raw)
+ if err != nil || identity.Namespace != pod.Namespace || identity.ServiceAccount != serviceAccount {
+ continue
+ }
+ for _, trustDomain := range trustDomains {
+ if identity.TrustDomain == trustDomain {
+ return nil
+ }
+ }
+ }
+ return fmt.Errorf("xDS identity does not match workload service account and trust domain")
+}
+
+func podHasIP(pod *corev1.Pod, ip string) bool {
+ if ip != "" && ip == pod.Status.PodIP {
+ return true
+ }
+ for _, address := range pod.Status.PodIPs {
+ if ip != "" && address.IP == ip {
+ return true
+ }
+ }
+ return false
+}
diff --git a/dubbod/discovery/pkg/bootstrap/xds_auth_test.go b/dubbod/discovery/pkg/bootstrap/xds_auth_test.go
new file mode 100644
index 00000000..5d8c396c
--- /dev/null
+++ b/dubbod/discovery/pkg/bootstrap/xds_auth_test.go
@@ -0,0 +1,77 @@
+//
+// 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 bootstrap
+
+import (
+ "testing"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/cluster"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/inject"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestAuthorizeXDSPod(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ change func(*model.Proxy, *corev1.Pod)
+ identity string
+ wantOK bool
+ }{
+ {name: "workload", wantOK: true},
+ {name: "gateway", change: func(p *model.Proxy, pod *corev1.Pod) {
+ p.Type = model.Router
+ pod.Labels = map[string]string{"gateway.networking.k8s.io/gateway-name": "edge"}
+ }, wantOK: true},
+ {name: "wrong service account", identity: "spiffe://cluster.local/ns/app/sa/other"},
+ {name: "wrong namespace", identity: "spiffe://cluster.local/ns/other/sa/default"},
+ {name: "wrong trust domain", identity: "spiffe://other/ns/app/sa/default"},
+ {name: "wrong IP", change: func(p *model.Proxy, _ *corev1.Pod) { p.IPAddresses = []string{"10.0.0.2"} }},
+ {name: "wrong pod", change: func(p *model.Proxy, _ *corev1.Pod) { p.ID = "other.app" }},
+ {name: "router impersonation", change: func(p *model.Proxy, _ *corev1.Pod) { p.Type = model.Router }},
+ {name: "unmanaged", change: func(_ *model.Proxy, pod *corev1.Pod) { pod.Annotations = nil }},
+ {name: "deleted", change: func(_ *model.Proxy, pod *corev1.Pod) { now := metav1.Now(); pod.DeletionTimestamp = &now }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod", Namespace: "app", Annotations: map[string]string{inject.InherentInjectTemplatesAnnoName: inject.InherentGRPCTemplateName}}, Status: corev1.PodStatus{PodIP: "10.0.0.1"}}
+ proxy := &model.Proxy{ID: "pod.app", Type: model.Inherent, Metadata: &model.NodeMetadata{Namespace: "app"}, IPAddresses: []string{"10.0.0.1"}}
+ if tc.change != nil {
+ tc.change(proxy, pod)
+ }
+ identity := tc.identity
+ if identity == "" {
+ identity = "spiffe://cluster.local/ns/app/sa/default"
+ }
+ err := authorizeXDSPod(proxy, []string{identity}, pod, []string{"cluster.local"})
+ if (err == nil) != tc.wantOK {
+ t.Fatalf("authorizeXDSPod = %v", err)
+ }
+ })
+ }
+}
+
+func TestXDSRejectsUnmanagedCluster(t *testing.T) {
+ s := &Server{clusterID: "local"}
+ for _, clusterID := range []string{"", "other", "local"} {
+ proxy := &model.Proxy{Metadata: &model.NodeMetadata{Namespace: "app"}}
+ proxy.Metadata.ClusterID = cluster.ID(clusterID)
+ if err := s.authorizeXDSWorkload(proxy, nil); err == nil {
+ t.Fatal("accepted workload without managed cluster")
+ }
+ }
+}
diff --git a/dubbod/discovery/pkg/bootstrap/xds_secrets.go b/dubbod/discovery/pkg/bootstrap/xds_secrets.go
new file mode 100644
index 00000000..b76be4cc
--- /dev/null
+++ b/dubbod/discovery/pkg/bootstrap/xds_secrets.go
@@ -0,0 +1,136 @@
+//
+// 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 bootstrap
+
+import (
+ "crypto/tls"
+ "crypto/x509"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/util/protoconv"
+ "github.com/apache/dubbo-kubernetes/pkg/config/constants"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/inject"
+ "github.com/apache/dubbo-kubernetes/pkg/security"
+ "github.com/apache/dubbo-kubernetes/pkg/spiffe"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ core "github.com/kdubbo/xds-api/core/v1"
+ tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
+ discovery "github.com/kdubbo/xds-api/service/discovery/v1"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+ corev1 "k8s.io/api/core/v1"
+)
+
+type workloadSecretGenerator struct {
+ lookup func(*model.Proxy) (*corev1.Pod, *corev1.Secret)
+}
+
+func (s *Server) lookupXDSWorkloadSecret(proxy *model.Proxy) (*corev1.Pod, *corev1.Secret) {
+ controller := s.xdsWorkloadController(proxy)
+ if controller == nil {
+ return nil, nil
+ }
+ pod := controller.pods.Get(strings.TrimSuffix(proxy.ID, "."+proxy.Metadata.Namespace), proxy.Metadata.Namespace)
+ if pod == nil {
+ return nil, nil
+ }
+ return pod, controller.secrets.Get(inject.InherentGRPCSecretNameForMeta(pod.ObjectMeta), pod.Namespace)
+}
+
+func (g workloadSecretGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, _ *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
+ if proxy == nil || !proxy.IsRouter() {
+ return nil, model.DefaultXdsLogDetails, status.Error(codes.PermissionDenied, "SDS is only available to managed gateways")
+ }
+ names := sets.New("default", security.RootCertReqResourceName)
+ if w != nil && len(w.ResourceNames) > 0 {
+ names = w.ResourceNames
+ }
+ for name := range names {
+ if name != "default" && name != security.RootCertReqResourceName {
+ return nil, model.DefaultXdsLogDetails, status.Errorf(codes.PermissionDenied, "SDS resource %q is not a workload credential", name)
+ }
+ }
+ pod, secret := g.lookup(proxy)
+ if err := validateWorkloadSecret(pod, secret); err != nil {
+ return nil, model.DefaultXdsLogDetails, status.Error(codes.FailedPrecondition, err.Error())
+ }
+ resources := make(model.Resources, 0, len(names))
+ for _, name := range sets.SortedList(names) {
+ resource := &tlsv1.Secret{Name: name}
+ if name == security.RootCertReqResourceName {
+ resource.Type = &tlsv1.Secret_ValidationContext{ValidationContext: &tlsv1.CertificateValidationContext{TrustedCa: secretBytes(secret.Data[constants.CACertNamespaceConfigMapDataName])}}
+ } else {
+ resource.Type = &tlsv1.Secret_TlsCertificate{TlsCertificate: &tlsv1.TlsCertificate{CertificateChain: secretBytes(secret.Data[constants.CertChainFilename]), PrivateKey: secretBytes(secret.Data[constants.KeyFilename])}}
+ }
+ resources = append(resources, &discovery.Resource{Name: name, Resource: protoconv.MessageToAny(resource)})
+ }
+ return resources, model.DefaultXdsLogDetails, nil
+}
+
+func secretBytes(value []byte) *core.DataSource {
+ return &core.DataSource{Specifier: &core.DataSource_InlineBytes{InlineBytes: append([]byte(nil), value...)}}
+}
+
+func validateWorkloadSecret(pod *corev1.Pod, secret *corev1.Secret) error {
+ if pod == nil || secret == nil || pod.DeletionTimestamp != nil {
+ return fmt.Errorf("workload credentials are not available")
+ }
+ owned := false
+ for _, owner := range secret.OwnerReferences {
+ if owner.APIVersion == "v1" && owner.Kind == "Pod" && owner.Name == pod.Name && owner.UID == pod.UID && pod.UID != "" {
+ owned = true
+ }
+ }
+ if !owned || secret.Namespace != pod.Namespace || secret.Name != inject.InherentGRPCSecretNameForMeta(pod.ObjectMeta) {
+ return fmt.Errorf("workload credentials do not belong to the requesting Pod")
+ }
+ pair, err := tls.X509KeyPair(secret.Data[constants.CertChainFilename], secret.Data[constants.KeyFilename])
+ if err != nil {
+ return fmt.Errorf("invalid workload certificate/key pair")
+ }
+ leaf, err := x509.ParseCertificate(pair.Certificate[0])
+ if err != nil || len(leaf.URIs) != 1 {
+ return fmt.Errorf("invalid workload certificate identity")
+ }
+ identity, err := spiffe.ParseIdentity(leaf.URIs[0].String())
+ sa := pod.Spec.ServiceAccountName
+ if sa == "" {
+ sa = "default"
+ }
+ if err != nil || identity.Namespace != pod.Namespace || identity.ServiceAccount != sa {
+ return fmt.Errorf("workload certificate does not match Pod identity")
+ }
+ roots := x509.NewCertPool()
+ if !roots.AppendCertsFromPEM(secret.Data[constants.CACertNamespaceConfigMapDataName]) {
+ return fmt.Errorf("invalid workload trust bundle")
+ }
+ intermediates := x509.NewCertPool()
+ for _, der := range pair.Certificate[1:] {
+ cert, err := x509.ParseCertificate(der)
+ if err != nil {
+ return fmt.Errorf("invalid workload intermediate certificate")
+ }
+ intermediates.AddCert(cert)
+ }
+ if _, err := leaf.Verify(x509.VerifyOptions{Roots: roots, Intermediates: intermediates, CurrentTime: time.Now(), KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}}); err != nil {
+ return fmt.Errorf("workload certificate is not valid against its trust bundle")
+ }
+ return nil
+}
diff --git a/dubbod/discovery/pkg/bootstrap/xds_secrets_test.go b/dubbod/discovery/pkg/bootstrap/xds_secrets_test.go
new file mode 100644
index 00000000..1ec61662
--- /dev/null
+++ b/dubbod/discovery/pkg/bootstrap/xds_secrets_test.go
@@ -0,0 +1,150 @@
+//
+// 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 bootstrap
+
+import (
+ "bytes"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "crypto/x509"
+ "encoding/pem"
+ "math/big"
+ "net/url"
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config/constants"
+ "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+ tlsv1 "github.com/kdubbo/xds-api/extensions/transport_sockets/tls/v1"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func testWorkloadSecret(t *testing.T, identity string) (*corev1.Pod, *corev1.Secret) {
+ t.Helper()
+ rootKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now()
+ root := &x509.Certificate{SerialNumber: big.NewInt(1), IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour)}
+ rootDER, err := x509.CreateCertificate(rand.Reader, root, root, &rootKey.PublicKey, rootKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ uri, err := url.Parse(identity)
+ if err != nil {
+ t.Fatal(err)
+ }
+ leaf := &x509.Certificate{SerialNumber: big.NewInt(2), URIs: []*url.URL{uri}, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour), KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}}
+ der, err := x509.CreateCertificate(rand.Reader, leaf, root, &key.PublicKey, rootKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ keyDER, err := x509.MarshalECPrivateKey(key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "edge", Namespace: "app", UID: "pod-uid"}}
+ secret := buildInherentGRPCSecret(pod, nil, nil, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: rootDER}))
+ return pod, secret
+}
+
+func TestWorkloadSDSRotationAndIsolation(t *testing.T) {
+ pod, secret := testWorkloadSecret(t, "spiffe://cluster.local/ns/app/sa/default")
+ g := workloadSecretGenerator{lookup: func(*model.Proxy) (*corev1.Pod, *corev1.Secret) { return pod, secret }}
+ proxy := &model.Proxy{Type: model.Router}
+ watch := &model.WatchedResource{ResourceNames: sets.New("default", "ROOTCA")}
+ first, _, err := g.Generate(proxy, watch, nil)
+ if err != nil || len(first) != 2 {
+ t.Fatalf("initial SDS = %d, %v", len(first), err)
+ }
+ var original []byte
+ for _, resource := range first {
+ decoded := &tlsv1.Secret{}
+ if err := resource.Resource.UnmarshalTo(decoded); err != nil {
+ t.Fatal(err)
+ }
+ if resource.Name == "default" {
+ original = decoded.GetTlsCertificate().GetCertificateChain().GetInlineBytes()
+ }
+ if resource.Name == "ROOTCA" && len(decoded.GetValidationContext().GetTrustedCa().GetInlineBytes()) == 0 {
+ t.Fatal("missing trust bundle")
+ }
+ }
+ _, secret = testWorkloadSecret(t, "spiffe://cluster.local/ns/app/sa/default")
+ rotated, _, err := g.Generate(proxy, watch, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, resource := range rotated {
+ if resource.Name != "default" {
+ continue
+ }
+ decoded := &tlsv1.Secret{}
+ if err := resource.Resource.UnmarshalTo(decoded); err != nil {
+ t.Fatal(err)
+ }
+ if bytes.Equal(original, decoded.GetTlsCertificate().GetCertificateChain().GetInlineBytes()) {
+ t.Fatal("rotation served old certificate")
+ }
+ }
+ if _, _, err := g.Generate(proxy, &model.WatchedResource{ResourceNames: sets.New("other-secret")}, nil); status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("foreign resource accepted: %v", err)
+ }
+ proxy.Type = model.Inherent
+ if _, _, err := g.Generate(proxy, watch, nil); status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("application SDS accepted: %v", err)
+ }
+}
+
+func TestWorkloadSDSRejectsInvalidCredentials(t *testing.T) {
+ pod, valid := testWorkloadSecret(t, "spiffe://cluster.local/ns/app/sa/default")
+ for _, tc := range []struct {
+ name string
+ change func(*corev1.Secret)
+ }{
+ {"foreign owner", func(s *corev1.Secret) { s.OwnerReferences[0].UID = "another-pod" }},
+ {"foreign namespace", func(s *corev1.Secret) { s.Namespace = "other" }},
+ {"foreign name", func(s *corev1.Secret) { s.Name = "other-secret" }},
+ {"invalid key", func(s *corev1.Secret) { s.Data[constants.KeyFilename] = []byte("invalid") }},
+ {"invalid root", func(s *corev1.Secret) { s.Data[constants.CACertNamespaceConfigMapDataName] = []byte("invalid") }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ s := valid.DeepCopy()
+ tc.change(s)
+ if err := validateWorkloadSecret(pod, s); err == nil {
+ t.Fatal("invalid secret accepted")
+ }
+ })
+ }
+ _, foreign := testWorkloadSecret(t, "spiffe://cluster.local/ns/app/sa/other")
+ if err := validateWorkloadSecret(pod, foreign); err == nil {
+ t.Fatal("foreign identity accepted")
+ }
+ if err := validateWorkloadSecret(pod, nil); err == nil {
+ t.Fatal("missing secret accepted")
+ }
+}
diff --git a/dubbod/discovery/pkg/config/kube/gateway/context.go b/dubbod/discovery/pkg/config/kube/gateway/context.go
index 9c61fa01..028ced30 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/context.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/context.go
@@ -22,10 +22,10 @@ import (
)
type Context struct {
- ps *model.PushContext
+ ps *model.ConfigSnapshot
cluster cluster.ID
}
-func NewGatewayContext(ps *model.PushContext, cluster cluster.ID) Context {
+func NewGatewayContext(ps *model.ConfigSnapshot, cluster cluster.ID) Context {
return Context{ps, cluster}
}
diff --git a/dubbod/discovery/pkg/config/kube/gateway/controller.go b/dubbod/discovery/pkg/config/kube/gateway/controller.go
index 37c95150..b2246061 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/controller.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/controller.go
@@ -161,7 +161,7 @@ func newKindStore(collection krt.Collection[config.Config]) kindStore {
}
}
-func (c *Controller) Reconcile(ps *model.PushContext) {
+func (c *Controller) Reconcile(ps *model.ConfigSnapshot) {
ctx := NewGatewayContext(ps, c.cluster)
c.gatewayContext.Modify(func(i **atomic.Pointer[Context]) {
(*i).Store(&ctx)
diff --git a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
index beabb6b8..0bcc4b5c 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller.go
@@ -24,6 +24,7 @@ import (
"encoding/json"
"fmt"
"net"
+ "net/url"
"sort"
"strconv"
"strings"
@@ -737,7 +738,7 @@ func (d *DeploymentController) buildDxgateBootstrapConfig(gw gateway.Gateway, se
if systemNamespace == "" {
systemNamespace = constants.DubboSystemNamespace
}
- xdsAddress := fmt.Sprintf("http://dubbod.%s.svc:26010", systemNamespace)
+ xdsAddress := fmt.Sprintf("https://dubbod.%s.svc:26012", systemNamespace)
if gw.Annotations[xdsAddressAnnotation] != "" {
xdsAddress = gw.Annotations[xdsAddressAnnotation]
}
@@ -753,6 +754,10 @@ func buildDxgateBootstrapConfig(xdsAddress string, listenerNames []string, clust
if xdsAddress == "" {
return "", "", fmt.Errorf("dxgate bootstrap xDS address is empty")
}
+ endpoint, err := url.Parse(xdsAddress)
+ if err != nil || endpoint.Scheme != "https" || endpoint.Hostname() == "" {
+ return "", "", fmt.Errorf("dxgate ADS address must be an https URL with workload credentials")
+ }
if clusterID == "" {
clusterID = string(cluster.ID("Kubernetes"))
}
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 e51b587d..5160dcec 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/deployment_controller_test.go
@@ -386,7 +386,7 @@ func TestBuildDxgateRuntimeConfigExternalNameBackendTLS(t *testing.T) {
func TestBuildDxgateBootstrapConfig(t *testing.T) {
raw, hash, err := buildDxgateBootstrapConfig(
- "http://dubbod.dubbo-system.svc:26010",
+ "https://dubbod.dubbo-system.svc:26012",
[]string{"public-dubbo.app.svc.cluster.local:80"},
"Kubernetes",
"cluster.local",
@@ -402,7 +402,7 @@ func TestBuildDxgateBootstrapConfig(t *testing.T) {
if err := json.Unmarshal([]byte(raw), &cfg); err != nil {
t.Fatal(err)
}
- if cfg.XDSAddress != "http://dubbod.dubbo-system.svc:26010" {
+ if cfg.XDSAddress != "https://dubbod.dubbo-system.svc:26012" {
t.Fatalf("unexpected xDS address: %s", cfg.XDSAddress)
}
if diff := cmp.Diff([]string{"public-dubbo.app.svc.cluster.local:80"}, cfg.ListenerNames); diff != "" {
@@ -413,6 +413,17 @@ func TestBuildDxgateBootstrapConfig(t *testing.T) {
}
}
+func TestManagedGatewayRequiresSecureADS(t *testing.T) {
+ controller := &DeploymentController{systemNamespace: "dubbo-system"}
+ raw, _, err := controller.buildDxgateBootstrapConfig(gatewayv1.Gateway{}, "gateway", nil)
+ if err != nil || !strings.Contains(raw, "https://dubbod.dubbo-system.svc:26012") {
+ t.Fatalf("managed gateway must default to authenticated ADS: %s, %v", raw, err)
+ }
+ if _, _, err := buildDxgateBootstrapConfig("http://dubbod.dubbo-system.svc:26010", nil, "", ""); err == nil {
+ t.Fatal("managed gateway must reject plaintext ADS")
+ }
+}
+
func TestDeploymentControllerBuildDxgateBootstrapConfigUsesXDSAddressAnnotation(t *testing.T) {
controller := &DeploymentController{
clusterID: "remote",
@@ -423,7 +434,7 @@ func TestDeploymentControllerBuildDxgateBootstrapConfigUsesXDSAddressAnnotation(
Name: "eastwest",
Namespace: "dubbo-system",
Annotations: map[string]string{
- xdsAddressAnnotation: "http://192.168.15.164:32010",
+ xdsAddressAnnotation: "https://dubbod.remote.example:32012",
},
},
}, "dxgate-gateway", []corev1.ServicePort{{Name: "http-eastwest", Port: 15443, TargetPort: intstr.FromInt(15080)}})
@@ -435,7 +446,7 @@ func TestDeploymentControllerBuildDxgateBootstrapConfigUsesXDSAddressAnnotation(
if err := json.Unmarshal([]byte(raw), &cfg); err != nil {
t.Fatal(err)
}
- if cfg.XDSAddress != "http://192.168.15.164:32010" {
+ if cfg.XDSAddress != "https://dubbod.remote.example:32012" {
t.Fatalf("xdsAddress = %q, want annotation", cfg.XDSAddress)
}
if cfg.ClusterID != "remote" {
@@ -676,7 +687,7 @@ func TestKubeGatewayTemplateRendersDxgateResources(t *testing.T) {
Ports: []corev1.ServicePort{{Name: "http", Port: 80, TargetPort: intstr.FromInt(15080)}},
ServiceType: corev1.ServiceTypeLoadBalancer,
Revision: "default",
- BootstrapConfig: "{\n \"xds_address\": \"http://dubbod.dubbo-system.svc:26010\",\n \"listener_names\": [\n \"public-dubbo.app.svc.cluster.local:80\"\n ],\n \"cluster_id\": \"Kubernetes\",\n \"dns_domain\": \"cluster.local\"\n}\n",
+ BootstrapConfig: "{\n \"xds_address\": \"https://dubbod.dubbo-system.svc:26012\",\n \"listener_names\": [\n \"public-dubbo.app.svc.cluster.local:80\"\n ],\n \"cluster_id\": \"Kubernetes\",\n \"dns_domain\": \"cluster.local\"\n}\n",
BootstrapConfigHash: "abc123",
DxgateImage: "kdubbo/dxgate:test",
SystemNamespace: "dubbo-system",
@@ -703,7 +714,7 @@ func TestKubeGatewayTemplateRendersDxgateResources(t *testing.T) {
if strings.Contains(strings.Join(rendered, "\n---\n"), "{{") {
t.Fatalf("hardcoded gateway template still contains template delimiters:\n%s", strings.Join(rendered, "\n---\n"))
}
- if !strings.Contains(rendered[0], "bootstrap.json") || !strings.Contains(rendered[0], `"xds_address": "http://dubbod.dubbo-system.svc:26010"`) {
+ if !strings.Contains(rendered[0], "bootstrap.json") || !strings.Contains(rendered[0], `"xds_address": "https://dubbod.dubbo-system.svc:26012"`) {
t.Fatalf("configmap did not render dxgate bootstrap:\n%s", rendered[0])
}
if !strings.Contains(rendered[1], "automountServiceAccountToken: true") ||
diff --git a/dubbod/discovery/pkg/features/dubbo.go b/dubbod/discovery/pkg/features/dubbo.go
index c90d9948..d7889080 100644
--- a/dubbod/discovery/pkg/features/dubbo.go
+++ b/dubbod/discovery/pkg/features/dubbo.go
@@ -27,15 +27,12 @@ var (
"Only works in kubernetes environment.").Get()
SharedMeshConfig = env.Register("SHARED_MESH_CONFIG", "",
"Additional config map to load for shared MeshConfig settings. The standard mesh config will take precedence.").Get()
- InformerWatchNamespace = env.Register("DUBBO_WATCH_NAMESPACE", "",
- "If set, limit Kubernetes watches to a single namespace. "+
- "Warning: only a single namespace can be set.").Get()
+
ClusterName = env.Register("CLUSTER_ID", constants.DefaultClusterName,
"Defines the cluster and service registry that this Dubbod instance belongs to").Get()
EnableVtprotobuf = env.Register("ENABLE_VTPROTOBUF", true,
"If true, will use optimized vtprotobuf based marshaling. Requires a build with -tags=vtprotobuf.").Get()
- KubernetesClientContentType = env.Register("DUBBO_KUBE_CLIENT_CONTENT_TYPE", "protobuf",
- "The content type to use for Kubernetes clients. Defaults to protobuf. Valid options: [protobuf, json]").Get()
+
EnableCAServer = env.Register("ENABLE_CA_SERVER", true,
"If this is set to false, will not create CA server in dubbod.").Get()
// EnableCACRL ToDo (nilekh): remove this feature flag once it's stable
diff --git a/dubbod/discovery/pkg/features/tuning.go b/dubbod/discovery/pkg/features/tuning.go
index 8acf1fdb..14be7b58 100644
--- a/dubbod/discovery/pkg/features/tuning.go
+++ b/dubbod/discovery/pkg/features/tuning.go
@@ -23,19 +23,6 @@ import (
)
var (
- MaxConcurrentStreams = env.Register(
- "DUBBO_GPRC_MAXSTREAMS",
- 100000,
- "Sets the maximum number of concurrent grpc streams.",
- ).Get()
-
- // MaxRecvMsgSize The max receive buffer size of gRPC received channel of Dubbo in bytes.
- MaxRecvMsgSize = env.Register(
- "DUBBO_GPRC_MAXRECVMSGSIZE",
- 4*1024*1024,
- "Sets the max receive buffer size of gRPC stream in bytes.",
- ).Get()
-
RequestLimit = func() float64 {
v := env.Register(
"DUBBO_MAX_REQUESTS_PER_SECOND",
diff --git a/dubbod/discovery/pkg/leaderelection/leaderelection_test.go b/dubbod/discovery/pkg/leaderelection/leaderelection_test.go
new file mode 100644
index 00000000..8f9fe1a4
--- /dev/null
+++ b/dubbod/discovery/pkg/leaderelection/leaderelection_test.go
@@ -0,0 +1,116 @@
+//
+// 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 leaderelection
+
+import (
+ "fmt"
+ "strconv"
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/pkg/kube"
+ corev1 "k8s.io/api/core/v1"
+ apierrors "k8s.io/apimachinery/pkg/api/errors"
+ "k8s.io/apimachinery/pkg/runtime"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/kubernetes/fake"
+ clienttesting "k8s.io/client-go/testing"
+)
+
+type electionTestClient struct {
+ kube.Client
+ client kubernetes.Interface
+}
+
+func (c electionTestClient) Kube() kubernetes.Interface { return c.client }
+
+func TestNamespaceElectionReplicasAndHandoff(t *testing.T) {
+ for _, remote := range []bool{false, true} {
+ t.Run(fmt.Sprintf("remote=%t", remote), func(t *testing.T) {
+ client := fake.NewSimpleClientset()
+ // The fake tracker does not implement API-server resource-version conflicts.
+ client.PrependReactor("update", "configmaps", func(action clienttesting.Action) (bool, runtime.Object, error) {
+ candidate := action.(clienttesting.UpdateAction).GetObject().(*corev1.ConfigMap)
+ stored, err := client.Tracker().Get(action.GetResource(), candidate.Namespace, candidate.Name)
+ if err != nil {
+ return true, nil, err
+ }
+ current := stored.(*corev1.ConfigMap)
+ if candidate.ResourceVersion != current.ResourceVersion {
+ return true, nil, apierrors.NewConflict(action.GetResource().GroupResource(), candidate.Name, fmt.Errorf("resource version changed"))
+ }
+ revision, _ := strconv.Atoi(current.ResourceVersion)
+ updated := candidate.DeepCopy()
+ updated.ResourceVersion = strconv.Itoa(revision + 1)
+ if err := client.Tracker().Update(action.GetResource(), updated, updated.Namespace); err != nil {
+ return true, nil, err
+ }
+ return true, updated, nil
+ })
+ started := make(chan int, 4)
+ stops := []chan struct{}{make(chan struct{}), make(chan struct{})}
+ finished := []chan struct{}{make(chan struct{}), make(chan struct{})}
+ closed := []bool{false, false}
+ for id := 0; id < 2; id++ {
+ election := NewLeaderElectionMulticluster("dubbo-system", fmt.Sprintf("dubbod-%d", id), NamespaceController, "default", remote, electionTestClient{client: client}).SetEnabled(true)
+ election.ttl = 2 * time.Second
+ election.AddRunFunction(func(stop <-chan struct{}) { started <- id; <-stop })
+ go func() { election.Run(stops[id]); close(finished[id]) }()
+ }
+ t.Cleanup(func() {
+ for id := range stops {
+ if !closed[id] {
+ close(stops[id])
+ }
+ }
+ for _, done := range finished {
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Error("election did not stop")
+ }
+ }
+ })
+ var leader int
+ select {
+ case leader = <-started:
+ case <-time.After(5 * time.Second):
+ t.Fatal("no namespace leader")
+ }
+ select {
+ case other := <-started:
+ t.Fatalf("two active namespace leaders: %d and %d", leader, other)
+ case <-time.After(700 * time.Millisecond):
+ }
+ close(stops[leader])
+ closed[leader] = true
+ select {
+ case <-finished[leader]:
+ case <-time.After(5 * time.Second):
+ t.Fatal("old leader did not stop")
+ }
+ select {
+ case next := <-started:
+ if next == leader {
+ t.Fatal("stopped replica regained leadership")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("standby did not take over namespace reconciliation")
+ }
+ })
+ }
+}
diff --git a/dubbod/discovery/pkg/model/push_context.go b/dubbod/discovery/pkg/model/config_snapshot.go
similarity index 71%
copy from dubbod/discovery/pkg/model/push_context.go
copy to dubbod/discovery/pkg/model/config_snapshot.go
index 3ad42cc3..d474aeea 100644
--- a/dubbod/discovery/pkg/model/push_context.go
+++ b/dubbod/discovery/pkg/model/config_snapshot.go
@@ -18,7 +18,6 @@ package model
import (
"cmp"
- "encoding/json"
"strings"
"sync"
"time"
@@ -28,7 +27,6 @@ import (
sigsk8siogatewayapiapisv1 "sigs.k8s.io/gateway-api/apis/v1"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/provider"
- "github.com/apache/dubbo-kubernetes/pkg/cluster"
"github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/host"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
@@ -36,31 +34,14 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/slices"
"github.com/apache/dubbo-kubernetes/pkg/spiffe"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
- "github.com/apache/dubbo-kubernetes/pkg/xds"
meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
"go.uber.org/atomic"
"google.golang.org/protobuf/types/known/wrapperspb"
)
-type TriggerReason string
-
-const (
- UnknownTrigger TriggerReason = "unknown"
- ProxyRequest TriggerReason = "proxyrequest"
- GlobalUpdate TriggerReason = "global"
- HeadlessEndpointUpdate TriggerReason = "headlessendpoint"
- EndpointUpdate TriggerReason = "endpoint"
- ProxyUpdate TriggerReason = "proxy"
- ConfigUpdate TriggerReason = "config"
- DependentResource TriggerReason = "depdendentresource"
-)
-
-var (
- LastPushStatus *PushContext
- LastPushMutex sync.Mutex
-)
-
-type PushContext struct {
+// ConfigSnapshot contains service and policy state shared by application and xDS configuration.
+// It owns no subscriptions, wire versions, acknowledgements, or connection status.
+type ConfigSnapshot struct {
Mesh *meshv1alpha1.MeshConfig `json:"-"`
initializeMutex sync.Mutex
InitDone atomic.Bool
@@ -78,40 +59,6 @@ type PushContext struct {
serviceActivationIndex serviceActivationPolicyIndex
serviceAccounts map[serviceAccountKey][]string
AuthenticationPolicies *AuthenticationPolicies
- PushVersion string
- ProxyStatus map[string]map[string]ProxyPushStatus
- proxyStatusMutex sync.RWMutex
-}
-
-type PushRequest struct {
- Reason ReasonStats
- ConfigsUpdated sets.Set[ConfigKey]
- ServiceActivationPolicyUpdates map[ConfigKey]ServiceActivationPolicyUpdate
- AddressesUpdated sets.Set[string]
- Forced bool
- Full bool
- Push *PushContext
- Start time.Time
- Delta ResourceDelta
-}
-
-type ServiceActivationPolicyUpdate struct {
- Config config.Config
- Previous config.Config
- Deleted bool
-}
-
-type XDSUpdater interface {
- ConfigUpdate(req *PushRequest)
- ServiceUpdate(shard ShardKey, hostname string, namespace string, event Event)
- EDSUpdate(shard ShardKey, hostname string, namespace string, entry []*DubboEndpoint)
- EDSCacheUpdate(shard ShardKey, hostname string, namespace string, entry []*DubboEndpoint)
- ProxyUpdate(clusterID cluster.ID, ip string)
-}
-
-type ProxyPushStatus struct {
- Proxy string `json:"proxy,omitempty"`
- Message string `json:"message,omitempty"`
}
type serviceAccountKey struct {
@@ -127,16 +74,6 @@ type serviceIndex struct {
instancesByPort map[string]map[int][]*DubboEndpoint
}
-type ReasonStats map[TriggerReason]int
-
-type ResourceDelta = xds.ResourceDelta
-
-type ConfigKey struct {
- Kind kind.Kind
- Name string
- Namespace string
-}
-
type exportToDefaults struct {
service sets.Set[visibility.Instance]
}
@@ -164,8 +101,8 @@ type serviceActivationPolicyIndex struct {
services map[string][]string
}
-func NewPushContext() *PushContext {
- return &PushContext{
+func NewConfigSnapshot() *ConfigSnapshot {
+ return &ConfigSnapshot{
ServiceIndex: newServiceIndex(),
dxgateServiceIndex: dxgateServiceIndex{byNamespace: map[string]map[string]config.Config{}},
backendTLSPolicyIndex: backendTLSPolicyIndex{serviceTLS: map[string]BackendTLSSettings{}},
@@ -173,7 +110,6 @@ func NewPushContext() *PushContext {
services: map[string][]string{},
},
serviceAccounts: map[serviceAccountKey][]string{},
- ProxyStatus: map[string]map[string]ProxyPushStatus{},
}
}
@@ -187,193 +123,7 @@ func newServiceIndex() serviceIndex {
}
}
-func NewReasonStats(reasons ...TriggerReason) ReasonStats {
- ret := make(ReasonStats)
- for _, reason := range reasons {
- ret.Add(reason)
- }
- return ret
-}
-
-func (r ReasonStats) Has(reason TriggerReason) bool {
- return r[reason] > 0
-}
-
-func (r ReasonStats) Add(reason TriggerReason) {
- r[reason]++
-}
-
-func (r ReasonStats) Merge(other ReasonStats) {
- for reason, count := range other {
- r[reason] += count
- }
-}
-
-func (r ReasonStats) Copy() ReasonStats {
- if len(r) == 0 {
- return nil
- }
- out := make(ReasonStats, len(r))
- for reason, count := range r {
- out[reason] = count
- }
- return out
-}
-
-func (r ReasonStats) Count() int {
- var ret int
- for _, count := range r {
- ret += count
- }
- return ret
-}
-
-func (pr *PushRequest) Merge(other *PushRequest) *PushRequest {
- if pr == nil {
- return other
- }
- if other == nil {
- return pr
- }
-
- // Keep the first (older) start time
-
- // Merge the two reasons. Note that we shouldn't deduplicate here, or we would under count
- if len(other.Reason) > 0 {
- if pr.Reason == nil {
- pr.Reason = make(map[TriggerReason]int)
- }
- pr.Reason.Merge(other.Reason)
- }
-
- // If either is full we need a full push
- pr.Full = pr.Full || other.Full
-
- // If either is forced we need a forced push
- pr.Forced = pr.Forced || other.Forced
-
- // The other push context is presumed to be later and more up to date
- if other.Push != nil {
- pr.Push = other.Push
- }
-
- if pr.Start.IsZero() {
- pr.Start = other.Start
- }
-
- if pr.ConfigsUpdated == nil {
- if other.ConfigsUpdated != nil {
- pr.ConfigsUpdated = other.ConfigsUpdated.Copy()
- }
- } else {
- pr.ConfigsUpdated.Merge(other.ConfigsUpdated)
- }
-
- if pr.AddressesUpdated == nil {
- if other.AddressesUpdated != nil {
- pr.AddressesUpdated = other.AddressesUpdated.Copy()
- }
- } else {
- pr.AddressesUpdated.Merge(other.AddressesUpdated)
- }
- if len(other.ServiceActivationPolicyUpdates) > 0 {
- if pr.ServiceActivationPolicyUpdates == nil {
- pr.ServiceActivationPolicyUpdates = make(map[ConfigKey]ServiceActivationPolicyUpdate)
- }
- for key, update := range other.ServiceActivationPolicyUpdates {
- if existing, found := pr.ServiceActivationPolicyUpdates[key]; found &&
- existing.Previous.Spec != nil {
- update.Previous = existing.Previous
- }
- pr.ServiceActivationPolicyUpdates[key] = update
- }
- }
-
- pr.Delta = mergeResourceDelta(pr.Delta, other.Delta)
-
- return pr
-}
-
-func mergeResourceDelta(first, second ResourceDelta) ResourceDelta {
- out := copyResourceDelta(first)
- if len(second.Subscribed) > 0 {
- if out.Subscribed == nil {
- out.Subscribed = second.Subscribed.Copy()
- } else {
- out.Subscribed.Merge(second.Subscribed)
- }
- }
- if len(second.Unsubscribed) > 0 {
- if out.Unsubscribed == nil {
- out.Unsubscribed = second.Unsubscribed.Copy()
- } else {
- out.Unsubscribed.Merge(second.Unsubscribed)
- }
- }
- return out
-}
-
-func copyResourceDelta(delta ResourceDelta) ResourceDelta {
- out := ResourceDelta{}
- if delta.Subscribed != nil {
- out.Subscribed = delta.Subscribed.Copy()
- }
- if delta.Unsubscribed != nil {
- out.Unsubscribed = delta.Unsubscribed.Copy()
- }
- return out
-}
-
-func (pr *PushRequest) Copy() *PushRequest {
- if pr == nil {
- return nil
- }
- out := *pr
- out.Reason = pr.Reason.Copy()
- if pr.ConfigsUpdated != nil {
- out.ConfigsUpdated = pr.ConfigsUpdated.Copy()
- }
- if pr.AddressesUpdated != nil {
- out.AddressesUpdated = pr.AddressesUpdated.Copy()
- }
- if pr.ServiceActivationPolicyUpdates != nil {
- out.ServiceActivationPolicyUpdates = make(map[ConfigKey]ServiceActivationPolicyUpdate, len(pr.ServiceActivationPolicyUpdates))
- for key, update := range pr.ServiceActivationPolicyUpdates {
- out.ServiceActivationPolicyUpdates[key] = update
- }
- }
- out.Delta = copyResourceDelta(pr.Delta)
- return &out
-}
-
-func (pr *PushRequest) CopyMerge(other *PushRequest) *PushRequest {
- if pr == nil {
- return other.Copy()
- }
- if other == nil {
- return pr.Copy()
- }
-
- merged := pr.Copy()
- return merged.Merge(other)
-}
-
-func (pr *PushRequest) IsProxyUpdate() bool {
- return pr.Reason.Has(ProxyUpdate)
-}
-
-func (pr *PushRequest) IsRequest() bool {
- return len(pr.Reason) == 1 && pr.Reason.Has(ProxyRequest)
-}
-
-func (pr *PushRequest) PushReason() string {
- if pr.IsRequest() {
- return " request"
- }
- return ""
-}
-
-func (ps *PushContext) initDefaultExportMaps() {
+func (ps *ConfigSnapshot) initDefaultExportMaps() {
ps.exportToDefaults.service = sets.New[visibility.Instance]()
if ps.Mesh.DefaultServiceExportTo != nil {
for _, e := range ps.Mesh.DefaultServiceExportTo {
@@ -385,7 +135,7 @@ func (ps *PushContext) initDefaultExportMaps() {
}
-func (ps *PushContext) InitContext(env *Environment, oldPushContext *PushContext, pushReq *PushRequest) {
+func (ps *ConfigSnapshot) Initialize(env *Environment, oldSnapshot *ConfigSnapshot, change *ConfigChange) {
ps.initializeMutex.Lock()
defer ps.initializeMutex.Unlock()
if ps.InitDone.Load() {
@@ -396,11 +146,11 @@ func (ps *PushContext) InitContext(env *Environment, oldPushContext *PushContext
ps.initDefaultExportMaps()
- if pushReq == nil || oldPushContext == nil || !oldPushContext.InitDone.Load() || pushReq.Forced {
+ if change == nil || oldSnapshot == nil || !oldSnapshot.InitDone.Load() || change.Forced {
ps.createNewContext(env)
- ps.applyServiceActivationPolicyUpdates(pushReq)
+ ps.applyServiceActivationPolicyUpdates(change)
} else {
- ps.updateContext(env, oldPushContext, pushReq)
+ ps.updateContext(env, oldSnapshot, change)
}
ps.clusterLocalHosts = env.ClusterLocal().GetClusterLocalHosts()
@@ -500,7 +250,7 @@ func resolveServiceAliases(allServices []*Service, configsUpdated sets.Set[Confi
}
}
-func (ps *PushContext) initServiceRegistry(env *Environment, configsUpdate sets.Set[ConfigKey]) {
+func (ps *ConfigSnapshot) initServiceRegistry(env *Environment, configsUpdate sets.Set[ConfigKey]) {
allServices := SortServicesByCreationTime(env.Services())
resolveServiceAliases(allServices, configsUpdate)
@@ -563,8 +313,8 @@ func (ps *PushContext) initServiceRegistry(env *Environment, configsUpdate sets.
ps.initServiceAccounts(env, allServices)
}
-func (ps *PushContext) createNewContext(env *Environment) {
- log.Debug("creating new PushContext (full initialization)")
+func (ps *ConfigSnapshot) createNewContext(env *Environment) {
+ log.Debug("creating configuration snapshot (full initialization)")
ps.initServiceRegistry(env, nil)
// Initialize Kubernetes Gateway API resources if the controller is enabled.
ps.initKubernetesGateways(env)
@@ -576,23 +326,23 @@ func (ps *PushContext) createNewContext(env *Environment) {
ps.initAuthenticationPolicies(env)
}
-func (ps *PushContext) updateContext(env *Environment, oldPushContext *PushContext, pushReq *PushRequest) {
+func (ps *ConfigSnapshot) updateContext(env *Environment, oldSnapshot *ConfigSnapshot, change *ConfigChange) {
// A Service may appear after a policy in the same informer burst. Rebuild
// the registry on the explicit Service event even when the total count was
// already visible through env.Services(); otherwise one HA replica can keep
// an activation RDS snapshot that permanently omits the new backend.
- servicesChanged := pushReq != nil &&
- (len(pushReq.AddressesUpdated) > 0 ||
- HasConfigsOfKind(pushReq.ConfigsUpdated, kind.Service) ||
- HasConfigsOfKind(pushReq.ConfigsUpdated, kind.ServiceEntry))
+ servicesChanged := change != nil &&
+ (len(change.AddressesUpdated) > 0 ||
+ HasConfigsOfKind(change.ConfigsUpdated, kind.Service) ||
+ HasConfigsOfKind(change.ConfigsUpdated, kind.ServiceEntry))
// Also check if the actual number of services has changed
// This handles cases where Kubernetes Services are added/removed without ServiceEntry updates
- if !servicesChanged && oldPushContext != nil {
+ if !servicesChanged && oldSnapshot != nil {
currentServices := env.Services()
// Count services in old ServiceIndex
oldServiceCount := 0
- for _, namespaces := range oldPushContext.ServiceIndex.HostnameAndNamespace {
+ for _, namespaces := range oldSnapshot.ServiceIndex.HostnameAndNamespace {
oldServiceCount += len(namespaces)
}
// If service count differs, services have changed
@@ -603,11 +353,11 @@ func (ps *PushContext) updateContext(env *Environment, oldPushContext *PushConte
if servicesChanged {
// Services have changed. initialize service registry
- ps.initServiceRegistry(env, pushReq.ConfigsUpdated)
+ ps.initServiceRegistry(env, change.ConfigsUpdated)
} else {
// make sure we copy over things that would be generated in initServiceRegistry
- ps.ServiceIndex = oldPushContext.ServiceIndex
- ps.serviceAccounts = oldPushContext.serviceAccounts
+ ps.ServiceIndex = oldSnapshot.ServiceIndex
+ ps.serviceAccounts = oldSnapshot.serviceAccounts
}
// Initialize or reuse Gateway API controller state.
@@ -616,58 +366,58 @@ func (ps *PushContext) updateContext(env *Environment, oldPushContext *PushConte
if servicesChanged {
ps.initKubernetesGateways(env)
} else {
- ps.GatewayAPIController = oldPushContext.GatewayAPIController
+ ps.GatewayAPIController = oldSnapshot.GatewayAPIController
}
- httpRoutesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.HTTPRoute)
+ httpRoutesChanged := change != nil && HasConfigsOfKind(change.ConfigsUpdated, kind.HTTPRoute)
if httpRoutesChanged {
log.Debugf("HTTPRoutes changed, re-initializing HTTPRoute index")
ps.initHTTPRoutes(env)
} else {
log.Debugf("HTTPRoutes unchanged, reusing old HTTPRoute index")
- ps.httpRouteIndex = oldPushContext.httpRouteIndex
+ ps.httpRouteIndex = oldSnapshot.httpRouteIndex
}
- dxgateServicesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.DxgateService)
+ dxgateServicesChanged := change != nil && HasConfigsOfKind(change.ConfigsUpdated, kind.DxgateService)
if dxgateServicesChanged {
ps.initDxgateServices(env)
} else {
- ps.dxgateServiceIndex = oldPushContext.dxgateServiceIndex
+ ps.dxgateServiceIndex = oldSnapshot.dxgateServiceIndex
}
- backendTLSPoliciesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.BackendTLSPolicy)
+ backendTLSPoliciesChanged := change != nil && HasConfigsOfKind(change.ConfigsUpdated, kind.BackendTLSPolicy)
if backendTLSPoliciesChanged {
log.Debugf("BackendTLSPolicies changed, re-initializing BackendTLSPolicy index")
ps.initBackendTLSPolicies(env)
} else {
log.Debugf("BackendTLSPolicies unchanged, reusing old BackendTLSPolicy index")
- ps.backendTLSPolicyIndex = oldPushContext.backendTLSPolicyIndex
+ ps.backendTLSPolicyIndex = oldSnapshot.backendTLSPolicyIndex
}
- faultInjectionPoliciesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.FaultInjectionPolicy)
+ faultInjectionPoliciesChanged := change != nil && HasConfigsOfKind(change.ConfigsUpdated, kind.FaultInjectionPolicy)
if faultInjectionPoliciesChanged {
log.Debugf("FaultInjectionPolicies changed, re-initializing FaultInjectionPolicy index")
ps.initFaultInjectionPolicies(env)
} else {
- ps.faultInjectionIndex = oldPushContext.faultInjectionIndex
+ ps.faultInjectionIndex = oldSnapshot.faultInjectionIndex
}
- ps.serviceActivationIndex = copyServiceActivationPolicyIndex(oldPushContext.serviceActivationIndex)
- ps.applyServiceActivationPolicyUpdates(pushReq)
+ ps.serviceActivationIndex = copyServiceActivationPolicyIndex(oldSnapshot.serviceActivationIndex)
+ ps.applyServiceActivationPolicyUpdates(change)
- authnPoliciesChanged := pushReq != nil && (pushReq.Full || authPolicyKindsChanged(pushReq.ConfigsUpdated))
- if authnPoliciesChanged || oldPushContext == nil || oldPushContext.AuthenticationPolicies == nil {
+ authnPoliciesChanged := change != nil && (change.Full || authPolicyKindsChanged(change.ConfigsUpdated))
+ if authnPoliciesChanged || oldSnapshot == nil || oldSnapshot.AuthenticationPolicies == nil {
log.Debugf("security authentication policy changed (full=%v, configsUpdatedContainingSecurityPolicy=%v), rebuilding authentication policies",
- pushReq != nil && pushReq.Full, func() bool {
- if pushReq == nil {
+ change != nil && change.Full, func() bool {
+ if change == nil {
return false
}
- return authPolicyKindsChanged(pushReq.ConfigsUpdated)
+ return authPolicyKindsChanged(change.ConfigsUpdated)
}())
ps.initAuthenticationPolicies(env)
} else {
- ps.AuthenticationPolicies = oldPushContext.AuthenticationPolicies
+ ps.AuthenticationPolicies = oldSnapshot.AuthenticationPolicies
}
}
@@ -677,7 +427,7 @@ func authPolicyKindsChanged(configs sets.Set[ConfigKey]) bool {
HasConfigsOfKind(configs, kind.AuthorizationPolicy)
}
-func (ps *PushContext) initAuthenticationPolicies(env *Environment) {
+func (ps *ConfigSnapshot) initAuthenticationPolicies(env *Environment) {
if env == nil {
ps.AuthenticationPolicies = nil
return
@@ -685,7 +435,7 @@ func (ps *PushContext) initAuthenticationPolicies(env *Environment) {
ps.AuthenticationPolicies = initAuthenticationPolicies(env)
}
-func (ps *PushContext) InboundMTLSModeForProxy(proxy *Proxy, port uint32) MutualTLSMode {
+func (ps *ConfigSnapshot) InboundMTLSModeForProxy(proxy *Proxy, port uint32) MutualTLSMode {
if ps == nil || proxy == nil || ps.AuthenticationPolicies == nil {
return MTLSUnknown
}
@@ -699,14 +449,14 @@ func (ps *PushContext) InboundMTLSModeForProxy(proxy *Proxy, port uint32) Mutual
return ps.AuthenticationPolicies.EffectiveMutualTLSMode(namespace, nil, port)
}
-func (ps *PushContext) RequestAuthenticationsForWorkload(namespace string, workloadLabels map[string]string) []config.Config {
+func (ps *ConfigSnapshot) RequestAuthenticationsForWorkload(namespace string, workloadLabels map[string]string) []config.Config {
if ps == nil || ps.AuthenticationPolicies == nil {
return nil
}
return ps.AuthenticationPolicies.RequestAuthenticationsForWorkload(namespace, workloadLabels)
}
-func (ps *PushContext) AuthorizationPoliciesForWorkload(namespace string, workloadLabels map[string]string) []config.Config {
+func (ps *ConfigSnapshot) AuthorizationPoliciesForWorkload(namespace string, workloadLabels map[string]string) []config.Config {
if ps == nil || ps.AuthenticationPolicies == nil {
return nil
}
@@ -716,7 +466,7 @@ func (ps *PushContext) AuthorizationPoliciesForWorkload(namespace string, worklo
// initKubernetesGateways initializes Kubernetes Gateway API objects by delegating
// to the GatewayAPIController, if it is present in the Environment.
// This closely follows Dubbo's initKubernetesGateways behavior.
-func (ps *PushContext) initKubernetesGateways(env *Environment) {
+func (ps *ConfigSnapshot) initKubernetesGateways(env *Environment) {
if env == nil || env.GatewayAPIController == nil {
return
}
@@ -724,7 +474,7 @@ func (ps *PushContext) initKubernetesGateways(env *Environment) {
env.GatewayAPIController.Reconcile(ps)
}
-func (ps *PushContext) ServiceForHostname(proxy *Proxy, hostname host.Name) *Service {
+func (ps *ConfigSnapshot) ServiceForHostname(proxy *Proxy, hostname host.Name) *Service {
for _, service := range ps.ServiceIndex.HostnameAndNamespace[hostname] {
return service
}
@@ -733,19 +483,7 @@ func (ps *PushContext) ServiceForHostname(proxy *Proxy, hostname host.Name) *Ser
return nil
}
-func (ps *PushContext) UpdateMetrics() {
- ps.proxyStatusMutex.RLock()
- defer ps.proxyStatusMutex.RUnlock()
-}
-
-func (ps *PushContext) OnConfigChange() {
- LastPushMutex.Lock()
- LastPushStatus = ps
- LastPushMutex.Unlock()
- ps.UpdateMetrics()
-}
-
-func (ps *PushContext) servicesExportedToNamespace(ns string) []*Service {
+func (ps *ConfigSnapshot) servicesExportedToNamespace(ns string) []*Service {
var out []*Service
// First add private services and explicitly exportedTo services
@@ -767,11 +505,11 @@ func (ps *PushContext) servicesExportedToNamespace(ns string) []*Service {
return out
}
-func (ps *PushContext) GetAllServices() []*Service {
+func (ps *ConfigSnapshot) GetAllServices() []*Service {
return ps.servicesExportedToNamespace(NamespaceAll)
}
-func (ps *PushContext) initHTTPRoutes(env *Environment) {
+func (ps *ConfigSnapshot) initHTTPRoutes(env *Environment) {
log.Debugf("starting HTTPRoute initialization")
httproutes := env.List(gvk.HTTPRoute, NamespaceAll)
log.Debugf("found %d HTTPRoute configs", len(httproutes))
@@ -815,7 +553,7 @@ func (ps *PushContext) initHTTPRoutes(env *Environment) {
}
}
-func (ps *PushContext) initDxgateServices(env *Environment) {
+func (ps *ConfigSnapshot) initDxgateServices(env *Environment) {
services := sortConfigByCreationTime(env.List(gvk.DxgateService, NamespaceAll))
index := make(map[string]map[string]config.Config)
for _, service := range services {
@@ -828,7 +566,7 @@ func (ps *PushContext) initDxgateServices(env *Environment) {
}
// DxgateService resolves the mesh-native backend named by an HTTPRoute.
-func (ps *PushContext) DxgateService(namespace, name string) (config.Config, bool) {
+func (ps *ConfigSnapshot) DxgateService(namespace, name string) (config.Config, bool) {
if ps == nil {
return config.Config{}, false
}
@@ -841,7 +579,7 @@ func (ps *PushContext) DxgateService(namespace, name string) (config.Config, boo
}
// HTTPRouteForHost returns HTTPRoutes that match the given hostname.
-func (ps *PushContext) HTTPRouteForHost(hostname host.Name) []config.Config {
+func (ps *ConfigSnapshot) HTTPRouteForHost(hostname host.Name) []config.Config {
var routes []config.Config
hostStr := string(hostname)
@@ -893,7 +631,7 @@ func (ps *PushContext) HTTPRouteForHost(hostname host.Name) []config.Config {
return routes
}
-func (ps *PushContext) initBackendTLSPolicies(env *Environment) {
+func (ps *ConfigSnapshot) initBackendTLSPolicies(env *Environment) {
policies := sortConfigByCreationTime(env.List(gvk.BackendTLSPolicy, NamespaceAll))
serviceTLS := map[string]BackendTLSSettings{}
for _, cfg := range policies {
@@ -920,7 +658,7 @@ func (ps *PushContext) initBackendTLSPolicies(env *Environment) {
log.Debugf("indexed BackendTLSPolicies for %d services", len(serviceTLS))
}
-func (ps *PushContext) BackendTLSForService(namespace, name string) (BackendTLSSettings, bool) {
+func (ps *ConfigSnapshot) BackendTLSForService(namespace, name string) (BackendTLSSettings, bool) {
if ps == nil || ps.backendTLSPolicyIndex.serviceTLS == nil {
return BackendTLSSettings{}, false
}
@@ -939,7 +677,7 @@ type faultInjectionPolicyIndex struct {
serviceFaults map[string]FaultInjectionSettings
}
-func (ps *PushContext) initFaultInjectionPolicies(env *Environment) {
+func (ps *ConfigSnapshot) initFaultInjectionPolicies(env *Environment) {
policies := sortConfigByCreationTime(env.List(gvk.FaultInjectionPolicy, NamespaceAll))
serviceFaults := map[string]FaultInjectionSettings{}
for _, cfg := range policies {
@@ -975,7 +713,7 @@ func (ps *PushContext) initFaultInjectionPolicies(env *Environment) {
log.Debugf("indexed FaultInjectionPolicies for %d service targets", len(serviceFaults))
}
-func (ps *PushContext) FaultInjectionForService(namespace, name, portName string) (FaultInjectionSettings, bool) {
+func (ps *ConfigSnapshot) FaultInjectionForService(namespace, name, portName string) (FaultInjectionSettings, bool) {
if ps == nil || ps.faultInjectionIndex.serviceFaults == nil {
return FaultInjectionSettings{}, false
}
@@ -1031,14 +769,14 @@ func backendTLSPolicyServiceKey(namespace, name string) string {
// ServiceAccounts returns the SPIFFE identities associated with the workloads
// backing the given service hostname in the given namespace. The returned list
// is sorted and already expanded with trust domain aliases.
-func (ps *PushContext) ServiceAccounts(hostname host.Name, namespace string) []string {
+func (ps *ConfigSnapshot) ServiceAccounts(hostname host.Name, namespace string) []string {
return ps.serviceAccounts[serviceAccountKey{hostname: hostname, namespace: namespace}]
}
// ServiceActivationEnabled reports whether a structurally valid policy targets
// this Service. The live ActivatorReady condition cannot gate routing: the first
// cold request is what makes an Activator report demand for this target.
-func (ps *PushContext) ServiceActivationEnabled(namespace, name string) bool {
+func (ps *ConfigSnapshot) ServiceActivationEnabled(namespace, name string) bool {
if ps == nil {
return false
}
@@ -1047,7 +785,7 @@ func (ps *PushContext) ServiceActivationEnabled(namespace, name string) bool {
}
// ActivationGatewayService returns the dedicated namespace-local Activator.
-func (ps *PushContext) ActivationGatewayService(namespace string) *Service {
+func (ps *ConfigSnapshot) ActivationGatewayService(namespace string) *Service {
if ps == nil {
return nil
}
@@ -1062,7 +800,7 @@ func (ps *PushContext) ActivationGatewayService(namespace string) *Service {
// ActivationGatewaySANs is stable across cold/hot EDS transitions. Keeping the
// backend and Activator identities in one CDS validation context avoids a TLS
// verification window while the endpoint assignment changes.
-func (ps *PushContext) ActivationGatewaySANs(namespace string) []string {
+func (ps *ConfigSnapshot) ActivationGatewaySANs(namespace string) []string {
if ps == nil || ps.Mesh == nil || namespace == "" {
return nil
}
@@ -1073,7 +811,7 @@ func (ps *PushContext) ActivationGatewaySANs(namespace string) []string {
// ActivationBackendSANs returns the backend identities declared by the policy.
// Unlike endpoint-derived identities, these remain available when dubbod starts
// while the target has zero endpoints.
-func (ps *PushContext) ActivationBackendSANs(namespace, name string) []string {
+func (ps *ConfigSnapshot) ActivationBackendSANs(namespace, name string) []string {
if ps == nil || ps.Mesh == nil || namespace == "" {
return nil
}
@@ -1099,7 +837,7 @@ func (ps *PushContext) ActivationBackendSANs(namespace, name string) []string {
return sets.SortedList(identities)
}
-func (ps *PushContext) ActivatedServices(namespace string) []*Service {
+func (ps *ConfigSnapshot) ActivatedServices(namespace string) []*Service {
if ps == nil {
return nil
}
@@ -1113,7 +851,7 @@ func (ps *PushContext) ActivatedServices(namespace string) []*Service {
return SortServicesByCreationTime(out)
}
-func (ps *PushContext) initServiceActivationPolicies(env *Environment) {
+func (ps *ConfigSnapshot) initServiceActivationPolicies(env *Environment) {
ps.serviceActivationIndex = serviceActivationPolicyIndex{services: map[string][]string{}}
if env == nil {
return
@@ -1133,11 +871,11 @@ func copyServiceActivationPolicyIndex(in serviceActivationPolicyIndex) serviceAc
return out
}
-func (ps *PushContext) applyServiceActivationPolicyUpdates(pushReq *PushRequest) {
- if pushReq == nil {
+func (ps *ConfigSnapshot) applyServiceActivationPolicyUpdates(change *ConfigChange) {
+ if change == nil {
return
}
- for _, update := range pushReq.ServiceActivationPolicyUpdates {
+ for _, update := range change.ServiceActivationPolicyUpdates {
if update.Previous.Spec != nil {
ps.deleteServiceActivationPolicy(update.Previous)
}
@@ -1149,7 +887,7 @@ func (ps *PushContext) applyServiceActivationPolicyUpdates(pushReq *PushRequest)
}
}
-func (ps *PushContext) deleteServiceActivationPolicy(cfg config.Config) {
+func (ps *ConfigSnapshot) deleteServiceActivationPolicy(cfg config.Config) {
spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
if ok && spec != nil && spec.GetTargetRef() != nil {
delete(ps.serviceActivationIndex.services,
@@ -1157,7 +895,7 @@ func (ps *PushContext) deleteServiceActivationPolicy(cfg config.Config) {
}
}
-func (ps *PushContext) upsertServiceActivationPolicy(cfg config.Config) {
+func (ps *ConfigSnapshot) upsertServiceActivationPolicy(cfg config.Config) {
spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
if !ok || spec == nil || spec.GetTargetRef() == nil || spec.GetAutoscalerRef() == nil {
return
@@ -1174,7 +912,7 @@ func (ps *PushContext) upsertServiceActivationPolicy(cfg config.Config) {
append([]string(nil), spec.GetBackendServiceAccounts()...)
}
-func (ps *PushContext) initServiceAccounts(env *Environment, services []*Service) {
+func (ps *ConfigSnapshot) initServiceAccounts(env *Environment, services []*Service) {
for _, svc := range services {
var accounts sets.String
// First get endpoint level service accounts
@@ -1199,15 +937,3 @@ func (ps *PushContext) initServiceAccounts(env *Environment, services []*Service
ps.serviceAccounts[key] = sa
}
}
-
-func (ps *PushContext) StatusJSON() ([]byte, error) {
- if ps == nil {
- return []byte{'{', '}'}, nil
- }
- ps.proxyStatusMutex.RLock()
- defer ps.proxyStatusMutex.RUnlock()
- if len(ps.ProxyStatus) == 0 {
- return []byte{'{', '}'}, nil
- }
- return json.MarshalIndent(ps.ProxyStatus, "", " ")
-}
diff --git a/dubbod/discovery/pkg/model/config_state.go b/dubbod/discovery/pkg/model/config_state.go
new file mode 100644
index 00000000..1df66999
--- /dev/null
+++ b/dubbod/discovery/pkg/model/config_state.go
@@ -0,0 +1,65 @@
+//
+// 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 model
+
+import "github.com/apache/dubbo-kubernetes/pkg/util/sets"
+
+// ConfigChange describes source changes without carrying an xDS connection or delivery state.
+type ConfigChange struct {
+ ConfigsUpdated sets.Set[ConfigKey]
+ ServiceActivationPolicyUpdates map[ConfigKey]ServiceActivationPolicyUpdate
+ AddressesUpdated sets.String
+ Full bool
+ Forced bool
+ EndpointsChanged bool
+ Global bool
+}
+
+func (e *Environment) ConfigSnapshot() *ConfigSnapshot {
+ e.mutex.RLock()
+ defer e.mutex.RUnlock()
+ return e.configSnapshot
+}
+
+// ReconcileConfig publishes one completed snapshot before notifying its consumers.
+func (e *Environment) ReconcileConfig(change *ConfigChange) *ConfigSnapshot {
+ e.configMutex.Lock()
+ defer e.configMutex.Unlock()
+ current := NewConfigSnapshot()
+ current.Initialize(e, e.ConfigSnapshot(), change)
+ e.mutex.Lock()
+ e.configSnapshot = current
+ e.mutex.Unlock()
+ e.NotifyConfigChange(change)
+ return current
+}
+
+func (e *Environment) AddConfigHandler(handler func(*ConfigChange)) {
+ e.mutex.Lock()
+ defer e.mutex.Unlock()
+ e.configHandlers = append(e.configHandlers, handler)
+}
+
+// NotifyConfigChange also covers endpoint changes that reuse the policy snapshot.
+func (e *Environment) NotifyConfigChange(change *ConfigChange) {
+ e.mutex.RLock()
+ handlers := append([]func(*ConfigChange){}, e.configHandlers...)
+ e.mutex.RUnlock()
+ for _, handler := range handlers {
+ handler(change)
+ }
+}
diff --git a/dubbod/discovery/pkg/model/context.go b/dubbod/discovery/pkg/model/context.go
index 5d56fd22..dc94fee5 100644
--- a/dubbod/discovery/pkg/model/context.go
+++ b/dubbod/discovery/pkg/model/context.go
@@ -91,6 +91,9 @@ type Environment struct {
ConfigStore
mutex sync.RWMutex
pushContext *PushContext
+ configSnapshot *ConfigSnapshot
+ configMutex sync.Mutex
+ configHandlers []func(*ConfigChange)
clusterLocalServices ClusterLocalProvider
DomainSuffix string
EndpointIndex *EndpointIndex
@@ -100,7 +103,7 @@ type Environment struct {
type GatewayController interface {
ConfigStoreController
- Reconcile(ctx *PushContext)
+ Reconcile(ctx *ConfigSnapshot)
SecretAllowed(ourKind config.GroupVersionKind, resourceName string, namespace string) bool
}
@@ -111,10 +114,12 @@ func NewEnvironment() *Environment {
} else {
cache = DisabledCache{}
}
+ push := NewPushContext()
return &Environment{
- pushContext: NewPushContext(),
- Cache: cache,
- EndpointIndex: NewEndpointIndex(cache),
+ pushContext: push,
+ configSnapshot: push.ConfigSnapshot,
+ Cache: cache,
+ EndpointIndex: NewEndpointIndex(cache),
}
}
@@ -148,6 +153,7 @@ func (e *Environment) SetPushContext(pc *PushContext) {
e.mutex.Lock()
defer e.mutex.Unlock()
e.pushContext = pc
+ e.configSnapshot = pc.ConfigSnapshot
}
func (e *Environment) Mesh() *meshv1alpha1.MeshConfig {
@@ -196,20 +202,19 @@ func (e *Environment) Init() {
type Proxy struct {
sync.RWMutex
- XdsResourceGenerator XdsResourceGenerator
- LastPushContext *PushContext
- LastPushTime time.Time
- Type NodeType
- WatchedResources map[string]*WatchedResource
- ID string
- DNSDomain string
- Metadata *NodeMetadata
- IPAddresses []string
- XdsNode *core.Node
- ConfigNamespace string
- ServiceTargets []ServiceTarget
- ipMode IPMode
- GlobalUnicastIP string
+ LastPushContext *PushContext
+ LastPushTime time.Time
+ Type NodeType
+ WatchedResources map[string]*WatchedResource
+ ID string
+ DNSDomain string
+ Metadata *NodeMetadata
+ IPAddresses []string
+ XdsNode *core.Node
+ ConfigNamespace string
+ ServiceTargets []ServiceTarget
+ ipMode IPMode
+ GlobalUnicastIP string
}
func (node *Proxy) GetWatchedResource(typeURL string) *WatchedResource {
diff --git a/dubbod/discovery/pkg/model/push_context.go b/dubbod/discovery/pkg/model/push_context.go
index 3ad42cc3..7cb5f9d0 100644
--- a/dubbod/discovery/pkg/model/push_context.go
+++ b/dubbod/discovery/pkg/model/push_context.go
@@ -17,29 +17,15 @@
package model
import (
- "cmp"
"encoding/json"
- "strings"
"sync"
"time"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
- networking "github.com/kdubbo/api/networking/v1alpha3"
- sigsk8siogatewayapiapisv1 "sigs.k8s.io/gateway-api/apis/v1"
-
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/serviceregistry/provider"
"github.com/apache/dubbo-kubernetes/pkg/cluster"
"github.com/apache/dubbo-kubernetes/pkg/config"
- "github.com/apache/dubbo-kubernetes/pkg/config/host"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
- "github.com/apache/dubbo-kubernetes/pkg/config/visibility"
- "github.com/apache/dubbo-kubernetes/pkg/slices"
- "github.com/apache/dubbo-kubernetes/pkg/spiffe"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/apache/dubbo-kubernetes/pkg/xds"
- meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
- "go.uber.org/atomic"
- "google.golang.org/protobuf/types/known/wrapperspb"
)
type TriggerReason string
@@ -61,26 +47,10 @@ var (
)
type PushContext struct {
- Mesh *meshv1alpha1.MeshConfig `json:"-"`
- initializeMutex sync.Mutex
- InitDone atomic.Bool
- // GatewayAPIController holds a reference to the Gateway API controller.
- // When enabled, this controller is responsible for translating Kubernetes
- // Gateway API resources into internal Dubbo resources during push.
- GatewayAPIController GatewayController
- clusterLocalHosts ClusterLocalHosts
- exportToDefaults exportToDefaults
- ServiceIndex serviceIndex
- httpRouteIndex httpRouteIndex
- dxgateServiceIndex dxgateServiceIndex
- backendTLSPolicyIndex backendTLSPolicyIndex
- faultInjectionIndex faultInjectionPolicyIndex
- serviceActivationIndex serviceActivationPolicyIndex
- serviceAccounts map[serviceAccountKey][]string
- AuthenticationPolicies *AuthenticationPolicies
- PushVersion string
- ProxyStatus map[string]map[string]ProxyPushStatus
- proxyStatusMutex sync.RWMutex
+ *ConfigSnapshot
+ PushVersion string
+ ProxyStatus map[string]map[string]ProxyPushStatus
+ proxyStatusMutex sync.RWMutex
}
type PushRequest struct {
@@ -114,19 +84,6 @@ type ProxyPushStatus struct {
Message string `json:"message,omitempty"`
}
-type serviceAccountKey struct {
- hostname host.Name
- namespace string
-}
-
-type serviceIndex struct {
- privateByNamespace map[string][]*Service
- public []*Service
- exportedToNamespace map[string][]*Service
- HostnameAndNamespace map[host.Name]map[string]*Service `json:"-"`
- instancesByPort map[string]map[int][]*DubboEndpoint
-}
-
type ReasonStats map[TriggerReason]int
type ResourceDelta = xds.ResourceDelta
@@ -137,54 +94,8 @@ type ConfigKey struct {
Namespace string
}
-type exportToDefaults struct {
- service sets.Set[visibility.Instance]
-}
-
-type httpRouteIndex struct {
- // hostToRoutes keeps the Gateway API HTTPRoutes keyed by hostname
- hostToRoutes map[host.Name][]config.Config
-}
-
-type dxgateServiceIndex struct {
- byNamespace map[string]map[string]config.Config
-}
-
-type backendTLSPolicyIndex struct {
- serviceTLS map[string]BackendTLSSettings
-}
-
-type BackendTLSSettings struct {
- SNI string
-}
-
-const ActivationGatewayServiceName = "dxgate-gateway"
-
-type serviceActivationPolicyIndex struct {
- services map[string][]string
-}
-
func NewPushContext() *PushContext {
- return &PushContext{
- ServiceIndex: newServiceIndex(),
- dxgateServiceIndex: dxgateServiceIndex{byNamespace: map[string]map[string]config.Config{}},
- backendTLSPolicyIndex: backendTLSPolicyIndex{serviceTLS: map[string]BackendTLSSettings{}},
- serviceActivationIndex: serviceActivationPolicyIndex{
- services: map[string][]string{},
- },
- serviceAccounts: map[serviceAccountKey][]string{},
- ProxyStatus: map[string]map[string]ProxyPushStatus{},
- }
-}
-
-func newServiceIndex() serviceIndex {
- return serviceIndex{
- public: []*Service{},
- privateByNamespace: map[string][]*Service{},
- exportedToNamespace: map[string][]*Service{},
- HostnameAndNamespace: map[host.Name]map[string]*Service{},
- instancesByPort: map[string]map[int][]*DubboEndpoint{},
- }
+ return &PushContext{ConfigSnapshot: NewConfigSnapshot(), ProxyStatus: map[string]map[string]ProxyPushStatus{}}
}
func NewReasonStats(reasons ...TriggerReason) ReasonStats {
@@ -373,366 +284,6 @@ func (pr *PushRequest) PushReason() string {
return ""
}
-func (ps *PushContext) initDefaultExportMaps() {
- ps.exportToDefaults.service = sets.New[visibility.Instance]()
- if ps.Mesh.DefaultServiceExportTo != nil {
- for _, e := range ps.Mesh.DefaultServiceExportTo {
- ps.exportToDefaults.service.Insert(visibility.Instance(e))
- }
- } else {
- ps.exportToDefaults.service.Insert(visibility.Public)
- }
-
-}
-
-func (ps *PushContext) InitContext(env *Environment, oldPushContext *PushContext, pushReq *PushRequest) {
- ps.initializeMutex.Lock()
- defer ps.initializeMutex.Unlock()
- if ps.InitDone.Load() {
- return
- }
-
- ps.Mesh = env.Mesh()
-
- ps.initDefaultExportMaps()
-
- if pushReq == nil || oldPushContext == nil || !oldPushContext.InitDone.Load() || pushReq.Forced {
- ps.createNewContext(env)
- ps.applyServiceActivationPolicyUpdates(pushReq)
- } else {
- ps.updateContext(env, oldPushContext, pushReq)
- }
-
- ps.clusterLocalHosts = env.ClusterLocal().GetClusterLocalHosts()
-
- ps.InitDone.Store(true)
-}
-
-func SortServicesByCreationTime(services []*Service) []*Service {
- slices.SortStableFunc(services, func(i, j *Service) int {
- if r := i.CreationTime.Compare(j.CreationTime); r != 0 {
- return r
- }
- if r := cmp.Compare(i.Attributes.Name, j.Attributes.Name); r != 0 {
- return r
- }
- return cmp.Compare(i.Attributes.Namespace, j.Attributes.Namespace)
- })
- return services
-}
-
-func resolveServiceAliases(allServices []*Service, configsUpdated sets.Set[ConfigKey]) {
- rawAlias := map[NamespacedHostname]host.Name{}
- for _, s := range allServices {
- if s.Resolution != Alias {
- continue
- }
- nh := NamespacedHostname{
- Hostname: s.Hostname,
- Namespace: s.Attributes.Namespace,
- }
- rawAlias[nh] = host.Name(s.Attributes.ExternalName)
- }
-
- unnamespacedRawAlias := make(map[host.Name]host.Name, len(rawAlias))
- for k, v := range rawAlias {
- unnamespacedRawAlias[k.Hostname] = v
- }
-
- resolvedAliases := make(map[NamespacedHostname]host.Name, len(rawAlias))
- for alias, referencedService := range rawAlias {
- if _, f := unnamespacedRawAlias[referencedService]; !f {
- // Common case: alias pointing to a concrete service
- resolvedAliases[alias] = referencedService
- continue
- }
- seen := sets.New(alias.Hostname, referencedService)
- for {
- n, f := unnamespacedRawAlias[referencedService]
- if !f {
- resolvedAliases[alias] = referencedService
- break
- }
- if seen.InsertContains(n) {
- break
- }
- referencedService = n
- }
- }
-
- aliasesForService := map[host.Name][]NamespacedHostname{}
- for alias, concrete := range resolvedAliases {
- aliasesForService[concrete] = append(aliasesForService[concrete], alias)
-
- aliasKey := ConfigKey{
- // Kind: kind.ServiceEntry,
- // Name: alias.Hostname.String(),
- // Namespace: alias.Namespace,
- }
- if configsUpdated.Contains(aliasKey) {
- for _, svc := range allServices {
- if svc.Hostname == concrete {
- configsUpdated.Insert(ConfigKey{
- // Kind: kind.ServiceEntry,
- // Name: concrete.String(),
- // Namespace: svc.Attributes.Namespace,
- })
- }
- }
- }
- }
- for _, v := range aliasesForService {
- slices.SortFunc(v, func(a, b NamespacedHostname) int {
- if r := cmp.Compare(a.Namespace, b.Namespace); r != 0 {
- return r
- }
- return cmp.Compare(a.Hostname, b.Hostname)
- })
- }
-
- for i, s := range allServices {
- if aliases, f := aliasesForService[s.Hostname]; f {
- // This service has an alias; set it. We need to make a copy since the underlying Service is shared
- s = s.DeepCopy()
- s.Attributes.Aliases = aliases
- allServices[i] = s
- }
- }
-}
-
-func (ps *PushContext) initServiceRegistry(env *Environment, configsUpdate sets.Set[ConfigKey]) {
- allServices := SortServicesByCreationTime(env.Services())
- resolveServiceAliases(allServices, configsUpdate)
-
- for _, s := range allServices {
- portMap := map[string]int{}
- ports := sets.New[int]()
- for _, port := range s.Ports {
- portMap[port.Name] = port.Port
- ports.Insert(port.Port)
- }
-
- svcKey := s.Key()
- if _, ok := ps.ServiceIndex.instancesByPort[svcKey]; !ok {
- ps.ServiceIndex.instancesByPort[svcKey] = make(map[int][]*DubboEndpoint)
- }
- shards, ok := env.EndpointIndex.ShardsForService(string(s.Hostname), s.Attributes.Namespace)
- if ok {
- instancesByPort := shards.CopyEndpoints(portMap, ports)
- // Iterate over the instances and add them to the service index to avoid overriding the existing port instances.
- for port, instances := range instancesByPort {
- ps.ServiceIndex.instancesByPort[svcKey][port] = instances
- }
- }
- if _, f := ps.ServiceIndex.HostnameAndNamespace[s.Hostname]; !f {
- ps.ServiceIndex.HostnameAndNamespace[s.Hostname] = map[string]*Service{}
- }
- if existing := ps.ServiceIndex.HostnameAndNamespace[s.Hostname][s.Attributes.Namespace]; existing != nil &&
- (existing.Attributes.ServiceRegistry == provider.Kubernetes || s.Attributes.ServiceRegistry != provider.Kubernetes) {
- log.Debugf("Service %s/%s from registry %s ignored by %s/%s/%s", s.Attributes.Namespace, s.Hostname, s.Attributes.ServiceRegistry,
- existing.Attributes.ServiceRegistry, existing.Attributes.Namespace, existing.Hostname)
- } else {
- ps.ServiceIndex.HostnameAndNamespace[s.Hostname][s.Attributes.Namespace] = s
- }
-
- ns := s.Attributes.Namespace
- if s.Attributes.ExportTo.IsEmpty() {
- if ps.exportToDefaults.service.Contains(visibility.Private) {
- ps.ServiceIndex.privateByNamespace[ns] = append(ps.ServiceIndex.privateByNamespace[ns], s)
- } else if ps.exportToDefaults.service.Contains(visibility.Public) {
- ps.ServiceIndex.public = append(ps.ServiceIndex.public, s)
- }
- } else {
- if s.Attributes.ExportTo.Contains(visibility.Public) {
- ps.ServiceIndex.public = append(ps.ServiceIndex.public, s)
- continue
- } else if s.Attributes.ExportTo.Contains(visibility.None) {
- continue
- }
- // . or other namespaces
- for exportTo := range s.Attributes.ExportTo {
- if exportTo == visibility.Private || string(exportTo) == ns {
- ps.ServiceIndex.privateByNamespace[ns] = append(ps.ServiceIndex.privateByNamespace[ns], s)
- } else {
- ps.ServiceIndex.exportedToNamespace[string(exportTo)] = append(ps.ServiceIndex.exportedToNamespace[string(exportTo)], s)
- }
- }
- }
- }
-
- ps.initServiceAccounts(env, allServices)
-}
-
-func (ps *PushContext) createNewContext(env *Environment) {
- log.Debug("creating new PushContext (full initialization)")
- ps.initServiceRegistry(env, nil)
- // Initialize Kubernetes Gateway API resources if the controller is enabled.
- ps.initKubernetesGateways(env)
- ps.initHTTPRoutes(env)
- ps.initDxgateServices(env)
- ps.initBackendTLSPolicies(env)
- ps.initFaultInjectionPolicies(env)
- ps.initServiceActivationPolicies(env)
- ps.initAuthenticationPolicies(env)
-}
-
-func (ps *PushContext) updateContext(env *Environment, oldPushContext *PushContext, pushReq *PushRequest) {
- // A Service may appear after a policy in the same informer burst. Rebuild
- // the registry on the explicit Service event even when the total count was
- // already visible through env.Services(); otherwise one HA replica can keep
- // an activation RDS snapshot that permanently omits the new backend.
- servicesChanged := pushReq != nil &&
- (len(pushReq.AddressesUpdated) > 0 ||
- HasConfigsOfKind(pushReq.ConfigsUpdated, kind.Service) ||
- HasConfigsOfKind(pushReq.ConfigsUpdated, kind.ServiceEntry))
-
- // Also check if the actual number of services has changed
- // This handles cases where Kubernetes Services are added/removed without ServiceEntry updates
- if !servicesChanged && oldPushContext != nil {
- currentServices := env.Services()
- // Count services in old ServiceIndex
- oldServiceCount := 0
- for _, namespaces := range oldPushContext.ServiceIndex.HostnameAndNamespace {
- oldServiceCount += len(namespaces)
- }
- // If service count differs, services have changed
- if len(currentServices) != oldServiceCount {
- servicesChanged = true
- }
- }
-
- if servicesChanged {
- // Services have changed. initialize service registry
- ps.initServiceRegistry(env, pushReq.ConfigsUpdated)
- } else {
- // make sure we copy over things that would be generated in initServiceRegistry
- ps.ServiceIndex = oldPushContext.ServiceIndex
- ps.serviceAccounts = oldPushContext.serviceAccounts
- }
-
- // Initialize or reuse Gateway API controller state.
- // Gateway status and derived configs depend on services, so recompute
- // if services have changed, otherwise carry over the previous controller.
- if servicesChanged {
- ps.initKubernetesGateways(env)
- } else {
- ps.GatewayAPIController = oldPushContext.GatewayAPIController
- }
-
- httpRoutesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.HTTPRoute)
-
- if httpRoutesChanged {
- log.Debugf("HTTPRoutes changed, re-initializing HTTPRoute index")
- ps.initHTTPRoutes(env)
- } else {
- log.Debugf("HTTPRoutes unchanged, reusing old HTTPRoute index")
- ps.httpRouteIndex = oldPushContext.httpRouteIndex
- }
-
- dxgateServicesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.DxgateService)
- if dxgateServicesChanged {
- ps.initDxgateServices(env)
- } else {
- ps.dxgateServiceIndex = oldPushContext.dxgateServiceIndex
- }
-
- backendTLSPoliciesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.BackendTLSPolicy)
- if backendTLSPoliciesChanged {
- log.Debugf("BackendTLSPolicies changed, re-initializing BackendTLSPolicy index")
- ps.initBackendTLSPolicies(env)
- } else {
- log.Debugf("BackendTLSPolicies unchanged, reusing old BackendTLSPolicy index")
- ps.backendTLSPolicyIndex = oldPushContext.backendTLSPolicyIndex
- }
-
- faultInjectionPoliciesChanged := pushReq != nil && HasConfigsOfKind(pushReq.ConfigsUpdated, kind.FaultInjectionPolicy)
- if faultInjectionPoliciesChanged {
- log.Debugf("FaultInjectionPolicies changed, re-initializing FaultInjectionPolicy index")
- ps.initFaultInjectionPolicies(env)
- } else {
- ps.faultInjectionIndex = oldPushContext.faultInjectionIndex
- }
-
- ps.serviceActivationIndex = copyServiceActivationPolicyIndex(oldPushContext.serviceActivationIndex)
- ps.applyServiceActivationPolicyUpdates(pushReq)
-
- authnPoliciesChanged := pushReq != nil && (pushReq.Full || authPolicyKindsChanged(pushReq.ConfigsUpdated))
- if authnPoliciesChanged || oldPushContext == nil || oldPushContext.AuthenticationPolicies == nil {
- log.Debugf("security authentication policy changed (full=%v, configsUpdatedContainingSecurityPolicy=%v), rebuilding authentication policies",
- pushReq != nil && pushReq.Full, func() bool {
- if pushReq == nil {
- return false
- }
- return authPolicyKindsChanged(pushReq.ConfigsUpdated)
- }())
- ps.initAuthenticationPolicies(env)
- } else {
- ps.AuthenticationPolicies = oldPushContext.AuthenticationPolicies
- }
-}
-
-func authPolicyKindsChanged(configs sets.Set[ConfigKey]) bool {
- return HasConfigsOfKind(configs, kind.PeerAuthentication) ||
- HasConfigsOfKind(configs, kind.RequestAuthentication) ||
- HasConfigsOfKind(configs, kind.AuthorizationPolicy)
-}
-
-func (ps *PushContext) initAuthenticationPolicies(env *Environment) {
- if env == nil {
- ps.AuthenticationPolicies = nil
- return
- }
- ps.AuthenticationPolicies = initAuthenticationPolicies(env)
-}
-
-func (ps *PushContext) InboundMTLSModeForProxy(proxy *Proxy, port uint32) MutualTLSMode {
- if ps == nil || proxy == nil || ps.AuthenticationPolicies == nil {
- return MTLSUnknown
- }
- var namespace string
- if proxy.Metadata != nil {
- namespace = proxy.Metadata.Namespace
- }
- if namespace == "" {
- namespace = proxy.ConfigNamespace
- }
- return ps.AuthenticationPolicies.EffectiveMutualTLSMode(namespace, nil, port)
-}
-
-func (ps *PushContext) RequestAuthenticationsForWorkload(namespace string, workloadLabels map[string]string) []config.Config {
- if ps == nil || ps.AuthenticationPolicies == nil {
- return nil
- }
- return ps.AuthenticationPolicies.RequestAuthenticationsForWorkload(namespace, workloadLabels)
-}
-
-func (ps *PushContext) AuthorizationPoliciesForWorkload(namespace string, workloadLabels map[string]string) []config.Config {
- if ps == nil || ps.AuthenticationPolicies == nil {
- return nil
- }
- return ps.AuthenticationPolicies.AuthorizationPoliciesForWorkload(namespace, workloadLabels)
-}
-
-// initKubernetesGateways initializes Kubernetes Gateway API objects by delegating
-// to the GatewayAPIController, if it is present in the Environment.
-// This closely follows Dubbo's initKubernetesGateways behavior.
-func (ps *PushContext) initKubernetesGateways(env *Environment) {
- if env == nil || env.GatewayAPIController == nil {
- return
- }
- ps.GatewayAPIController = env.GatewayAPIController
- env.GatewayAPIController.Reconcile(ps)
-}
-
-func (ps *PushContext) ServiceForHostname(proxy *Proxy, hostname host.Name) *Service {
- for _, service := range ps.ServiceIndex.HostnameAndNamespace[hostname] {
- return service
- }
-
- // No service found
- return nil
-}
-
func (ps *PushContext) UpdateMetrics() {
ps.proxyStatusMutex.RLock()
defer ps.proxyStatusMutex.RUnlock()
@@ -745,469 +296,37 @@ func (ps *PushContext) OnConfigChange() {
ps.UpdateMetrics()
}
-func (ps *PushContext) servicesExportedToNamespace(ns string) []*Service {
- var out []*Service
-
- // First add private services and explicitly exportedTo services
- if ns == NamespaceAll {
- out = make([]*Service, 0, len(ps.ServiceIndex.privateByNamespace)+len(ps.ServiceIndex.public))
- for _, privateServices := range ps.ServiceIndex.privateByNamespace {
- out = append(out, privateServices...)
- }
- } else {
- out = make([]*Service, 0, len(ps.ServiceIndex.privateByNamespace[ns])+
- len(ps.ServiceIndex.exportedToNamespace[ns])+len(ps.ServiceIndex.public))
- out = append(out, ps.ServiceIndex.privateByNamespace[ns]...)
- out = append(out, ps.ServiceIndex.exportedToNamespace[ns]...)
- }
-
- // Second add public services
- out = append(out, ps.ServiceIndex.public...)
-
- return out
-}
-
-func (ps *PushContext) GetAllServices() []*Service {
- return ps.servicesExportedToNamespace(NamespaceAll)
-}
-
-func (ps *PushContext) initHTTPRoutes(env *Environment) {
- log.Debugf("starting HTTPRoute initialization")
- httproutes := env.List(gvk.HTTPRoute, NamespaceAll)
- log.Debugf("found %d HTTPRoute configs", len(httproutes))
-
- hostToRoutes := make(map[host.Name][]config.Config)
- for _, hr := range httproutes {
- hrSpec, ok := hr.Spec.(*sigsk8siogatewayapiapisv1.HTTPRouteSpec)
- if !ok {
- log.Debugf("HTTPRoute %s/%s spec is not HTTPRouteSpec", hr.Namespace, hr.Name)
- continue
- }
-
- // Process hostnames from HTTPRoute
- if len(hrSpec.Hostnames) == 0 {
- // If no hostnames specified, match all
- hostToRoutes["*"] = append(hostToRoutes["*"], hr)
- log.Debugf("indexed HTTPRoute %s/%s for wildcard hostname (no hostnames specified)", hr.Namespace, hr.Name)
- } else {
- for _, hostname := range hrSpec.Hostnames {
- if hostname == "" {
- // Empty hostname means match all
- hostToRoutes["*"] = append(hostToRoutes["*"], hr)
- log.Debugf("indexed HTTPRoute %s/%s for wildcard hostname", hr.Namespace, hr.Name)
- } else {
- hostStr := string(hostname)
- // Resolve shortname to FQDN
- resolvedHost := string(ResolveShortnameToFQDN(hostStr, hr.Meta))
- hostName := host.Name(resolvedHost)
- hostToRoutes[hostName] = append(hostToRoutes[hostName], hr)
- log.Debugf("indexed HTTPRoute %s/%s for hostname %s", hr.Namespace, hr.Name, hostName)
- }
- }
- }
- }
- ps.httpRouteIndex.hostToRoutes = hostToRoutes
- log.Debugf("indexed HTTPRoutes for %d hostnames", len(hostToRoutes))
- if len(hostToRoutes) > 0 {
- for hostname, routes := range hostToRoutes {
- log.Infof("hostname %s has %d HTTPRoute(s)", hostname, len(routes))
- }
- }
-}
-
-func (ps *PushContext) initDxgateServices(env *Environment) {
- services := sortConfigByCreationTime(env.List(gvk.DxgateService, NamespaceAll))
- index := make(map[string]map[string]config.Config)
- for _, service := range services {
- if index[service.Namespace] == nil {
- index[service.Namespace] = make(map[string]config.Config)
- }
- index[service.Namespace][service.Name] = service
- }
- ps.dxgateServiceIndex = dxgateServiceIndex{byNamespace: index}
-}
-
-// DxgateService resolves the mesh-native backend named by an HTTPRoute.
-func (ps *PushContext) DxgateService(namespace, name string) (config.Config, bool) {
- if ps == nil {
- return config.Config{}, false
- }
- services := ps.dxgateServiceIndex.byNamespace[namespace]
- if services == nil {
- return config.Config{}, false
- }
- service, found := services[name]
- return service, found
-}
-
-// HTTPRouteForHost returns HTTPRoutes that match the given hostname.
-func (ps *PushContext) HTTPRouteForHost(hostname host.Name) []config.Config {
- var routes []config.Config
- hostStr := string(hostname)
-
- // Special case: if hostname is "*", return ALL HTTPRoutes
- // This is needed for Gateway Pod inbound listeners that need to route traffic based on HTTPRoute hostnames
- if hostname == "*" {
- for _, routeList := range ps.httpRouteIndex.hostToRoutes {
- routes = append(routes, routeList...)
- }
- if len(routes) == 0 {
- log.Debugf("no HTTPRoute found for wildcard hostname")
- return nil
- }
- log.Infof("found %d HTTPRoute(s) for wildcard hostname", len(routes))
- return routes
- }
-
- // First check exact match
- if exactRoutes, ok := ps.httpRouteIndex.hostToRoutes[hostname]; ok {
- routes = append(routes, exactRoutes...)
- }
-
- // Check wildcard patterns (e.g., *.example.com)
- for patternHost, patternRoutes := range ps.httpRouteIndex.hostToRoutes {
- if patternHost == "*" {
- continue // Skip global wildcard, handle separately
- }
- patternStr := string(patternHost)
- if strings.HasPrefix(patternStr, "*.") {
- // Wildcard pattern like *.example.com
- suffix := patternStr[2:] // Remove "*."
- if strings.HasSuffix(hostStr, suffix) {
- routes = append(routes, patternRoutes...)
- }
- }
- }
-
- // Then check global wildcard
- if wildcardRoutes, ok := ps.httpRouteIndex.hostToRoutes["*"]; ok {
- routes = append(routes, wildcardRoutes...)
- }
-
- if len(routes) == 0 {
- log.Debugf("no HTTPRoute found for hostname %s", hostname)
- return nil
- }
-
- log.Infof("found %d HTTPRoute(s) for hostname %s", len(routes), hostname)
- return routes
-}
-
-func (ps *PushContext) initBackendTLSPolicies(env *Environment) {
- policies := sortConfigByCreationTime(env.List(gvk.BackendTLSPolicy, NamespaceAll))
- serviceTLS := map[string]BackendTLSSettings{}
- for _, cfg := range policies {
- spec, ok := cfg.Spec.(*sigsk8siogatewayapiapisv1.BackendTLSPolicySpec)
- if !ok {
- log.Debugf("BackendTLSPolicy %s/%s spec is not BackendTLSPolicySpec", cfg.Namespace, cfg.Name)
- continue
- }
- if !supportsSystemBackendTLS(spec) {
- continue
- }
- settings := BackendTLSSettings{SNI: string(spec.Validation.Hostname)}
- for _, target := range spec.TargetRefs {
- if !isBackendTLSPolicyServiceTarget(target) {
- continue
- }
- key := backendTLSPolicyServiceKey(cfg.Namespace, string(target.Name))
- if _, found := serviceTLS[key]; !found {
- serviceTLS[key] = settings
- }
- }
- }
- ps.backendTLSPolicyIndex.serviceTLS = serviceTLS
- log.Debugf("indexed BackendTLSPolicies for %d services", len(serviceTLS))
-}
-
-func (ps *PushContext) BackendTLSForService(namespace, name string) (BackendTLSSettings, bool) {
- if ps == nil || ps.backendTLSPolicyIndex.serviceTLS == nil {
- return BackendTLSSettings{}, false
- }
- settings, found := ps.backendTLSPolicyIndex.serviceTLS[backendTLSPolicyServiceKey(namespace, name)]
- return settings, found
-}
-
-type FaultInjectionSettings struct {
- Delay time.Duration
- DelayPercentage uint32
- AbortStatus uint32
- AbortPercentage uint32
-}
-
-type faultInjectionPolicyIndex struct {
- serviceFaults map[string]FaultInjectionSettings
-}
-
-func (ps *PushContext) initFaultInjectionPolicies(env *Environment) {
- policies := sortConfigByCreationTime(env.List(gvk.FaultInjectionPolicy, NamespaceAll))
- serviceFaults := map[string]FaultInjectionSettings{}
- for _, cfg := range policies {
- spec, ok := cfg.Spec.(*networking.FaultInjectionPolicy)
- if !ok || spec == nil {
- continue
- }
- settings := FaultInjectionSettings{}
- if delay := spec.GetDelay(); delay != nil {
- if value := delay.GetFixedDelay(); value != nil && value.CheckValid() == nil && value.AsDuration() > 0 {
- settings.Delay = value.AsDuration()
- settings.DelayPercentage = faultPercentage(delay.GetPercentage())
- }
- }
- if abort := spec.GetAbort(); abort != nil && abort.GetHttpStatus() >= 400 && abort.GetHttpStatus() <= 599 {
- settings.AbortStatus = abort.GetHttpStatus()
- settings.AbortPercentage = faultPercentage(abort.GetPercentage())
- }
- if settings.Delay == 0 && settings.AbortStatus == 0 {
- continue
- }
- for _, target := range spec.GetTargetRefs() {
- if !isFaultInjectionServiceTarget(target) {
- continue
- }
- key := faultInjectionServiceKey(cfg.Namespace, target.GetName(), target.GetSectionName())
- if _, found := serviceFaults[key]; !found {
- serviceFaults[key] = settings
- }
- }
- }
- ps.faultInjectionIndex.serviceFaults = serviceFaults
- log.Debugf("indexed FaultInjectionPolicies for %d service targets", len(serviceFaults))
-}
-
-func (ps *PushContext) FaultInjectionForService(namespace, name, portName string) (FaultInjectionSettings, bool) {
- if ps == nil || ps.faultInjectionIndex.serviceFaults == nil {
- return FaultInjectionSettings{}, false
- }
- if portName != "" {
- if settings, found := ps.faultInjectionIndex.serviceFaults[faultInjectionServiceKey(namespace, name, portName)]; found {
- return settings, true
- }
- }
- settings, found := ps.faultInjectionIndex.serviceFaults[faultInjectionServiceKey(namespace, name, "")]
- return settings, found
-}
-
-func faultPercentage(value *wrapperspb.UInt32Value) uint32 {
- if value == nil {
- return 100
- }
- return value.GetValue()
-}
-
-func isFaultInjectionServiceTarget(target *networking.PolicyTargetReference) bool {
- if target == nil || target.GetName() == "" {
- return false
- }
- group := strings.TrimSpace(target.GetGroup())
- return (group == "" || group == "core") && target.GetKind() == "Service"
-}
-
-func faultInjectionServiceKey(namespace, name, section string) string {
- return namespace + "/" + name + "#" + section
-}
-
-func supportsSystemBackendTLS(spec *sigsk8siogatewayapiapisv1.BackendTLSPolicySpec) bool {
- if spec == nil || spec.Validation.Hostname == "" {
- return false
- }
- wellKnown := spec.Validation.WellKnownCACertificates
- return wellKnown != nil && *wellKnown == sigsk8siogatewayapiapisv1.WellKnownCACertificatesSystem
-}
-
-func isBackendTLSPolicyServiceTarget(target sigsk8siogatewayapiapisv1.LocalPolicyTargetReferenceWithSectionName) bool {
- if target.SectionName != nil {
- return false
- }
- group := strings.TrimSpace(string(target.Group))
- kind := strings.TrimSpace(string(target.Kind))
- return (group == "" || group == "core") && strings.EqualFold(kind, "Service")
-}
-
-func backendTLSPolicyServiceKey(namespace, name string) string {
- return namespace + "/" + name
-}
-
-// ServiceAccounts returns the SPIFFE identities associated with the workloads
-// backing the given service hostname in the given namespace. The returned list
-// is sorted and already expanded with trust domain aliases.
-func (ps *PushContext) ServiceAccounts(hostname host.Name, namespace string) []string {
- return ps.serviceAccounts[serviceAccountKey{hostname: hostname, namespace: namespace}]
-}
-
-// ServiceActivationEnabled reports whether a structurally valid policy targets
-// this Service. The live ActivatorReady condition cannot gate routing: the first
-// cold request is what makes an Activator report demand for this target.
-func (ps *PushContext) ServiceActivationEnabled(namespace, name string) bool {
- if ps == nil {
- return false
- }
- _, found := ps.serviceActivationIndex.services[backendTLSPolicyServiceKey(namespace, name)]
- return found
-}
-
-// ActivationGatewayService returns the dedicated namespace-local Activator.
-func (ps *PushContext) ActivationGatewayService(namespace string) *Service {
+func (ps *PushContext) StatusJSON() ([]byte, error) {
if ps == nil {
- return nil
- }
- for _, namespaces := range ps.ServiceIndex.HostnameAndNamespace {
- if svc := namespaces[namespace]; svc != nil && svc.Attributes.Name == ActivationGatewayServiceName {
- return svc
- }
+ return []byte{'{', '}'}, nil
}
- return nil
-}
-
-// ActivationGatewaySANs is stable across cold/hot EDS transitions. Keeping the
-// backend and Activator identities in one CDS validation context avoids a TLS
-// verification window while the endpoint assignment changes.
-func (ps *PushContext) ActivationGatewaySANs(namespace string) []string {
- if ps == nil || ps.Mesh == nil || namespace == "" {
- return nil
+ ps.proxyStatusMutex.RLock()
+ defer ps.proxyStatusMutex.RUnlock()
+ if len(ps.ProxyStatus) == 0 {
+ return []byte{'{', '}'}, nil
}
- identities := sets.New(spiffe.MustGenSpiffeURI(ps.Mesh, namespace, ActivationGatewayServiceName))
- return sets.SortedList(spiffe.ExpandWithTrustDomains(identities, ps.Mesh.TrustDomainAliases))
+ return json.MarshalIndent(ps.ProxyStatus, "", " ")
}
-// ActivationBackendSANs returns the backend identities declared by the policy.
-// Unlike endpoint-derived identities, these remain available when dubbod starts
-// while the target has zero endpoints.
-func (ps *PushContext) ActivationBackendSANs(namespace, name string) []string {
- if ps == nil || ps.Mesh == nil || namespace == "" {
- return nil
+func (ps *PushContext) InitContext(env *Environment, old *PushContext, req *PushRequest) {
+ var previous *ConfigSnapshot
+ if old != nil {
+ previous = old.ConfigSnapshot
}
- accounts, found := ps.serviceActivationIndex.services[backendTLSPolicyServiceKey(namespace, name)]
- if !found {
- return nil
- }
- identities := sets.New[string]()
- for _, account := range accounts {
- account = strings.TrimSpace(account)
- if account == "" {
- continue
- }
- if strings.HasPrefix(account, "spiffe://") {
- identities.Insert(account)
- continue
- }
- generated := sets.New(spiffe.MustGenSpiffeURI(ps.Mesh, namespace, account))
- identities.InsertAll(sets.SortedList(
- spiffe.ExpandWithTrustDomains(generated, ps.Mesh.TrustDomainAliases),
- )...)
- }
- return sets.SortedList(identities)
+ ps.Initialize(env, previous, req.ConfigChange())
}
-func (ps *PushContext) ActivatedServices(namespace string) []*Service {
- if ps == nil {
+func (req *PushRequest) ConfigChange() *ConfigChange {
+ if req == nil {
return nil
}
- out := make([]*Service, 0)
- for _, namespaces := range ps.ServiceIndex.HostnameAndNamespace {
- svc := namespaces[namespace]
- if svc != nil && ps.ServiceActivationEnabled(namespace, svc.Attributes.Name) {
- out = append(out, svc)
- }
- }
- return SortServicesByCreationTime(out)
-}
-
-func (ps *PushContext) initServiceActivationPolicies(env *Environment) {
- ps.serviceActivationIndex = serviceActivationPolicyIndex{services: map[string][]string{}}
- if env == nil {
- return
- }
- configs := sortConfigByCreationTime(env.List(gvk.ServiceActivationPolicy, NamespaceAll))
- for _, cfg := range configs {
- ps.upsertServiceActivationPolicy(cfg)
- }
- log.Debugf("activation policy index rebuilt: configs=%d services=%d", len(configs), len(ps.serviceActivationIndex.services))
-}
-
-func copyServiceActivationPolicyIndex(in serviceActivationPolicyIndex) serviceActivationPolicyIndex {
- out := serviceActivationPolicyIndex{services: make(map[string][]string, len(in.services))}
- for key, accounts := range in.services {
- out.services[key] = append([]string(nil), accounts...)
- }
- return out
-}
-
-func (ps *PushContext) applyServiceActivationPolicyUpdates(pushReq *PushRequest) {
- if pushReq == nil {
- return
- }
- for _, update := range pushReq.ServiceActivationPolicyUpdates {
- if update.Previous.Spec != nil {
- ps.deleteServiceActivationPolicy(update.Previous)
- }
- if update.Deleted {
- ps.deleteServiceActivationPolicy(update.Config)
- continue
- }
- ps.upsertServiceActivationPolicy(update.Config)
- }
-}
-
-func (ps *PushContext) deleteServiceActivationPolicy(cfg config.Config) {
- spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
- if ok && spec != nil && spec.GetTargetRef() != nil {
- delete(ps.serviceActivationIndex.services,
- backendTLSPolicyServiceKey(cfg.Namespace, spec.GetTargetRef().GetName()))
- }
-}
-
-func (ps *PushContext) upsertServiceActivationPolicy(cfg config.Config) {
- spec, ok := cfg.Spec.(*networking.ServiceActivationPolicy)
- if !ok || spec == nil || spec.GetTargetRef() == nil || spec.GetAutoscalerRef() == nil {
- return
- }
- target := spec.GetTargetRef()
- if strings.TrimSpace(target.GetName()) == "" ||
- (target.GetKind() != "" && !strings.EqualFold(target.GetKind(), "Service")) ||
- strings.TrimSpace(target.GetGroup()) != "" ||
- strings.TrimSpace(spec.GetAutoscalerRef().GetName()) == "" ||
- len(spec.GetBackendServiceAccounts()) == 0 {
- return
+ return &ConfigChange{
+ ConfigsUpdated: req.ConfigsUpdated,
+ ServiceActivationPolicyUpdates: req.ServiceActivationPolicyUpdates,
+ AddressesUpdated: req.AddressesUpdated,
+ Full: req.Full,
+ Forced: req.Forced,
+ EndpointsChanged: req.Reason.Has(EndpointUpdate) || req.Reason.Has(HeadlessEndpointUpdate),
+ Global: req.Reason.Has(GlobalUpdate),
}
- ps.serviceActivationIndex.services[backendTLSPolicyServiceKey(cfg.Namespace, target.GetName())] =
- append([]string(nil), spec.GetBackendServiceAccounts()...)
-}
-
-func (ps *PushContext) initServiceAccounts(env *Environment, services []*Service) {
- for _, svc := range services {
- var accounts sets.String
- // First get endpoint level service accounts
- shard, f := env.EndpointIndex.ShardsForService(string(svc.Hostname), svc.Attributes.Namespace)
- if f {
- shard.RLock()
- accounts = shard.ServiceAccounts.Copy()
- shard.RUnlock()
- }
- if len(svc.ServiceAccounts) > 0 {
- if accounts == nil {
- accounts = sets.New(svc.ServiceAccounts...)
- } else {
- accounts = accounts.InsertAll(svc.ServiceAccounts...)
- }
- }
- sa := sets.SortedList(spiffe.ExpandWithTrustDomains(accounts, ps.Mesh.TrustDomainAliases))
- key := serviceAccountKey{
- hostname: svc.Hostname,
- namespace: svc.Attributes.Namespace,
- }
- ps.serviceAccounts[key] = sa
- }
-}
-
-func (ps *PushContext) StatusJSON() ([]byte, error) {
- if ps == nil {
- return []byte{'{', '}'}, nil
- }
- ps.proxyStatusMutex.RLock()
- defer ps.proxyStatusMutex.RUnlock()
- if len(ps.ProxyStatus) == 0 {
- return []byte{'{', '}'}, nil
- }
- return json.MarshalIndent(ps.ProxyStatus, "", " ")
}
diff --git a/dubbod/discovery/pkg/model/push_context_test.go b/dubbod/discovery/pkg/model/push_context_test.go
index d22cdc56..679dac76 100644
--- a/dubbod/discovery/pkg/model/push_context_test.go
+++ b/dubbod/discovery/pkg/model/push_context_test.go
@@ -127,7 +127,7 @@ func TestApplyServiceActivationPolicyUpdates(t *testing.T) {
}
push := NewPushContext()
- push.applyServiceActivationPolicyUpdates(add)
+ push.applyServiceActivationPolicyUpdates(add.ConfigChange())
if !push.ServiceActivationEnabled("app", "payment") {
t.Fatal("policy event did not add payment to activation index")
}
@@ -143,7 +143,7 @@ func TestApplyServiceActivationPolicyUpdates(t *testing.T) {
}
next := NewPushContext()
next.serviceActivationIndex = copyServiceActivationPolicyIndex(push.serviceActivationIndex)
- next.applyServiceActivationPolicyUpdates(merged)
+ next.applyServiceActivationPolicyUpdates(merged.ConfigChange())
if next.ServiceActivationEnabled("app", "payment") {
t.Fatal("policy delete did not remove payment from activation index")
}
diff --git a/dubbod/discovery/pkg/networking/apigen/apigen.go b/dubbod/discovery/pkg/networking/apigen/apigen.go
deleted file mode 100644
index bfc8cb8a..00000000
--- a/dubbod/discovery/pkg/networking/apigen/apigen.go
+++ /dev/null
@@ -1,36 +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 apigen
-
-import (
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
-)
-
-type APIGenerator struct {
- store model.ConfigStore
-}
-
-func NewGenerator(store model.ConfigStore) *APIGenerator {
- return &APIGenerator{
- store: store,
- }
-}
-
-func (g *APIGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
- resp := model.Resources{}
- return resp, model.DefaultXdsLogDetails, nil
-}
diff --git a/dubbod/discovery/pkg/networking/core/configgen.go b/dubbod/discovery/pkg/networking/core/configgen.go
deleted file mode 100644
index 61baa216..00000000
--- a/dubbod/discovery/pkg/networking/core/configgen.go
+++ /dev/null
@@ -1,39 +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 core
-
-import (
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
- serviceRouteIndex "github.com/kdubbo/api/mesh/v1alpha1"
-)
-
-type ConfigGenerator interface {
- serviceRouteIndexChanged(mesh *serviceRouteIndex.MeshConfig)
-}
-
-type ConfigGeneratorImpl struct {
- Cache model.XdsCache
-}
-
-func NewConfigGenerator(cache model.XdsCache) *ConfigGeneratorImpl {
- return &ConfigGeneratorImpl{
- Cache: cache,
- }
-}
-
-func (configgen *ConfigGeneratorImpl) serviceRouteIndexChanged(_ *serviceRouteIndex.MeshConfig) {
-}
diff --git a/dubbod/discovery/pkg/serviceregistry/kube/controller/multicluster.go b/dubbod/discovery/pkg/serviceregistry/kube/controller/multicluster.go
index 37de6327..14ca5866 100644
--- a/dubbod/discovery/pkg/serviceregistry/kube/controller/multicluster.go
+++ b/dubbod/discovery/pkg/serviceregistry/kube/controller/multicluster.go
@@ -103,22 +103,11 @@ func (m *Multicluster) initializeCluster(cluster *multicluster.Cluster, kubeCont
m.opts.ServiceController.AddRegistryAndRun(kubeRegistry, clusterStopCh)
go func() {
- var shouldLead bool
- if !configCluster {
- shouldLead = m.checkShouldLead()
- log.Infof("should join leader-election for cluster %s: %t", cluster.ID, shouldLead)
- }
-
- if m.distributeCACert && (shouldLead || configCluster) {
+ if m.distributeCACert {
// Block server exit on graceful termination of the leader controller.
m.s.RunComponentAsyncAndWait("namespace controller", func(_ <-chan struct{}) error {
election := leaderelection.
NewLeaderElectionMulticluster(options.SystemNamespace, m.serverID, leaderelection.NamespaceController, m.revision, !configCluster, client)
- // For config cluster (single-node deployment), disable leader election even if globally enabled
- // because there's only one instance and no need for election.
- if configCluster {
- election.SetEnabled(false)
- }
election.AddRunFunction(func(leaderStop <-chan struct{}) {
log.Infof("starting namespace controller for cluster %s", cluster.ID)
nc := NewNamespaceController(client, m.caBundleWatcher)
@@ -136,8 +125,3 @@ func (m *Multicluster) initializeCluster(cluster *multicluster.Cluster, kubeCont
}
}()
}
-
-func (m *Multicluster) checkShouldLead() bool {
- var res bool
- return res
-}
diff --git a/dubbod/discovery/pkg/serviceregistry/kube/controller/namespace_controller.go b/dubbod/discovery/pkg/serviceregistry/kube/controller/namespace_controller.go
index 8fb059bf..59716114 100644
--- a/dubbod/discovery/pkg/serviceregistry/kube/controller/namespace_controller.go
+++ b/dubbod/discovery/pkg/serviceregistry/kube/controller/namespace_controller.go
@@ -27,6 +27,7 @@ import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/keycertbundle"
"github.com/apache/dubbo-kubernetes/pkg/config/constants"
+ commonfeatures "github.com/apache/dubbo-kubernetes/pkg/features"
"github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/inject"
@@ -92,7 +93,7 @@ func NewNamespaceController(kubeClient kube.Client, caBundleWatcher *keycertbund
}))
c.namespaces.AddEventHandler(controllers.FilteredObjectSpecHandler(c.queue.AddObject, func(o controllers.Object) bool {
- if features.InformerWatchNamespace != "" && features.InformerWatchNamespace != o.GetName() {
+ if commonfeatures.InformerWatchNamespace != "" && commonfeatures.InformerWatchNamespace != o.GetName() {
// We are only watching one namespace, and its not this one
return false
}
diff --git a/dubbod/discovery/pkg/util/network/ip.go b/dubbod/discovery/pkg/util/network/ip.go
index 69b99b1b..4303a97f 100644
--- a/dubbod/discovery/pkg/util/network/ip.go
+++ b/dubbod/discovery/pkg/util/network/ip.go
@@ -30,36 +30,6 @@ const (
waitTimeout = 2 * time.Minute
)
-func AllIPv4(ipAddrs []string) bool {
- for i := 0; i < len(ipAddrs); i++ {
- addr, err := netip.ParseAddr(ipAddrs[i])
- if err != nil {
- // Should not happen, invalid IP in proxy's IPAddresses slice should have been caught earlier,
- // skip it to prevent a panic.
- continue
- }
- if !addr.Is4() && addr.Is6() {
- return false
- }
- }
- return true
-}
-
-func AllIPv6(ipAddrs []string) bool {
- for i := 0; i < len(ipAddrs); i++ {
- addr, err := netip.ParseAddr(ipAddrs[i])
- if err != nil {
- // Should not happen, invalid IP in proxy's IPAddresses slice should have been caught earlier,
- // skip it to prevent a panic.
- continue
- }
- if addr.Is4() {
- return false
- }
- }
- return true
-}
-
func GlobalUnicastIP(ipAddrs []string) string {
for i := 0; i < len(ipAddrs); i++ {
addr, err := netip.ParseAddr(ipAddrs[i])
diff --git a/dubbod/discovery/pkg/xds/ads.go b/dubbod/discovery/pkg/xds/ads.go
index 72ff10a6..3c0ed67a 100644
--- a/dubbod/discovery/pkg/xds/ads.go
+++ b/dubbod/discovery/pkg/xds/ads.go
@@ -55,6 +55,7 @@ type Connection struct {
deltaReqChan chan *discovery.DeltaDiscoveryRequest
s *DiscoveryServer
ids []string
+ authExpiry time.Time
}
type Event struct {
@@ -180,8 +181,14 @@ func (s *DiscoveryServer) initConnection(node *core.Node, con *Connection, ident
con.SetID(connectionID(proxy.ID))
con.node = node
con.proxy = proxy
+ con.ids = identities
- // TODO authorize
+ if s.Authorize == nil {
+ return status.Error(codes.PermissionDenied, "xDS workload authorization is not configured")
+ }
+ if err := s.Authorize(proxy, identities); err != nil {
+ return status.Error(codes.PermissionDenied, err.Error())
+ }
s.addCon(con.ID(), con)
currentCount := s.adsClientCount()
@@ -223,15 +230,13 @@ func (s *DiscoveryServer) initializeProxy(con *Connection) error {
proxy.WatchedResources = map[string]*model.WatchedResource{}
- // Based on node metadata and version, we can associate a different generator.
- if proxy.Metadata.Generator != "" {
- proxy.XdsResourceGenerator = s.Generators[proxy.Metadata.Generator]
- }
-
return nil
}
func (s *DiscoveryServer) initProxyMetadata(node *core.Node) (*model.Proxy, error) {
+ if node == nil {
+ return nil, status.Error(codes.InvalidArgument, "xDS node is required")
+ }
meta, err := model.ParseMetadata(node.Metadata)
if err != nil {
return nil, status.New(codes.InvalidArgument, err.Error()).Err()
@@ -240,6 +245,13 @@ func (s *DiscoveryServer) initProxyMetadata(node *core.Node) (*model.Proxy, erro
if err != nil {
return nil, status.New(codes.InvalidArgument, err.Error()).Err()
}
+ // Both supported runtime roles use the gRPC resource compiler.
+ if meta.Generator == "" {
+ meta.Generator = "grpc"
+ }
+ if meta.Generator != "grpc" {
+ return nil, status.Errorf(codes.InvalidArgument, "unsupported xDS generator %q", meta.Generator)
+ }
// Update the config namespace associated with this proxy
proxy.ConfigNamespace = model.GetProxyConfigNamespace(proxy)
proxy.XdsNode = node
@@ -302,11 +314,16 @@ func (s *DiscoveryServer) Stream(stream DiscoveryStream) error {
return status.Errorf(codes.ResourceExhausted, "request rate limit exceeded: %v", err)
}
- // TODO authenticate
+ ids, err := s.authenticate(ctx)
+ if err != nil {
+ return status.Error(codes.Unauthenticated, err.Error())
+ }
s.globalPushContext().InitContext(s.Env, nil, nil)
con := newConnection(peerAddr, stream)
con.s = s
+ con.ids = ids
+ con.authExpiry = certificateExpiry(ctx)
return xds.Stream(con)
}
@@ -355,9 +372,12 @@ func (s *DiscoveryServer) pushConnection(con *Connection, pushEv *Event) error {
}
func (s *DiscoveryServer) processRequest(req *discovery.DiscoveryRequest, con *Connection) error {
+ if err := s.authorizeConnection(con); err != nil {
+ return err
+ }
stype := v1.GetShortType(req.TypeUrl)
- if req.TypeUrl == v1.HealthInfoType {
- return nil
+ if s.findGenerator(req.TypeUrl, con) == nil {
+ return status.Errorf(codes.InvalidArgument, "unsupported xDS resource type %q", req.TypeUrl)
}
shouldRespond, delta := xds.ShouldRespond(con.proxy, con.ID(), req)
@@ -391,20 +411,6 @@ func (s *DiscoveryServer) processRequest(req *discovery.DiscoveryRequest, con *C
return nil
}
- // For Inherent gRPC, if client sends wildcard (empty ResourceNames) after receiving specific resources,
- // this is likely an ACK and we should NOT push all resources again
- // Check if this is a wildcard request after specific resources were sent
- watchedResource := con.proxy.GetWatchedResource(req.TypeUrl)
- if con.proxy.IsInherentGrpc() && len(req.ResourceNames) == 0 && watchedResource != nil && len(watchedResource.ResourceNames) > 0 && watchedResource.NonceSent != "" {
- // This is a wildcard ACK after specific resources were sent
- // ShouldRespond should have returned false, but we check here as safety net
- // Update the WatchedResource to reflect the ACK, but don't push
- log.Debugf("%s: Inherent gRPC wildcard ACK after specific resources (prev: %d resources, nonce: %s), skipping push",
- stype, len(watchedResource.ResourceNames), watchedResource.NonceSent)
- // ShouldRespond should have already handled this, but we ensure no push happens
- return nil
- }
-
// Log PUSH request BEFORE calling pushXds with detailed information
// This helps track what resources are being requested for each push
pushResourceNamesStr := ""
@@ -446,12 +452,15 @@ func (s *DiscoveryServer) processRequest(req *discovery.DiscoveryRequest, con *C
}
func (s *DiscoveryServer) processDeltaRequest(req *discovery.DeltaDiscoveryRequest, con *Connection) error {
+ if err := s.authorizeConnection(con); err != nil {
+ return err
+ }
stype := v1.GetShortType(req.TypeUrl)
deltaLog.Debugf("%s: REQ %s resources sub:%d unsub:%d nonce:%s", stype,
con.ID(), len(req.ResourceNamesSubscribe), len(req.ResourceNamesUnsubscribe), req.ResponseNonce)
- if req.TypeUrl == v1.HealthInfoType {
- return nil
+ if s.findGenerator(req.TypeUrl, con) == nil {
+ return status.Errorf(codes.InvalidArgument, "unsupported xDS resource type %q", req.TypeUrl)
}
shouldRespond := shouldRespondDelta(con, req)
diff --git a/dubbod/discovery/pkg/xds/ads_benchmark_test.go b/dubbod/discovery/pkg/xds/ads_benchmark_test.go
index 33359f12..695a4081 100644
--- a/dubbod/discovery/pkg/xds/ads_benchmark_test.go
+++ b/dubbod/discovery/pkg/xds/ads_benchmark_test.go
@@ -230,7 +230,8 @@ func newInherentPushBenchmarkFixture(tb testing.TB, scale inherentPushScale) *in
env.SetPushContext(push)
server := NewDiscoveryServer(env, nil, nil)
- server.Generators[v1.EndpointType] = &EdsGenerator{
+ server.Authorize = allowTestWorkload
+ server.Generators["grpc/"+v1.EndpointType] = &EdsGenerator{
Cache: model.DisabledCache{},
EndpointIndex: env.EndpointIndex,
}
diff --git a/dubbod/discovery/pkg/xds/ads_test.go b/dubbod/discovery/pkg/xds/ads_test.go
index 32d5b341..b0518e00 100644
--- a/dubbod/discovery/pkg/xds/ads_test.go
+++ b/dubbod/discovery/pkg/xds/ads_test.go
@@ -146,10 +146,10 @@ func TestDiscoveryPushQueuesAndNotifies(t *testing.T) {
server, con, _ := newInherentXDSTestServer(t)
server.addCon("inherent-1", con)
- var notified *model.PushRequest
- server.RuntimeConfigUpdate = func(req *model.PushRequest) {
- notified = req
- }
+ var notified *model.ConfigChange
+ server.Env.AddConfigHandler(func(change *model.ConfigChange) {
+ notified = change
+ })
req := &model.PushRequest{
Full: full,
Reason: model.NewReasonStats(model.ConfigUpdate),
@@ -172,8 +172,8 @@ func TestDiscoveryPushQueuesAndNotifies(t *testing.T) {
if req.Push == nil || req.Start.IsZero() {
t.Fatalf("Push() left request without context or start time: %+v", req)
}
- if notified != req {
- t.Fatalf("RuntimeConfigUpdate received %p, want %p", notified, req)
+ if notified == nil || notified.Full != full || !notified.ConfigsUpdated.Equals(req.ConfigsUpdated) {
+ t.Fatalf("configuration notification = %+v, want change for %+v", notified, req)
}
if full && req.Push.PushVersion == "test-version" {
t.Fatalf("full Push() reused old push version %q", req.Push.PushVersion)
diff --git a/dubbod/discovery/pkg/xds/auth.go b/dubbod/discovery/pkg/xds/auth.go
index 8add1bb8..40982660 100644
--- a/dubbod/discovery/pkg/xds/auth.go
+++ b/dubbod/discovery/pkg/xds/auth.go
@@ -18,9 +18,64 @@ package xds
import (
"context"
+ "fmt"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/pkg/security"
+ "github.com/apache/dubbo-kubernetes/pkg/spiffe"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/peer"
+ "google.golang.org/grpc/status"
)
func (s *DiscoveryServer) authenticate(ctx context.Context) ([]string, error) {
- // TODO
- return nil, nil
+ for _, authenticator := range s.Authenticators {
+ caller, err := authenticator.Authenticate(security.AuthContext{GrpcContext: ctx})
+ if err != nil || caller == nil {
+ continue
+ }
+ var identities []string
+ for _, raw := range caller.Identities {
+ identity, err := spiffe.ParseIdentity(raw)
+ if err == nil && identity.TrustDomain != "" && identity.Namespace != "" && identity.ServiceAccount != "" {
+ identities = append(identities, raw)
+ }
+ }
+ if len(identities) > 0 {
+ return identities, nil
+ }
+ }
+ return nil, fmt.Errorf("xDS requires an authenticated workload identity")
+}
+
+func (s *DiscoveryServer) authorizeConnection(con *Connection) error {
+ if con == nil || con.proxy == nil || s.Authorize == nil {
+ return status.Error(codes.PermissionDenied, "xDS workload authorization is not configured")
+ }
+ if !con.authExpiry.IsZero() && !time.Now().Before(con.authExpiry) {
+ return status.Error(codes.Unauthenticated, "xDS client certificate has expired; reconnect with renewed credentials")
+ }
+ if err := s.Authorize(con.proxy, con.ids); err != nil {
+ return status.Error(codes.PermissionDenied, err.Error())
+ }
+ return nil
+}
+
+func certificateExpiry(ctx context.Context) time.Time {
+ p, ok := peer.FromContext(ctx)
+ if !ok {
+ return time.Time{}
+ }
+ info, ok := p.AuthInfo.(credentials.TLSInfo)
+ if !ok || len(info.State.VerifiedChains) == 0 {
+ return time.Time{}
+ }
+ var expires time.Time
+ for _, cert := range info.State.VerifiedChains[0] {
+ if expires.IsZero() || cert.NotAfter.Before(expires) {
+ expires = cert.NotAfter
+ }
+ }
+ return expires
}
diff --git a/dubbod/discovery/pkg/xds/auth_test.go b/dubbod/discovery/pkg/xds/auth_test.go
new file mode 100644
index 00000000..938d78d8
--- /dev/null
+++ b/dubbod/discovery/pkg/xds/auth_test.go
@@ -0,0 +1,160 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package xds
+
+import (
+ "context"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/security"
+ discovery "github.com/kdubbo/xds-api/service/discovery/v1"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+type testAuthenticator struct {
+ identities []string
+ err error
+}
+
+func (testAuthenticator) AuthenticatorType() string { return "test" }
+func (a testAuthenticator) Authenticate(security.AuthContext) (*security.Caller, error) {
+ return &security.Caller{Identities: a.identities}, a.err
+}
+
+func allowTestWorkload(*model.Proxy, []string) error { return nil }
+
+func TestAuthenticateRequiresWorkloadIdentity(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ authenticators []security.Authenticator
+ wantOK bool
+ }{
+ {name: "unconfigured"},
+ {name: "failed", authenticators: []security.Authenticator{testAuthenticator{err: fmt.Errorf("invalid certificate")}}},
+ {name: "empty", authenticators: []security.Authenticator{testAuthenticator{}}},
+ {name: "dns identity", authenticators: []security.Authenticator{testAuthenticator{identities: []string{"pod.app.svc"}}}},
+ {name: "empty service account", authenticators: []security.Authenticator{testAuthenticator{identities: []string{"spiffe://cluster.local/ns/app/sa/"}}}},
+ {name: "verified workload", authenticators: []security.Authenticator{testAuthenticator{identities: []string{"spiffe://cluster.local/ns/app/sa/default"}}}, wantOK: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ s := &DiscoveryServer{Authenticators: tc.authenticators}
+ ids, err := s.authenticate(context.Background())
+ if (err == nil) != tc.wantOK || (len(ids) > 0) != tc.wantOK {
+ t.Fatalf("authenticate = %v, %v", ids, err)
+ }
+ })
+ }
+}
+
+func TestADSRejectsUnauthenticatedStreams(t *testing.T) {
+ s, _, stream := newInherentXDSTestServer(t)
+ s.Authenticators = nil
+ s.serverReady.Store(true)
+ if got := status.Code(s.Stream(stream)); got != codes.Unauthenticated {
+ t.Fatalf("SotW status = %v", got)
+ }
+ if got := status.Code(s.StreamDeltas(newFakeDeltaADSStream())); got != codes.Unauthenticated {
+ t.Fatalf("Delta status = %v", got)
+ }
+ if len(s.AllClients()) != 0 {
+ t.Fatal("unauthenticated client registered")
+ }
+}
+
+func TestADSRejectsUnauthorizedWorkload(t *testing.T) {
+ for _, delta := range []bool{false, true} {
+ t.Run(fmt.Sprintf("delta=%t", delta), func(t *testing.T) {
+ s, _, stream := newInherentXDSTestServer(t)
+ s.serverReady.Store(true)
+ s.Authorize = func(*model.Proxy, []string) error { return fmt.Errorf("wrong service account") }
+ done := make(chan error, 1)
+ if delta {
+ ds := newFakeDeltaADSStream()
+ ds.recvCh <- &discovery.DeltaDiscoveryRequest{Node: testInherentNode(t), TypeUrl: "unknown"}
+ close(ds.recvCh)
+ go func() { done <- s.StreamDeltas(ds) }()
+ } else {
+ stream.recvCh <- &discovery.DiscoveryRequest{Node: testInherentNode(t), TypeUrl: "unknown"}
+ close(stream.recvCh)
+ go func() { done <- s.Stream(stream) }()
+ }
+ select {
+ case err := <-done:
+ if status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("status = %v", err)
+ }
+ case <-time.After(3 * time.Second):
+ t.Fatal("unauthorized stream did not terminate")
+ }
+ if len(s.AllClients()) != 0 {
+ t.Fatal("unauthorized client registered")
+ }
+ })
+ }
+}
+
+func TestADSRejectsUnknownResourceWithoutFallback(t *testing.T) {
+ s, con, _ := newInherentXDSTestServer(t)
+ s.Generators["api"] = staticResourceGenerator{}
+ for _, typeURL := range []string{"unknown", "type.googleapis.com/dubbo.v1.HealthInformation"} {
+ if err := s.processRequest(&discovery.DiscoveryRequest{TypeUrl: typeURL}, con); status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("SotW accepted %s: %v", typeURL, err)
+ }
+ if err := s.processDeltaRequest(&discovery.DeltaDiscoveryRequest{TypeUrl: typeURL}, con); status.Code(err) != codes.InvalidArgument {
+ t.Fatalf("Delta accepted %s: %v", typeURL, err)
+ }
+ }
+}
+
+func TestADSRechecksAuthorizationAfterConnection(t *testing.T) {
+ s, con, _ := newInherentXDSTestServer(t)
+ s.Authorize = func(*model.Proxy, []string) error { return fmt.Errorf("workload deleted") }
+ if err := s.processRequest(&discovery.DiscoveryRequest{TypeUrl: "unknown"}, con); status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("SotW did not recheck identity: %v", err)
+ }
+ if err := s.processDeltaRequest(&discovery.DeltaDiscoveryRequest{TypeUrl: "unknown"}, con); status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("Delta did not recheck identity: %v", err)
+ }
+ if err := s.pushXds(con, nil, nil); status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("SotW push did not recheck identity: %v", err)
+ }
+ if err := s.pushDeltaXds(con, nil, nil); status.Code(err) != codes.PermissionDenied {
+ t.Fatalf("Delta push did not recheck identity: %v", err)
+ }
+}
+
+func TestProxyUpdateNotifiesEveryWorkloadConnection(t *testing.T) {
+ s, _, _ := newInherentXDSTestServer(t)
+ for _, id := range []string{"first", "second", "unrelated"} {
+ con := newConnection("10.0.0.1", newFakeADSStream())
+ con.proxy = &model.Proxy{Metadata: &model.NodeMetadata{ClusterID: "local"}, IPAddresses: []string{"10.0.0.1"}}
+ con.SetID(id)
+ con.MarkInitialized()
+ if id == "unrelated" {
+ con.proxy.Metadata.ClusterID = "other"
+ }
+ s.addCon(id, con)
+ }
+ s.ProxyUpdate("local", "10.0.0.1")
+ if stats := s.pushQueue.Stats(); stats.Pending != 2 {
+ t.Fatalf("queued = %+v, want two connections", stats)
+ }
+}
diff --git a/dubbod/discovery/pkg/xds/cds.go b/dubbod/discovery/pkg/xds/cds.go
deleted file mode 100644
index 84749ea7..00000000
--- a/dubbod/discovery/pkg/xds/cds.go
+++ /dev/null
@@ -1,60 +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 xds
-
-import (
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/core"
-)
-
-type CdsGenerator struct {
- ConfigGenerator core.ConfigGenerator
-}
-
-var _ model.XdsDeltaResourceGenerator = &CdsGenerator{}
-
-func cdsNeedsPush(req *model.PushRequest, proxy *model.Proxy) (*model.PushRequest, bool) {
- if res, ok := xdsNeedsPush(req, proxy); ok {
- return req, res
- }
-
- if !req.Full {
- return req, false
- }
- // TODO Gateway
- return nil, false
-}
-
-func (c CdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
- _, needsPush := cdsNeedsPush(req, proxy)
- if !needsPush {
- return nil, model.DefaultXdsLogDetails, nil
- }
- return nil, model.DefaultXdsLogDetails, nil
-}
-
-// GenerateDeltas for CDS currently only builds deltas when services change.
-func (c CdsGenerator) GenerateDeltas(proxy *model.Proxy, req *model.PushRequest,
- w *model.WatchedResource,
-) (model.Resources, model.DeletedResources, model.XdsLogDetails, bool, error) {
- _, needsPush := cdsNeedsPush(req, proxy)
- if !needsPush {
- return nil, nil, model.DefaultXdsLogDetails, false, nil
- }
-
- return nil, nil, model.DefaultXdsLogDetails, false, nil
-}
diff --git a/dubbod/discovery/pkg/xds/delta.go b/dubbod/discovery/pkg/xds/delta.go
index bb305e6b..c625ef65 100644
--- a/dubbod/discovery/pkg/xds/delta.go
+++ b/dubbod/discovery/pkg/xds/delta.go
@@ -22,9 +22,9 @@ import (
"strings"
"time"
- dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/grpc"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
+ dubbogrpc "github.com/apache/dubbo-kubernetes/pkg/grpc"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
"github.com/google/uuid"
discovery "github.com/kdubbo/xds-api/service/discovery/v1"
@@ -74,6 +74,7 @@ func (s *DiscoveryServer) StreamDeltas(stream DeltaDiscoveryStream) error {
s.globalPushContext().InitContext(s.Env, nil, nil)
con := newDeltaConnection(peerAddr, stream)
con.s = s
+ con.authExpiry = certificateExpiry(ctx)
go s.receiveDelta(con, ids)
@@ -139,10 +140,6 @@ func (s *DiscoveryServer) receiveDelta(con *Connection, identities []string) {
return
}
if firstRequest {
- if req.TypeUrl == v1.HealthInfoType {
- deltaLog.Warnf("%q %s send health check probe before normal xDS request", con.Peer(), con.ID())
- continue
- }
firstRequest = false
if req.Node == nil || req.Node.Id == "" {
con.ErrorCh() <- status.New(codes.InvalidArgument, "missing node information").Err()
diff --git a/dubbod/discovery/pkg/xds/delta_test.go b/dubbod/discovery/pkg/xds/delta_test.go
index 0b49b046..6c7533c8 100644
--- a/dubbod/discovery/pkg/xds/delta_test.go
+++ b/dubbod/discovery/pkg/xds/delta_test.go
@@ -90,8 +90,9 @@ func newDeltaXDSTestServer(generator model.XdsResourceGenerator) (*DiscoveryServ
push := model.NewPushContext()
push.PushVersion = "test-version"
server := NewDiscoveryServer(model.NewEnvironment(), nil, nil)
+ server.Authorize = allowTestWorkload
server.Generators = map[string]model.XdsResourceGenerator{
- v1.ClusterType: generator,
+ "grpc/" + v1.ClusterType: generator,
}
stream := newFakeDeltaADSStream()
con := newDeltaConnection("127.0.0.1:26010", stream)
diff --git a/dubbod/discovery/pkg/xds/discovery.go b/dubbod/discovery/pkg/xds/discovery.go
index 88ad434c..b22642d2 100644
--- a/dubbod/discovery/pkg/xds/discovery.go
+++ b/dubbod/discovery/pkg/xds/discovery.go
@@ -62,7 +62,7 @@ type DiscoveryServer struct {
Generators map[string]model.XdsResourceGenerator
pushVersion atomic.Uint64
Authenticators []security.Authenticator
- RuntimeConfigUpdate func(*model.PushRequest)
+ Authorize func(*model.Proxy, []string) error
}
type DebounceOptions struct {
@@ -121,12 +121,12 @@ func (s *DiscoveryServer) Shutdown() {
s.pushQueue.ShutDown()
}
-func (s *DiscoveryServer) initPushContext(req *model.PushRequest, oldPushContext *model.PushContext, version string) *model.PushContext {
+func (s *DiscoveryServer) initPushContext(req *model.PushRequest, version string) *model.PushContext {
startTime := time.Now()
push := model.NewPushContext()
push.PushVersion = version
- push.InitContext(s.Env, oldPushContext, req)
+ push.ConfigSnapshot = s.Env.ReconcileConfig(req.ConfigChange())
s.dropCacheForRequest(req)
s.Env.SetPushContext(push)
@@ -149,7 +149,7 @@ func (s *DiscoveryServer) Push(req *model.PushRequest) {
req.Push = s.globalPushContext()
s.dropCacheForRequest(req)
s.AdsPushAll(req)
- s.notifyRuntimeConfigUpdate(req)
+ s.Env.NotifyConfigChange(req.ConfigChange())
return
}
@@ -161,16 +161,9 @@ func (s *DiscoveryServer) Push(req *model.PushRequest) {
// PushContext is reset after a config change. Previous status is
// saved.
versionLocal := s.NextVersion()
- push := s.initPushContext(req, oldPushContext, versionLocal)
+ push := s.initPushContext(req, versionLocal)
req.Push = push
s.AdsPushAll(req)
- s.notifyRuntimeConfigUpdate(req)
-}
-
-func (s *DiscoveryServer) notifyRuntimeConfigUpdate(req *model.PushRequest) {
- if s.RuntimeConfigUpdate != nil {
- s.RuntimeConfigUpdate(req)
- }
}
func debounce(ch chan *model.PushRequest, stopCh <-chan struct{}, opts DebounceOptions, pushFn func(req *model.PushRequest), updateSent *atomic.Int64) {
@@ -418,27 +411,17 @@ func (s *DiscoveryServer) Clients() []*Connection {
}
func (s *DiscoveryServer) ProxyUpdate(clusterID cluster.ID, ip string) {
- var connection *Connection
-
for _, v := range s.Clients() {
- // Check IPAddresses is not empty before accessing index 0
if len(v.proxy.IPAddresses) > 0 && v.proxy.Metadata.ClusterID == clusterID && v.proxy.IPAddresses[0] == ip {
- connection = v
- break
+ s.pushQueue.Enqueue(v, &model.PushRequest{
+ Full: true,
+ Push: s.globalPushContext(),
+ Start: time.Now(),
+ Reason: model.NewReasonStats(model.ProxyUpdate),
+ Forced: true,
+ })
}
}
-
- if connection == nil {
- return
- }
-
- s.pushQueue.Enqueue(connection, &model.PushRequest{
- Full: true,
- Push: s.globalPushContext(),
- Start: time.Now(),
- Reason: model.NewReasonStats(model.ProxyUpdate),
- Forced: true,
- })
}
func (s *DiscoveryServer) EDSUpdate(shard model.ShardKey, serviceName string, namespace string, dubboEndpoints []*model.DubboEndpoint) {
diff --git a/dubbod/discovery/pkg/xds/lds.go b/dubbod/discovery/pkg/xds/lds.go
deleted file mode 100644
index 614dc707..00000000
--- a/dubbod/discovery/pkg/xds/lds.go
+++ /dev/null
@@ -1,47 +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 xds
-
-import (
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/core"
-)
-
-type LdsGenerator struct {
- ConfigGenerator core.ConfigGenerator
-}
-
-var _ model.XdsResourceGenerator = &LdsGenerator{}
-
-func ldsNeedsPush(proxy *model.Proxy, req *model.PushRequest) bool {
- if res, ok := xdsNeedsPush(req, proxy); ok {
- return res
- }
- // For full pushes, always push LDS to ensure listeners are generated
- if req.Full {
- return true
- }
- return false
-}
-
-func (l LdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
- if !ldsNeedsPush(proxy, req) {
- return nil, model.DefaultXdsLogDetails, nil
- }
- resources := model.Resources{}
- return resources, model.DefaultXdsLogDetails, nil
-}
diff --git a/dubbod/discovery/pkg/xds/rds.go b/dubbod/discovery/pkg/xds/rds.go
deleted file mode 100644
index 54e9c200..00000000
--- a/dubbod/discovery/pkg/xds/rds.go
+++ /dev/null
@@ -1,65 +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 xds
-
-import (
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/networking/core"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/kind"
- "github.com/apache/dubbo-kubernetes/pkg/util/sets"
-)
-
-type RdsGenerator struct {
- ConfigGenerator core.ConfigGenerator
-}
-
-var _ model.XdsResourceGenerator = &RdsGenerator{}
-
-var skippedRdsConfigs = sets.New[kind.Kind](
- kind.PeerAuthentication,
- kind.Secret,
-)
-
-// HTTPRoute changes should trigger RDS push for Gateway Pods
-// Gateway Pods use RDS to route traffic based on HTTPRoute rules
-
-func rdsNeedsPush(req *model.PushRequest, proxy *model.Proxy) bool {
- if res, ok := xdsNeedsPush(req, proxy); ok {
- return res
- }
- if !req.Full {
- return false
- }
- for config := range req.ConfigsUpdated {
- if !skippedRdsConfigs.Contains(config.Kind) {
- return true
- }
- }
- return false
-}
-
-func (c RdsGenerator) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
- log.Infof("node=%s, routeNames=%v, Full=%v, ConfigsUpdated=%v", proxy.ID, w.ResourceNames.UnsortedList(), req.Full, req.ConfigsUpdated)
- needsPush := rdsNeedsPush(req, proxy)
- log.Infof("rdsNeedsPush=%v for node=%s", needsPush, proxy.ID)
- if !needsPush {
- log.Debugf("skipping push for node=%s (rdsNeedsPush=false)", proxy.ID)
- return nil, model.DefaultXdsLogDetails, nil
- }
- resources := model.Resources{}
- return resources, model.DefaultXdsLogDetails, nil
-}
diff --git a/dubbod/discovery/pkg/xds/statusgen.go b/dubbod/discovery/pkg/xds/statusgen.go
deleted file mode 100644
index 7ad03eff..00000000
--- a/dubbod/discovery/pkg/xds/statusgen.go
+++ /dev/null
@@ -1,45 +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 xds
-
-import (
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
-)
-
-type StatusGen struct {
- Server *DiscoveryServer
-}
-
-func NewStatusGen(s *DiscoveryServer) *StatusGen {
- return &StatusGen{
- Server: s,
- }
-}
-
-func (sg *StatusGen) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
- return sg.handleInternalRequest(proxy, w, req)
-}
-
-func (sg *StatusGen) GenerateDeltas(proxy *model.Proxy, req *model.PushRequest, w *model.WatchedResource) (model.Resources, model.DeletedResources, model.XdsLogDetails, bool, error) {
- res, detail, err := sg.handleInternalRequest(proxy, w, req)
- return res, nil, detail, true, err
-}
-
-func (sg *StatusGen) handleInternalRequest(_ *model.Proxy, _ *model.WatchedResource, _ *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
- res := model.Resources{}
- return res, model.DefaultXdsLogDetails, nil
-}
diff --git a/dubbod/discovery/pkg/xds/stream_test.go b/dubbod/discovery/pkg/xds/stream_test.go
index 111b4c36..0ad38a28 100644
--- a/dubbod/discovery/pkg/xds/stream_test.go
+++ b/dubbod/discovery/pkg/xds/stream_test.go
@@ -29,6 +29,7 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/config/mesh/meshwatcher"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ "github.com/apache/dubbo-kubernetes/pkg/security"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
cluster "github.com/kdubbo/xds-api/cluster/v1"
@@ -133,10 +134,12 @@ func newInherentXDSTestServer(t *testing.T) (*DiscoveryServer, *Connection, *fak
env.SetPushContext(push)
server := NewDiscoveryServer(env, nil, nil)
+ server.Authenticators = []security.Authenticator{testAuthenticator{identities: []string{"spiffe://cluster.local/ns/app/sa/default"}}}
+ server.Authorize = allowTestWorkload
server.Generators = map[string]model.XdsResourceGenerator{
- v1.ListenerType: staticResourceGenerator{resources: inherentListenerResources()},
- v1.ClusterType: staticResourceGenerator{resources: inherentClusterResources()},
- v1.RouteType: staticResourceGenerator{resources: inherentRouteResources()},
+ "grpc/" + v1.ListenerType: staticResourceGenerator{resources: inherentListenerResources()},
+ "grpc/" + v1.ClusterType: staticResourceGenerator{resources: inherentClusterResources()},
+ "grpc/" + v1.RouteType: staticResourceGenerator{resources: inherentRouteResources()},
}
stream := newFakeADSStream()
@@ -164,6 +167,37 @@ type staticResourceGenerator struct {
resources model.Resources
}
+func TestSotWEmptySnapshotPreservesSubscriptionAndACKDoesNotLoop(t *testing.T) {
+ server, con, stream := newInherentXDSTestServer(t)
+ server.Generators["grpc/"+v1.ListenerType] = staticResourceGenerator{resources: model.Resources{}}
+ push := &model.PushRequest{Push: con.proxy.LastPushContext, Full: true}
+ if err := server.pushXds(con, con.proxy.GetWatchedResource(v1.ListenerType), push); err != nil {
+ t.Fatal(err)
+ }
+ resp := stream.takeAllResponses(t, 1)[0]
+ if len(resp.Resources) != 0 || resp.Nonce == "" {
+ t.Fatalf("expected empty deletion snapshot with nonce, got %v", resp)
+ }
+ if !con.proxy.GetWatchedResource(v1.ListenerType).ResourceNames.Contains(testListenerName) {
+ t.Fatal("deleting a resource must preserve the client's subscription")
+ }
+ if err := server.processRequest(&discovery.DiscoveryRequest{TypeUrl: v1.ListenerType, ResourceNames: []string{testListenerName}, ResponseNonce: resp.Nonce, VersionInfo: resp.VersionInfo}, con); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case <-stream.sendCh:
+ t.Fatal("ACK of an empty snapshot triggered another response")
+ default:
+ }
+ server.Generators["grpc/"+v1.ListenerType] = staticResourceGenerator{resources: inherentListenerResources()}
+ if err := server.pushXds(con, con.proxy.GetWatchedResource(v1.ListenerType), push); err != nil {
+ t.Fatal(err)
+ }
+ if len(stream.takeAllResponses(t, 1)[0].Resources) == 0 {
+ t.Fatal("recreated resource was not published")
+ }
+}
+
func (g staticResourceGenerator) Generate(*model.Proxy, *model.WatchedResource, *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
return g.resources, model.DefaultXdsLogDetails, nil
}
diff --git a/dubbod/discovery/pkg/xds/tls_stream_test.go b/dubbod/discovery/pkg/xds/tls_stream_test.go
new file mode 100644
index 00000000..6106b408
--- /dev/null
+++ b/dubbod/discovery/pkg/xds/tls_stream_test.go
@@ -0,0 +1,155 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package xds
+
+import (
+ "context"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "crypto/tls"
+ "crypto/x509"
+ "encoding/pem"
+ "fmt"
+ "math/big"
+ "net"
+ "net/url"
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ v1 "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/xds/v1"
+ "github.com/apache/dubbo-kubernetes/dubbod/security/pkg/server/ca/authenticate"
+ "github.com/apache/dubbo-kubernetes/pkg/security"
+ discovery "github.com/kdubbo/xds-api/service/discovery/v1"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/status"
+ "google.golang.org/grpc/test/bufconn"
+)
+
+func TestADSAuthenticatesTLSStreams(t *testing.T) {
+ rootKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now()
+ root := &x509.Certificate{SerialNumber: big.NewInt(1), IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour)}
+ rootDER, err := x509.CreateCertificate(rand.Reader, root, root, &rootKey.PublicKey, rootKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ roots := x509.NewCertPool()
+ roots.AppendCertsFromPEM(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: rootDER}))
+ issue := func(identity string, server bool) tls.Certificate {
+ t.Helper()
+ key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ leaf := &x509.Certificate{SerialNumber: big.NewInt(2), NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour), KeyUsage: x509.KeyUsageDigitalSignature}
+ if server {
+ leaf.DNSNames = []string{"dubbod.test"}
+ leaf.ExtKeyUsage = []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}
+ } else {
+ uri, err := url.Parse(identity)
+ if err != nil {
+ t.Fatal(err)
+ }
+ leaf.URIs = []*url.URL{uri}
+ leaf.ExtKeyUsage = []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}
+ }
+ der, err := x509.CreateCertificate(rand.Reader, leaf, root, &key.PublicKey, rootKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return tls.Certificate{Certificate: [][]byte{der, rootDER}, PrivateKey: key}
+ }
+ s, _, _ := newInherentXDSTestServer(t)
+ s.Authenticators = []security.Authenticator{&authenticate.ClientCertAuthenticator{}}
+ s.Authorize = func(_ *model.Proxy, identities []string) error {
+ if len(identities) != 1 || identities[0] != "spiffe://cluster.local/ns/app/sa/default" {
+ return fmt.Errorf("workload identity mismatch")
+ }
+ return nil
+ }
+ s.serverReady.Store(true)
+ listener := bufconn.Listen(1024 * 1024)
+ server := grpc.NewServer(grpc.Creds(credentials.NewTLS(&tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{issue("", true)}, ClientCAs: roots, ClientAuth: tls.VerifyClientCertIfGiven})))
+ s.Register(server)
+ go func() { _ = server.Serve(listener) }()
+ t.Cleanup(func() { server.Stop(); _ = listener.Close() })
+ for _, delta := range []bool{false, true} {
+ for _, tc := range []struct {
+ name, identity string
+ want codes.Code
+ }{
+ {"valid", "spiffe://cluster.local/ns/app/sa/default", codes.OK},
+ {"anonymous", "", codes.Unauthenticated},
+ {"other workload", "spiffe://cluster.local/ns/other/sa/default", codes.PermissionDenied},
+ } {
+ t.Run(fmt.Sprintf("delta=%t/%s", delta, tc.name), func(t *testing.T) {
+ cfg := &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: roots, ServerName: "dubbod.test"}
+ if tc.identity != "" {
+ cfg.Certificates = []tls.Certificate{issue(tc.identity, false)}
+ }
+ conn, err := grpc.NewClient("passthrough:///dubbod.test", grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), grpc.WithTransportCredentials(credentials.NewTLS(cfg)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close()
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ client := discovery.NewAggregatedDiscoveryServiceClient(conn)
+ if delta {
+ stream, e := client.DeltaAggregatedResources(ctx)
+ err = e
+ if err == nil {
+ err = stream.Send(&discovery.DeltaDiscoveryRequest{Node: testInherentNode(t), TypeUrl: v1.ListenerType, ResourceNamesSubscribe: []string{testListenerName}})
+ }
+ if err == nil {
+ _, err = stream.Recv()
+ }
+ } else {
+ stream, e := client.StreamAggregatedResources(ctx)
+ err = e
+ if err == nil {
+ err = stream.Send(&discovery.DiscoveryRequest{Node: testInherentNode(t), TypeUrl: v1.ListenerType, ResourceNames: []string{testListenerName}})
+ }
+ if err == nil {
+ _, err = stream.Recv()
+ }
+ }
+ if got := status.Code(err); got != tc.want {
+ t.Fatalf("TLS ADS status=%v, want %v: %v", got, tc.want, err)
+ }
+ })
+ }
+ }
+}
+
+func TestADSExpiredConnectionCannotReceiveResources(t *testing.T) {
+ s, con, _ := newInherentXDSTestServer(t)
+ con.authExpiry = time.Now().Add(-time.Second)
+ if err := s.pushXds(con, nil, nil); status.Code(err) != codes.Unauthenticated {
+ t.Fatalf("SotW expiry ignored: %v", err)
+ }
+ if err := s.pushDeltaXds(con, nil, nil); status.Code(err) != codes.Unauthenticated {
+ t.Fatalf("Delta expiry ignored: %v", err)
+ }
+}
diff --git a/dubbod/discovery/pkg/xds/v1/model.go b/dubbod/discovery/pkg/xds/v1/model.go
index fcc5068d..31d81f75 100644
--- a/dubbod/discovery/pkg/xds/v1/model.go
+++ b/dubbod/discovery/pkg/xds/v1/model.go
@@ -19,12 +19,11 @@ package v1
import "github.com/apache/dubbo-kubernetes/pkg/model"
const (
- ClusterType = model.ClusterType
- ListenerType = model.ListenerType
- EndpointType = model.EndpointType
- RouteType = model.RouteType
- SecretType = model.SecretType
- HealthInfoType = model.HealthInfoType
+ ClusterType = model.ClusterType
+ ListenerType = model.ListenerType
+ EndpointType = model.EndpointType
+ RouteType = model.RouteType
+ SecretType = model.SecretType
)
func GetShortType(typeURL string) string {
diff --git a/dubbod/discovery/pkg/xds/xdsgen.go b/dubbod/discovery/pkg/xds/xdsgen.go
index 31d11176..f0e664a3 100644
--- a/dubbod/discovery/pkg/xds/xdsgen.go
+++ b/dubbod/discovery/pkg/xds/xdsgen.go
@@ -68,6 +68,9 @@ func xdsNeedsPush(req *model.PushRequest, _ *model.Proxy) (needsPush, definitive
}
func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req *model.PushRequest) error {
+ if err := s.authorizeConnection(con); err != nil {
+ return err
+ }
if w == nil {
return nil
}
@@ -76,22 +79,9 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
return nil
}
- // For Inherent gRPC, handle wildcard (empty ResourceNames) requests correctly
- // When client sends empty ResourceNames after receiving specific resources, it's likely an ACK
- // We should NOT generate all resources, but instead return the last sent resources
- // However, for initial wildcard requests, we need to extract resource names from parent resources
var requestedResourceNames sets.String
- var useLastSentResources bool
if con.proxy.IsInherentGrpc() {
- // Check if this is a wildcard request (empty ResourceNames) but we have previously sent resources
- if len(w.ResourceNames) == 0 && w.NonceSent != "" {
- // This is likely an ACK after receiving specific resources
- // Use the last sent resources instead of generating all
- useLastSentResources = true
- // Get the last sent resource names from WatchedResource
- // We'll populate this from the last sent resources after generation
- log.Debugf("Inherent gRPC wildcard request with NonceSent=%s, will use last sent resources", w.NonceSent)
- } else if len(w.ResourceNames) == 0 && w.NonceSent == "" {
+ if len(w.ResourceNames) == 0 {
// Initial wildcard request - need to extract resource names from parent resources
// For CDS: extract cluster names from LDS
// For EDS: extract cluster names from CDS
@@ -201,18 +191,7 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
}
}
- // For Inherent gRPC wildcard requests with previous NonceSent, use last sent resources
- var res model.Resources
- var logdata model.XdsLogDetails
- var err error
- if useLastSentResources {
- // Don't generate new resources, return empty and we'll handle it below
- res = nil
- logdata = model.DefaultXdsLogDetails
- err = nil
- } else {
- res, logdata, err = gen.Generate(con.proxy, w, req)
- }
+ res, logdata, err := gen.Generate(con.proxy, w, req)
info := ""
if len(logdata.AdditionalInfo) > 0 {
@@ -225,15 +204,7 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
return err
}
- // For Inherent gRPC wildcard requests with previous NonceSent, return last sent resources
- if useLastSentResources && res == nil {
- // This is a wildcard ACK - client is acknowledging previous push
- // We should NOT push again, as the client already has the resources
- // The ShouldRespond logic should have prevented this, but we handle it here as safety
- log.Debugf("Inherent gRPC wildcard ACK with NonceSent=%s, skipping push (client already has resources from previous push)", w.NonceSent)
- return nil
- }
-
+ // nil means no change; a non-nil empty slice is a deletion snapshot.
if res == nil {
return nil
}
@@ -256,14 +227,6 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
info += " filtered:" + strconv.Itoa(len(res)-len(filteredRes))
res = filteredRes
}
- // If filtering resulted in 0 resources but client requested specific resources,
- // this means the requested resources don't exist. Don't send empty response to avoid loop.
- // Instead, log and return nil to prevent push.
- if len(res) == 0 && len(requestedResourceNames) > 0 {
- log.Warnf("Inherent gRPC requested %d resources but none matched after filtering (requested: %v, generated before filter: %d). Skipping push to avoid loop.",
- len(requestedResourceNames), requestedResourceNames.UnsortedList(), len(filteredRes)+len(res))
- return nil
- }
} else if len(w.ResourceNames) == 0 {
// Wildcard request without previous NonceSent - this is initial request
// Allow generating all resources for initial connection
@@ -271,13 +234,6 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
}
}
- // Never send empty response for Inherent gRPC - this causes push loops
- // If we have no resources to send, return nil instead of sending empty response
- if len(res) == 0 {
- log.Debugf("no resources to send for %s (proxy: %s), skipping push", w.TypeUrl, con.proxy.ID)
- return nil
- }
-
nonceValue := nonce(req.Push.PushVersion)
resp := &discovery.DiscoveryResponse{
ControlPlane: ControlPlane(),
@@ -302,32 +258,7 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
return nil
}
wr.NonceSent = nonceValue
- // Also update ResourceNames to match what we actually sent (for Inherent gRPC)
- if con.proxy.IsInherentGrpc() && res != nil {
- sentNames := sets.New[string]()
- for _, r := range res {
- sentNames.Insert(r.Name)
- }
- // Only update if we sent different resources than requested
- if requestedResourceNames != nil {
- // Compare sets by checking if they have the same size and all elements match
- if sentNames.Len() != requestedResourceNames.Len() {
- wr.ResourceNames = sentNames
- } else {
- // Check if all sent names are in requested names
- allMatch := true
- for name := range sentNames {
- if !requestedResourceNames.Contains(name) {
- allMatch = false
- break
- }
- }
- if !allMatch {
- wr.ResourceNames = sentNames
- }
- }
- }
- }
+
return wr
})
@@ -527,6 +458,9 @@ func (s *DiscoveryServer) pushXds(con *Connection, w *model.WatchedResource, req
}
func (s *DiscoveryServer) pushDeltaXds(con *Connection, w *model.WatchedResource, req *model.PushRequest) error {
+ if err := s.authorizeConnection(con); err != nil {
+ return err
+ }
if w == nil {
return nil
}
@@ -626,21 +560,10 @@ func (s *DiscoveryServer) pushDeltaXds(con *Connection, w *model.WatchedResource
}
func (s *DiscoveryServer) findGenerator(typeURL string, con *Connection) model.XdsResourceGenerator {
- if g, f := s.Generators[con.proxy.Metadata.Generator+"/"+typeURL]; f {
- return g
- }
- if g, f := s.Generators[typeURL]; f {
- return g
- }
-
- // XdsResourceGenerator is the default generator for this connection. We want to allow
- // some types to use custom generators - for example EDS.
- g := con.proxy.XdsResourceGenerator
- if g == nil {
- // TODO move this to just directly using the resource TypeUrl
- g = s.Generators["api"] // default to "MCP" generators - any type supported by store
+ if con == nil || con.proxy == nil || con.proxy.Metadata == nil {
+ return nil
}
- return g
+ return s.Generators[con.proxy.Metadata.Generator+"/"+typeURL]
}
func resourceNamesSet(res model.Resources) sets.String {
diff --git a/dubbod/security/pkg/nodeagent/caclient/providers/aegis/client.go b/dubbod/security/pkg/nodeagent/caclient/providers/aegis/client.go
index fb429bc2..c1cd6087 100644
--- a/dubbod/security/pkg/nodeagent/caclient/providers/aegis/client.go
+++ b/dubbod/security/pkg/nodeagent/caclient/providers/aegis/client.go
@@ -24,8 +24,8 @@ import (
"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"
+ dubbogrpc "github.com/apache/dubbo-kubernetes/pkg/grpc"
"github.com/apache/dubbo-kubernetes/pkg/security"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
diff --git a/dubbod/security/pkg/server/ca/authenticate/cert_authenticator.go b/dubbod/security/pkg/server/ca/authenticate/cert_authenticator.go
index 14530aee..1f101947 100644
--- a/dubbod/security/pkg/server/ca/authenticate/cert_authenticator.go
+++ b/dubbod/security/pkg/server/ca/authenticate/cert_authenticator.go
@@ -20,6 +20,7 @@ import (
"context"
"fmt"
"net/http"
+ "time"
"github.com/apache/dubbo-kubernetes/dubbod/security/pkg/pki/util"
"github.com/apache/dubbo-kubernetes/pkg/security"
@@ -63,11 +64,20 @@ func (cca *ClientCertAuthenticator) authenticateGrpc(ctx context.Context) (*secu
return nil, fmt.Errorf("unsupported auth type: %q", authType)
}
- tlsInfo := peer.AuthInfo.(credentials.TLSInfo)
+ tlsInfo, ok := peer.AuthInfo.(credentials.TLSInfo)
+ if !ok {
+ return nil, fmt.Errorf("no verified TLS information is available")
+ }
chains := tlsInfo.State.VerifiedChains
if len(chains) == 0 || len(chains[0]) == 0 {
return nil, fmt.Errorf("no verified chain is found")
}
+ now := time.Now()
+ for _, cert := range chains[0] {
+ if now.Before(cert.NotBefore) || !now.Before(cert.NotAfter) {
+ return nil, fmt.Errorf("client certificate chain is outside its validity period")
+ }
+ }
ids, err := util.ExtractIDs(chains[0][0].Extensions)
if err != nil {
diff --git a/dubbod/security/pkg/server/ca/server.go b/dubbod/security/pkg/server/ca/server.go
index e9347112..4b27df6b 100644
--- a/dubbod/security/pkg/server/ca/server.go
+++ b/dubbod/security/pkg/server/ca/server.go
@@ -123,14 +123,14 @@ func (s *Server) CreateCertificate(ctx context.Context, request *pb.DubboCertifi
func (s *Server) authenticate(ctx context.Context) (*security.Caller, error) {
if len(s.Authenticators) == 0 {
- return nil, nil
+ return nil, status.Error(codes.Unauthenticated, "CA authentication is not configured")
}
authCtx := security.AuthContext{
GrpcContext: ctx,
}
for _, authenticator := range s.Authenticators {
caller, err := authenticator.Authenticate(authCtx)
- if err == nil && caller != nil {
+ if err == nil && caller != nil && len(caller.Identities) > 0 {
return caller, nil
}
}
diff --git a/pkg/features/transport.go b/pkg/features/transport.go
new file mode 100644
index 00000000..cff0615c
--- /dev/null
+++ b/pkg/features/transport.go
@@ -0,0 +1,40 @@
+//
+// 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 features
+
+import "github.com/apache/dubbo-kubernetes/pkg/env"
+
+var (
+ MaxConcurrentStreams = env.Register(
+ "DUBBO_GPRC_MAXSTREAMS",
+ 100000,
+ "Sets the maximum number of concurrent grpc streams.",
+ ).Get()
+
+ // MaxRecvMsgSize The max receive buffer size of gRPC received channel of Dubbo in bytes.
+ MaxRecvMsgSize = env.Register(
+ "DUBBO_GPRC_MAXRECVMSGSIZE",
+ 4*1024*1024,
+ "Sets the max receive buffer size of gRPC stream in bytes.",
+ ).Get()
+
+ InformerWatchNamespace = env.Register("DUBBO_WATCH_NAMESPACE", "",
+ "If set, limit Kubernetes watches to a single namespace. "+
+ "Warning: only a single namespace can be set.").Get()
+ KubernetesClientContentType = env.Register("DUBBO_KUBE_CLIENT_CONTENT_TYPE", "protobuf",
+ "The content type to use for Kubernetes clients. Defaults to protobuf. Valid options: [protobuf, json]").Get()
+)
diff --git a/dubbod/discovery/pkg/grpc/grpc.go b/pkg/grpc/grpc.go
similarity index 98%
rename from dubbod/discovery/pkg/grpc/grpc.go
rename to pkg/grpc/grpc.go
index c2017de7..4fb27c20 100644
--- a/dubbod/discovery/pkg/grpc/grpc.go
+++ b/pkg/grpc/grpc.go
@@ -21,7 +21,7 @@ import (
"math"
"strings"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
+ "github.com/apache/dubbo-kubernetes/pkg/features"
dubbokeepalive "github.com/apache/dubbo-kubernetes/pkg/keepalive"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
middleware "github.com/grpc-ecosystem/go-grpc-middleware"
diff --git a/dubbod/discovery/pkg/grpc/tls.go b/pkg/grpc/tls.go
similarity index 100%
rename from dubbod/discovery/pkg/grpc/tls.go
rename to pkg/grpc/tls.go
diff --git a/pkg/introspection/options_test.go b/pkg/introspection/options_test.go
index 044ec14b..ee4f0244 100644
--- a/pkg/introspection/options_test.go
+++ b/pkg/introspection/options_test.go
@@ -21,12 +21,30 @@ import (
"net"
"os"
"strings"
+ "sync"
"testing"
dubbolog "github.com/apache/dubbo-kubernetes/pkg/log"
"github.com/spf13/cobra"
)
+type synchronizedBuffer struct {
+ mu sync.Mutex
+ buffer bytes.Buffer
+}
+
+func (b *synchronizedBuffer) Write(p []byte) (int, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.buffer.Write(p)
+}
+
+func (b *synchronizedBuffer) String() string {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.buffer.String()
+}
+
func legacyIntrospectionTokens() []string {
return []string{"ctrl" + "z", "control" + "z"}
}
@@ -81,7 +99,7 @@ func TestRunLogsHaveNoLegacyIntrospectionIdentifiers(t *testing.T) {
if scope == nil {
t.Fatal("introspection log scope missing")
}
- var out bytes.Buffer
+ var out synchronizedBuffer
scope.SetOutput(&out)
defer scope.SetOutput(os.Stderr)
diff --git a/pkg/kube/inject/inherent.go b/pkg/kube/inject/inherent.go
index 2f13e76a..d9a7405b 100644
--- a/pkg/kube/inject/inherent.go
+++ b/pkg/kube/inject/inherent.go
@@ -23,6 +23,7 @@ import (
"github.com/kdubbo/api/annotation"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/types"
)
const (
@@ -76,9 +77,14 @@ func InherentGRPCSecretName(podName string) string {
}
func InherentGRPCSecretNameForMeta(meta metav1.ObjectMeta) string {
- name := meta.Name
- if meta.GenerateName != "" {
- name = meta.GenerateName
+ return InherentGRPCSecretName(meta.Name)
+}
+
+func inherentGRPCPodName(prefix string, requestUID types.UID) string {
+ const suffixLength = 16
+ if len(prefix) > 63-suffixLength {
+ prefix = prefix[:63-suffixLength]
}
- return InherentGRPCSecretName(name)
+ sum := sha256.Sum256([]byte(requestUID))
+ return prefix + hex.EncodeToString(sum[:suffixLength/2])
}
diff --git a/pkg/kube/inject/inherent_test.go b/pkg/kube/inject/inherent_test.go
index cfba8750..9fa7d9a9 100644
--- a/pkg/kube/inject/inherent_test.go
+++ b/pkg/kube/inject/inherent_test.go
@@ -97,7 +97,7 @@ func TestInstallerGRPCEngineTemplateInjectsDirectXDSConnection(t *testing.T) {
assertDirectXDSConnection(t, mergedPod, "app", InherentGRPCSecretNameForMeta(pod.ObjectMeta))
}
-func TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t *testing.T) {
+func TestInstallerGRPCEngineTemplateAllocatesUniqueNameForDeploymentPods(t *testing.T) {
_, currentFile, _, ok := runtime.Caller(0)
if !ok {
t.Fatalf("runtime.Caller() failed")
@@ -136,6 +136,7 @@ func TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t *testing
}
req := InjectionParameters{
pod: pod,
+ requestUID: "admission-1",
templates: templates,
valuesConfig: valuesConfig,
meshConfig: &meshv1alpha1.MeshConfig{
@@ -156,9 +157,26 @@ func TestInstallerGRPCEngineTemplateUsesGenerateNameForDeploymentPods(t *testing
if len(mergedPod.Spec.Containers) != 1 {
t.Fatalf("containers = %d, want original application container only", len(mergedPod.Spec.Containers))
}
- assertDirectXDSConnection(t, mergedPod, "nginx", InherentGRPCSecretNameForMeta(pod.ObjectMeta))
- if got := mergedPod.Spec.Volumes[0].Secret.SecretName; got == InherentGRPCSecretName("") {
- t.Fatalf("secret name = %q, want generateName-based secret", got)
+ assertDirectXDSConnection(t, mergedPod, "nginx", InherentGRPCSecretNameForMeta(mergedPod.ObjectMeta))
+ if mergedPod.Name == "" || !strings.HasPrefix(mergedPod.Name, pod.GenerateName) {
+ t.Fatalf("Pod name = %q, want allocated name with deployment prefix", mergedPod.Name)
+ }
+ patch, err := injectPod(req)
+ if err != nil || !strings.Contains(string(patch), `"path":"/metadata/name"`) {
+ t.Fatalf("injection must patch the allocated Pod name: patch=%s error=%v", patch, err)
+ }
+ other := pod.DeepCopy()
+ otherReq := req
+ otherReq.requestUID = "admission-2"
+ if err := postProcessPod(other, *injectedPod, otherReq); err != nil {
+ t.Fatal(err)
+ }
+ if InherentGRPCSecretNameForMeta(other.ObjectMeta) == InherentGRPCSecretNameForMeta(mergedPod.ObjectMeta) {
+ t.Fatal("replicas must not share credentials")
+ }
+ name := mergedPod.Name
+ if err := postProcessPod(mergedPod, *injectedPod, otherReq); err != nil || mergedPod.Name != name {
+ t.Fatalf("reinvocation must preserve Pod identity: name=%q error=%v", mergedPod.Name, err)
}
}
@@ -540,9 +558,9 @@ func TestInherentGRPCSecretNameFitsKubernetesLengthLimit(t *testing.T) {
}
}
-func TestInherentGRPCSecretNameForMetaPrefersGenerateName(t *testing.T) {
+func TestInherentGRPCSecretNameForMetaUsesPodName(t *testing.T) {
meta := metav1.ObjectMeta{Name: "nginx-95575cc5d-kh98x", GenerateName: "nginx-95575cc5d-"}
- if got, want := InherentGRPCSecretNameForMeta(meta), InherentGRPCSecretName(meta.GenerateName); got != want {
+ if got, want := InherentGRPCSecretNameForMeta(meta), InherentGRPCSecretName(meta.Name); got != want {
t.Fatalf("secret name = %q, want %q", got, want)
}
}
diff --git a/pkg/kube/inject/webhook.go b/pkg/kube/inject/webhook.go
index 1d7a0c86..38f5e398 100644
--- a/pkg/kube/inject/webhook.go
+++ b/pkg/kube/inject/webhook.go
@@ -92,6 +92,7 @@ type ValuesConfig struct {
type InjectionParameters struct {
pod *corev1.Pod
+ requestUID types.UID
deployMeta types.NamespacedName
typeMeta metav1.TypeMeta
templates map[string]*template.Template
@@ -286,6 +287,7 @@ func (wh *Webhook) injectPod(ar *kube.AdmissionReview, path string) *kube.Admiss
deploy, typeMeta := kube.GetDeployMetaFromPod(&pod)
params := InjectionParameters{
pod: &pod,
+ requestUID: req.UID,
deployMeta: deploy,
typeMeta: typeMeta,
templates: wh.Config.Templates,
@@ -464,11 +466,15 @@ func addApplicationContainerConfig(pod *corev1.Pod, req InjectionParameters) err
discoveryAddress = req.proxyConfig.GetDiscoveryAddress()
}
- meta := pod.ObjectMeta
- if req.pod != nil {
- meta = req.pod.ObjectMeta
+ if pod.Name == "" {
+ if pod.GenerateName == "" || req.requestUID == "" {
+ return fmt.Errorf("inherent credentials require a Pod name or an admission request UID and generateName")
+ }
+ // Admission runs before Kubernetes assigns a generated Pod name. Allocate
+ // it here so the immutable volume reference is unique to this Pod.
+ pod.Name = inherentGRPCPodName(pod.GenerateName, req.requestUID)
}
- secretName := InherentGRPCSecretNameForMeta(meta)
+ secretName := InherentGRPCSecretNameForMeta(pod.ObjectMeta)
desiredVolume := corev1.Volume{
Name: InherentXDSVolumeName,
VolumeSource: corev1.VolumeSource{
diff --git a/pkg/kube/kclient/client.go b/pkg/kube/kclient/client.go
index 4cd8115d..ed6b95bd 100644
--- a/pkg/kube/kclient/client.go
+++ b/pkg/kube/kclient/client.go
@@ -22,10 +22,10 @@ import (
"github.com/apache/dubbo-kubernetes/pkg/util/ptr"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
dubbogvr "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvr"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/kubeclient"
types "github.com/apache/dubbo-kubernetes/pkg/config/schema/kubetypes"
+ "github.com/apache/dubbo-kubernetes/pkg/features"
"github.com/apache/dubbo-kubernetes/pkg/kube"
"github.com/apache/dubbo-kubernetes/pkg/kube/controllers"
"github.com/apache/dubbo-kubernetes/pkg/kube/informerfactory"
diff --git a/pkg/kube/util.go b/pkg/kube/util.go
index 85ef7531..1e6a481c 100644
--- a/pkg/kube/util.go
+++ b/pkg/kube/util.go
@@ -18,7 +18,7 @@ package kube
import (
"fmt"
- "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
+ "github.com/apache/dubbo-kubernetes/pkg/features"
"io"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
diff --git a/pkg/model/proxy.go b/pkg/model/proxy.go
index 8f7de132..618f90b0 100644
--- a/pkg/model/proxy.go
+++ b/pkg/model/proxy.go
@@ -23,7 +23,7 @@ import (
core "github.com/kdubbo/xds-api/core/v1"
"strconv"
- networkutil "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/util/network"
+ networkutil "github.com/apache/dubbo-kubernetes/pkg/util/net"
"github.com/apache/dubbo-kubernetes/pkg/util/protomarshal"
meshv1alpha1 "github.com/kdubbo/api/mesh/v1alpha1"
)
diff --git a/pkg/model/xds.go b/pkg/model/xds.go
index 453a24b3..3cb02a8e 100644
--- a/pkg/model/xds.go
+++ b/pkg/model/xds.go
@@ -23,9 +23,6 @@ const (
ListenerType = APITypePrefix + "listener.v1.Listener"
RouteType = APITypePrefix + "route.v1.RouteConfiguration"
SecretType = APITypePrefix + "extensions.transport_sockets.tls.v1.Secret"
-
- HealthInfoType = APITypePrefix + "dubbo.v1.HealthInformation"
- ProxyConfigType = APITypePrefix + "dubbo.mesh.v1alpha1.ProxyConfig"
)
func GetShortType(typeURL string) string {
@@ -40,8 +37,6 @@ func GetShortType(typeURL string) string {
return "EDS"
case SecretType:
return "SDS"
- case ProxyConfigType:
- return "PCDS"
default:
return typeURL
}
@@ -59,8 +54,6 @@ func GetMetricType(typeURL string) string {
return "eds"
case SecretType:
return "sds"
- case ProxyConfigType:
- return "pcds"
default:
return typeURL
}
diff --git a/pkg/spiffe/spiffe.go b/pkg/spiffe/spiffe.go
index 7da58660..4bed5776 100644
--- a/pkg/spiffe/spiffe.go
+++ b/pkg/spiffe/spiffe.go
@@ -292,6 +292,7 @@ func (v *PeerCertVerifier) VerifyPeerCert(rawCerts [][]byte, _ [][]*x509.Certifi
_, err = peerCert.Verify(x509.VerifyOptions{
Roots: rootCertPool,
Intermediates: intCertPool,
+ KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
})
return err
}
diff --git a/pkg/util/net/ip.go b/pkg/util/net/ip.go
index 94344856..af6a68bb 100644
--- a/pkg/util/net/ip.go
+++ b/pkg/util/net/ip.go
@@ -27,3 +27,33 @@ func IsValidIPAddress(ip string) bool {
}
return ipa.IsValid()
}
+
+func AllIPv4(ipAddrs []string) bool {
+ for i := 0; i < len(ipAddrs); i++ {
+ addr, err := netip.ParseAddr(ipAddrs[i])
+ if err != nil {
+ // Should not happen, invalid IP in proxy's IPAddresses slice should have been caught earlier,
+ // skip it to prevent a panic.
+ continue
+ }
+ if !addr.Is4() && addr.Is6() {
+ return false
+ }
+ }
+ return true
+}
+
+func AllIPv6(ipAddrs []string) bool {
+ for i := 0; i < len(ipAddrs); i++ {
+ addr, err := netip.ParseAddr(ipAddrs[i])
+ if err != nil {
+ // Should not happen, invalid IP in proxy's IPAddresses slice should have been caught earlier,
+ // skip it to prevent a panic.
+ continue
+ }
+ if addr.Is4() {
+ return false
+ }
+ }
+ return true
+}
diff --git a/pkg/xds/server.go b/pkg/xds/server.go
index 9a955d79..d848c8b0 100644
--- a/pkg/xds/server.go
+++ b/pkg/xds/server.go
@@ -20,7 +20,7 @@ import (
"strings"
"time"
- dubbogrpc "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/grpc"
+ dubbogrpc "github.com/apache/dubbo-kubernetes/pkg/grpc"
"github.com/apache/dubbo-kubernetes/pkg/model"
"github.com/apache/dubbo-kubernetes/pkg/util/sets"
core "github.com/kdubbo/xds-api/core/v1"
@@ -174,10 +174,6 @@ func Receive(ctx ConnectionContext) {
return
}
if firstRequest {
- if req.TypeUrl == model.HealthInfoType {
- xdsLog.Warnf("%q %s send health check probe before normal xDS request", con.peerAddr, con.conID)
- continue
- }
firstRequest = false
if req.Node == nil || req.Node.Id == "" {
con.errorChan <- status.New(codes.InvalidArgument, "missing node information").Err()
diff --git a/samples/ai-mesh/README.md b/samples/ai/README.md
similarity index 100%
rename from samples/ai-mesh/README.md
rename to samples/ai/README.md
diff --git a/samples/ai-mesh/backends.yaml b/samples/ai/backends.yaml
similarity index 100%
rename from samples/ai-mesh/backends.yaml
rename to samples/ai/backends.yaml
diff --git a/samples/ai-mesh/gateway.yaml b/samples/ai/gateway.yaml
similarity index 100%
rename from samples/ai-mesh/gateway.yaml
rename to samples/ai/gateway.yaml
diff --git a/samples/ai-mesh/routes.yaml b/samples/ai/routes.yaml
similarity index 100%
rename from samples/ai-mesh/routes.yaml
rename to samples/ai/routes.yaml
diff --git a/samples/ai-mesh/services.yaml b/samples/ai/services.yaml
similarity index 100%
rename from samples/ai-mesh/services.yaml
rename to samples/ai/services.yaml
diff --git a/tests/e2e/xdscheck/main.go b/tests/e2e/xdscheck/main.go
new file mode 100644
index 00000000..8d03648c
--- /dev/null
+++ b/tests/e2e/xdscheck/main.go
@@ -0,0 +1,158 @@
+// 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.
+
+// xdscheck verifies a running dubbod using a managed gateway Pod's credentials.
+package main
+
+import (
+ "context"
+ "crypto/tls"
+ "crypto/x509"
+ "encoding/json"
+ "flag"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/pkg/model"
+ core "github.com/kdubbo/xds-api/core/v1"
+ discovery "github.com/kdubbo/xds-api/service/discovery/v1"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/status"
+ "google.golang.org/protobuf/encoding/protojson"
+ "google.golang.org/protobuf/proto"
+)
+
+func main() {
+ dir := flag.String("credentials", "", "directory containing the gateway Pod Secret data")
+ target := flag.String("target", "127.0.0.1:36012", "forwarded secure ADS address")
+ serverName := flag.String("server-name", "dubbod.dubbo-system.svc", "TLS server name")
+ flag.Parse()
+ if err := verify(*dir, *target, *serverName); err != nil {
+ fmt.Fprintln(os.Stderr, err)
+ os.Exit(1)
+ }
+}
+
+func verify(dir, target, serverName string) error {
+ bootstrap, err := os.ReadFile(filepath.Join(dir, "grpc-bootstrap.json"))
+ if err != nil {
+ return err
+ }
+ var envelope struct{ Node json.RawMessage }
+ if err := json.Unmarshal(bootstrap, &envelope); err != nil {
+ return err
+ }
+ node := &core.Node{}
+ if err := protojson.Unmarshal(envelope.Node, node); err != nil {
+ return err
+ }
+ // dxgate builds a Router node from the shared workload bootstrap.
+ parts := strings.Split(node.Id, "~")
+ if len(parts) != 4 {
+ return fmt.Errorf("unexpected bootstrap node ID format")
+ }
+ parts[0] = "router"
+ node.Id = strings.Join(parts, "~")
+ pair, err := tls.LoadX509KeyPair(filepath.Join(dir, "cert-chain.pem"), filepath.Join(dir, "key.pem"))
+ if err != nil {
+ return err
+ }
+ root, err := os.ReadFile(filepath.Join(dir, "root-cert.pem"))
+ if err != nil {
+ return err
+ }
+ roots := x509.NewCertPool()
+ if !roots.AppendCertsFromPEM(root) {
+ return fmt.Errorf("invalid root bundle")
+ }
+ for _, delta := range []bool{false, true} {
+ for _, scenario := range []string{"valid", "anonymous", "unknown pod", "unknown type"} {
+ tlsConfig := &tls.Config{MinVersion: tls.VersionTLS12, ServerName: serverName, RootCAs: roots, Certificates: []tls.Certificate{pair}}
+ n := proto.Clone(node).(*core.Node)
+ typeURL := model.SecretType
+ want := codes.OK
+ switch scenario {
+ case "anonymous":
+ tlsConfig.Certificates = nil
+ want = codes.Unauthenticated
+ case "unknown pod":
+ parts := strings.Split(n.Id, "~")
+ if len(parts) != 4 {
+ return fmt.Errorf("unexpected node ID format")
+ }
+ parts[2] = "nonexistent.e2e"
+ n.Id = strings.Join(parts, "~")
+ want = codes.PermissionDenied
+ case "unknown type":
+ typeURL = "type.googleapis.com/unsupported.Resource"
+ want = codes.InvalidArgument
+ }
+ err := query(target, tlsConfig, n, typeURL, delta)
+ if status.Code(err) != want {
+ return fmt.Errorf("delta=%v %s: got %v, want %s", delta, scenario, err, want)
+ }
+ fmt.Printf("PASS delta=%v %s: %s\n", delta, scenario, want)
+ }
+ }
+ return nil
+}
+
+func query(target string, tlsConfig *tls.Config, node *core.Node, typeURL string, delta bool) error {
+ conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)))
+ if err != nil {
+ return err
+ }
+ defer conn.Close()
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ client := discovery.NewAggregatedDiscoveryServiceClient(conn)
+ if delta {
+ stream, err := client.DeltaAggregatedResources(ctx)
+ if err != nil {
+ return err
+ }
+ if err := stream.Send(&discovery.DeltaDiscoveryRequest{Node: node, TypeUrl: typeURL, ResourceNamesSubscribe: []string{"default", "ROOTCA"}}); err != nil {
+ return err
+ }
+ res, err := stream.Recv()
+ if err != nil {
+ return err
+ }
+ if len(res.Resources) != 2 || res.Nonce == "" {
+ return fmt.Errorf("expected two SDS resources and a nonce")
+ }
+ return stream.Send(&discovery.DeltaDiscoveryRequest{TypeUrl: typeURL, ResponseNonce: res.Nonce})
+ }
+ stream, err := client.StreamAggregatedResources(ctx)
+ if err != nil {
+ return err
+ }
+ if err := stream.Send(&discovery.DiscoveryRequest{Node: node, TypeUrl: typeURL, ResourceNames: []string{"default", "ROOTCA"}}); err != nil {
+ return err
+ }
+ res, err := stream.Recv()
+ if err != nil {
+ return err
+ }
+ if len(res.Resources) != 2 || res.Nonce == "" {
+ return fmt.Errorf("expected two SDS resources and a nonce")
+ }
+ return stream.Send(&discovery.DiscoveryRequest{TypeUrl: typeURL, ResourceNames: []string{"default", "ROOTCA"}, ResponseNonce: res.Nonce, VersionInfo: res.VersionInfo})
+}