test: adds initial translation e2e tests with testupstream

Makefile

@@ -18,6 +18,10 @@ TAG ?= latest
 ENABLE_MULTI_PLATFORMS ?= false
 HELM_CHART_VERSION ?= v0.0.0-latest
 
+# Arguments for go test. This can be used, for example, to run specific tests via
+# `GO_TEST_EXTRA_ARGS="-run TestName/foo/etc"`.
+GO_TEST_EXTRA_ARGS ?=
+
 # This will print out the help message for contributing to the project.
 .PHONY: help
 help:
@@ -104,7 +108,7 @@ test-cel: envtest apigen
 	@for k8sVersion in $(ENVTEST_K8S_VERSIONS); do \
   		echo "Run CEL Validation on k8s $$k8sVersion"; \
         KUBEBUILDER_ASSETS="$$($(ENVTEST) use $$k8sVersion -p path)" \
-                 go test ./tests/cel-validation --tags test_cel_validation -v -count=1; \
+                 go test ./tests/cel-validation $(GO_TEST_EXTRA_ARGS) --tags test_cel_validation -v -count=1; \
     done
 
 # This runs the end-to-end tests for extproc without controller or k8s at all.
@@ -116,15 +120,15 @@ test-extproc: build.extproc
 	@$(MAKE) build.extproc_custom_router CMD_PATH_PREFIX=examples
 	@$(MAKE) build.testupstream CMD_PATH_PREFIX=tests
 	@echo "Run ExtProc test"
-	@go test ./tests/extproc/... -tags test_extproc -v -count=1
+	@go test ./tests/extproc/... $(GO_TEST_EXTRA_ARGS) -tags test_extproc -v -count=1
 
 # This runs the end-to-end tests for the controller with EnvTest.
 .PHONY: test-controller
 test-controller: envtest apigen
 	@for k8sVersion in $(ENVTEST_K8S_VERSIONS); do \
   		echo "Run Controller tests on k8s $$k8sVersion"; \
         KUBEBUILDER_ASSETS="$$($(ENVTEST) use $$k8sVersion -p path)" \
-                 go test ./tests/controller --tags test_controller -v -count=1; \
+                 go test ./tests/controller $(GO_TEST_EXTRA_ARGS) --tags test_controller -v -count=1; \
     done
 
 # This runs the end-to-end tests for the controller and extproc with a local kind cluster.
@@ -133,8 +137,9 @@ test-controller: envtest apigen
 .PHONY: test-e2e
 test-e2e: kind
 	@$(MAKE) docker-build DOCKER_BUILD_ARGS="--load"
+	@$(MAKE) docker-build.testupstream CMD_PATH_PREFIX=tests DOCKER_BUILD_ARGS="--load"
 	@echo "Run E2E tests"
-	@go test ./tests/e2e/... -tags test_e2e -v -count=1
+	@go test ./tests/e2e/... $(GO_TEST_EXTRA_ARGS) -tags test_e2e -v -count=1
 
 # This builds a binary for the given command under the internal/cmd directory.
 #
@@ -187,6 +192,12 @@ build.%:
 #
 # Example:
 # - `make docker-build.controller TAG=v1.2.3`
+#
+# To build the main functions outside cmd/ directory, set CMD_PATH_PREFIX to the directory containing the main function.
+#
+# Example:
+# - `make docker-build.extproc_custom_router CMD_PATH_PREFIX=examples`
+# - `make docker-build.testupstream CMD_PATH_PREFIX=tests`
 .PHONY: docker-build.%
 docker-build.%:
 	$(eval COMMAND_NAME := $(subst docker-build.,,$@))
@@ -198,7 +209,7 @@ docker-build.%:
 	@$(MAKE) build.$(COMMAND_NAME) GOOS_LIST="linux"
 	docker buildx build . -t $(OCI_REGISTRY)/$(COMMAND_NAME):$(TAG) --build-arg COMMAND_NAME=$(COMMAND_NAME) $(PLATFORMS) $(DOCKER_BUILD_ARGS)
 
-# This builds docker images for all commands. All options for `docker-build.%` apply.
+# This builds docker images for all commands under cmd/ directory. All options for `docker-build.%` apply.
 #
 # Example:
 # - `make docker-build`

internal/controller/ai_gateway_route.go

