Repository: envoyproxy/ai-gateway
Pull request #434 (2025-02-26 17:27:47)

controller: removes the leftover from config sink
**Commit Message**

This removes the left over implementation from when we had config sink. 

**Related Issues/PRs (if applicable)**

Follow up on #364 

mathetake: @aabchoo PTAL!

Patch:
diff --git a/internal/controller/ai_gateway_route.go b/internal/controller/ai_gateway_route.go
index 9fb496c..4e3d401 100644
--- a/internal/controller/ai_gateway_route.go
+++ b/internal/controller/ai_gateway_route.go
@@ -7,7 +7,6 @@ package controller
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"path"
 	"sort"
@@ -93,23 +92,6 @@ func (c *AIGatewayRouteController) Reconcile(ctx context.Context, req reconcile.
 		return ctrl.Result{}, err
 	}
 
-	// TODO: merge this into syncAIGatewayRoute. This is a left over from the previous sink based implementation.
-	c.logger.Info("Ensuring extproc configmap exists", "namespace", aiGatewayRoute.Namespace, "name", aiGatewayRoute.Name)
-	if err := c.ensuresExtProcConfigMapExists(ctx, &aiGatewayRoute); err != nil {
-		c.logger.Error(err, "failed to reconcile extProc config map")
-		message := fmt.Sprintf("failed to ensure extproc configmap exists: %v", err)
-		c.updateAIGatewayRouteStatus(ctx, &aiGatewayRoute, false, message)
-		return ctrl.Result{}, errors.New(message)
-	}
-	// TODO: merge this into syncAIGatewayRoute. This is a left over from the previous sink based implementation.
-	c.logger.Info("Reconciling extension policy", "namespace", aiGatewayRoute.Namespace, "name", aiGatewayRoute.Name)
-	if err := c.reconcileExtProcExtensionPolicy(ctx, &aiGatewayRoute); err != nil {
-		c.logger.Error(err, "failed to reconcile extension policy")
-		message := fmt.Sprintf("failed to reconcile extension policy: %v", err)
-		c.updateAIGatewayRouteStatus(ctx, &aiGatewayRoute, false, message)
-		return ctrl.Result{}, errors.New(message)
-	}
-
 	if err := c.syncAIGatewayRoute(ctx, &aiGatewayRoute); err != nil {
 		c.logger.Error(err, "failed to sync AIGatewayRoute")
 		c.updateAIGatewayRouteStatus(ctx, &aiGatewayRoute, false, err.Error())
@@ -168,28 +150,6 @@ func (c *AIGatewayRouteController) reconcileExtProcExtensionPolicy(ctx context.C
 	return
 }
 
-// ensuresExtProcConfigMapExists ensures that a configmap exists for the external process.
-// This must happen before the external processor deployment is created.
-func (c *AIGatewayRouteController) ensuresExtProcConfigMapExists(ctx context.Context, aiGatewayRoute *aigv1a1.AIGatewayRoute) (err error) {
-	name := extProcName(aiGatewayRoute)
-	// Check if a configmap exists for extproc exists, and if not, create one with the default config.
-	_, err = c.kube.CoreV1().ConfigMaps(aiGatewayRoute.Namespace).Get(ctx, name, metav1.GetOptions{})
-	if apierrors.IsNotFound(err) {
-		configMap := &corev1.ConfigMap{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      name,
-				Namespace: aiGatewayRoute.Namespace,
-			},
-			Data: map[string]string{expProcConfigFileName: filterapi.DefaultConfig},
-		}
-		if err = ctrlutil.SetControllerReference(aiGatewayRoute, configMap, c.client.Scheme()); err != nil {
-			panic(fmt.Errorf("BUG: failed to set controller reference for extproc configmap: %w", err))
-		}
-		_, err = c.kube.CoreV1().ConfigMaps(aiGatewayRoute.Namespace).Create(ctx, configMap, metav1.CreateOptions{})
-	}
-	return
-}
-
 func extProcName(route *aigv1a1.AIGatewayRoute) string {
 	return fmt.Sprintf("ai-eg-route-extproc-%s", route.Name)
 }
@@ -236,6 +196,10 @@ func (c *AIGatewayRouteController) syncAIGatewayRoute(ctx context.Context, aiGat
 		return fmt.Errorf("failed to get HTTPRouteFilter: %w", err)
 	}
 
+	if err = c.reconcileExtProcExtensionPolicy(ctx, aiGatewayRoute); err != nil {
+		return fmt.Errorf("failed to reconcile extension policy: %w", err)
+	}
+
 	// Check if the HTTPRoute exists.
 	c.logger.Info("syncing AIGatewayRoute", "namespace", aiGatewayRoute.Namespace, "name", aiGatewayRoute.Name)
 	var httpRoute gwapiv1.HTTPRoute
