move hermetic test for extproc into a new integration package and move k8s API logic into a BeforeSuite() function run before the hermetic tests

Author: BenjaminBraunDev

pkg/ext-proc/test/artifacts/inferencepool-with-model-hermetic.yaml

@@ -1,5 +1,5 @@
 // Package test contains e2e tests for the ext proc while faking the backend pods.
-package test
+package integration
 
 import (
 	"bufio"
@@ -24,11 +24,11 @@ import (
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/api/v1alpha1"
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/backend"
 	runserver "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/server"
+	extprocutils "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/test"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	k8syaml "k8s.io/apimachinery/pkg/util/yaml"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/rest"
 	klog "k8s.io/klog/v2"
 	k8sclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -40,6 +40,7 @@ const (
 )
 
 var (
+	runner    *runserver.ExtProcServerRunner
 	k8sClient k8sclient.Client
 	testEnv   *envtest.Environment
 	scheme    = runtime.NewScheme()
@@ -57,7 +58,7 @@ func SKIPTestHandleRequestBody(t *testing.T) {
 	}{
 		{
 			name: "success",
-			req:  GenerateRequest("my-model"),
+			req:  extprocutils.GenerateRequest("my-model"),
 			models: map[string]*v1alpha1.InferenceModel{
 				"my-model": {
 					Spec: v1alpha1.InferenceModelSpec{
@@ -75,7 +76,7 @@ func SKIPTestHandleRequestBody(t *testing.T) {
 			// model being active, and has low KV cache.
 			pods: []*backend.PodMetrics{
 				{
-					Pod: FakePod(0),
+					Pod: extprocutils.FakePod(0),
 					Metrics: backend.Metrics{
 						WaitingQueueSize:    0,
 						KVCacheUsagePercent: 0.2,
@@ -86,7 +87,7 @@ func SKIPTestHandleRequestBody(t *testing.T) {
 					},
 				},
 				{
-					Pod: FakePod(1),
+					Pod: extprocutils.FakePod(1),
 					Metrics: backend.Metrics{
 						WaitingQueueSize:    0,
 						KVCacheUsagePercent: 0.1,
@@ -97,7 +98,7 @@ func SKIPTestHandleRequestBody(t *testing.T) {
 					},
 				},
 				{
-					Pod: FakePod(2),
+					Pod: extprocutils.FakePod(2),
 					Metrics: backend.Metrics{
 						WaitingQueueSize:    10,
 						KVCacheUsagePercent: 0.2,
@@ -172,7 +173,7 @@ func TestKubeInferenceModelRequest(t *testing.T) {
 	}{
 		{
 			name: "success",
-			req:  GenerateRequest("sql-lora"),
+			req:  extprocutils.GenerateRequest("sql-lora"),
 			// pod-1 will be picked because it has relatively low queue size, with the requested
 			// model being active, and has low KV cache.
 			wantHeaders: []*configPb.HeaderValueOption{
@@ -194,29 +195,9 @@ func TestKubeInferenceModelRequest(t *testing.T) {
 		},
 	}
 
-	// Set up mock k8s API Client
-	testEnv = &envtest.Environment{
-		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "..", "config", "crd", "bases")},
-		ErrorIfCRDPathMissing: true,
-	}
-	cfg, err := testEnv.Start()
-	if err != nil {
-		log.Fatalf("Failed to start test environment, cfg: %v error: %v", cfg, err)
-	}
-
-	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
-	utilruntime.Must(v1alpha1.AddToScheme(scheme))
-
-	k8sClient, err = k8sclient.New(cfg, k8sclient.Options{Scheme: scheme})
-	if err != nil {
-		log.Fatalf("Failed to start k8s Client: %v", err)
-	} else if k8sClient == nil {
-		log.Fatalf("No error, but returned kubernetes client is nil, cfg: %v", cfg)
-	}
-
 	pods := []*backend.PodMetrics{
 		{
-			Pod: FakePod(0),
+			Pod: extprocutils.FakePod(0),
 			Metrics: backend.Metrics{
 				WaitingQueueSize:    0,
 				KVCacheUsagePercent: 0.2,
@@ -227,7 +208,7 @@ func TestKubeInferenceModelRequest(t *testing.T) {
 			},
 		},
 		{
-			Pod: FakePod(1),
+			Pod: extprocutils.FakePod(1),
 			Metrics: backend.Metrics{
 				WaitingQueueSize:    0,
 				KVCacheUsagePercent: 0.1,
@@ -238,7 +219,7 @@ func TestKubeInferenceModelRequest(t *testing.T) {
 			},
 		},
 		{
-			Pod: FakePod(2),
+			Pod: extprocutils.FakePod(2),
 			Metrics: backend.Metrics{
 				WaitingQueueSize:    10,
 				KVCacheUsagePercent: 0.2,
@@ -248,9 +229,13 @@ func TestKubeInferenceModelRequest(t *testing.T) {
 			},
 		},
 	}
+
+	// Set up global k8sclient and extproc server runner with test environment config
+	BeforeSuit()
+
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			client, cleanup := setUpHermeticServer(t, cfg, pods)
+			client, cleanup := setUpHermeticServer(t, pods)
 			t.Cleanup(cleanup)
 			want := &extProcPb.ProcessingResponse{
 				Response: &extProcPb.ProcessingResponse_RequestBody{
@@ -283,7 +268,7 @@ func TestKubeInferenceModelRequest(t *testing.T) {
 
 func setUpServer(t *testing.T, pods []*backend.PodMetrics, models map[string]*v1alpha1.InferenceModel) (client extProcPb.ExternalProcessor_ProcessClient, cleanup func()) {
 	t.Logf("Setting up ExtProc server")
-	server := StartExtProc(port, time.Second, time.Second, pods, models)
+	server := extprocutils.StartExtProc(port, time.Second, time.Second, pods, models)
 
 	address := fmt.Sprintf("localhost:%v", port)
 	// Create a grpc connection
@@ -304,33 +289,13 @@ func setUpServer(t *testing.T, pods []*backend.PodMetrics, models map[string]*v1
 	}
 }
 
-func setUpHermeticServer(t *testing.T, cfg *rest.Config, pods []*backend.PodMetrics) (client extProcPb.ExternalProcessor_ProcessClient, cleanup func()) {
+func setUpHermeticServer(t *testing.T, pods []*backend.PodMetrics) (client extProcPb.ExternalProcessor_ProcessClient, cleanup func()) {
 	t.Logf("Setting up hermetic ExtProc server")
 	klog.InitFlags(nil)
 	flag.Parse()
 	// Configure klog verbosity levels to print ext proc logs.
 	_ = flag.Lookup("v").Value.Set("3")
 
-	runner := &runserver.ExtProcServerRunner{
-		GrpcPort:               port,
-		TargetPodHeader:        "target-pod",
-		PoolName:               "vllm-llama2-7b-pool",
-		PoolNamespace:          "default",
-		ServiceName:            "",
-		Zone:                   "",
-		RefreshPodsInterval:    10 * time.Second,
-		RefreshMetricsInterval: 50 * time.Millisecond,
-		Scheme:                 scheme,
-		Config:                 cfg,
-		Datastore:              backend.NewK8sDataStore(),
-	}
-	runner.Setup()
-
-	// Start the controller manager in go routine, not blocking
-	go func() {
-		runner.StartManager()
-	}()
-
 	// Unmarshal CRDs from file into structs
 	manifestsPath := filepath.Join(".", "artifacts", "inferencepool-with-model-hermetic.yaml")
 	docs, err := readDocuments(manifestsPath)
@@ -392,6 +357,51 @@ func setUpHermeticServer(t *testing.T, cfg *rest.Config, pods []*backend.PodMetr
 	}
 }
 
+// Sets up a test environment and returns the runner struct
+func BeforeSuit() {
+	// Set up mock k8s API Client
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "config", "crd", "bases")},
+		ErrorIfCRDPathMissing: true,
+	}
+	cfg, err := testEnv.Start()
+
+	if err != nil {
+		log.Fatalf("Failed to start test environment, cfg: %v error: %v", cfg, err)
+	}
+
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+	utilruntime.Must(v1alpha1.AddToScheme(scheme))
+
+	k8sClient, err = k8sclient.New(cfg, k8sclient.Options{Scheme: scheme})
+	if err != nil {
+		log.Fatalf("Failed to start k8s Client: %v", err)
+	} else if k8sClient == nil {
+		log.Fatalf("No error, but returned kubernetes client is nil, cfg: %v", cfg)
+	}
+
+	runner = &runserver.ExtProcServerRunner{
+		GrpcPort:               port,
+		TargetPodHeader:        "target-pod",
+		PoolName:               "vllm-llama2-7b-pool",
+		PoolNamespace:          "default",
+		ServiceName:            "",
+		Zone:                   "",
+		RefreshPodsInterval:    10 * time.Second,
+		RefreshMetricsInterval: 50 * time.Millisecond,
+		Scheme:                 scheme,
+		Config:                 cfg,
+		Datastore:              backend.NewK8sDataStore(),
+	}
+
+	runner.Setup()
+
+	// Start the controller manager in go routine, not blocking
+	go func() {
+		runner.StartManager()
+	}()
+}
+
 func sendRequest(t *testing.T, client extProcPb.ExternalProcessor_ProcessClient, req *extProcPb.ProcessingRequest) (*extProcPb.ProcessingResponse, error) {
 	t.Logf("Sending request: %v", req)
 	if err := client.Send(req); err != nil {