@@ -46,7 +46,6 @@ type aiGatewayRouteController struct {
 	client              client.Client
 	kube                kubernetes.Interface
 	logger              logr.Logger
-	logLevel            string
 	defaultExtProcImage string
 	eventChan           chan ConfigSinkEvent
 }
@@ -59,7 +58,7 @@ func NewAIGatewayRouteController(
 	return &aiGatewayRouteController{
 		client:              client,
 		kube:                kube,
-		logger:              logger.WithName("ai-gateway-route-controller"),
+		logger:              logger.WithName("eaig-route-controller"),
 		defaultExtProcImage: options.ExtProcImage,
 		eventChan:           ch,
 	}
@@ -204,7 +203,7 @@ func (c *aiGatewayRouteController) reconcileExtProcDeployment(ctx context.Contex
 									Ports:           []corev1.ContainerPort{{Name: "grpc", ContainerPort: 1063}},
 									Args: []string{
 										"-configPath", "/etc/ai-gateway/extproc/" + expProcConfigFileName,
-										"-logLevel", c.logLevel,
+										"-logLevel", "info", // TODO: this should be configurable via FilterConfig API.
 									},
 									VolumeMounts: []corev1.VolumeMount{
 										{Name: "config", MountPath: "/etc/ai-gateway/extproc"},
@@ -268,7 +267,7 @@ func (c *aiGatewayRouteController) reconcileExtProcDeployment(ctx context.Contex
 }
 
 func extProcName(route *aigv1a1.AIGatewayRoute) string {
-	return fmt.Sprintf("ai-gateway-ai-gateway-route-extproc-%s", route.Name)
+	return fmt.Sprintf("eaig-route-extproc-%s", route.Name)
 }
 
 func ownerReferenceForAIGatewayRoute(aiGatewayRoute *aigv1a1.AIGatewayRoute) []metav1.OwnerReference {

internal/controller/ai_gateway_route_test.go

@@ -27,7 +27,7 @@ func Test_extProcName(t *testing.T) {
 			Name: "myroute",
 		},
 	})
-	require.Equal(t, "ai-gateway-ai-gateway-route-extproc-myroute", actual)
+	require.Equal(t, "eaig-route-extproc-myroute", actual)
 }
 
 func TestAIGatewayRouteController_ensuresExtProcConfigMapExists(t *testing.T) {

internal/controller/sink.go

@@ -7,6 +7,7 @@ import (
 	"github.com/go-logr/logr"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 	"sigs.k8s.io/yaml"
@@ -89,6 +90,7 @@ func (c *configSink) handleEvent(event ConfigSinkEvent) {
 
 func (c *configSink) syncAIGatewayRoute(aiGatewayRoute *aigv1a1.AIGatewayRoute) {
 	// Check if the HTTPRoute exists.
+	c.logger.Info("syncing AIGatewayRoute", "namespace", aiGatewayRoute.Namespace, "name", aiGatewayRoute.Name)
 	var httpRoute gwapiv1.HTTPRoute
 	err := c.client.Get(context.Background(), client.ObjectKey{Name: aiGatewayRoute.Name, Namespace: aiGatewayRoute.Namespace}, &httpRoute)
 	existingRoute := err == nil
@@ -115,11 +117,13 @@ func (c *configSink) syncAIGatewayRoute(aiGatewayRoute *aigv1a1.AIGatewayRoute)
 	}
 
 	if existingRoute {
+		c.logger.Info("updating HTTPRoute", "namespace", httpRoute.Namespace, "name", httpRoute.Name)
 		if err := c.client.Update(context.Background(), &httpRoute); err != nil {
 			c.logger.Error(err, "failed to update HTTPRoute", "namespace", httpRoute.Namespace, "name", httpRoute.Name)
 			return
 		}
 	} else {
+		c.logger.Info("creating HTTPRoute", "namespace", httpRoute.Namespace, "name", httpRoute.Name)
 		if err := c.client.Create(context.Background(), &httpRoute); err != nil {
 			c.logger.Error(err, "failed to create HTTPRoute", "namespace", httpRoute.Namespace, "name", httpRoute.Name)
 			return
@@ -142,6 +146,10 @@ func (c *configSink) syncAIServiceBackend(aiBackend *aigv1a1.AIServiceBackend) {
 		return
 	}
 	for _, aiGatewayRoute := range aiGatewayRoutes.Items {
+		c.logger.Info("syncing AIGatewayRoute",
+			"namespace", aiGatewayRoute.Namespace, "name", aiGatewayRoute.Name,
+			"referenced_backend", aiBackend.Name, "referenced_backend_namespace", aiBackend.Namespace,
+		)
 		c.syncAIGatewayRoute(&aiGatewayRoute)
 	}
 }
@@ -229,6 +237,17 @@ func (c *configSink) newHTTPRoute(dst *gwapiv1.HTTPRoute, aiGatewayRoute *aigv1a
 		}
 		rules[i] = rule
 	}
+
+	// Adds the default route rule with "/" path.
+	rules = append(rules, gwapiv1.HTTPRouteRule{
+		Matches: []gwapiv1.HTTPRouteMatch{
+			{Path: &gwapiv1.HTTPPathMatch{Value: ptr.To("/")}},
+		},
+		BackendRefs: []gwapiv1.HTTPBackendRef{
+			{BackendRef: gwapiv1.BackendRef{BackendObjectReference: backends[0].Spec.BackendRef.BackendObjectReference}},
+		},
+	})
+
 	dst.Spec.Rules = rules
 
 	targetRefs := aiGatewayRoute.Spec.TargetRefs

internal/controller/sink_test.go

@@ -86,12 +86,15 @@ func TestConfigSink_syncAIGatewayRoute(t *testing.T) {
 		var updatedHTTPRoute gwapiv1.HTTPRoute
 		err = fakeClient.Get(context.Background(), client.ObjectKey{Name: "route1", Namespace: "ns1"}, &updatedHTTPRoute)
 		require.NoError(t, err)
-		require.Len(t, updatedHTTPRoute.Spec.Rules, 2)
+		require.Len(t, updatedHTTPRoute.Spec.Rules, 3) // 2 backends + 1 for the default rule.
 		require.Len(t, updatedHTTPRoute.Spec.Rules[0].BackendRefs, 1)
 		require.Equal(t, "some-backend1", string(updatedHTTPRoute.Spec.Rules[0].BackendRefs[0].BackendRef.Name))
 		require.Equal(t, "apple.ns1", updatedHTTPRoute.Spec.Rules[0].Matches[0].Headers[0].Value)
 		require.Equal(t, "some-backend2", string(updatedHTTPRoute.Spec.Rules[1].BackendRefs[0].BackendRef.Name))
 		require.Equal(t, "orange.ns1", updatedHTTPRoute.Spec.Rules[1].Matches[0].Headers[0].Value)
+		// Defaulting to the first backend.
+		require.Equal(t, "some-backend1", string(updatedHTTPRoute.Spec.Rules[2].BackendRefs[0].BackendRef.Name))
+		require.Equal(t, "/", *updatedHTTPRoute.Spec.Rules[2].Matches[0].Path.Value)
 	})
 }
 
@@ -196,11 +199,17 @@ func Test_newHTTPRoute(t *testing.T) {
 			BackendRefs: []gwapiv1.HTTPBackendRef{{BackendRef: gwapiv1.BackendRef{BackendObjectReference: gwapiv1.BackendObjectReference{Name: "some-backend4", Namespace: ptr.To[gwapiv1.Namespace]("ns1")}}}},
 		},
 	}
-	require.Len(t, httpRoute.Spec.Rules, 4)
+	require.Len(t, httpRoute.Spec.Rules, 5) // 4 backends + 1 for the default rule.
 	for i, r := range httpRoute.Spec.Rules {
 		t.Run(fmt.Sprintf("rule-%d", i), func(t *testing.T) {
-			require.Equal(t, expRules[i].Matches, r.Matches)
-			require.Equal(t, expRules[i].BackendRefs, r.BackendRefs)
+			if i == 4 {
+				require.Equal(t, expRules[0].BackendRefs, r.BackendRefs)
+				require.NotNil(t, r.Matches[0].Path)
+				require.Equal(t, "/", *r.Matches[0].Path.Value)
+			} else {
+				require.Equal(t, expRules[i].Matches, r.Matches)
+				require.Equal(t, expRules[i].BackendRefs, r.BackendRefs)
+			}
 		})
 	}
 }