@@ -276,7 +240,7 @@ func (c *AIGatewayRouteController) syncAIGatewayRoute(ctx context.Context, aiGat
 
 	// Update the extproc configmap.
 	uuid := string(uuid2.NewUUID())
-	if err = c.updateExtProcConfigMap(ctx, aiGatewayRoute, uuid); err != nil {
+	if err = c.reconcileExtProcConfigMap(ctx, aiGatewayRoute, uuid); err != nil {
 		return fmt.Errorf("failed to update extproc configmap: %w", err)
 	}
 
@@ -294,14 +258,8 @@ func (c *AIGatewayRouteController) syncAIGatewayRoute(ctx context.Context, aiGat
 	return nil
 }
 
-// updateExtProcConfigMap updates the external processor configmap with the new AIGatewayRoute.
-func (c *AIGatewayRouteController) updateExtProcConfigMap(ctx context.Context, aiGatewayRoute *aigv1a1.AIGatewayRoute, uuid string) error {
-	configMap, err := c.kube.CoreV1().ConfigMaps(aiGatewayRoute.Namespace).Get(ctx, extProcName(aiGatewayRoute), metav1.GetOptions{})
-	if err != nil {
-		// This is a bug since we should have created the configmap before sending the AIGatewayRoute to the configSink.
-		panic(fmt.Errorf("failed to get configmap %s: %w", extProcName(aiGatewayRoute), err))
-	}
-
+// reconcileExtProcConfigMap updates the external processor configmap with the new AIGatewayRoute.
+func (c *AIGatewayRouteController) reconcileExtProcConfigMap(ctx context.Context, aiGatewayRoute *aigv1a1.AIGatewayRoute, uuid string) error {
 	ec := &filterapi.Config{UUID: uuid}
 	spec := &aiGatewayRoute.Spec
 
@@ -310,6 +268,7 @@ func (c *AIGatewayRouteController) updateExtProcConfigMap(ctx context.Context, a
 	ec.ModelNameHeaderKey = aigv1a1.AIModelHeaderKey
 	ec.SelectedBackendHeaderKey = selectedBackendHeaderKey
 	ec.Rules = make([]filterapi.RouteRule, len(spec.Rules))
+	var err error
 	for i := range spec.Rules {
 		rule := &spec.Rules[i]
 		ec.Rules[i].Backends = make([]filterapi.Backend, len(rule.BackendRefs))
@@ -395,9 +354,27 @@ func (c *AIGatewayRouteController) updateExtProcConfigMap(ctx context.Context, a
 	if err != nil {
 		return fmt.Errorf("failed to marshal extproc config: %w", err)
 	}
-	if configMap.Data == nil {
-		configMap.Data = make(map[string]string)
+
+	name := extProcName(aiGatewayRoute)
+	configMap, err := c.kube.CoreV1().ConfigMaps(aiGatewayRoute.Namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			configMap = &corev1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: aiGatewayRoute.Namespace},
+				Data:       map[string]string{expProcConfigFileName: string(marshaled)},
+			}
+			if err = ctrlutil.SetControllerReference(aiGatewayRoute, configMap, c.client.Scheme()); err != nil {
+				panic(fmt.Errorf("BUG: failed to set controller reference for extproc configmap: %w", err))
+			}
+			_, err = c.kube.CoreV1().ConfigMaps(aiGatewayRoute.Namespace).Create(ctx, configMap, metav1.CreateOptions{})
+			if err != nil {
+				return fmt.Errorf("failed to create configmap %s: %w", name, err)
+			}
+			return nil
+		}
+		return fmt.Errorf("failed to get configmap %s: %w", name, err)
 	}
+
 	configMap.Data[expProcConfigFileName] = string(marshaled)
 	if _, err := c.kube.CoreV1().ConfigMaps(aiGatewayRoute.Namespace).Update(ctx, configMap, metav1.UpdateOptions{}); err != nil {
 		return fmt.Errorf("failed to update configmap %s: %w", configMap.Name, err)


Test patch:
diff --git a/internal/controller/ai_gateway_route_test.go b/internal/controller/ai_gateway_route_test.go
index 25d60aa..33ea848 100644
--- a/internal/controller/ai_gateway_route_test.go
+++ b/internal/controller/ai_gateway_route_test.go
@@ -10,7 +10,6 @@ import (
 	"fmt"
 	"strconv"
 	"testing"
-	"time"
 
 	egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1"
 	"github.com/go-logr/logr"
@@ -80,30 +79,6 @@ func Test_extProcName(t *testing.T) {
 	require.Equal(t, "ai-eg-route-extproc-myroute", actual)
 }
 
-func TestAIGatewayRouteController_ensuresExtProcConfigMapExists(t *testing.T) {
-	c := &AIGatewayRouteController{client: fake.NewClientBuilder().WithScheme(scheme).Build()}
-	c.kube = fake2.NewClientset()
-	name := "myroute"
-	ownerRef := []metav1.OwnerReference{
-		{APIVersion: "aigateway.envoyproxy.io/v1alpha1", Kind: "AIGatewayRoute", Name: name, Controller: ptr.To(true), BlockOwnerDeletion: ptr.To(true)},
-	}
-	aiGatewayRoute := &aigv1a1.AIGatewayRoute{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"}}
-
-	err := c.ensuresExtProcConfigMapExists(t.Context(), aiGatewayRoute)
-	require.NoError(t, err)
-
-	configMap, err := c.kube.CoreV1().ConfigMaps("default").Get(t.Context(), extProcName(aiGatewayRoute), metav1.GetOptions{})
-	require.NoError(t, err)
-	require.Equal(t, extProcName(aiGatewayRoute), configMap.Name)
-	require.Equal(t, "default", configMap.Namespace)
-	require.Equal(t, ownerRef, configMap.OwnerReferences)
-	require.Equal(t, filterapi.DefaultConfig, configMap.Data[expProcConfigFileName])
-
-	// Doing it again should not fail.
-	err = c.ensuresExtProcConfigMapExists(t.Context(), aiGatewayRoute)
-	require.NoError(t, err)
-}
-
 func TestAIGatewayRouteController_reconcileExtProcExtensionPolicy(t *testing.T) {
 	c := &AIGatewayRouteController{client: fake.NewClientBuilder().WithScheme(scheme).Build()}
 	name := "myroute"
@@ -291,12 +266,6 @@ func TestAIGatewayRouterController_syncAIGatewayRoute(t *testing.T) {
 		err = fakeClient.Create(t.Context(), httpRoute, &client.CreateOptions{})
 		require.NoError(t, err)
 
-		// Create the initial configmap.
-		_, err = kube.CoreV1().ConfigMaps(route.Namespace).Create(t.Context(), &corev1.ConfigMap{
-			ObjectMeta: metav1.ObjectMeta{Name: extProcName(route), Namespace: route.Namespace},
-		}, metav1.CreateOptions{})
-		require.NoError(t, err)
-
 		// Then sync, which should update the HTTPRoute.
 		require.NoError(t, s.syncAIGatewayRoute(t.Context(), route))
 		var updatedHTTPRoute gwapiv1.HTTPRoute
@@ -447,7 +416,7 @@ func Test_newHTTPRoute(t *testing.T) {
 	}
 }
 
-func TestAIGatewayRouteController_updateExtProcConfigMap(t *testing.T) {
+func TestAIGatewayRouteController_reconcileExtProcConfigMap(t *testing.T) {
 	fakeClient := requireNewFakeClientWithIndexes(t)
 	kube := fake2.NewClientset()
 
@@ -656,12 +625,7 @@ func TestAIGatewayRouteController_updateExtProcConfigMap(t *testing.T) {
 		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			_, err := s.kube.CoreV1().ConfigMaps(tc.route.Namespace).Create(t.Context(), &corev1.ConfigMap{
-				ObjectMeta: metav1.ObjectMeta{Name: extProcName(tc.route), Namespace: tc.route.Namespace},
-			}, metav1.CreateOptions{})
-			require.NoError(t, err)
-
-			err = s.updateExtProcConfigMap(t.Context(), tc.route, tc.exp.UUID)
+			err := s.reconcileExtProcConfigMap(t.Context(), tc.route, tc.exp.UUID)
 			require.NoError(t, err)
 
 			cm, err := s.kube.CoreV1().ConfigMaps(tc.route.Namespace).Get(t.Context(), extProcName(tc.route), metav1.GetOptions{})
@@ -677,7 +641,6 @@ func TestAIGatewayRouteController_updateExtProcConfigMap(t *testing.T) {
 }
 
 func TestAIGatewayRouteController_syncExtProcDeployment(t *testing.T) {
-	t.Skip()
 	fakeClient := requireNewFakeClientWithIndexes(t)
 	kube := fake2.NewClientset()
 
@@ -786,20 +749,14 @@ func TestAIGatewayRouteController_syncExtProcDeployment(t *testing.T) {
 				corev1.ResourceMemory: resource.MustParse("100Mi"),
 			},
 		}
-		require.Eventually(t, func() bool {
-			extProcDeployment, err := s.kube.AppsV1().Deployments("ns").Get(t.Context(), extProcName(aiGatewayRoute), metav1.GetOptions{})
-			if err != nil {
-				t.Logf("failed to get deployment %s: %v", extProcName(aiGatewayRoute), err)
-				return false
-			}
-			require.Equal(t, "envoyproxy/ai-gateway-extproc:foo", extProcDeployment.Spec.Template.Spec.Containers[0].Image)
-			require.Len(t, extProcDeployment.OwnerReferences, 1)
-			require.Equal(t, "myroute", extProcDeployment.OwnerReferences[0].Name)
-			require.Equal(t, "AIGatewayRoute", extProcDeployment.OwnerReferences[0].Kind)
-			require.Equal(t, int32(123), *extProcDeployment.Spec.Replicas)
-			require.Equal(t, resourceLimits, &extProcDeployment.Spec.Template.Spec.Containers[0].Resources)
-			return true
-		}, 30*time.Second, 200*time.Millisecond)
+		extProcDeployment, err := s.kube.AppsV1().Deployments("ns").Get(t.Context(), extProcName(aiGatewayRoute), metav1.GetOptions{})
+		require.NoError(t, err)
+		require.Equal(t, "envoyproxy/ai-gateway-extproc:foo", extProcDeployment.Spec.Template.Spec.Containers[0].Image)
+		require.Len(t, extProcDeployment.OwnerReferences, 1)
+		require.Equal(t, "myroute", extProcDeployment.OwnerReferences[0].Name)
+		require.Equal(t, "AIGatewayRoute", extProcDeployment.OwnerReferences[0].Kind)
+		require.Equal(t, int32(123), *extProcDeployment.Spec.Replicas)
+		require.Equal(t, resourceLimits, &extProcDeployment.Spec.Template.Spec.Containers[0].Resources)
 
 		service, err := s.kube.CoreV1().Services("ns").Get(t.Context(), extProcName(aiGatewayRoute), metav1.GetOptions{})
 		require.NoError(t, err)
@@ -820,24 +777,17 @@ func TestAIGatewayRouteController_syncExtProcDeployment(t *testing.T) {
 
 		require.NoError(t, s.syncExtProcDeployment(t.Context(), aiGatewayRoute))
 		// Check the deployment is updated.
-		require.Eventually(t, func() bool {
-			extProcDeployment, err := s.kube.AppsV1().Deployments("ns").Get(t.Context(), extProcName(aiGatewayRoute), metav1.GetOptions{})
-			if err != nil {
-				t.Logf("failed to get deployment %s: %v", extProcName(aiGatewayRoute), err)
-				return false
-			}
-			require.Equal(t, "envoyproxy/ai-gateway-extproc:foo", extProcDeployment.Spec.Template.Spec.Containers[0].Image)
-			require.Len(t, extProcDeployment.OwnerReferences, 1)
-			require.Equal(t, "myroute", extProcDeployment.OwnerReferences[0].Name)
-			require.Equal(t, "AIGatewayRoute", extProcDeployment.OwnerReferences[0].Kind)
-			require.Equal(t, int32(456), *extProcDeployment.Spec.Replicas)
-			require.Equal(t, newResourceLimits, &extProcDeployment.Spec.Template.Spec.Containers[0].Resources)
-
-			for _, v := range extProcDeployment.Spec.Template.Spec.Containers[0].VolumeMounts {
-				require.True(t, v.ReadOnly)
-			}
-			return true
-		}, 30*time.Second, 200*time.Millisecond)
+		extProcDeployment, err := s.kube.AppsV1().Deployments("ns").Get(t.Context(), extProcName(aiGatewayRoute), metav1.GetOptions{})
+		require.NoError(t, err)
+		require.Equal(t, "envoyproxy/ai-gateway-extproc:foo", extProcDeployment.Spec.Template.Spec.Containers[0].Image)
+		require.Len(t, extProcDeployment.OwnerReferences, 1)
+		require.Equal(t, "myroute", extProcDeployment.OwnerReferences[0].Name)
+		require.Equal(t, "AIGatewayRoute", extProcDeployment.OwnerReferences[0].Kind)
+		require.Equal(t, int32(456), *extProcDeployment.Spec.Replicas)
+		require.Equal(t, newResourceLimits, &extProcDeployment.Spec.Template.Spec.Containers[0].Resources)
+		for _, v := range extProcDeployment.Spec.Template.Spec.Containers[0].VolumeMounts {
+			require.True(t, v.ReadOnly)
+		}
 	})
 }