scaffolding for the gateway api controller

Change-Id: I2bca38e888a281a8a325e8e242ce11b5edebc45d

Author: aojea

go.mod

@@ -15,6 +15,7 @@ require (
 	k8s.io/controller-manager v0.32.3
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
+	sigs.k8s.io/gateway-api v1.2.1
 	sigs.k8s.io/kind v0.27.0
 )
 

go.sum

@@ -200,6 +200,8 @@ k8s.io/kube-openapi v0.0.0-20241212222426-2c72e554b1e7 h1:hcha5B1kVACrLujCKLbr8X
 k8s.io/kube-openapi v0.0.0-20241212222426-2c72e554b1e7/go.mod h1:GewRfANuJ70iYzvn+i4lezLDAFzvjxZYK1gn1lWcfas=
 k8s.io/utils v0.0.0-20241210054802-24370beab758 h1:sdbE21q2nlQtFh65saZY+rRM6x6aJJI8IUa1AmH/qa0=
 k8s.io/utils v0.0.0-20241210054802-24370beab758/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
+sigs.k8s.io/gateway-api v1.2.1 h1:fZZ/+RyRb+Y5tGkwxFKuYuSRQHu9dZtbjenblleOLHM=
+sigs.k8s.io/gateway-api v1.2.1/go.mod h1:EpNfEXNjiYfUJypf0eZ0P5iXA9ekSGWaS1WgPaM42X0=
 sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 h1:gBQPwqORJ8d8/YNZWEjoZs7npUVDpVXUUOFfW6CgAqE=
 sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
 sigs.k8s.io/kind v0.27.0 h1:PQ3f0iAWNIj66LYkZ1ivhEg/+Zb6UPMbO+qVei/INZA=

pkg/controller/controller.go

