Fix status propagation and global topology (#54)

* Fix status propagation and global topology

Signed-off-by: Waleed Malik <ahmedwaleedmalik@gmail.com>

* Update docs

Signed-off-by: Waleed Malik <ahmedwaleedmalik@gmail.com>

---------

Signed-off-by: Waleed Malik <ahmedwaleedmalik@gmail.com>

api/kubelb.k8c.io/v1alpha1/sync_secret_types.go

@@ -31,6 +31,8 @@ type SyncSecret struct {
 
 	// Source: https://pkg.go.dev/k8s.io/api/core/v1#Secret
 
+	// +optional
+	Immutable *bool `json:"immutable,omitempty" protobuf:"varint,5,opt,name=immutable"`
 	// +optional
 	Data map[string][]byte `json:"data,omitempty" protobuf:"bytes,2,rep,name=data"`
 

charts/kubelb-ccm/crds/kubelb.k8c.io_syncsecrets.yaml

@@ -33,6 +33,8 @@ spec:
               format: byte
               type: string
             type: object
+          immutable:
+            type: boolean
           kind:
             description: |-
               Kind is a string value representing the REST resource this object represents.

charts/kubelb-ccm/templates/clusterrole.yaml

@@ -25,6 +25,7 @@ rules:
   - patch
   - update
   - watch
+  - delete
 - apiGroups:
   - ""
   resources:

charts/kubelb-manager/crds/kubelb.k8c.io_syncsecrets.yaml

@@ -33,6 +33,8 @@ spec:
               format: byte
               type: string
             type: object
+          immutable:
+            type: boolean
           kind:
             description: |-
               Kind is a string value representing the REST resource this object represents.

charts/kubelb-manager/templates/clusterrole.yaml

@@ -53,6 +53,18 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - gateway.networking.k8s.io
   resources:

cmd/ccm/main.go

@@ -232,7 +232,7 @@ func main() {
 	if !disableIngressController {
 		if err = (&ccm.IngressReconciler{
 			Client:          mgr.GetClient(),
-			LBClient:        kubeLBMgr.GetClient(),
+			LBManager:       kubeLBMgr,
 			ClusterName:     clusterName,
 			Log:             ctrl.Log.WithName("controllers").WithName(ccm.IngressControllerName),
 			Scheme:          mgr.GetScheme(),
@@ -247,7 +247,7 @@ func main() {
 	if !disableGatewayController && !disableGatewayAPI {
 		if err = (&ccm.GatewayReconciler{
 			Client:          mgr.GetClient(),
-			LBClient:        kubeLBMgr.GetClient(),
+			LBManager:       kubeLBMgr,
 			ClusterName:     clusterName,
 			Log:             ctrl.Log.WithName("controllers").WithName(ccm.GatewayControllerName),
 			Scheme:          mgr.GetScheme(),
@@ -262,7 +262,7 @@ func main() {
 	if !disableHTTPRouteController && !disableGatewayAPI {
 		if err = (&ccm.HTTPRouteReconciler{
 			Client:      mgr.GetClient(),
-			LBClient:    kubeLBMgr.GetClient(),
+			LBManager:   kubeLBMgr,
 			ClusterName: clusterName,
 			Log:         ctrl.Log.WithName("controllers").WithName(ccm.GatewayHTTPRouteControllerName),
 			Scheme:      mgr.GetScheme(),
@@ -276,7 +276,7 @@ func main() {
 	if !disableGRPCRouteController && !disableGatewayAPI {
 		if err = (&ccm.GRPCRouteReconciler{
 			Client:      mgr.GetClient(),
-			LBClient:    kubeLBMgr.GetClient(),
+			LBManager:   kubeLBMgr,
 			ClusterName: clusterName,
 			Log:         ctrl.Log.WithName("controllers").WithName(ccm.GatewayGRPCRouteControllerName),
 			Scheme:      mgr.GetScheme(),

cmd/kubelb/main.go

@@ -240,6 +240,19 @@ func main() {
 		}
 	}
 
+	// This is only required when using global topology.
+	if conf.IsGlobalTopology() {
+		if err = (&kubelb.BridgeServiceReconciler{
+			Client:   mgr.GetClient(),
+			Scheme:   mgr.GetScheme(),
+			Log:      ctrl.Log.WithName("controllers").WithName(kubelb.BridgeServiceControllerName),
+			Recorder: mgr.GetEventRecorderFor(kubelb.BridgeServiceControllerName),
+		}).SetupWithManager(mgr); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", kubelb.BridgeServiceControllerName)
+			os.Exit(1)
+		}
+	}
+
 	go func() {
 		setupLog.Info("starting kubelb envoy manager")
 

config/crd/bases/kubelb.k8c.io_syncsecrets.yaml

@@ -33,6 +33,8 @@ spec:
               format: byte
               type: string
             type: object
+          immutable:
+            type: boolean
           kind:
             description: |-
               Kind is a string value representing the REST resource this object represents.

config/kubelb/rbac/role.yaml

@@ -84,6 +84,18 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - gateway.networking.k8s.io
   resources:

docs/api-reference.md

@@ -453,7 +453,7 @@ _Appears in:_
 | --- | --- | --- | --- |
 | `name` _string_ | The name of this port within the service. This must be a DNS_LABEL.<br />All ports within a ServiceSpec must have unique names. When considering<br />the endpoints for a Service, this must match the 'name' field in the<br />EndpointPort.<br />Optional if only one ServicePort is defined on this service. |  |  |
 | `protocol` _[Protocol](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#protocol-v1-core)_ | The IP protocol for this port. Supports "TCP", "UDP", and "SCTP".<br />Default is TCP. |  |  |
-| `appProtocol` _string_ | The application protocol for this port.<br />This is used as a hint for implementations to offer richer behavior for protocols that they understand.<br />This field follows standard Kubernetes label syntax.<br />Valid values are either:<br /><br />*Un-prefixed protocol names - reserved for IANA standard service names (as per<br />RFC-6335 and <https://www.iana.org/assignments/service-names>).<br /><br />* Kubernetes-defined prefixed names:<br />  *'kubernetes.io/h2c' - HTTP/2 prior knowledge over cleartext as described in <https://www.rfc-editor.org/rfc/rfc9113.html#name-starting-http-2-with-prior-><br />* 'kubernetes.io/ws'  - WebSocket over cleartext as described in <https://www.rfc-editor.org/rfc/rfc6455><br />  *'kubernetes.io/wss' - WebSocket over TLS as described in <https://www.rfc-editor.org/rfc/rfc6455><br /><br />* Other protocols should use implementation-defined prefixed names such as<br />mycompany.com/my-custom-protocol. |  |  |
+| `appProtocol` _string_ | The application protocol for this port.<br />This is used as a hint for implementations to offer richer behavior for protocols that they understand.<br />This field follows standard Kubernetes label syntax.<br />Valid values are either:<br /><br />_Un-prefixed protocol names - reserved for IANA standard service names (as per<br />RFC-6335 and <https://www.iana.org/assignments/service-names>).<br /><br />_ Kubernetes-defined prefixed names:<br />  _'kubernetes.io/h2c' - HTTP/2 prior knowledge over cleartext as described in <https://www.rfc-editor.org/rfc/rfc9113.html#name-starting-http-2-with-prior-><br />_ 'kubernetes.io/ws'  - WebSocket over cleartext as described in <https://www.rfc-editor.org/rfc/rfc6455><br />  _'kubernetes.io/wss' - WebSocket over TLS as described in <https://www.rfc-editor.org/rfc/rfc6455><br /><br />_ Other protocols should use implementation-defined prefixed names such as<br />mycompany.com/my-custom-protocol. |  |  |
 | `port` _integer_ | The port that will be exposed by this service. |  |  |
 | `targetPort` _[IntOrString](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#intorstring-intstr-util)_ | Number or name of the port to access on the pods targeted by the service.<br />Number must be in the range 1 to 65535. Name must be an IANA_SVC_NAME.<br />If this is a string, it will be looked up as a named port in the<br />target Pod's container ports. If this is not specified, the value<br />of the 'port' field is used (an identity map).<br />This field is ignored for services with clusterIP=None, and should be<br />omitted or set equal to the 'port' field.<br />More info: <https://kubernetes.io/docs/concepts/services-networking/service/#defining-a-service> |  |  |
 | `nodePort` _integer_ | The port on each node on which this service is exposed when type is<br />NodePort or LoadBalancer.  Usually assigned by the system. If a value is<br />specified, in-range, and not in use it will be used, otherwise the<br />operation will fail.  If not specified, a port will be allocated if this<br />Service requires one.  If this field is specified when creating a<br />Service which does not need it, creation will fail. This field will be<br />wiped when updating a Service to no longer need it (e.g. changing type<br />from NodePort to ClusterIP).<br />More info: <https://kubernetes.io/docs/concepts/services-networking/service/#type-nodeport> |  |  |
@@ -482,6 +482,7 @@ _Appears in:_
 | `apiVersion` _string_ | `kubelb.k8c.io/v1alpha1` | | |
 | `kind` _string_ | `SyncSecret` | | |
 | `metadata` _[ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#objectmeta-v1-meta)_ | Refer to Kubernetes API documentation for fields of `metadata`. |  |  |
+| `immutable` _boolean_ |  |  |  |
 | `data` _object (keys:string, values:integer array)_ |  |  |  |
 | `stringData` _object (keys:string, values:string)_ |  |  |  |
 | `type` _[SecretType](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#secrettype-v1-core)_ |  |  |  |

internal/controllers/ccm/gateway_controller.go

@@ -37,8 +37,10 @@ import (
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 	"sigs.k8s.io/yaml"
 )
@@ -53,7 +55,7 @@ const (
 type GatewayReconciler struct {
 	ctrlclient.Client
 
-	LBClient        ctrlclient.Client
+	LBManager       ctrl.Manager
 	ClusterName     string
 	UseGatewayClass bool
 
@@ -117,14 +119,14 @@ func (r *GatewayReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 
 func (r *GatewayReconciler) reconcile(ctx context.Context, log logr.Logger, gateway *gwapiv1.Gateway) error {
 	// Create/update the corresponding Route in LB cluster.
-	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBClient, gateway, nil, nil, r.ClusterName)
+	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBManager.GetClient(), gateway, nil, nil, r.ClusterName)
 	if err != nil {
 		return fmt.Errorf("failed to reconcile source for route: %w", err)
 	}
 
 	// Route was reconciled successfully, now we need to update the status of the Resource.
 	route := kubelbv1alpha1.Route{}
-	err = r.LBClient.Get(ctx, types.NamespacedName{Name: string(gateway.UID), Namespace: r.ClusterName}, &route)
+	err = r.LBManager.GetClient().Get(ctx, types.NamespacedName{Name: string(gateway.UID), Namespace: r.ClusterName}, &route)
 	if err != nil {
 		return fmt.Errorf("failed to get Route from LB cluster: %w", err)
 	}
@@ -164,7 +166,7 @@ func (r *GatewayReconciler) reconcile(ctx context.Context, log logr.Logger, gate
 
 func (r *GatewayReconciler) cleanup(ctx context.Context, gateway *gwapiv1.Gateway) (ctrl.Result, error) {
 	// Find the Route in LB cluster and delete it
-	err := cleanupRoute(ctx, r.LBClient, string(gateway.UID), r.ClusterName)
+	err := cleanupRoute(ctx, r.LBManager.GetClient(), string(gateway.UID), r.ClusterName)
 	if err != nil {
 		return reconcile.Result{}, fmt.Errorf("failed to cleanup route: %w", err)
 	}
@@ -223,5 +225,9 @@ func (r *GatewayReconciler) shouldReconcile(gateway *gwapiv1.Gateway) bool {
 func (r *GatewayReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&gwapiv1.Gateway{}, builder.WithPredicates(r.resourceFilter())).
+		WatchesRawSource(
+			source.Kind(r.LBManager.GetCache(), &kubelbv1alpha1.Route{},
+				handler.TypedEnqueueRequestsFromMapFunc[*kubelbv1alpha1.Route](enqueueRoutes("Gateway.gateway.networking.k8s.io", r.ClusterName))),
+		).
 		Complete(r)
 }

internal/controllers/ccm/gateway_grpcroute_controller.go

@@ -43,6 +43,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 	"sigs.k8s.io/yaml"
 )
@@ -55,7 +56,7 @@ const (
 type GRPCRouteReconciler struct {
 	ctrlclient.Client
 
-	LBClient    ctrlclient.Client
+	LBManager   ctrl.Manager
 	ClusterName string
 
 	Log      logr.Logger
@@ -115,14 +116,14 @@ func (r *GRPCRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 func (r *GRPCRouteReconciler) reconcile(ctx context.Context, log logr.Logger, grpcRoute *gwapiv1.GRPCRoute) error {
 	// We need to traverse the GRPCRoute, find all the services associated with it, create/update the corresponding Route in LB cluster.
 	originalServices := grpcrouteHelpers.GetServicesFromGRPCRoute(grpcRoute)
-	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBClient, grpcRoute, originalServices, nil, r.ClusterName)
+	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBManager.GetClient(), grpcRoute, originalServices, nil, r.ClusterName)
 	if err != nil {
 		return fmt.Errorf("failed to reconcile source for route: %w", err)
 	}
 
 	// Route was reconciled successfully, now we need to update the status of the Resource.
 	route := kubelbv1alpha1.Route{}
-	err = r.LBClient.Get(ctx, types.NamespacedName{Name: string(grpcRoute.UID), Namespace: r.ClusterName}, &route)
+	err = r.LBManager.GetClient().Get(ctx, types.NamespacedName{Name: string(grpcRoute.UID), Namespace: r.ClusterName}, &route)
 	if err != nil {
 		return fmt.Errorf("failed to get Route from LB cluster: %w", err)
 	}
@@ -186,7 +187,7 @@ func (r *GRPCRouteReconciler) cleanup(ctx context.Context, grpcRoute *gwapiv1.GR
 	}
 
 	// Find the Route in LB cluster and delete it
-	err = cleanupRoute(ctx, r.LBClient, string(grpcRoute.UID), r.ClusterName)
+	err = cleanupRoute(ctx, r.LBManager.GetClient(), string(grpcRoute.UID), r.ClusterName)
 	if err != nil {
 		return reconcile.Result{}, fmt.Errorf("failed to cleanup route: %w", err)
 	}
@@ -278,5 +279,9 @@ func (r *GRPCRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			&corev1.Service{},
 			handler.EnqueueRequestsFromMapFunc(r.enqueueResources()),
 		).
+		WatchesRawSource(
+			source.Kind(r.LBManager.GetCache(), &kubelbv1alpha1.Route{},
+				handler.TypedEnqueueRequestsFromMapFunc[*kubelbv1alpha1.Route](enqueueRoutes("GRPCRoute.gateway.networking.k8s.io", r.ClusterName))),
+		).
 		Complete(r)
 }

internal/controllers/ccm/gateway_httproute_controller.go

@@ -43,6 +43,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 	"sigs.k8s.io/yaml"
 )
@@ -55,7 +56,7 @@ const (
 type HTTPRouteReconciler struct {
 	ctrlclient.Client
 
-	LBClient    ctrlclient.Client
+	LBManager   ctrl.Manager
 	ClusterName string
 
 	Log      logr.Logger
@@ -119,14 +120,14 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 func (r *HTTPRouteReconciler) reconcile(ctx context.Context, log logr.Logger, httpRoute *gwapiv1.HTTPRoute) error {
 	// We need to traverse the HTTPRoute, find all the services associated with it, create/update the corresponding Route in LB cluster.
 	originalServices := httprouteHelpers.GetServicesFromHTTPRoute(httpRoute)
-	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBClient, httpRoute, originalServices, nil, r.ClusterName)
+	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBManager.GetClient(), httpRoute, originalServices, nil, r.ClusterName)
 	if err != nil {
 		return fmt.Errorf("failed to reconcile source for route: %w", err)
 	}
 
 	// Route was reconciled successfully, now we need to update the status of the Resource.
 	route := kubelbv1alpha1.Route{}
-	err = r.LBClient.Get(ctx, types.NamespacedName{Name: string(httpRoute.UID), Namespace: r.ClusterName}, &route)
+	err = r.LBManager.GetClient().Get(ctx, types.NamespacedName{Name: string(httpRoute.UID), Namespace: r.ClusterName}, &route)
 	if err != nil {
 		return fmt.Errorf("failed to get Route from LB cluster: %w", err)
 	}
@@ -190,7 +191,7 @@ func (r *HTTPRouteReconciler) cleanup(ctx context.Context, httpRoute *gwapiv1.HT
 	}
 
 	// Find the Route in LB cluster and delete it
-	err = cleanupRoute(ctx, r.LBClient, string(httpRoute.UID), r.ClusterName)
+	err = cleanupRoute(ctx, r.LBManager.GetClient(), string(httpRoute.UID), r.ClusterName)
 	if err != nil {
 		return reconcile.Result{}, fmt.Errorf("failed to cleanup route: %w", err)
 	}
@@ -282,5 +283,9 @@ func (r *HTTPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			&corev1.Service{},
 			handler.EnqueueRequestsFromMapFunc(r.enqueueResources()),
 		).
+		WatchesRawSource(
+			source.Kind(r.LBManager.GetCache(), &kubelbv1alpha1.Route{},
+				handler.TypedEnqueueRequestsFromMapFunc[*kubelbv1alpha1.Route](enqueueRoutes("HTTPRoute.gateway.networking.k8s.io", r.ClusterName))),
+		).
 		Complete(r)
 }

internal/controllers/ccm/ingress_controller.go

@@ -44,6 +44,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 	"sigs.k8s.io/yaml"
 )
 
@@ -56,7 +57,7 @@ const (
 type IngressReconciler struct {
 	ctrlclient.Client
 
-	LBClient        ctrlclient.Client
+	LBManager       ctrl.Manager
 	ClusterName     string
 	UseIngressClass bool
 
@@ -121,14 +122,14 @@ func (r *IngressReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 func (r *IngressReconciler) reconcile(ctx context.Context, log logr.Logger, ingress *networkingv1.Ingress) error {
 	// We need to traverse the Ingress, find all the services associated with it, create/update the corresponding Route in LB cluster.
 	originalServices := ingressHelpers.GetServicesFromIngress(*ingress)
-	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBClient, ingress, originalServices, nil, r.ClusterName)
+	err := reconcileSourceForRoute(ctx, log, r.Client, r.LBManager.GetClient(), ingress, originalServices, nil, r.ClusterName)
 	if err != nil {
 		return fmt.Errorf("failed to reconcile source for route: %w", err)
 	}
 
 	// Route was reconciled successfully, now we need to update the status of the Ingress.
 	route := kubelbv1alpha1.Route{}
-	err = r.LBClient.Get(ctx, types.NamespacedName{Name: string(ingress.UID), Namespace: r.ClusterName}, &route)
+	err = r.LBManager.GetClient().Get(ctx, types.NamespacedName{Name: string(ingress.UID), Namespace: r.ClusterName}, &route)
 	if err != nil {
 		return fmt.Errorf("failed to get Route from LB cluster: %w", err)
 	}
@@ -192,7 +193,7 @@ func (r *IngressReconciler) cleanup(ctx context.Context, ingress *networkingv1.I
 	}
 
 	// Find the Route in LB cluster and delete it
-	err = cleanupRoute(ctx, r.LBClient, string(ingress.UID), r.ClusterName)
+	err = cleanupRoute(ctx, r.LBManager.GetClient(), string(ingress.UID), r.ClusterName)
 	if err != nil {
 		return reconcile.Result{}, fmt.Errorf("failed to cleanup route: %w", err)
 	}
@@ -279,5 +280,9 @@ func (r *IngressReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			&corev1.Service{},
 			handler.EnqueueRequestsFromMapFunc(r.enqueueResources()),
 		).
+		WatchesRawSource(
+			source.Kind(r.LBManager.GetCache(), &kubelbv1alpha1.Route{},
+				handler.TypedEnqueueRequestsFromMapFunc[*kubelbv1alpha1.Route](enqueueRoutes("Ingress.networking.k8s.io", r.ClusterName))),
+		).
 		Complete(r)
 }

internal/controllers/ccm/secret_conversion_controller.go

@@ -108,6 +108,7 @@ func (r *SecretConversionReconciler) reconcile(ctx context.Context, _ logr.Logge
 	syncSecret.Labels[kubelb.LabelOriginName] = secret.Name
 	syncSecret.Data = secret.Data
 	syncSecret.StringData = secret.StringData
+	syncSecret.Immutable = secret.Immutable
 	syncSecret.Type = secret.Type
 	return CreateOrUpdateSyncSecret(ctx, r.Client, syncSecret)
 }