feat: add TCPRoute controller (#55)

This primarily adds the `TCPRoute` controller and samples. Note however
that the dataplane does not have complete TCP support yet (though some
scaffolding for it is added as of this PR) so the dataplane
configuration produced is valid but the samples wont work yet. This
scaffolding however will help to support #45 and #15 which are
responsible for completing this.

Resolves #16

This also includes some refactoring and cleanup, notably it refactors
the dataplane code into modules.

Makefile

@@ -268,3 +268,8 @@ KIND_CLUSTER ?= blixt-dev
 .PHONY: build.cluster
 build.cluster: $(KTF) # builds a KIND cluster which can be used for testing and development
 	PATH="$(LOCALBIN):${PATH}" $(KTF) env create --name $(KIND_CLUSTER) --addon metallb
+
+.PHONY: load.image
+load.image: build.image
+	kind load docker-image $(IMAGE):$(TAG) --name $(KIND_CLUSTER) && \
+		kubectl -n blixt-system rollout restart deployment blixt-controlplane

PROJECT

@@ -22,4 +22,9 @@ resources:
   group: gateway
   kind: UDPRoute
   version: v1alpha2
+- controller: true
+  domain: networking.k8s.io
+  group: gateway
+  kind: TCPRoute
+  version: v1alpha2
 version: "3"

config/rbac/role.yaml

@@ -121,6 +121,32 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - gateway.networking.k8s.io
+  resources:
+  - tcproutes
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - gateway.networking.k8s.io
+  resources:
+  - tcproutes/finalizers
+  verbs:
+  - update
+- apiGroups:
+  - gateway.networking.k8s.io
+  resources:
+  - tcproutes/status
+  verbs:
+  - get
+  - patch
+  - update
 - apiGroups:
   - gateway.networking.k8s.io
   resources:

config/samples/tcproute/gateway.yaml

@@ -0,0 +1,17 @@
+kind: GatewayClass
+apiVersion: gateway.networking.k8s.io/v1beta1
+metadata:
+  name: blixt-tcproute-sample
+spec:
+  controllerName: konghq.com/blixt
+---
+kind: Gateway
+apiVersion: gateway.networking.k8s.io/v1beta1
+metadata:
+  name: blixt-tcproute-sample
+spec:
+  gatewayClassName: blixt-tcproute-sample
+  listeners:
+  - name: tcp
+    protocol: TCP
+    port: 9080

config/samples/tcproute/kustomization.yaml

@@ -0,0 +1,5 @@
+resources:
+- gateway.yaml
+- server.yaml
+- tcproute.yaml
+#+kubebuilder:scaffold:manifestskustomizesamples

config/samples/tcproute/server.yaml

@@ -0,0 +1,39 @@
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: blixt-tcproute-sample
+  labels:
+    app: blixt-tcproute-sample
+spec:
+  selector:
+    matchLabels:
+      app: blixt-tcproute-sample
+  template:
+    metadata:
+      labels:
+        app: blixt-tcproute-sample
+    spec:
+      containers:
+      - name: server
+        image: nginx
+        imagePullPolicy: IfNotPresent
+        ports:
+        - containerPort: 9080
+          protocol: TCP
+---
+apiVersion: v1
+kind: Service
+metadata:
+  labels:
+    app: blixt-tcproute-sample
+  name: blixt-tcproute-sample
+spec:
+  ports:
+  - name: tcp
+    port: 9080
+    protocol: TCP
+    targetPort: 9080
+  selector:
+    app: blixt-tcproute-sample
+  type: ClusterIP

config/samples/tcproute/tcproute.yaml

@@ -0,0 +1,12 @@
+apiVersion: gateway.networking.k8s.io/v1alpha2
+kind: TCPRoute
+metadata:
+  name: blixt-tcproute-sample
+spec:
+  parentRefs:
+  - name: blixt-tcproute-sample
+    port: 9080
+  rules:
+  - backendRefs:
+    - name: blixt-tcproute-sample
+      port: 9080

controllers/tcproute_controller.go

@@ -0,0 +1,236 @@
+package controllers
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-logr/logr"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+	gatewayv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
+	gatewayv1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
+
+	dataplane "github.com/kong/blixt/internal/dataplane/client"
+	"github.com/kong/blixt/pkg/vars"
+)
+
+//+kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=tcproutes,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=tcproutes/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=gateway.networking.k8s.io,resources=tcproutes/finalizers,verbs=update
+//+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
+//+kubebuilder:rbac:groups=core,resources=pods/status,verbs=get
+//+kubebuilder:rbac:groups=apps,resources=daemonsets,verbs=get;list;watch
+//+kubebuilder:rbac:groups=apps,resources=daemonsets/status,verbs=get
+
+// TCPRouteReconciler reconciles a TCPRoute object
+type TCPRouteReconciler struct {
+	client.Client
+	Scheme *runtime.Scheme
+
+	log logr.Logger
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *TCPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.log = log.FromContext(context.Background())
+
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&gatewayv1alpha2.TCPRoute{}).
+		Watches(
+			&source.Kind{Type: &appsv1.DaemonSet{}},
+			handler.EnqueueRequestsFromMapFunc(r.mapDataPlaneDaemonsetToTCPRoutes),
+		).
+		Watches(
+			&source.Kind{Type: &gatewayv1beta1.Gateway{}},
+			handler.EnqueueRequestsFromMapFunc(r.mapGatewayToTCPRoutes),
+		).
+		Complete(r)
+}
+
+// TCProuteReconciler reconciles TCPRoute object
+func (r *TCPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	tcproute := new(gatewayv1alpha2.TCPRoute)
+	if err := r.Get(ctx, req.NamespacedName, tcproute); err != nil {
+		if errors.IsNotFound(err) {
+			r.log.Info("object enqueued no longer exists, skipping")
+			return ctrl.Result{}, nil
+		}
+		r.log.Info("Error retrieving tcp route", "Err : ", err)
+		return ctrl.Result{}, err
+	}
+
+	isManaged, gateway, err := r.isTCPRouteManaged(ctx, *tcproute)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+	if !isManaged {
+		// TODO: enable orphan checking https://github.com/Kong/blixt/issues/47
+		return ctrl.Result{}, nil
+	}
+
+	if !controllerutil.ContainsFinalizer(tcproute, DataPlaneFinalizer) {
+		if tcproute.DeletionTimestamp != nil {
+			// if the finalizer isn't set, AND the object is being deleted then there's
+			// no reason to bother with dataplane configuration for it its already
+			// handled.
+			return ctrl.Result{}, nil
+		}
+		// if the finalizer is not set, and the object is not being deleted, set the
+		// finalizer before we do anything else to ensure we don't lose track of
+		// dataplane configuration.
+		return ctrl.Result{}, setDataPlaneFinalizer(ctx, r.Client, tcproute)
+	}
+
+	// if the TCPRoute is being deleted, remove it from the DataPlane
+	// TODO: enable deletion grace period https://github.com/Kong/blixt/issues/48
+	if tcproute.DeletionTimestamp != nil {
+		return ctrl.Result{}, r.ensureTCPRouteDeletedInDataPlane(ctx, tcproute, gateway)
+	}
+
+	// in all other cases ensure the TCPRoute is configured in the dataplane
+	if err := r.ensureTCPRouteConfiguredInDataPlane(ctx, tcproute, gateway); err != nil {
+		if err.Error() == "endpoints not ready" {
+			r.log.Info("endpoints not yet ready for TCPRoute, retrying", "namespace", tcproute.Namespace, "name", tcproute.Name)
+			return ctrl.Result{RequeueAfter: time.Second}, nil
+		}
+		return ctrl.Result{}, err
+	}
+
+	return ctrl.Result{}, nil
+}
+
+// isTCPRouteManaged verifies wether a provided TCPRoute is managed by this
+// controller, according to it's Gateway and GatewayClass.
+func (r *TCPRouteReconciler) isTCPRouteManaged(ctx context.Context, tcproute gatewayv1alpha2.TCPRoute) (bool, *gatewayv1beta1.Gateway, error) {
+	var supportedGateways []gatewayv1beta1.Gateway
+
+	//Use the retrieve objects its parent ref to look for the gateway.
+	for _, parentRef := range tcproute.Spec.ParentRefs {
+		//Build Gateway object to retrieve
+		gw := new(gatewayv1beta1.Gateway)
+
+		ns := tcproute.Namespace
+		if parentRef.Namespace != nil {
+			ns = string(*parentRef.Namespace)
+		}
+
+		//Get Gateway for TCP Route
+		if err := r.Get(ctx, types.NamespacedName{Name: string(parentRef.Name), Namespace: ns}, gw); err != nil {
+			if !errors.IsNotFound(err) {
+				return false, nil, err
+			}
+			continue
+		}
+
+		//Get GatewayClass for the Gateway and match to our name of controler
+		gwc := new(gatewayv1beta1.GatewayClass)
+		if err := r.Get(ctx, types.NamespacedName{Name: string(gw.Spec.GatewayClassName), Namespace: ns}, gwc); err != nil {
+			if !errors.IsNotFound(err) {
+				return false, nil, err
+			}
+			continue
+		}
+
+		if gwc.Spec.ControllerName != vars.GatewayClassControllerName {
+			// not managed by this implementation, check the next parent ref
+			continue
+		}
+
+		//Check if referred gateway has the at least one listener with properties defined from TCPRoute parentref.
+		if err := r.verifyListener(ctx, gw, parentRef); err != nil {
+			// until the Gateway has a relevant listener, we can't operate on the route.
+			// Updates to the relevant Gateway will re-enqueue the TCPRoute reconcilation to retry.
+			r.log.Info("No matching listener found for referred gateway", "GatewayName", parentRef.Name, "GatewayPort", parentRef.Port)
+			//Check next parent ref.
+			continue
+		}
+
+		supportedGateways = append(supportedGateways, *gw)
+	}
+
+	if len(supportedGateways) < 1 {
+		return false, nil, nil
+	}
+
+	// TODO: support multiple gateways https://github.com/Kong/blixt/issues/40
+	referredGateway := &supportedGateways[0]
+	r.log.Info("TCP Route appeared referring to Gateway", "Gateway ", referredGateway.Name, "GatewayClass Name", referredGateway.Spec.GatewayClassName)
+
+	return true, referredGateway, nil
+}
+
+// verifyListener verifies that the provided gateway has at least one listener
+// matching the provided ParentReference.
+func (r *TCPRouteReconciler) verifyListener(ctx context.Context, gw *gatewayv1beta1.Gateway, tcprouteSpec gatewayv1alpha2.ParentReference) error {
+	for _, listener := range gw.Spec.Listeners {
+		if (listener.Protocol == gatewayv1beta1.TCPProtocolType) && (listener.Port == gatewayv1beta1.PortNumber(*tcprouteSpec.Port)) {
+			return nil
+		}
+	}
+	return fmt.Errorf("No matching Gateway listener found for defined Parentref")
+}
+
+func (r *TCPRouteReconciler) ensureTCPRouteConfiguredInDataPlane(ctx context.Context, tcproute *gatewayv1alpha2.TCPRoute, gateway *gatewayv1beta1.Gateway) error {
+	// build the dataplane configuration from the TCPRoute and its Gateway
+	targets, err := dataplane.CompileTCPRouteToDataPlaneBackend(ctx, r.Client, tcproute, gateway)
+	if err != nil {
+		return err
+	}
+
+	// TODO: add multiple endpoint support https://github.com/Kong/blixt/issues/46
+	dataplaneClient, err := dataplane.NewDataPlaneClient(context.Background(), r.Client)
+	if err != nil {
+		return err
+	}
+
+	confirmation, err := dataplaneClient.Update(context.Background(), targets)
+	if err != nil {
+		return err
+	}
+
+	r.log.Info(fmt.Sprintf("successful data-plane UPDATE, confirmation: %s", confirmation.String()))
+
+	return nil
+}
+
+func (r *TCPRouteReconciler) ensureTCPRouteDeletedInDataPlane(ctx context.Context, tcproute *gatewayv1alpha2.TCPRoute, gateway *gatewayv1beta1.Gateway) error {
+	// build the dataplane configuration from the TCPRoute and its Gateway
+	targets, err := dataplane.CompileTCPRouteToDataPlaneBackend(ctx, r.Client, tcproute, gateway)
+	if err != nil {
+		return err
+	}
+
+	// TODO: add multiple endpoint support https://github.com/Kong/blixt/issues/46
+	dataplaneClient, err := dataplane.NewDataPlaneClient(context.Background(), r.Client)
+	if err != nil {
+		return err
+	}
+
+	// delete the target from the dataplane
+	confirmation, err := dataplaneClient.Delete(context.Background(), targets.Vip)
+	if err != nil {
+		return err
+	}
+
+	r.log.Info(fmt.Sprintf("successful data-plane DELETE, confirmation: %s", confirmation.String()))
+
+	oldFinalizers := tcproute.GetFinalizers()
+	newFinalizers := make([]string, 0, len(oldFinalizers)-1)
+	for _, finalizer := range oldFinalizers {
+		if finalizer != DataPlaneFinalizer {
+			newFinalizers = append(newFinalizers, finalizer)
+		}
+	}
+	tcproute.SetFinalizers(newFinalizers)
+
+	return r.Client.Update(ctx, tcproute)
+}