@@ -20,9 +20,14 @@ import (
 	controllersmetrics "k8s.io/component-base/metrics/prometheus/controllers"
 	ccmfeatures "k8s.io/controller-manager/pkg/features"
 	"k8s.io/klog/v2"
+
+	gatewayclient "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned"
+	gatewayinformers "sigs.k8s.io/gateway-api/pkg/client/informers/externalversions"
+
 	cpkconfig "sigs.k8s.io/cloud-provider-kind/pkg/config"
 	"sigs.k8s.io/cloud-provider-kind/pkg/constants"
 	"sigs.k8s.io/cloud-provider-kind/pkg/container"
+	"sigs.k8s.io/cloud-provider-kind/pkg/gateway"
 	"sigs.k8s.io/cloud-provider-kind/pkg/loadbalancer"
 	"sigs.k8s.io/cloud-provider-kind/pkg/provider"
 	"sigs.k8s.io/kind/pkg/cluster"
@@ -39,6 +44,7 @@ type ccm struct {
 	factory           informers.SharedInformerFactory
 	serviceController *servicecontroller.Controller
 	nodeController    *nodecontroller.CloudNodeController
+	gatewayController *gateway.Controller
 	cancelFn          context.CancelFunc
 }
 
@@ -74,15 +80,15 @@ func (c *Controller) Run(ctx context.Context) {
 				continue
 			}
 
-			kubeClient, err := c.getKubeClient(ctx, cluster)
+			restConfig, err := c.getRestConfig(ctx, cluster)
 			if err != nil {
 				klog.Errorf("Failed to create kubeClient for cluster %s: %v", cluster, err)
 				continue
 			}
 
 			klog.V(2).Infof("Creating new cloud provider for cluster %s", cluster)
 			cloud := provider.New(cluster, c.kind)
-			ccm, err := startCloudControllerManager(ctx, cluster, kubeClient, cloud)
+			ccm, err := startCloudControllerManager(ctx, cluster, restConfig, cloud)
 			if err != nil {
 				klog.Errorf("Failed to start cloud controller for cluster %s: %v", cluster, err)
 				continue
@@ -123,9 +129,9 @@ func (c *Controller) getKubeConfig(cluster string, internal bool) (*rest.Config,
 	return config, nil
 }
 
-// getKubeClient returns a kubeclient for the cluster passed as argument
+// getRestConfig returns a valid rest.Config for the cluster passed as argument
 // It tries first to connect to the internal endpoint.
-func (c *Controller) getKubeClient(ctx context.Context, cluster string) (kubernetes.Interface, error) {
+func (c *Controller) getRestConfig(ctx context.Context, cluster string) (*rest.Config, error) {
 	addresses := []string{}
 	internalConfig, err := c.getKubeConfig(cluster, true)
 	if err != nil {
@@ -172,17 +178,12 @@ func (c *Controller) getKubeClient(ctx context.Context, cluster string) (kuberne
 		return nil, fmt.Errorf("restConfig for host %s not avaliable", host)
 	}
 
-	kubeClient, err := kubernetes.NewForConfig(config)
-	if err != nil {
-		klog.Errorf("Failed to create kubeClient for cluster %s: %v", cluster, err)
-		return kubeClient, err
-	}
-	return kubeClient, nil
+	return config, nil
 }
 
 // TODO: implement leader election to not have problems with  multiple providers
 // ref: https://github.com/kubernetes/kubernetes/blob/d97ea0f705847f90740cac3bc3dd8f6a4026d0b5/cmd/kube-scheduler/app/server.go#L211
-func startCloudControllerManager(ctx context.Context, clusterName string, kubeClient kubernetes.Interface, cloud cloudprovider.Interface) (*ccm, error) {
+func startCloudControllerManager(ctx context.Context, clusterName string, config *rest.Config, cloud cloudprovider.Interface) (*ccm, error) {
 	// TODO: we need to set up the ccm specific feature gates
 	// but try to avoid to expose this to users
 	featureGates := utilfeature.DefaultMutableFeatureGate
@@ -191,6 +192,12 @@ func startCloudControllerManager(ctx context.Context, clusterName string, kubeCl
 		return nil, err
 	}
 
+	kubeClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		klog.Errorf("Failed to create kubeClient for cluster %s: %v", clusterName, err)
+		return nil, err
+	}
+
 	client := kubeClient.Discovery().RESTClient()
 	// wait for health
 	err = wait.PollUntilContextTimeout(ctx, 1*time.Second, 30*time.Second, true, func(ctx context.Context) (bool, error) {
@@ -246,6 +253,33 @@ func startCloudControllerManager(ctx context.Context, clusterName string, kubeCl
 
 	sharedInformers.Start(ctx.Done())
 
+	// Gateway setup
+	gwClient, err := gatewayclient.NewForConfig(config)
+	if err != nil {
+		// This error shouldn't fail. It lives like this as a legacy.
+		klog.Errorf("Failed to create Gateway API client: %v", err)
+		cancel()
+		return nil, err
+	}
+
+	sharedGwInformers := gatewayinformers.NewSharedInformerFactory(gwClient, 60*time.Second)
+
+	gatewayController, err := gateway.New(
+		gwClient,
+		sharedGwInformers.Gateway().V1().Gateways(),
+		sharedGwInformers.Gateway().V1().HTTPRoutes(),
+		sharedGwInformers.Gateway().V1().GRPCRoutes(),
+	)
+	if err != nil {
+		klog.Errorf("Failed to start gateway controller: %v", err)
+		cancel()
+		return nil, err
+	}
+
+	go gatewayController.Run(ctx)
+
+	sharedGwInformers.Start(ctx.Done())
+
 	// This has to cleanup all the resources allocated by the cloud provider in this cluster
 	// - containers as loadbalancers
 	// - in windows and darwin ip addresses on the loopback interface
@@ -290,6 +324,7 @@ func startCloudControllerManager(ctx context.Context, clusterName string, kubeCl
 		factory:           sharedInformers,
 		serviceController: serviceController,
 		nodeController:    nodeController,
+		gatewayController: gatewayController,
 		cancelFn:          cancelFn}, nil
 }
 

pkg/gateway/controller.go

@@ -0,0 +1,146 @@
+package gateway
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
+
+	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+	gatewayclient "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned"
+	gatewayinformers "sigs.k8s.io/gateway-api/pkg/client/informers/externalversions/apis/v1"
+	gatewaylisters "sigs.k8s.io/gateway-api/pkg/client/listers/apis/v1"
+)
+
+const (
+	controllerName = "kind.sigs.k8s.io/gateway-controller"
+	gwClassName    = "cloud-provider-kind"
+	maxRetries     = 5
+	workers        = 5
+)
+
+type Controller struct {
+	gwClient            gatewayclient.Interface
+	gatewayLister       gatewaylisters.GatewayLister
+	gatewayListerSynced cache.InformerSynced
+	gatewayqueue        workqueue.TypedRateLimitingInterface[string]
+
+	httprouteLister       gatewaylisters.HTTPRouteLister
+	httprouteListerSynced cache.InformerSynced
+	httproutequeue        workqueue.TypedRateLimitingInterface[string]
+
+	grpcrouteLister       gatewaylisters.GRPCRouteLister
+	grpcrouteListerSynced cache.InformerSynced
+	grpcroutequeue        workqueue.TypedRateLimitingInterface[string]
+}
+
+func New(
+	gwClient *gatewayclient.Clientset,
+	gatewayInformer gatewayinformers.GatewayInformer,
+	httprouteInformer gatewayinformers.HTTPRouteInformer,
+	grpcrouteInformer gatewayinformers.GRPCRouteInformer,
+) (*Controller, error) {
+	c := &Controller{
+		gwClient:            gwClient,
+		gatewayLister:       gatewayInformer.Lister(),
+		gatewayListerSynced: gatewayInformer.Informer().HasSynced,
+		gatewayqueue: workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{Name: "gateway"},
+		),
+		httprouteLister:       httprouteInformer.Lister(),
+		httprouteListerSynced: httprouteInformer.Informer().HasSynced,
+		httproutequeue: workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{Name: "httproute"},
+		),
+		grpcrouteLister:       grpcrouteInformer.Lister(),
+		grpcrouteListerSynced: grpcrouteInformer.Informer().HasSynced,
+		grpcroutequeue: workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{Name: "grpcroute"},
+		),
+	}
+
+	_, err := gatewayInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) {},
+		UpdateFunc: func(oldObj, newObj interface{}) {},
+		DeleteFunc: func(obj interface{}) {},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = httprouteInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) {},
+		UpdateFunc: func(oldObj, newObj interface{}) {},
+		DeleteFunc: func(obj interface{}) {},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = grpcrouteInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) {},
+		UpdateFunc: func(oldObj, newObj interface{}) {},
+		DeleteFunc: func(obj interface{}) {},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}
+
+// Run begins watching and syncing.
+func (c *Controller) Run(ctx context.Context) error {
+	defer runtime.HandleCrashWithContext(ctx)
+
+	// Let the workers stop when we are done
+	defer c.gatewayqueue.ShutDown()
+	defer c.httproutequeue.ShutDown()
+	defer c.grpcroutequeue.ShutDown()
+	klog.Info("Starting Gateway API controller")
+
+	// Create GatewayClass if it does not exist
+	gwClass := gatewayv1.GatewayClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: gwClassName,
+		},
+		Spec: gatewayv1.GatewayClassSpec{
+			ControllerName: controllerName,
+			Description:    ptr.To("cloud-provider-kind gateway API"),
+		},
+	}
+
+	if _, err := c.gwClient.GatewayV1().GatewayClasses().Get(ctx, gwClassName, metav1.GetOptions{}); apierrors.IsNotFound(err) {
+		_, err := c.gwClient.GatewayV1().GatewayClasses().Create(ctx, &gwClass, metav1.CreateOptions{})
+		if err != nil {
+			klog.Infof("faile to create cloud-provider-kind GatewayClass: %v")
+			return err
+		}
+	}
+
+	// Wait for all involved caches to be synced, before processing items from the queue is started
+	if !cache.WaitForNamedCacheSync(controllerName, ctx.Done(), c.gatewayListerSynced, c.httprouteListerSynced, c.grpcrouteListerSynced) {
+		return fmt.Errorf("Timed out waiting for caches to sync")
+	}
+
+	for i := 0; i < workers; i++ {
+		go wait.UntilWithContext(ctx, c.runGatewayWorker, time.Second)
+		go wait.UntilWithContext(ctx, c.runHTTPRouteWorker, time.Second)
+		go wait.UntilWithContext(ctx, c.runGRPCrouteWorker, time.Second)
+	}
+
+	<-ctx.Done()
+	klog.Info("Stopping Gateway API controller")
+	return nil
+}

