Set up constants for main flags in extproc server package, improve formatting and helper functions for runserver.

Author: BenjaminBraunDev

pkg/ext-proc/main.go

@@ -7,7 +7,6 @@ import (
 	"net"
 	"net/http"
 	"strconv"
-	"time"
 
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -36,7 +35,7 @@ const (
 var (
 	grpcPort = flag.Int(
 		"grpcPort",
-		9002,
+		runserver.DefaultGrpcPort,
 		"The gRPC port used for communicating with Envoy proxy")
 	grpcHealthPort = flag.Int(
 		"grpcHealthPort",
@@ -46,31 +45,31 @@ var (
 		"metricsPort", 9090, "The metrics port")
 	targetPodHeader = flag.String(
 		"targetPodHeader",
-		"target-pod",
+		runserver.DefaultTargetPodHeader,
 		"Header key used by Envoy to route to the appropriate pod. This must match Envoy configuration.")
 	poolName = flag.String(
 		"poolName",
-		"",
+		runserver.DefaultPoolName,
 		"Name of the InferencePool this Endpoint Picker is associated with.")
 	poolNamespace = flag.String(
 		"poolNamespace",
-		"default",
+		runserver.DefaultPoolNamespace,
 		"Namespace of the InferencePool this Endpoint Picker is associated with.")
 	serviceName = flag.String(
 		"serviceName",
-		"",
+		runserver.DefaultServiceName,
 		"Name of the Service that will be used to read EndpointSlices from")
 	zone = flag.String(
 		"zone",
-		"",
+		runserver.DefaultZone,
 		"The zone that this instance is created in. Will be passed to the corresponding endpointSlice. ")
 	refreshPodsInterval = flag.Duration(
 		"refreshPodsInterval",
-		10*time.Second,
+		runserver.DefaultRefreshPodsInterval,
 		"interval to refresh pods")
 	refreshMetricsInterval = flag.Duration(
 		"refreshMetricsInterval",
-		50*time.Millisecond,
+		runserver.DefaultRefreshMetricsInterval,
 		"interval to refresh metrics")
 
 	scheme = runtime.NewScheme()
@@ -104,7 +103,7 @@ func main() {
 
 	datastore := backend.NewK8sDataStore()
 
-	runner := &runserver.ExtProcServerRunner{
+	serverRunner := &runserver.ExtProcServerRunner{
 		GrpcPort:               *grpcPort,
 		TargetPodHeader:        *targetPodHeader,
 		PoolName:               *poolName,
@@ -117,11 +116,11 @@ func main() {
 		Config:                 ctrl.GetConfigOrDie(),
 		Datastore:              datastore,
 	}
-	runner.Setup()
+	serverRunner.Setup()
 
 	// Start health and ext-proc servers in goroutines
 	healthSvr := startHealthServer(datastore, *grpcHealthPort)
-	extProcSvr := runner.Start(
+	extProcSvr := serverRunner.Start(
 		datastore,
 		*grpcPort,
 		*refreshPodsInterval,
@@ -132,7 +131,7 @@ func main() {
 	metricsSvr := startMetricsHandler(*metricsPort, cfg)
 
 	// Start manager, blocking
-	runner.StartManager()
+	serverRunner.StartManager()
 
 	// Gracefully shutdown servers
 	if healthSvr != nil {

pkg/ext-proc/server/runserver.go

@@ -30,7 +30,33 @@ type ExtProcServerRunner struct {
 	Scheme                 *runtime.Scheme
 	Config                 *rest.Config
 	Datastore              *backend.K8sDatastore
-	manager                *ctrl.Manager
+	manager                ctrl.Manager
+}
+
+// Default values for CLI flags in main
+const (
+	DefaultGrpcPort               = 9002                  // default for --grpcPort
+	DefaultTargetPodHeader        = "target-pod"          // default for --targetPodHeader
+	DefaultPoolName               = ""                    // required but no default
+	DefaultPoolNamespace          = "default"             // default for --poolNamespace
+	DefaultServiceName            = ""                    // required but no default
+	DefaultZone                   = ""                    // default for --zone
+	DefaultRefreshPodsInterval    = 10 * time.Second      // default for --refreshPodsInterval
+	DefaultRefreshMetricsInterval = 50 * time.Millisecond // default for --refreshMetricsInterval
+)
+
+func NewDefaultExtProcServerRunner() *ExtProcServerRunner {
+	return &ExtProcServerRunner{
+		GrpcPort:               DefaultGrpcPort,
+		TargetPodHeader:        DefaultTargetPodHeader,
+		PoolName:               DefaultPoolName,
+		PoolNamespace:          DefaultPoolNamespace,
+		ServiceName:            DefaultServiceName,
+		Zone:                   DefaultZone,
+		RefreshPodsInterval:    DefaultRefreshPodsInterval,
+		RefreshMetricsInterval: DefaultRefreshMetricsInterval,
+		// Scheme, Config, and Datastore can be assigned later.
+	}
 }
 
 // Setup creates the reconcilers for pools, models, and endpointSlices and starts the manager.
@@ -40,7 +66,7 @@ func (r *ExtProcServerRunner) Setup() {
 	if err != nil {
 		klog.Fatalf("Failed to create controller manager: %v", err)
 	}
-	r.manager = &mgr
+	r.manager = mgr
 
 	// Create the controllers and register them with the manager
 	if err := (&backend.InferencePoolReconciler{
@@ -122,7 +148,7 @@ func (r *ExtProcServerRunner) StartManager() {
 	}
 	// Start the controller manager. Blocking and will return when shutdown is complete.
 	klog.Infof("Starting controller manager")
-	mgr := *r.manager
+	mgr := r.manager
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		klog.Fatalf("Error starting controller manager: %v", err)
 	}

test/integration/hermetic_test.go

@@ -36,14 +36,14 @@ import (
 )
 
 const (
-	port = 9002
+	port = runserver.DefaultGrpcPort
 )
 
 var (
-	runner    *runserver.ExtProcServerRunner
-	k8sClient k8sclient.Client
-	testEnv   *envtest.Environment
-	scheme    = runtime.NewScheme()
+	serverRunner *runserver.ExtProcServerRunner
+	k8sClient    k8sclient.Client
+	testEnv      *envtest.Environment
+	scheme       = runtime.NewScheme()
 )
 
 func SKIPTestHandleRequestBody(t *testing.T) {
@@ -128,9 +128,6 @@ func SKIPTestHandleRequestBody(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-
-			// log.Fatalf("inference model: %v", *test.models["my-model"]) // TEMP
-
 			client, cleanup := setUpServer(t, test.pods, test.models)
 			t.Cleanup(cleanup)
 			want := &extProcPb.ProcessingResponse{
@@ -297,7 +294,7 @@ func setUpHermeticServer(t *testing.T, pods []*backend.PodMetrics) (client extPr
 	_ = flag.Lookup("v").Value.Set("3")
 
 	// Unmarshal CRDs from file into structs
-	manifestsPath := filepath.Join(".", "artifacts", "inferencepool-with-model-hermetic.yaml")
+	manifestsPath := filepath.Join("..", "testdata", "inferencepool-with-model-hermetic.yaml")
 	docs, err := readDocuments(manifestsPath)
 	if err != nil {
 		log.Fatalf("Can't read object manifests at path %v, %v", manifestsPath, err)
@@ -330,7 +327,7 @@ func setUpHermeticServer(t *testing.T, pods []*backend.PodMetrics) (client extPr
 	}
 	pmc := &backend.FakePodMetricsClient{Res: pms}
 
-	server := runner.Start(backend.NewK8sDataStore(backend.WithPods(pods)), pmc)
+	server := serverRunner.Start(backend.NewK8sDataStore(backend.WithPods(pods)), pmc)
 	if err != nil {
 		log.Fatalf("Ext-proc failed with the err: %v", err)
 	}
@@ -380,25 +377,18 @@ func BeforeSuit() {
 		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(),
-	}
+	serverRunner = runserver.NewDefaultExtProcServerRunner()
+	// Adjust from defaults
+	serverRunner.PoolName = "vllm-llama2-7b-pool"
+	serverRunner.Scheme = scheme
+	serverRunner.Config = cfg
+	serverRunner.Datastore = backend.NewK8sDataStore()
 
-	runner.Setup()
+	serverRunner.Setup()
 
 	// Start the controller manager in go routine, not blocking
 	go func() {
-		runner.StartManager()
+		serverRunner.StartManager()
 	}()
 }