Complete KRT controllers framework

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

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)
 }

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)
+	}
+}

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 {

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)
+	}
+}