internal/extproc/mocks_test.go

@@ -3,6 +3,7 @@ package extproc
 import (
 	"context"
 	"io"
+	"log/slog"
 	"testing"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -23,7 +24,7 @@ var (
 	_ extprocapi.Router     = &mockRouter{}
 )
 
-func newMockProcessor(_ *processorConfig) *mockProcessor {
+func newMockProcessor(_ *processorConfig, _ *slog.Logger) *mockProcessor {
 	return &mockProcessor{}
 }
 

internal/extproc/processor.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"log/slog"
 	"unicode/utf8"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -44,12 +45,13 @@ type ProcessorIface interface {
 }
 
 // NewProcessor creates a new processor.
-func NewProcessor(config *processorConfig) *Processor {
-	return &Processor{config: config}
+func NewProcessor(config *processorConfig, logger *slog.Logger) *Processor {
+	return &Processor{config: config, logger: logger}
 }
 
 // Processor handles the processing of the request and response messages for a single stream.
 type Processor struct {
+	logger           *slog.Logger
 	config           *processorConfig
 	requestHeaders   map[string]string
 	responseEncoding string
@@ -72,12 +74,14 @@ func (p *Processor) ProcessRequestBody(_ context.Context, rawBody *extprocv3.Htt
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse request body: %w", err)
 	}
+	p.logger.Info("Processing request", "path", path, "model", model)
 
 	p.requestHeaders[p.config.ModelNameHeaderKey] = model
 	b, err := p.config.router.Calculate(p.requestHeaders)
 	if err != nil {
 		return nil, fmt.Errorf("failed to calculate route: %w", err)
 	}
+	p.logger.Info("Selected backend", "backend", b.Name)
 
 	factory, ok := p.config.factories[b.OutputSchema]
 	if !ok {

internal/extproc/processor_test.go

@@ -3,6 +3,7 @@ package extproc
 import (
 	"context"
 	"errors"
+	"log/slog"
 	"testing"
 
 	corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -86,7 +87,7 @@ func TestProcessor_ProcessRequestBody(t *testing.T) {
 		headers := map[string]string{":path": "/foo"}
 		rbp := mockRequestBodyParser{t: t, retModelName: "some-model", expPath: "/foo"}
 		rt := mockRouter{t: t, expHeaders: headers, retErr: errors.New("test error")}
-		p := &Processor{config: &processorConfig{bodyParser: rbp.impl, router: rt}, requestHeaders: headers}
+		p := &Processor{config: &processorConfig{bodyParser: rbp.impl, router: rt}, requestHeaders: headers, logger: slog.Default()}
 		_, err := p.ProcessRequestBody(context.Background(), &extprocv3.HttpBody{})
 		require.ErrorContains(t, err, "failed to calculate route: test error")
 	})
@@ -100,7 +101,7 @@ func TestProcessor_ProcessRequestBody(t *testing.T) {
 		p := &Processor{config: &processorConfig{
 			bodyParser: rbp.impl, router: rt,
 			factories: make(map[filterconfig.VersionedAPISchema]translator.Factory),
-		}, requestHeaders: headers}
+		}, requestHeaders: headers, logger: slog.Default()}
 		_, err := p.ProcessRequestBody(context.Background(), &extprocv3.HttpBody{})
 		require.ErrorContains(t, err, "failed to find factory for output schema {\"some-schema\" \"v10.0\"}")
 	})
@@ -117,7 +118,7 @@ func TestProcessor_ProcessRequestBody(t *testing.T) {
 			factories: map[filterconfig.VersionedAPISchema]translator.Factory{
 				{Schema: "some-schema", Version: "v10.0"}: factory.impl,
 			},
-		}, requestHeaders: headers}
+		}, requestHeaders: headers, logger: slog.Default()}
 		_, err := p.ProcessRequestBody(context.Background(), &extprocv3.HttpBody{})
 		require.ErrorContains(t, err, "failed to create translator: test error")
 	})
