Merge pull request #93 from datum-cloud/feat/connector-envoypatch

feat: Add connector EnvoyPatchPolicy for HTTPProxy backends

Author: zachsmith1

api/v1alpha/httpproxy_types.go

@@ -218,6 +218,10 @@ const (
 	// This condition is present and true when a hostname defined in an HTTPProxy
 	// is in use by another resource.
 	HTTPProxyConditionHostnamesInUse = "HostnamesInUse"
+
+	// This condition is true when connector metadata has been programmed
+	// via the downstream EnvoyPatchPolicy.
+	HTTPProxyConditionConnectorMetadataProgrammed = "ConnectorMetadataProgrammed"
 )
 
 const (
@@ -228,6 +232,9 @@ const (
 	// HTTPProxyReasonProgrammed indicates that the HTTP proxy has been programmed.
 	HTTPProxyReasonProgrammed = "Programmed"
 
+	// HTTPProxyReasonConnectorMetadataApplied indicates connector metadata has been applied.
+	HTTPProxyReasonConnectorMetadataApplied = "ConnectorMetadataApplied"
+
 	// HTTPProxyReasonConflict indicates that the HTTP proxy encountered a conflict
 	// when being programmed.
 	HTTPProxyReasonConflict = "Conflict"

cmd/main.go

@@ -218,7 +218,8 @@ func main() {
 	}
 
 	if err := (&controller.HTTPProxyReconciler{
-		Config: serverConfig,
+		Config:            serverConfig,
+		DownstreamCluster: downstreamCluster,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "HTTPProxy")
 		os.Exit(1)

config/rbac/role.yaml

@@ -127,6 +127,18 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - gateway.envoyproxy.io
+  resources:
+  - httproutefilters
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - gateway.networking.k8s.io
   resources:

internal/config/config.go

@@ -470,6 +470,12 @@ type GatewayConfig struct {
 	// +default={"gateway.networking.datumapis.com/certificate-issuer": "auto"}
 	ListenerTLSOptions map[gatewayv1.AnnotationKey]gatewayv1.AnnotationValue `json:"listenerTLSOptions"`
 
+	// ConnectorInternalListenerName is the Envoy internal listener name used by
+	// connector tunnel routing patches.
+	//
+	// +default="connector-tunnel"
+	ConnectorInternalListenerName string `json:"connectorInternalListenerName,omitempty"`
+
 	// Coraza specifies configuration for the Coraza WAF.
 	Coraza CorazaConfig `json:"coraza,omitempty"`
 
@@ -527,6 +533,13 @@ func (c *GatewayConfig) GatewayDNSAddress(gateway *gatewayv1.Gateway) string {
 	return fmt.Sprintf("%s.%s", strings.ReplaceAll(string(gateway.UID), "-", ""), c.TargetDomain)
 }
 
+func (c *GatewayConfig) ConnectorTunnelListenerName() string {
+	if c.ConnectorInternalListenerName == "" {
+		return "connector-tunnel"
+	}
+	return c.ConnectorInternalListenerName
+}
+
 // +k8s:deepcopy-gen=true
 
 type CorazaConfig struct {

internal/config/zz_generated.defaults.go

@@ -16,6 +16,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -105,6 +106,19 @@ func (r *ConnectorReconciler) Reconcile(ctx context.Context, req mcreconcile.Req
 	apimeta.SetStatusCondition(&connector.Status.Conditions, *acceptedCondition)
 	apimeta.SetStatusCondition(&connector.Status.Conditions, *readyCondition)
 
+	if acceptedCondition.Status != metav1.ConditionTrue {
+		readyCondition.Status = metav1.ConditionFalse
+		readyCondition.Reason = networkingv1alpha1.ConnectorReasonNotReady
+		readyCondition.Message = "Waiting for ConnectorClass to be resolved."
+		apimeta.SetStatusCondition(&connector.Status.Conditions, *readyCondition)
+		if !equality.Semantic.DeepEqual(*originalStatus, connector.Status) {
+			if statusErr := cl.GetClient().Status().Update(ctx, &connector); statusErr != nil {
+				return ctrl.Result{}, fmt.Errorf("failed updating connector status: %w", statusErr)
+			}
+		}
+		return ctrl.Result{}, nil
+	}
+
 	leaseDurationSeconds := r.connectorLeaseDurationSeconds()
 	if connector.Status.LeaseRef == nil || connector.Status.LeaseRef.Name == "" {
 		lease := &coordinationv1.Lease{
@@ -214,6 +228,41 @@ func (r *ConnectorReconciler) SetupWithManager(mgr mcmanager.Manager) error {
 
 	return mcbuilder.ControllerManagedBy(mgr).
 		For(&networkingv1alpha1.Connector{}).
+		Watches(
+			&networkingv1alpha1.ConnectorClass{},
+			func(clusterName string, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
+				return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
+					logger := log.FromContext(ctx)
+
+					connectorClass, ok := obj.(*networkingv1alpha1.ConnectorClass)
+					if !ok {
+						return nil
+					}
+
+					var connectors networkingv1alpha1.ConnectorList
+					if err := cl.GetClient().List(ctx, &connectors); err != nil {
+						logger.Error(err, "failed to list Connectors for ConnectorClass watch", "connectorClass", connectorClass.Name)
+						return nil
+					}
+
+					var requests []mcreconcile.Request
+					for i := range connectors.Items {
+						connector := &connectors.Items[i]
+						if connector.Spec.ConnectorClassName != connectorClass.Name {
+							continue
+						}
+						requests = append(requests, mcreconcile.Request{
+							ClusterName: clusterName,
+							Request: ctrl.Request{
+								NamespacedName: client.ObjectKeyFromObject(connector),
+							},
+						})
+					}
+
+					return requests
+				})
+			},
+		).
 		Watches(
 			&coordinationv1.Lease{},
 			mchandler.EnqueueRequestForOwner(&networkingv1alpha1.Connector{}, handler.OnlyControllerOwner()),