pkg/gateway/gateway.go

@@ -0,0 +1,84 @@
+package gateway
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+func (c *Controller) processNextGatewayItem() bool {
+	// Wait until there is a new item in the working queue
+	key, quit := c.gatewayqueue.Get()
+	if quit {
+		return false
+	}
+	defer c.gatewayqueue.Done(key)
+
+	err := c.syncGateway(key)
+
+	c.handleGatewayErr(err, key)
+	return true
+}
+
+// syncToStdout is the business logic of the controller. In this controller it simply prints
+// information about the pod to stdout. In case an error happened, it has to simply return the error.
+// The retry logic should not be part of the business logic.
+func (c *Controller) syncGateway(key string) error {
+	startTime := time.Now()
+	defer func() {
+		klog.V(4).Infof("Finished syncing gateway %q (%v)", key, time.Since(startTime))
+	}()
+
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return err
+	}
+
+	gw, err := c.gatewayLister.Gateways(namespace).Get(name)
+	if err != nil {
+		klog.Errorf("Fetching object with key %s from store failed with %v", key, err)
+		return err
+	}
+
+	if apierrors.IsNotFound(err) {
+		// Below we will warm up our cache with a Pod, so that we will see a delete for one pod
+		fmt.Printf("Gateway %s does not exist anymore\n", key)
+	} else {
+		// Note that you also have to check the uid if you have a local controlled resource, which
+		// is dependent on the actual instance, to detect that a Pod was recreated with the same name
+		fmt.Printf("Sync/Add/Update for Gateway %s\n", gw.GetName())
+	}
+	return nil
+}
+
+// handleErr checks if an error happened and makes sure we will retry later.
+func (c *Controller) handleGatewayErr(err error, key string) {
+	if err == nil {
+		c.gatewayqueue.Forget(key)
+		return
+	}
+
+	if c.gatewayqueue.NumRequeues(key) < maxRetries {
+		klog.Infof("Error syncing Gateway %v: %v", key, err)
+
+		// Re-enqueue the key rate limited. Based on the rate limiter on the
+		// queue and the re-enqueue history, the key will be processed later again.
+		c.gatewayqueue.AddRateLimited(key)
+		return
+	}
+
+	c.gatewayqueue.Forget(key)
+	// Report to an external entity that, even after several retries, we could not successfully process this key
+	runtime.HandleError(err)
+	klog.Infof("Dropping Gateway %q out of the queue: %v", key, err)
+}
+
+func (c *Controller) runGatewayWorker(ctx context.Context) {
+	for c.processNextGatewayItem() {
+	}
+}