This is an automated email from the ASF dual-hosted git repository.
mfordjody pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new 8052b378 Complete KRT controllers framework (#932)
8052b378 is described below
commit 8052b3789b895bf61df69a934343217d6f51d8db
Author: mfordjody <[email protected]>
AuthorDate: Tue Jun 30 09:18:41 2026 +0800
Complete KRT controllers framework (#932)
---
dubbod/discovery/pkg/bootstrap/server.go | 3 +
.../discovery/pkg/config/kube/crdclient/client.go | 9 +-
.../pkg/config/kube/crdclient/client_test.go | 72 +++++
.../pkg/config/kube/crdclient/types.gen.go | 46 +++
.../discovery/pkg/config/kube/file/controller.go | 3 +-
.../pkg/config/kube/file/controller_test.go | 92 ++++++
.../pkg/config/kube/gateway/controller.go | 137 ++++++---
.../pkg/config/kube/gateway/controller_test.go | 336 +++++++++++++++++++++
.../pkg/config/kube/gateway/gateways_collection.go | 38 ++-
.../pkg/config/kube/gateway/referencegrants.go | 187 ++++++++++++
dubbod/discovery/pkg/model/context.go | 2 +
pkg/config/schema/collections/collections.gen.go | 23 ++
pkg/config/schema/gvk/resources.gen.go | 10 +
pkg/config/schema/gvr/resources.gen.go | 6 +
pkg/config/schema/kind/resources.gen.go | 5 +
pkg/config/schema/kubeclient/resources.gen.go | 14 +
pkg/config/schema/kubetypes/resources.gen.go | 3 +
pkg/config/schema/metadata.yaml | 9 +
18 files changed, 938 insertions(+), 57 deletions(-)
diff --git a/dubbod/discovery/pkg/bootstrap/server.go
b/dubbod/discovery/pkg/bootstrap/server.go
index d904759d..fd19f73a 100644
--- a/dubbod/discovery/pkg/bootstrap/server.go
+++ b/dubbod/discovery/pkg/bootstrap/server.go
@@ -505,6 +505,8 @@ func (s *Server) initRegistryEventHandlers() {
configKind = kind.PeerAuthentication
case "RequestAuthentication":
configKind = kind.RequestAuthentication
+ case "ReferenceGrant":
+ configKind = kind.ReferenceGrant
case "GatewayClass":
configKind = kind.GatewayClass
case "KubernetesGateway":
@@ -538,6 +540,7 @@ func (s *Server) initRegistryEventHandlers() {
configKind == kind.AuthorizationPolicy ||
configKind == kind.HTTPRoute ||
configKind == kind.BackendTLSPolicy ||
+ configKind == kind.ReferenceGrant ||
configKind == kind.CircuitBreakerPolicy
// Trigger ConfigUpdate to push changes to all connected proxies
diff --git a/dubbod/discovery/pkg/config/kube/crdclient/client.go
b/dubbod/discovery/pkg/config/kube/crdclient/client.go
index 6e6f7cb7..28760a93 100644
--- a/dubbod/discovery/pkg/config/kube/crdclient/client.go
+++ b/dubbod/discovery/pkg/config/kube/crdclient/client.go
@@ -19,10 +19,11 @@ package crdclient
import (
"encoding/json"
"fmt"
- "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
"sync"
"time"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/resource"
+
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
"github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/schema/collection"
@@ -249,7 +250,8 @@ func (cl *Client) Schemas() collection.Schemas {
}
func (cl *Client) RegisterEventHandler(kind config.GroupVersionKind, handler
model.EventHandler) {
- if c, ok := cl.kind(kind); ok {
+ cl.kindsMu.Lock()
+ if c, ok := cl.kinds[kind]; ok {
c.handlers = append(c.handlers,
c.collection.RegisterBatch(func(o []krt.Event[config.Config]) {
for _, event := range o {
switch event.Event {
@@ -262,8 +264,11 @@ func (cl *Client) RegisterEventHandler(kind
config.GroupVersionKind, handler mod
}
}
}, false))
+ cl.kinds[kind] = c
+ cl.kindsMu.Unlock()
return
}
+ cl.kindsMu.Unlock()
cl.logger.Warnf("unknown type: %s", kind)
}
diff --git a/dubbod/discovery/pkg/config/kube/crdclient/client_test.go
b/dubbod/discovery/pkg/config/kube/crdclient/client_test.go
new file mode 100644
index 00000000..c3439123
--- /dev/null
+++ b/dubbod/discovery/pkg/config/kube/crdclient/client_test.go
@@ -0,0 +1,72 @@
+// 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 crdclient
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+func TestRegisterEventHandlerPersistsRegistration(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+
+ routes := krt.NewStaticCollection[config.Config](nil, nil,
krt.WithStop(stop))
+ cl := &Client{
+ kinds: map[config.GroupVersionKind]nsStore{
+ gvk.HTTPRoute: {
+ collection: routes,
+ index: krt.NewNamespaceIndex(routes),
+ },
+ },
+ }
+ events := make(chan model.Event, 1)
+ cl.RegisterEventHandler(gvk.HTTPRoute, func(_, _ config.Config, event
model.Event) {
+ events <- event
+ })
+
+ if got := len(cl.kinds[gvk.HTTPRoute].handlers); got != 1 {
+ t.Fatalf("registered handlers = %d, want 1", got)
+ }
+
+ routes.UpdateObject(config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "orders",
+ Namespace: "app",
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{},
+ })
+ expectEvent(t, events, model.EventAdd)
+}
+
+func expectEvent(t *testing.T, events <-chan model.Event, want model.Event) {
+ t.Helper()
+ select {
+ case got := <-events:
+ if got != want {
+ t.Fatalf("event = %v, want %v", got, want)
+ }
+ case <-time.After(time.Second):
+ t.Fatalf("expected event handler to receive %v event", want)
+ }
+}
diff --git a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
index 3d9de0e7..66927ce2 100755
--- a/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
+++ b/dubbod/discovery/pkg/config/kube/crdclient/types.gen.go
@@ -28,6 +28,7 @@ import (
k8sioapipolicyv1 "k8s.io/api/policy/v1"
k8sioapiextensionsapiserverpkgapisapiextensionsv1
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
sigsk8siogatewayapiapisv1 "sigs.k8s.io/gateway-api/apis/v1"
+ sigsk8siogatewayapiapisv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
)
func create(c kube.Client, cfg config.Config, objMeta metav1.ObjectMeta)
(metav1.Object, error) {
@@ -67,6 +68,11 @@ func create(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapisecurityv1alpha3.PeerAuthentication)),
}, metav1.CreateOptions{})
+ case gvk.ReferenceGrant:
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(cfg.Namespace).Create(context.TODO(),
&sigsk8siogatewayapiapisv1beta1.ReferenceGrant{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*sigsk8siogatewayapiapisv1beta1.ReferenceGrantSpec)),
+ }, metav1.CreateOptions{})
case gvk.RequestAuthentication:
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(cfg.Namespace).Create(context.TODO(),
&apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication{
ObjectMeta: objMeta,
@@ -114,6 +120,11 @@ func update(c kube.Client, cfg config.Config, objMeta
metav1.ObjectMeta) (metav1
ObjectMeta: objMeta,
Spec:
*(cfg.Spec.(*githubcomkdubboapisecurityv1alpha3.PeerAuthentication)),
}, metav1.UpdateOptions{})
+ case gvk.ReferenceGrant:
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(cfg.Namespace).Update(context.TODO(),
&sigsk8siogatewayapiapisv1beta1.ReferenceGrant{
+ ObjectMeta: objMeta,
+ Spec:
*(cfg.Spec.(*sigsk8siogatewayapiapisv1beta1.ReferenceGrantSpec)),
+ }, metav1.UpdateOptions{})
case gvk.RequestAuthentication:
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(cfg.Namespace).Update(context.TODO(),
&apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication{
ObjectMeta: objMeta,
@@ -281,6 +292,21 @@ func patch(c kube.Client, orig config.Config, origMeta
metav1.ObjectMeta, mod co
}
return
c.Dubbo().SecurityV1alpha3().PeerAuthentications(orig.Namespace).
Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
+ case gvk.ReferenceGrant:
+ oldRes := &sigsk8siogatewayapiapisv1beta1.ReferenceGrant{
+ ObjectMeta: origMeta,
+ Spec:
*(orig.Spec.(*sigsk8siogatewayapiapisv1beta1.ReferenceGrantSpec)),
+ }
+ modRes := &sigsk8siogatewayapiapisv1beta1.ReferenceGrant{
+ ObjectMeta: modMeta,
+ Spec:
*(mod.Spec.(*sigsk8siogatewayapiapisv1beta1.ReferenceGrantSpec)),
+ }
+ patchBytes, err := genPatchBytes(oldRes, modRes, typ)
+ if err != nil {
+ return nil, err
+ }
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(orig.Namespace).
+ Patch(context.TODO(), orig.Name, typ, patchBytes,
metav1.PatchOptions{FieldManager: "pilot-discovery"})
case gvk.RequestAuthentication:
oldRes :=
&apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication{
ObjectMeta: origMeta,
@@ -321,6 +347,8 @@ func delete(c kube.Client, typ config.GroupVersionKind,
name, namespace string,
return
c.GatewayAPI().GatewayV1().Gateways(namespace).Delete(context.TODO(), name,
deleteOptions)
case gvk.PeerAuthentication:
return
c.Dubbo().SecurityV1alpha3().PeerAuthentications(namespace).Delete(context.TODO(),
name, deleteOptions)
+ case gvk.ReferenceGrant:
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(namespace).Delete(context.TODO(),
name, deleteOptions)
case gvk.RequestAuthentication:
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(namespace).Delete(context.TODO(),
name, deleteOptions)
default:
@@ -698,6 +726,24 @@ var translationMap = map[config.GroupVersionKind]func(r
runtime.Object) config.C
Status: &obj.Status,
}
},
+ gvk.ReferenceGrant: func(r runtime.Object) config.Config {
+ obj := r.(*sigsk8siogatewayapiapisv1beta1.ReferenceGrant)
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.ReferenceGrant,
+ Name: obj.Name,
+ Namespace: obj.Namespace,
+ Labels: obj.Labels,
+ Annotations: obj.Annotations,
+ ResourceVersion: obj.ResourceVersion,
+ CreationTimestamp: obj.CreationTimestamp.Time,
+ OwnerReferences: obj.OwnerReferences,
+ UID: string(obj.UID),
+ Generation: obj.Generation,
+ },
+ Spec: &obj.Spec,
+ }
+ },
gvk.RequestAuthentication: func(r runtime.Object) config.Config {
obj :=
r.(*apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication)
return config.Config{
diff --git a/dubbod/discovery/pkg/config/kube/file/controller.go
b/dubbod/discovery/pkg/config/kube/file/controller.go
index 43ef5ded..6c7613e0 100644
--- a/dubbod/discovery/pkg/config/kube/file/controller.go
+++ b/dubbod/discovery/pkg/config/kube/file/controller.go
@@ -154,6 +154,7 @@ func (c *Controller) RegisterEventHandler(typ
config.GroupVersionKind, handler m
}
}, false),
)
+ c.data[typ] = data
}
}
@@ -183,7 +184,7 @@ func (c ConfigKind) ResourceName() string {
return c.GroupVersionKind.String() + "/" + c.Name
}
- return c.GroupVersionKind.String() + "/" + c.Namespace + c.Name
+ return c.GroupVersionKind.String() + "/" + c.Namespace + "/" + c.Name
}
func (c ConfigKind) Equals(other ConfigKind) bool {
diff --git a/dubbod/discovery/pkg/config/kube/file/controller_test.go
b/dubbod/discovery/pkg/config/kube/file/controller_test.go
new file mode 100644
index 00000000..c61d97d0
--- /dev/null
+++ b/dubbod/discovery/pkg/config/kube/file/controller_test.go
@@ -0,0 +1,92 @@
+// 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 file
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+func TestRegisterEventHandlerPersistsRegistration(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+
+ routes := krt.NewStaticCollection[config.Config](nil, nil,
krt.WithStop(stop))
+ c := &Controller{
+ data: map[config.GroupVersionKind]kindStore{
+ gvk.HTTPRoute: {
+ collection: routes,
+ index: krt.NewNamespaceIndex(routes),
+ },
+ },
+ }
+ events := make(chan model.Event, 1)
+ c.RegisterEventHandler(gvk.HTTPRoute, func(_, _ config.Config, event
model.Event) {
+ events <- event
+ })
+
+ if got := len(c.data[gvk.HTTPRoute].handlers); got != 1 {
+ t.Fatalf("registered handlers = %d, want 1", got)
+ }
+
+ routes.UpdateObject(config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "orders",
+ Namespace: "app",
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{},
+ })
+ expectEvent(t, events, model.EventAdd)
+}
+
+func TestConfigKindResourceNameUsesNamespaceSeparator(t *testing.T) {
+ first := ConfigKind{&config.Config{Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "c",
+ Namespace: "ab",
+ }}}
+ second := ConfigKind{&config.Config{Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "bc",
+ Namespace: "a",
+ }}}
+
+ if got, want := first.ResourceName(), gvk.HTTPRoute.String()+"/ab/c";
got != want {
+ t.Fatalf("resource name = %q, want %q", got, want)
+ }
+ if first.ResourceName() == second.ResourceName() {
+ t.Fatalf("resource names collided: %q", first.ResourceName())
+ }
+}
+
+func expectEvent(t *testing.T, events <-chan model.Event, want model.Event) {
+ t.Helper()
+ select {
+ case got := <-events:
+ if got != want {
+ t.Fatalf("event = %v, want %v", got, want)
+ }
+ case <-time.After(time.Second):
+ t.Fatalf("expected event handler to receive %v event", want)
+ }
+}
diff --git a/dubbod/discovery/pkg/config/kube/gateway/controller.go
b/dubbod/discovery/pkg/config/kube/gateway/controller.go
index 36eb733a..37c95150 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/controller.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/controller.go
@@ -41,6 +41,7 @@ import (
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
gateway "sigs.k8s.io/gateway-api/apis/v1"
+ gatewayv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
)
var log = dubbolog.RegisterScope("gateway", "gateway-api controller")
@@ -55,24 +56,33 @@ type Controller struct {
gatewayContext krt.RecomputeProtected[*atomic.Pointer[Context]]
stop chan struct{}
xdsUpdater model.XDSUpdater
- handlers []krt.HandlerRegistration
outputs Outputs
+ data map[config.GroupVersionKind]kindStore
domainSuffix string
status *status.StatusCollections
inputs Inputs
}
type Outputs struct {
- Gateways krt.Collection[Gateway]
+ Gateways krt.Collection[config.Config]
+ HTTPRoutes krt.Collection[config.Config]
+ ReferenceGrants referenceGrantStore
+}
+
+type kindStore struct {
+ collection krt.Collection[config.Config]
+ index krt.Index[string, config.Config]
+ handlers []krt.HandlerRegistration
}
type Inputs struct {
Services krt.Collection[*corev1.Service]
Namespaces krt.Collection[*corev1.Namespace]
- GatewayClasses krt.Collection[*gateway.GatewayClass]
- Gateways krt.Collection[*gateway.Gateway]
- HTTPRoutes krt.Collection[*gateway.HTTPRoute]
+ GatewayClasses krt.Collection[*gateway.GatewayClass]
+ Gateways krt.Collection[*gateway.Gateway]
+ HTTPRoutes krt.Collection[*gateway.HTTPRoute]
+ ReferenceGrants krt.Collection[*gatewayv1beta1.ReferenceGrant]
}
func NewController(kc kube.Client, waitForCRD func(class
schema.GroupVersionResource, stop <-chan struct{}) bool, options
controller.Options, xdsUpdater model.XDSUpdater) *Controller {
@@ -99,6 +109,13 @@ func NewController(kc kube.Client, waitForCRD func(class
schema.GroupVersionReso
GatewayClasses: buildClient[*gateway.GatewayClass](c, kc,
gvr.GatewayClass, opts, "informer/GatewayClasses"),
Gateways: buildClient[*gateway.Gateway](c, kc,
gvr.KubernetesGateway, opts, "informer/Gateways"),
HTTPRoutes: buildClient[*gateway.HTTPRoute](c, kc,
gvr.HTTPRoute, opts, "informer/HTTPRoutes"),
+ ReferenceGrants: buildClient[*gatewayv1beta1.ReferenceGrant](
+ c,
+ kc,
+ gvr.ReferenceGrant,
+ opts,
+ "informer/ReferenceGrants",
+ ),
}
GatewayClassStatus, GatewayClasses :=
GatewayClassesCollection(inputs.GatewayClasses, opts)
@@ -113,18 +130,37 @@ func NewController(kc kube.Client, waitForCRD func(class
schema.GroupVersionReso
GatewayFinalStatus := FinalGatewayStatusCollection(GatewaysStatus, opts)
status.RegisterStatus(c.status, GatewayFinalStatus, GetStatus)
- handlers := []krt.HandlerRegistration{}
+ ReferenceGrants :=
newReferenceGrantStore(ReferenceGrantsCollection(inputs.ReferenceGrants, opts))
+
+ HTTPRoutes := krt.NewCollection(inputs.HTTPRoutes, func(ctx
krt.HandlerContext, hr *gateway.HTTPRoute) *config.Config {
+ cfg := convertHTTPRouteToConfig(hr)
+ return &cfg
+ }, opts.WithName("HTTPRoutes")...)
+
outputs := Outputs{
- Gateways: Gateways,
+ Gateways: Gateways,
+ HTTPRoutes: HTTPRoutes,
+ ReferenceGrants: ReferenceGrants,
+ }
+ data := map[config.GroupVersionKind]kindStore{
+ gvk.KubernetesGateway: newKindStore(Gateways),
+ gvk.HTTPRoute: newKindStore(HTTPRoutes),
}
c.outputs = outputs
- c.handlers = handlers
+ c.data = data
c.inputs = inputs
return c
}
+func newKindStore(collection krt.Collection[config.Config]) kindStore {
+ return kindStore{
+ collection: collection,
+ index: krt.NewNamespaceIndex(collection),
+ }
+}
+
func (c *Controller) Reconcile(ps *model.PushContext) {
ctx := NewGatewayContext(ps, c.cluster)
c.gatewayContext.Modify(func(i **atomic.Pointer[Context]) {
@@ -145,25 +181,23 @@ func (c *Controller) SetStatusWrite(enabled bool,
statusManager *status.Manager)
}
func (c *Controller) Get(typ config.GroupVersionKind, name, namespace string)
*config.Config {
+ if data, ok := c.data[typ]; ok {
+ if namespace == "" {
+ return data.collection.GetKey(name)
+ }
+ return data.collection.GetKey(namespace + "/" + name)
+ }
return nil
}
func (c *Controller) List(typ config.GroupVersionKind, namespace string)
[]config.Config {
- switch typ {
- case gvk.HTTPRoute:
- // Convert HTTPRoute collection to config.Config list
- httpRoutes := c.inputs.HTTPRoutes.List()
- result := make([]config.Config, 0, len(httpRoutes))
- for _, hr := range httpRoutes {
- cfg := convertHTTPRouteToConfig(hr)
- if namespace == "" || cfg.Namespace == namespace {
- result = append(result, cfg)
- }
+ if data, ok := c.data[typ]; ok {
+ if namespace == "" {
+ return data.collection.List()
}
- return result
- default:
- return nil
+ return data.index.Lookup(namespace)
}
+ return nil
}
func convertHTTPRouteToConfig(hr *gateway.HTTPRoute) config.Config {
@@ -180,8 +214,8 @@ func convertHTTPRouteToConfig(hr *gateway.HTTPRoute)
config.Config {
UID: string(hr.UID),
Generation: hr.Generation,
},
- Spec: &hr.Spec,
- Status: &hr.Status,
+ Spec: hr.Spec.DeepCopy(),
+ Status: hr.Status.DeepCopy(),
}
}
@@ -206,36 +240,57 @@ func (c *Controller) Delete(typ config.GroupVersionKind,
name, namespace string,
}
func (c *Controller) HasSynced() bool {
- // Check if all input collections are synced
- if !c.inputs.Services.HasSynced() {
- return false
- }
- if !c.inputs.Namespaces.HasSynced() {
- return false
- }
- if !c.inputs.GatewayClasses.HasSynced() {
+ if !collectionHasSynced(c.inputs.Services) ||
+ !collectionHasSynced(c.inputs.Namespaces) ||
+ !collectionHasSynced(c.inputs.GatewayClasses) ||
+ !collectionHasSynced(c.inputs.Gateways) ||
+ !collectionHasSynced(c.inputs.HTTPRoutes) ||
+ !collectionHasSynced(c.inputs.ReferenceGrants) {
return false
}
- if !c.inputs.Gateways.HasSynced() {
- return false
- }
- if !c.inputs.HTTPRoutes.HasSynced() {
+
+ if c.outputs.ReferenceGrants.Collection != nil &&
!c.outputs.ReferenceGrants.Collection.HasSynced() {
return false
}
- for _, h := range c.handlers {
- if !h.HasSynced() {
+
+ for _, data := range c.data {
+ if !data.collection.HasSynced() {
return false
}
+ for _, handler := range data.handlers {
+ if !handler.HasSynced() {
+ return false
+ }
+ }
}
return true
}
+func collectionHasSynced[T any](collection krt.Collection[T]) bool {
+ return collection == nil || collection.HasSynced()
+}
+
func (c *Controller) RegisterEventHandler(typ config.GroupVersionKind, handler
model.EventHandler) {
- // We do not do event handler registration this way, and instead
directly call the XDS Updated.
+ if data, ok := c.data[typ]; ok {
+ data.handlers = append(data.handlers,
data.collection.RegisterBatch(func(evs []krt.Event[config.Config]) {
+ for _, event := range evs {
+ switch event.Event {
+ case controllers.EventAdd:
+ handler(config.Config{}, *event.New,
model.EventAdd)
+ case controllers.EventUpdate:
+ handler(*event.Old, *event.New,
model.EventUpdate)
+ case controllers.EventDelete:
+ handler(config.Config{}, *event.Old,
model.EventDelete)
+ }
+ }
+ }, false))
+ c.data[typ] = data
+ }
}
func (c *Controller) Schemas() collection.Schemas {
return collection.SchemasFor(
+ collections.KubernetesGateway,
collections.HTTPRoute,
)
}
@@ -255,7 +310,7 @@ func (c *Controller) Run(stop <-chan struct{}) {
}
func (c *Controller) SecretAllowed(ourKind config.GroupVersionKind,
resourceName string, namespace string) bool {
- return true // TODO
+ return c.outputs.ReferenceGrants.SecretAllowed(nil, ourKind,
resourceName, namespace)
}
func buildClient[I controllers.ComparableObject](
@@ -269,8 +324,8 @@ func buildClient[I controllers.ComparableObject](
ObjectFilter: kubetypes.ComposeFilters(kc.ObjectFilter(),
c.inRevision),
}
- // Gateway and HTTPRoute are not filtered by revision, they need to
select tags as well
- if res == gvr.KubernetesGateway || res == gvr.HTTPRoute {
+ // Gateway API resources select revisions through their
parent/attachment semantics.
+ if res == gvr.KubernetesGateway || res == gvr.HTTPRoute || res ==
gvr.ReferenceGrant {
filter.ObjectFilter = kc.ObjectFilter()
}
diff --git a/dubbod/discovery/pkg/config/kube/gateway/controller_test.go
b/dubbod/discovery/pkg/config/kube/gateway/controller_test.go
new file mode 100644
index 00000000..c24afa58
--- /dev/null
+++ b/dubbod/discovery/pkg/config/kube/gateway/controller_test.go
@@ -0,0 +1,336 @@
+//
+// 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 gateway
+
+import (
+ "testing"
+ "time"
+
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
+ "github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+ gatewayv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
+)
+
+func TestGatewayControllerConfigStore(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+
+ gatewayConfigs := krt.NewStaticCollection[config.Config](nil,
[]config.Config{
+ {
+ Meta: config.Meta{
+ GroupVersionKind: gvk.KubernetesGateway,
+ Name: "public",
+ Namespace: "app",
+ },
+ Spec: &gatewayv1.GatewaySpec{},
+ },
+ }, krt.WithStop(stop))
+ httpRouteConfigs := krt.NewStaticCollection[config.Config](nil,
[]config.Config{
+ {
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "orders",
+ Namespace: "app",
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{},
+ },
+ {
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "admin",
+ Namespace: "ops",
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{},
+ },
+ }, krt.WithStop(stop))
+ c := &Controller{
+ data: map[config.GroupVersionKind]kindStore{
+ gvk.KubernetesGateway: newKindStore(gatewayConfigs),
+ gvk.HTTPRoute: newKindStore(httpRouteConfigs),
+ },
+ }
+
+ if _, ok := c.Schemas().FindByGroupVersionKind(gvk.KubernetesGateway);
!ok {
+ t.Fatal("schemas missing KubernetesGateway")
+ }
+ if _, ok := c.Schemas().FindByGroupVersionKind(gvk.HTTPRoute); !ok {
+ t.Fatal("schemas missing HTTPRoute")
+ }
+
+ if got := c.Get(gvk.KubernetesGateway, "public", "app"); got == nil ||
got.Name != "public" {
+ t.Fatalf("gateway Get() = %#v, want public", got)
+ }
+ if got := c.List(gvk.HTTPRoute, "app"); len(got) != 1 || got[0].Name !=
"orders" {
+ t.Fatalf("HTTPRoute namespace List() = %#v, want only
app/orders", got)
+ }
+ if got := c.List(gvk.HTTPRoute, metav1.NamespaceAll); len(got) != 2 {
+ t.Fatalf("HTTPRoute all-namespaces List() = %d, want 2",
len(got))
+ }
+}
+
+func TestGatewayControllerRegisterEventHandler(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+
+ routes := krt.NewStaticCollection[config.Config](nil, nil,
krt.WithStop(stop))
+ c := &Controller{
+ data: map[config.GroupVersionKind]kindStore{
+ gvk.HTTPRoute: newKindStore(routes),
+ },
+ }
+ events := make(chan model.Event, 1)
+ c.RegisterEventHandler(gvk.HTTPRoute, func(_, _ config.Config, event
model.Event) {
+ events <- event
+ })
+
+ routes.UpdateObject(config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "orders",
+ Namespace: "app",
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{},
+ })
+ expectEvent(t, events, model.EventAdd)
+
+ routes.UpdateObject(config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.HTTPRoute,
+ Name: "orders",
+ Namespace: "app",
+ Generation: 2,
+ },
+ Spec: &gatewayv1.HTTPRouteSpec{},
+ })
+ expectEvent(t, events, model.EventUpdate)
+
+ routes.DeleteObject("app/orders")
+ expectEvent(t, events, model.EventDelete)
+}
+
+func TestGatewayControllerSecretAllowed(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+ opts := krt.NewOptionsBuilder(stop, "test", nil)
+
+ secretName := gatewayv1.ObjectName("tls-cert")
+ grants := krt.NewStaticCollection[*gatewayv1beta1.ReferenceGrant](nil,
[]*gatewayv1beta1.ReferenceGrant{
+ {
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "allow-specific",
+ Namespace: "certs",
+ },
+ Spec: gatewayv1beta1.ReferenceGrantSpec{
+ From: []gatewayv1beta1.ReferenceGrantFrom{{
+ Group:
gatewayv1.Group(gvk.KubernetesGateway.Group),
+ Kind:
gatewayv1.Kind(gvk.KubernetesGateway.Kind),
+ Namespace: "ingress",
+ }},
+ To: []gatewayv1beta1.ReferenceGrantTo{{
+ Group:
gatewayv1.Group(gvk.Secret.Group),
+ Kind: gatewayv1.Kind(gvk.Secret.Kind),
+ Name: &secretName,
+ }},
+ },
+ },
+ {
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "allow-all",
+ Namespace: "shared-certs",
+ },
+ Spec: gatewayv1beta1.ReferenceGrantSpec{
+ From: []gatewayv1beta1.ReferenceGrantFrom{{
+ Group:
gatewayv1.Group(gvk.KubernetesGateway.Group),
+ Kind:
gatewayv1.Kind(gvk.KubernetesGateway.Kind),
+ Namespace: "ingress",
+ }},
+ To: []gatewayv1beta1.ReferenceGrantTo{{
+ Group:
gatewayv1.Group(gvk.Secret.Group),
+ Kind: gatewayv1.Kind(gvk.Secret.Kind),
+ }},
+ },
+ },
+ }, krt.WithStop(stop))
+ referenceGrants :=
newReferenceGrantStore(ReferenceGrantsCollection(grants, opts))
+ waitSynced(t, referenceGrants.Collection)
+ c := &Controller{
+ outputs: Outputs{
+ ReferenceGrants: referenceGrants,
+ },
+ }
+
+ if !c.SecretAllowed(gvk.KubernetesGateway,
"kubernetes-gateway://certs/tls-cert", "ingress") {
+ t.Fatal("SecretAllowed() = false, want true for specifically
granted Secret")
+ }
+ if !c.SecretAllowed(gvk.KubernetesGateway,
"kubernetes-gateway://shared-certs/any-cert", "ingress") {
+ t.Fatal("SecretAllowed() = false, want true for wildcard Secret
grant")
+ }
+ if c.SecretAllowed(gvk.KubernetesGateway,
"kubernetes-gateway://certs/other-cert", "ingress") {
+ t.Fatal("SecretAllowed() = true, want false for ungranted
Secret name")
+ }
+ if c.SecretAllowed(gvk.KubernetesGateway,
"kubernetes-gateway://certs/tls-cert", "other") {
+ t.Fatal("SecretAllowed() = true, want false for ungranted
source namespace")
+ }
+ if c.SecretAllowed(gvk.HTTPRoute,
"kubernetes-gateway://certs/tls-cert", "ingress") {
+ t.Fatal("SecretAllowed() = true, want false for ungranted
source kind")
+ }
+}
+
+func TestGatewayControllerHasSyncedChecksInputCollections(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+
+ services := krt.NewStaticCollection[*corev1.Service](unsyncedSyncer{},
nil, krt.WithStop(stop))
+ routes := krt.NewStaticCollection[config.Config](nil, nil,
krt.WithStop(stop))
+ c := &Controller{
+ inputs: Inputs{
+ Services: services,
+ },
+ data: map[config.GroupVersionKind]kindStore{
+ gvk.HTTPRoute: newKindStore(routes),
+ },
+ }
+
+ if c.HasSynced() {
+ t.Fatal("HasSynced() = true, want false while input collection
is still syncing")
+ }
+}
+
+func TestGatewaysCollectionEmitsManagedGatewayConfig(t *testing.T) {
+ stop := make(chan struct{})
+ t.Cleanup(func() { close(stop) })
+ opts := krt.NewOptionsBuilder(stop, "test", nil)
+
+ gateways := krt.NewStaticCollection[*gatewayv1.Gateway](nil,
[]*gatewayv1.Gateway{
+ {
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "public",
+ Namespace: "app",
+ },
+ Spec: gatewayv1.GatewaySpec{
+ GatewayClassName:
gatewayv1.ObjectName(features.GatewayAPIDefaultGatewayClass),
+ },
+ },
+ {
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "foreign",
+ Namespace: "app",
+ },
+ Spec: gatewayv1.GatewaySpec{
+ GatewayClassName: "external",
+ },
+ },
+ }, krt.WithStop(stop))
+ gatewayClasses := krt.NewStaticCollection[GatewayClass](nil, nil,
krt.WithStop(stop))
+
+ statuses, configs := GatewaysCollection(gateways, gatewayClasses, opts)
+ waitSynced(t, configs)
+ waitSynced(t, statuses)
+
+ got := configs.List()
+ if len(got) != 1 {
+ t.Fatalf("gateway configs = %#v, want only managed gateway",
got)
+ }
+ cfg := got[0]
+ if cfg.GroupVersionKind != gvk.KubernetesGateway || cfg.Name !=
"public" || cfg.Namespace != "app" {
+ t.Fatalf("gateway config meta = %#v, want app/public
KubernetesGateway", cfg.Meta)
+ }
+ if _, ok := cfg.Spec.(*gatewayv1.GatewaySpec); !ok {
+ t.Fatalf("gateway config spec type = %T, want
*gatewayv1.GatewaySpec", cfg.Spec)
+ }
+
+ statusList := statuses.List()
+ if len(statusList) != 1 {
+ t.Fatalf("gateway statuses = %#v, want only managed gateway
status", statusList)
+ }
+ status := statusList[0].Status
+ if !hasGatewayCondition(status,
string(gatewayv1.GatewayConditionAccepted), metav1.ConditionTrue) {
+ t.Fatalf("gateway status missing Accepted=True: %#v",
status.Conditions)
+ }
+ if !hasGatewayCondition(status,
string(gatewayv1.GatewayConditionProgrammed), metav1.ConditionTrue) {
+ t.Fatalf("gateway status missing Programmed=True: %#v",
status.Conditions)
+ }
+}
+
+func TestConvertHTTPRouteToConfigCopiesSpec(t *testing.T) {
+ route := &gatewayv1.HTTPRoute{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: "orders",
+ Namespace: "app",
+ },
+ Spec: gatewayv1.HTTPRouteSpec{
+ Hostnames: []gatewayv1.Hostname{"orders.example.com"},
+ },
+ }
+ cfg := convertHTTPRouteToConfig(route)
+ route.Spec.Hostnames[0] = "mutated.example.com"
+
+ spec, ok := cfg.Spec.(*gatewayv1.HTTPRouteSpec)
+ if !ok {
+ t.Fatalf("HTTPRoute config spec type = %T, want
*gatewayv1.HTTPRouteSpec", cfg.Spec)
+ }
+ if got := spec.Hostnames[0]; got != "orders.example.com" {
+ t.Fatalf("HTTPRoute config hostname = %q, want deep-copied
value", got)
+ }
+}
+
+func expectEvent(t *testing.T, events <-chan model.Event, want model.Event) {
+ t.Helper()
+ select {
+ case got := <-events:
+ if got != want {
+ t.Fatalf("event = %v, want %v", got, want)
+ }
+ case <-time.After(time.Second):
+ t.Fatalf("expected event handler to receive %v event", want)
+ }
+}
+
+func waitSynced(t *testing.T, syncer krt.Syncer) {
+ t.Helper()
+ timeout := make(chan struct{})
+ timer := time.AfterFunc(2*time.Second, func() { close(timeout) })
+ defer timer.Stop()
+ if !syncer.WaitUntilSynced(timeout) {
+ t.Fatal("collection did not sync before timeout")
+ }
+}
+
+func hasGatewayCondition(status gatewayv1.GatewayStatus, conditionType string,
conditionStatus metav1.ConditionStatus) bool {
+ for _, condition := range status.Conditions {
+ if condition.Type == conditionType && condition.Status ==
conditionStatus {
+ return true
+ }
+ }
+ return false
+}
+
+type unsyncedSyncer struct{}
+
+func (unsyncedSyncer) HasSynced() bool {
+ return false
+}
+
+func (unsyncedSyncer) WaitUntilSynced(<-chan struct{}) bool {
+ return false
+}
diff --git a/dubbod/discovery/pkg/config/kube/gateway/gateways_collection.go
b/dubbod/discovery/pkg/config/kube/gateway/gateways_collection.go
index 9ef76500..1fb44b4e 100644
--- a/dubbod/discovery/pkg/config/kube/gateway/gateways_collection.go
+++ b/dubbod/discovery/pkg/config/kube/gateway/gateways_collection.go
@@ -19,29 +19,21 @@ package gateway
import (
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/model/kstatus"
"github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
"github.com/apache/dubbo-kubernetes/pkg/kube/krt"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
gateway "sigs.k8s.io/gateway-api/apis/v1"
)
-type Gateway struct {
- *config.Config `json:"config"`
-}
-
-func (g Gateway) ResourceName() string {
- return config.NamespacedName(g.Config).String()
-}
-
func GatewaysCollection(
gateways krt.Collection[*gateway.Gateway],
gatewayClasses krt.Collection[GatewayClass],
opts krt.OptionsBuilder,
) (
krt.StatusCollection[*gateway.Gateway, gateway.GatewayStatus],
- krt.Collection[Gateway],
+ krt.Collection[config.Config],
) {
- statusCol, gw := krt.NewStatusManyCollection(gateways, func(ctx
krt.HandlerContext, obj *gateway.Gateway) (*gateway.GatewayStatus, []Gateway) {
- result := []Gateway{}
+ statusCol, gw := krt.NewStatusCollection(gateways, func(ctx
krt.HandlerContext, obj *gateway.Gateway) (*gateway.GatewayStatus,
*config.Config) {
kgw := obj.Spec
status := obj.Status.DeepCopy()
class := fetchClass(ctx, gatewayClasses, kgw.GatewayClassName)
@@ -56,18 +48,38 @@ func GatewaysCollection(
if classInfo.disableRouteGeneration {
// For now we still mark the Gateway as accepted, but
let higher layers control route generation.
status = setGatewayConditions(status, obj.Generation,
true, true)
- return status, result
+ return status, nil
}
// Default behavior: GatewayClass is known and managed by this
controller.
// Mark Accepted/Programmed to ensure status no longer stays at
"Waiting for controller".
status = setGatewayConditions(status, obj.Generation, true,
true)
- return status, result
+ cfg := convertGatewayToConfig(obj)
+ return status, &cfg
}, opts.WithName("KubernetesGateway")...)
return statusCol, gw
}
+func convertGatewayToConfig(gw *gateway.Gateway) config.Config {
+ return config.Config{
+ Meta: config.Meta{
+ GroupVersionKind: gvk.KubernetesGateway,
+ Name: gw.Name,
+ Namespace: gw.Namespace,
+ Labels: gw.Labels,
+ Annotations: gw.Annotations,
+ ResourceVersion: gw.ResourceVersion,
+ CreationTimestamp: gw.CreationTimestamp.Time,
+ OwnerReferences: gw.OwnerReferences,
+ UID: string(gw.UID),
+ Generation: gw.Generation,
+ },
+ Spec: gw.Spec.DeepCopy(),
+ Status: gw.Status.DeepCopy(),
+ }
+}
+
func setGatewayConditions(
existing *gateway.GatewayStatus,
gen int64,
diff --git a/dubbod/discovery/pkg/config/kube/gateway/referencegrants.go
b/dubbod/discovery/pkg/config/kube/gateway/referencegrants.go
new file mode 100644
index 00000000..e71d46e8
--- /dev/null
+++ b/dubbod/discovery/pkg/config/kube/gateway/referencegrants.go
@@ -0,0 +1,187 @@
+// 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 gateway
+
+import (
+ "fmt"
+ "strings"
+
+ "github.com/apache/dubbo-kubernetes/pkg/config"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/collections"
+ "github.com/apache/dubbo-kubernetes/pkg/config/schema/gvk"
+ "github.com/apache/dubbo-kubernetes/pkg/kube/krt"
+ "k8s.io/apimachinery/pkg/types"
+ gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+ gatewayv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
+)
+
+const (
+ kubernetesSecretPrefix = "kubernetes://"
+ kubernetesGatewaySecretPrefix = "kubernetes-gateway://"
+)
+
+type referenceGrantStore struct {
+ Collection krt.Collection[ReferenceGrant]
+ Index krt.Index[ReferencePair, ReferenceGrant]
+}
+
+type Reference struct {
+ Kind config.GroupVersionKind
+ Namespace gatewayv1.Namespace
+}
+
+func (r Reference) String() string {
+ return r.Kind.String() + "/" + string(r.Namespace)
+}
+
+type ReferencePair struct {
+ To, From Reference
+}
+
+func (p ReferencePair) String() string {
+ return fmt.Sprintf("%s->%s", p.From, p.To)
+}
+
+type ReferenceGrant struct {
+ Source types.NamespacedName
+ From Reference
+ To Reference
+ AllowAll bool
+ AllowedName string
+}
+
+func (g ReferenceGrant) ResourceName() string {
+ nameKey := "*"
+ if !g.AllowAll {
+ nameKey = g.AllowedName
+ }
+ return g.Source.String() + "/" + g.From.Kind.String() + "/" +
string(g.From.Namespace) + "/" + g.To.Kind.String() + "/" +
string(g.To.Namespace) + "/" + nameKey
+}
+
+func ReferenceGrantsCollection(
+ referenceGrants krt.Collection[*gatewayv1beta1.ReferenceGrant],
+ opts krt.OptionsBuilder,
+) krt.Collection[ReferenceGrant] {
+ return krt.NewManyCollection(referenceGrants, func(ctx
krt.HandlerContext, obj *gatewayv1beta1.ReferenceGrant) []ReferenceGrant {
+ result := make([]ReferenceGrant, 0,
len(obj.Spec.From)*len(obj.Spec.To))
+ for _, from := range obj.Spec.From {
+ fromKind := normalizeReference(&from.Group, &from.Kind,
config.GroupVersionKind{})
+ if fromKind != gvk.KubernetesGateway && fromKind !=
gvk.HTTPRoute {
+ continue
+ }
+ fromRef := Reference{
+ Kind: fromKind,
+ Namespace: gatewayv1.Namespace(from.Namespace),
+ }
+ for _, to := range obj.Spec.To {
+ toKind := normalizeReference(&to.Group,
&to.Kind, config.GroupVersionKind{})
+ if toKind != gvk.Secret {
+ continue
+ }
+ grant := ReferenceGrant{
+ Source: types.NamespacedName{
+ Name: obj.Name,
+ Namespace: obj.Namespace,
+ },
+ From: fromRef,
+ To: Reference{
+ Kind: toKind,
+ Namespace:
gatewayv1.Namespace(obj.Namespace),
+ },
+ AllowAll: to.Name == nil,
+ }
+ if to.Name != nil {
+ grant.AllowedName = string(*to.Name)
+ }
+ result = append(result, grant)
+ }
+ }
+ return result
+ }, opts.WithName("ReferenceGrants")...)
+}
+
+func newReferenceGrantStore(collection krt.Collection[ReferenceGrant])
referenceGrantStore {
+ return referenceGrantStore{
+ Collection: collection,
+ Index: krt.NewIndex(collection, "toFrom", func(o
ReferenceGrant) []ReferencePair {
+ return []ReferencePair{{To: o.To, From: o.From}}
+ }),
+ }
+}
+
+func (refs referenceGrantStore) SecretAllowed(ctx krt.HandlerContext, kind
config.GroupVersionKind, resourceName string, namespace string) bool {
+ if refs.Collection == nil {
+ return false
+ }
+ name, targetNamespace, targetKind, ok :=
parseSecretResourceName(resourceName, namespace)
+ if !ok {
+ return false
+ }
+ pair := ReferencePair{
+ From: Reference{Kind: kind, Namespace:
gatewayv1.Namespace(namespace)},
+ To: Reference{Kind: targetKind, Namespace:
gatewayv1.Namespace(targetNamespace)},
+ }
+ for _, grant := range krt.FetchOrList(ctx, refs.Collection,
krt.FilterIndex(refs.Index, pair)) {
+ if grant.AllowAll || grant.AllowedName == name {
+ return true
+ }
+ }
+ return false
+}
+
+func parseSecretResourceName(resourceName string, defaultNamespace string)
(name string, namespace string, kind config.GroupVersionKind, ok bool) {
+ kind = gvk.Secret
+ switch {
+ case strings.HasPrefix(resourceName, kubernetesGatewaySecretPrefix):
+ parts := strings.Split(strings.TrimPrefix(resourceName,
kubernetesGatewaySecretPrefix), "/")
+ if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
+ return "", "", config.GroupVersionKind{}, false
+ }
+ return parts[1], parts[0], kind, true
+ case strings.HasPrefix(resourceName, kubernetesSecretPrefix):
+ parts := strings.Split(strings.TrimPrefix(resourceName,
kubernetesSecretPrefix), "/")
+ if len(parts) == 1 && parts[0] != "" {
+ return parts[0], defaultNamespace, kind, true
+ }
+ if len(parts) == 2 && parts[0] != "" && parts[1] != "" {
+ return parts[1], parts[0], kind, true
+ }
+ return "", "", config.GroupVersionKind{}, false
+ default:
+ parts := strings.Split(resourceName, "/")
+ if len(parts) == 1 && parts[0] != "" {
+ return parts[0], defaultNamespace, kind, true
+ }
+ if len(parts) == 2 && parts[0] != "" && parts[1] != "" {
+ return parts[1], parts[0], kind, true
+ }
+ return "", "", config.GroupVersionKind{}, false
+ }
+}
+
+func normalizeReference[G ~string, K ~string](group *G, kind *K, fallback
config.GroupVersionKind) config.GroupVersionKind {
+ out := fallback
+ if group != nil {
+ out.Group = string(*group)
+ }
+ if kind != nil {
+ out.Kind = string(*kind)
+ }
+ if schema, found := collections.All.FindByGroupKind(out); found {
+ return schema.GroupVersionKind()
+ }
+ return out
+}
diff --git a/dubbod/discovery/pkg/model/context.go
b/dubbod/discovery/pkg/model/context.go
index 52e41281..d0c30453 100644
--- a/dubbod/discovery/pkg/model/context.go
+++ b/dubbod/discovery/pkg/model/context.go
@@ -29,6 +29,7 @@ import (
networkutil
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/util/network"
"github.com/apache/dubbo-kubernetes/dubbod/discovery/pkg/features"
+ "github.com/apache/dubbo-kubernetes/pkg/config"
"github.com/apache/dubbo-kubernetes/pkg/config/constants"
"github.com/apache/dubbo-kubernetes/pkg/config/host"
"github.com/apache/dubbo-kubernetes/pkg/config/mesh"
@@ -100,6 +101,7 @@ type Environment struct {
type GatewayController interface {
ConfigStoreController
Reconcile(ctx *PushContext)
+ SecretAllowed(ourKind config.GroupVersionKind, resourceName string,
namespace string) bool
}
func NewEnvironment() *Environment {
diff --git a/pkg/config/schema/collections/collections.gen.go
b/pkg/config/schema/collections/collections.gen.go
index c454cc1a..7f317175 100755
--- a/pkg/config/schema/collections/collections.gen.go
+++ b/pkg/config/schema/collections/collections.gen.go
@@ -24,6 +24,7 @@ import (
k8sioapipolicyv1 "k8s.io/api/policy/v1"
k8sioapiextensionsapiserverpkgapisapiextensionsv1
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
sigsk8siogatewayapiapisv1 "sigs.k8s.io/gateway-api/apis/v1"
+ sigsk8siogatewayapiapisv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
)
var (
@@ -354,6 +355,24 @@ var (
ValidateProto: validation.EmptyValidate,
}.MustBuild()
+ ReferenceGrant = resource.Builder{
+ Identifier: "ReferenceGrant",
+ Group: "gateway.networking.k8s.io",
+ Kind: "ReferenceGrant",
+ Plural: "referencegrants",
+ Version: "v1beta1",
+ VersionAliases: []string{
+ "v1beta1",
+ },
+ Proto:
"k8s.io.gateway_api.api.v1alpha1.ReferenceGrantSpec",
+ ReflectType:
reflect.TypeOf(&sigsk8siogatewayapiapisv1beta1.ReferenceGrantSpec{}).Elem(),
+ ProtoPackage: "sigs.k8s.io/gateway-api/apis/v1beta1",
+ ClusterScoped: false,
+ Synthetic: false,
+ Builtin: false,
+ ValidateProto: validation.EmptyValidate,
+ }.MustBuild()
+
RequestAuthentication = resource.Builder{
Identifier: "RequestAuthentication",
Group: "security.dubbo.apache.org",
@@ -467,6 +486,7 @@ var (
MustAdd(PeerAuthentication).
MustAdd(Pod).
MustAdd(PodDisruptionBudget).
+ MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
MustAdd(Secret).
MustAdd(Service).
@@ -494,6 +514,7 @@ var (
MustAdd(Node).
MustAdd(Pod).
MustAdd(PodDisruptionBudget).
+ MustAdd(ReferenceGrant).
MustAdd(Secret).
MustAdd(Service).
MustAdd(ServiceAccount).
@@ -518,6 +539,7 @@ var (
MustAdd(HTTPRoute).
MustAdd(KubernetesGateway).
MustAdd(PeerAuthentication).
+ MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
Build()
@@ -530,6 +552,7 @@ var (
MustAdd(HTTPRoute).
MustAdd(KubernetesGateway).
MustAdd(PeerAuthentication).
+ MustAdd(ReferenceGrant).
MustAdd(RequestAuthentication).
Build()
)
diff --git a/pkg/config/schema/gvk/resources.gen.go
b/pkg/config/schema/gvk/resources.gen.go
index 5e3768ba..eae934d9 100755
--- a/pkg/config/schema/gvk/resources.gen.go
+++ b/pkg/config/schema/gvk/resources.gen.go
@@ -35,6 +35,8 @@ var (
PeerAuthentication = config.GroupVersionKind{Group:
"security.dubbo.apache.org", Version: "v1alpha3", Kind: "PeerAuthentication"}
Pod = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "Pod"}
PodDisruptionBudget = config.GroupVersionKind{Group:
"policy", Version: "v1", Kind: "PodDisruptionBudget"}
+ ReferenceGrant = config.GroupVersionKind{Group:
"gateway.networking.k8s.io", Version: "v1beta1", Kind: "ReferenceGrant"}
+ ReferenceGrant_v1beta1 = config.GroupVersionKind{Group:
"gateway.networking.k8s.io", Version: "v1beta1", Kind: "ReferenceGrant"}
RequestAuthentication = config.GroupVersionKind{Group:
"security.dubbo.apache.org", Version: "v1alpha3", Kind: "RequestAuthentication"}
Secret = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "Secret"}
Service = config.GroupVersionKind{Group: "",
Version: "v1", Kind: "Service"}
@@ -96,6 +98,10 @@ func ToGVR(g config.GroupVersionKind)
(schema.GroupVersionResource, bool) {
return gvr.Pod, true
case PodDisruptionBudget:
return gvr.PodDisruptionBudget, true
+ case ReferenceGrant:
+ return gvr.ReferenceGrant, true
+ case ReferenceGrant_v1beta1:
+ return gvr.ReferenceGrant_v1beta1, true
case RequestAuthentication:
return gvr.RequestAuthentication, true
case Secret:
@@ -157,6 +163,8 @@ func MustToKind(g config.GroupVersionKind) kind.Kind {
return kind.Pod
case PodDisruptionBudget:
return kind.PodDisruptionBudget
+ case ReferenceGrant:
+ return kind.ReferenceGrant
case RequestAuthentication:
return kind.RequestAuthentication
case Secret:
@@ -229,6 +237,8 @@ func FromGVR(g schema.GroupVersionResource)
(config.GroupVersionKind, bool) {
return Pod, true
case gvr.PodDisruptionBudget:
return PodDisruptionBudget, true
+ case gvr.ReferenceGrant:
+ return ReferenceGrant, true
case gvr.RequestAuthentication:
return RequestAuthentication, true
case gvr.Secret:
diff --git a/pkg/config/schema/gvr/resources.gen.go
b/pkg/config/schema/gvr/resources.gen.go
index 21955199..a37e5a65 100755
--- a/pkg/config/schema/gvr/resources.gen.go
+++ b/pkg/config/schema/gvr/resources.gen.go
@@ -30,6 +30,8 @@ var (
PeerAuthentication = schema.GroupVersionResource{Group:
"security.dubbo.apache.org", Version: "v1alpha3", Resource:
"peerauthentications"}
Pod = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "pods"}
PodDisruptionBudget = schema.GroupVersionResource{Group:
"policy", Version: "v1", Resource: "poddisruptionbudgets"}
+ ReferenceGrant = schema.GroupVersionResource{Group:
"gateway.networking.k8s.io", Version: "v1beta1", Resource: "referencegrants"}
+ ReferenceGrant_v1beta1 = schema.GroupVersionResource{Group:
"gateway.networking.k8s.io", Version: "v1beta1", Resource: "referencegrants"}
RequestAuthentication = schema.GroupVersionResource{Group:
"security.dubbo.apache.org", Version: "v1alpha3", Resource:
"requestauthentications"}
Secret = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "secrets"}
Service = schema.GroupVersionResource{Group: "",
Version: "v1", Resource: "services"}
@@ -88,6 +90,10 @@ func IsClusterScoped(g schema.GroupVersionResource) bool {
return false
case PodDisruptionBudget:
return false
+ case ReferenceGrant:
+ return false
+ case ReferenceGrant_v1beta1:
+ return false
case RequestAuthentication:
return false
case Secret:
diff --git a/pkg/config/schema/kind/resources.gen.go
b/pkg/config/schema/kind/resources.gen.go
index 40ab69ab..268c1cf0 100755
--- a/pkg/config/schema/kind/resources.gen.go
+++ b/pkg/config/schema/kind/resources.gen.go
@@ -27,6 +27,7 @@ const (
PeerAuthentication
Pod
PodDisruptionBudget
+ ReferenceGrant
RequestAuthentication
Secret
Service
@@ -83,6 +84,8 @@ func (k Kind) String() string {
return "Pod"
case PodDisruptionBudget:
return "PodDisruptionBudget"
+ case ReferenceGrant:
+ return "ReferenceGrant"
case RequestAuthentication:
return "RequestAuthentication"
case Secret:
@@ -148,6 +151,8 @@ func FromString(s string) Kind {
return Pod
case "PodDisruptionBudget":
return PodDisruptionBudget
+ case "ReferenceGrant":
+ return ReferenceGrant
case "RequestAuthentication":
return RequestAuthentication
case "Secret":
diff --git a/pkg/config/schema/kubeclient/resources.gen.go
b/pkg/config/schema/kubeclient/resources.gen.go
index c571b09b..fbf1bf6e 100755
--- a/pkg/config/schema/kubeclient/resources.gen.go
+++ b/pkg/config/schema/kubeclient/resources.gen.go
@@ -28,6 +28,7 @@ import (
k8sioapipolicyv1 "k8s.io/api/policy/v1"
k8sioapiextensionsapiserverpkgapisapiextensionsv1
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
sigsk8siogatewayapiapisv1 "sigs.k8s.io/gateway-api/apis/v1"
+ sigsk8siogatewayapiapisv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
)
func GetWriteClient[T runtime.Object](c ClientGetter, namespace string)
ktypes.WriteAPI[T] {
@@ -72,6 +73,8 @@ func GetWriteClient[T runtime.Object](c ClientGetter,
namespace string) ktypes.W
return c.Kube().CoreV1().Pods(namespace).(ktypes.WriteAPI[T])
case *k8sioapipolicyv1.PodDisruptionBudget:
return
c.Kube().PolicyV1().PodDisruptionBudgets(namespace).(ktypes.WriteAPI[T])
+ case *sigsk8siogatewayapiapisv1beta1.ReferenceGrant:
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(namespace).(ktypes.WriteAPI[T])
case
*apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication:
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(namespace).(ktypes.WriteAPI[T])
case *k8sioapicorev1.Secret:
@@ -131,6 +134,8 @@ func GetClient[T, TL runtime.Object](c ClientGetter,
namespace string) ktypes.Re
return
c.Kube().CoreV1().Pods(namespace).(ktypes.ReadWriteAPI[T, TL])
case *k8sioapipolicyv1.PodDisruptionBudget:
return
c.Kube().PolicyV1().PodDisruptionBudgets(namespace).(ktypes.ReadWriteAPI[T, TL])
+ case *sigsk8siogatewayapiapisv1beta1.ReferenceGrant:
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(namespace).(ktypes.ReadWriteAPI[T,
TL])
case
*apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication:
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(namespace).(ktypes.ReadWriteAPI[T,
TL])
case *k8sioapicorev1.Secret:
@@ -190,6 +195,8 @@ func gvrToObject(g schema.GroupVersionResource)
runtime.Object {
return &k8sioapicorev1.Pod{}
case gvr.PodDisruptionBudget:
return &k8sioapipolicyv1.PodDisruptionBudget{}
+ case gvr.ReferenceGrant:
+ return &sigsk8siogatewayapiapisv1beta1.ReferenceGrant{}
case gvr.RequestAuthentication:
return
&apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication{}
case gvr.Secret:
@@ -352,6 +359,13 @@ func getInformerFiltered(c ClientGetter, opts
ktypes.InformerOptions, g schema.G
w = func(options metav1.ListOptions) (watch.Interface, error) {
return
c.Kube().PolicyV1().PodDisruptionBudgets(opts.Namespace).Watch(context.Background(),
options)
}
+ case gvr.ReferenceGrant:
+ l = func(options metav1.ListOptions) (runtime.Object, error) {
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(opts.Namespace).List(context.Background(),
options)
+ }
+ w = func(options metav1.ListOptions) (watch.Interface, error) {
+ return
c.GatewayAPI().GatewayV1beta1().ReferenceGrants(opts.Namespace).Watch(context.Background(),
options)
+ }
case gvr.RequestAuthentication:
l = func(options metav1.ListOptions) (runtime.Object, error) {
return
c.Dubbo().SecurityV1alpha3().RequestAuthentications(opts.Namespace).List(context.Background(),
options)
diff --git a/pkg/config/schema/kubetypes/resources.gen.go
b/pkg/config/schema/kubetypes/resources.gen.go
index 38722483..01799d04 100755
--- a/pkg/config/schema/kubetypes/resources.gen.go
+++ b/pkg/config/schema/kubetypes/resources.gen.go
@@ -19,6 +19,7 @@ import (
k8sioapipolicyv1 "k8s.io/api/policy/v1"
k8sioapiextensionsapiserverpkgapisapiextensionsv1
"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
sigsk8siogatewayapiapisv1 "sigs.k8s.io/gateway-api/apis/v1"
+ sigsk8siogatewayapiapisv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
)
func getGvk(obj any) (config.GroupVersionKind, bool) {
@@ -71,6 +72,8 @@ func getGvk(obj any) (config.GroupVersionKind, bool) {
return gvk.Pod, true
case *k8sioapipolicyv1.PodDisruptionBudget:
return gvk.PodDisruptionBudget, true
+ case *sigsk8siogatewayapiapisv1beta1.ReferenceGrant:
+ return gvk.ReferenceGrant, true
case *githubcomkdubboapisecurityv1alpha3.RequestAuthentication:
return gvk.RequestAuthentication, true
case
*apigithubcomapachedubbokubernetesapisecurityv1alpha3.RequestAuthentication:
diff --git a/pkg/config/schema/metadata.yaml b/pkg/config/schema/metadata.yaml
index 5359af4a..a68dec0d 100644
--- a/pkg/config/schema/metadata.yaml
+++ b/pkg/config/schema/metadata.yaml
@@ -202,6 +202,15 @@ resources:
statusProto: "k8s.io.gateway_api.api.v1alpha1.PolicyStatus"
statusProtoPackage: "sigs.k8s.io/gateway-api/apis/v1"
+ - kind: "ReferenceGrant"
+ plural: "referencegrants"
+ group: "gateway.networking.k8s.io"
+ version: "v1beta1"
+ versionAliases:
+ - "v1beta1"
+ protoPackage: "sigs.k8s.io/gateway-api/apis/v1beta1"
+ proto: "k8s.io.gateway_api.api.v1alpha1.ReferenceGrantSpec"
+
## Dubbo resources
- kind: AuthorizationPolicy
plural: "authorizationpolicies"