@@ -134,7 +135,7 @@ func TestProcessor_ProcessRequestBody(t *testing.T) {
 			factories: map[filterconfig.VersionedAPISchema]translator.Factory{
 				{Schema: "some-schema", Version: "v10.0"}: factory.impl,
 			},
-		}, requestHeaders: headers}
+		}, requestHeaders: headers, logger: slog.Default()}
 		_, err := p.ProcessRequestBody(context.Background(), &extprocv3.HttpBody{})
 		require.ErrorContains(t, err, "failed to transform request: test error")
 	})
@@ -157,7 +158,7 @@ func TestProcessor_ProcessRequestBody(t *testing.T) {
 			},
 			selectedBackendHeaderKey: "x-ai-gateway-backend-key",
 			ModelNameHeaderKey:       "x-ai-gateway-model-key",
-		}, requestHeaders: headers}
+		}, requestHeaders: headers, logger: slog.Default()}
 		resp, err := p.ProcessRequestBody(context.Background(), &extprocv3.HttpBody{})
 		require.NoError(t, err)
 		require.Equal(t, mt, p.translator)

internal/extproc/server.go

@@ -23,11 +23,11 @@ import (
 type Server[P ProcessorIface] struct {
 	logger       *slog.Logger
 	config       *processorConfig
-	newProcessor func(*processorConfig) P
+	newProcessor func(*processorConfig, *slog.Logger) P
 }
 
 // NewServer creates a new external processor server.
-func NewServer[P ProcessorIface](logger *slog.Logger, newProcessor func(*processorConfig) P) (*Server[P], error) {
+func NewServer[P ProcessorIface](logger *slog.Logger, newProcessor func(*processorConfig, *slog.Logger) P) (*Server[P], error) {
 	srv := &Server[P]{logger: logger, newProcessor: newProcessor}
 	return srv, nil
 }
@@ -78,7 +78,7 @@ func (s *Server[P]) LoadConfig(config *filterconfig.Config) error {
 
 // Process implements [extprocv3.ExternalProcessorServer].
 func (s *Server[P]) Process(stream extprocv3.ExternalProcessor_ProcessServer) error {
-	p := s.newProcessor(s.config)
+	p := s.newProcessor(s.config, s.logger)
 	return s.process(p, stream)
 }
 

internal/extproc/server_test.go

@@ -104,13 +104,13 @@ func TestServer_Watch(t *testing.T) {
 func TestServer_processMsg(t *testing.T) {
 	t.Run("unknown request type", func(t *testing.T) {
 		s := requireNewServerWithMockProcessor(t)
-		p := s.newProcessor(nil)
+		p := s.newProcessor(nil, slog.Default())
 		_, err := s.processMsg(context.Background(), p, &extprocv3.ProcessingRequest{})
 		require.ErrorContains(t, err, "unknown request type")
 	})
 	t.Run("request headers", func(t *testing.T) {
 		s := requireNewServerWithMockProcessor(t)
-		p := s.newProcessor(nil)
+		p := s.newProcessor(nil, slog.Default())
 
 		hm := &corev3.HeaderMap{Headers: []*corev3.HeaderValue{{Key: "foo", Value: "bar"}}}
 		expResponse := &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_RequestHeaders{}}
@@ -127,7 +127,7 @@ func TestServer_processMsg(t *testing.T) {
 	})
 	t.Run("request body", func(t *testing.T) {
 		s := requireNewServerWithMockProcessor(t)
-		p := s.newProcessor(nil)
+		p := s.newProcessor(nil, slog.Default())
 
 		reqBody := &extprocv3.HttpBody{}
 		expResponse := &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_RequestBody{}}
@@ -144,7 +144,7 @@ func TestServer_processMsg(t *testing.T) {
 	})
 	t.Run("response headers", func(t *testing.T) {
 		s := requireNewServerWithMockProcessor(t)
-		p := s.newProcessor(nil)
+		p := s.newProcessor(nil, slog.Default())
 
 		hm := &corev3.HeaderMap{Headers: []*corev3.HeaderValue{{Key: "foo", Value: "bar"}}}
 		expResponse := &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_ResponseHeaders{}}
@@ -161,7 +161,7 @@ func TestServer_processMsg(t *testing.T) {
 	})
 	t.Run("response body", func(t *testing.T) {
 		s := requireNewServerWithMockProcessor(t)
-		p := s.newProcessor(nil)
+		p := s.newProcessor(nil, slog.Default())
 
 		reqBody := &extprocv3.HttpBody{}
 		expResponse := &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_ResponseBody{}}
@@ -213,7 +213,7 @@ func TestServer_Process(t *testing.T) {
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
 
-		p := s.newProcessor(nil)
+		p := s.newProcessor(nil, slog.Default())
 		hm := &corev3.HeaderMap{Headers: []*corev3.HeaderValue{{Key: "foo", Value: "bar"}}}
 		expResponse := &extprocv3.ProcessingResponse{Response: &extprocv3.ProcessingResponse_RequestHeaders{}}
 		p.t = t