Merge pull request #17 from nirrozenbaum/migrate/cmd-config-charts-body-based-routing-internal-pkg-bbr-pkg-common-test-integration-test-integration-util.go-test-testdata-test-utils-version

migrate: sigs.k8s.io/gateway-api-inference-extension @ dc908434

Author: nirrozenbaum

cmd/main.go

@@ -0,0 +1,32 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"os"
+
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/llm-d/llm-d-inference-payload-processor/cmd/runner"
+)
+
+func main() {
+	if err := runner.NewRunner().Run(ctrl.SetupSignalHandler()); err != nil {
+		os.Exit(1)
+	}
+
+}

cmd/runner/health.go

@@ -0,0 +1,61 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package runner
+
+import (
+	"context"
+
+	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"google.golang.org/grpc/codes"
+	healthPb "google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/status"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	logutil "github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/logging"
+)
+
+type healthServer struct{}
+
+func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckRequest) (*healthPb.HealthCheckResponse, error) {
+	// TODO: we're accepting ANY service name for now as a temporary hack in alignment with
+	// upstream issues. See https://github.com/kubernetes-sigs/gateway-api-inference-extension/pull/788
+	// if in.Service != extProcPb.ExternalProcessor_ServiceDesc.ServiceName {
+	// 	s.logger.V(logutil.DEFAULT).Info("gRPC health check requested unknown service", "available-services", []string{extProcPb.ExternalProcessor_ServiceDesc.ServiceName}, "requested-service", in.Service)
+	// 	return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVICE_UNKNOWN}, nil
+	// }
+
+	log.FromContext(ctx).V(logutil.DEBUG).Info("gRPC health check serving", "service", in.Service)
+	return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVING}, nil
+}
+
+func (s *healthServer) List(ctx context.Context, _ *healthPb.HealthListRequest) (*healthPb.HealthListResponse, error) {
+	// currently only the ext_proc service is provided
+	serviceHealthResponse, err := s.Check(ctx, &healthPb.HealthCheckRequest{Service: extProcPb.ExternalProcessor_ServiceDesc.ServiceName})
+	if err != nil {
+		return nil, err
+	}
+
+	return &healthPb.HealthListResponse{
+		Statuses: map[string]*healthPb.HealthCheckResponse{
+			extProcPb.ExternalProcessor_ServiceDesc.ServiceName: serviceHealthResponse,
+		},
+	}, nil
+}
+
+func (s *healthServer) Watch(in *healthPb.HealthCheckRequest, srv healthPb.Health_WatchServer) error {
+	return status.Error(codes.Unimplemented, "Watch is not implemented")
+}

cmd/runner/runner.go

@@ -0,0 +1,266 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package runner
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/spf13/pflag"
+	"google.golang.org/grpc"
+	healthPb "google.golang.org/grpc/health/grpc_health_v1"
+	"k8s.io/client-go/rest"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	"github.com/llm-d/llm-d-inference-payload-processor/internal/runnable"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/bbr/framework"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/bbr/metrics"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/bbr/plugins/basemodelextractor"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/bbr/plugins/bodyfieldtoheader"
+	runserver "github.com/llm-d/llm-d-inference-payload-processor/pkg/bbr/server"
+	logutil "github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/logging"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/profiling"
+	"github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/tracing"
+	"github.com/llm-d/llm-d-inference-payload-processor/version"
+)
+
+const modelField = "model"
+
+var setupLog = ctrl.Log.WithName("setup")
+
+func NewRunner() *Runner {
+	return &Runner{
+		bbrExecutableName: "BBR",
+		requestPlugins:    []framework.RequestProcessor{},
+		responsePlugins:   []framework.ResponseProcessor{},
+		customCollectors:  []prometheus.Collector{},
+	}
+}
+
+// Runner is used to run bbr with its plugins
+type Runner struct {
+	bbrExecutableName string
+	// The slice of BBR plugin instances executed by the request handler,
+	// in the same order the plugin flags are provided.
+	requestPlugins []framework.RequestProcessor
+	// The slice of BBR plugin instances executed by the response handler,
+	// in the same order the plugin flags are provided.
+	responsePlugins []framework.ResponseProcessor
+
+	customCollectors []prometheus.Collector
+}
+
+// WithExecutableName sets the name of the executable containing the runner.
+// The name is used in the version log upon startup and is otherwise opaque.
+func (r *Runner) WithExecutableName(exeName string) *Runner {
+	r.bbrExecutableName = exeName
+	return r
+}
+
+func (r *Runner) WithCustomCollectors(collectors ...prometheus.Collector) *Runner {
+	r.customCollectors = collectors
+	return r
+}
+
+func (r *Runner) Run(ctx context.Context) error {
+	// Setup a basic logger in case command-line argument parsing fails.
+	logutil.InitSetupLogging()
+
+	setupLog.Info(r.bbrExecutableName+" build", "commit-sha", version.CommitSHA, "build-ref", version.BuildRef)
+
+	opts := runserver.NewOptions()
+	opts.AddFlags(pflag.CommandLine)
+	pflag.Parse()
+
+	if err := opts.Complete(); err != nil {
+		return err
+	}
+	if err := opts.Validate(); err != nil {
+		setupLog.Error(err, "Failed to validate flags")
+		return err
+	}
+
+	// Print all flag values.
+	flags := make(map[string]any)
+	pflag.VisitAll(func(f *pflag.Flag) {
+		flags[f.Name] = f.Value
+	})
+
+	if opts.Tracing {
+		err := tracing.InitTracing(ctx, setupLog, "gateway-api-inference-extension/bbr")
+		if err != nil {
+			setupLog.Error(err, "failed to initialize tracing")
+			return err
+		}
+	}
+
+	setupLog.Info("Flags processed", "flags", flags)
+
+	logutil.InitLogging(&opts.ZapOptions)
+
+	// Init runtime.
+	cfg, err := ctrl.GetConfig()
+	if err != nil {
+		setupLog.Error(err, "Failed to get rest config")
+		return err
+	}
+
+	// --- Setup Metrics Server ---
+	metrics.Register(r.customCollectors...)
+	metrics.RecordBBRInfo(version.CommitSHA, version.BuildRef)
+	// Register metrics handler.
+	// Metrics endpoint is enabled in 'config/default/kustomization.yaml'. The Metrics options configure the server.
+	// More info:
+	// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.19.1/pkg/metrics/server
+	// - https://book.kubebuilder.io/reference/metrics.html
+	metricsServerOptions := metricsserver.Options{
+		BindAddress: fmt.Sprintf(":%d", opts.MetricsPort),
+		FilterProvider: func() func(c *rest.Config, httpClient *http.Client) (metricsserver.Filter, error) {
+			if opts.MetricsEndpointAuth {
+				return filters.WithAuthenticationAndAuthorization
+			}
+
+			return nil
+		}(),
+	}
+	cacheOptions := cache.Options{}
+	// Apply namespace filtering only if env var is set.
+	namespace := os.Getenv("NAMESPACE")
+	if namespace != "" {
+		cacheOptions.DefaultNamespaces = map[string]cache.Config{
+			namespace: {},
+		}
+	}
+
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{Cache: cacheOptions, Metrics: metricsServerOptions})
+	if err != nil {
+		setupLog.Error(err, "Failed to create manager", "config", cfg)
+		return err
+	}
+
+	if opts.EnablePprof {
+		setupLog.Info("Setting pprof handlers")
+		if err = profiling.SetupPprofHandlers(mgr); err != nil {
+			setupLog.Error(err, "Failed to setup pprof handlers")
+			return err
+		}
+	}
+
+	bbrHandle := framework.NewBbrHandle(ctx, mgr)
+
+	// Register factories for all known in-tree BBR plugins
+	r.registerInTreePlugins()
+
+	// Construct BBR plugin instances for the in-tree plugins that are (1) registered and (2) requested via the --plugin flags
+	if len(opts.PluginSpecs) == 0 {
+		setupLog.Info("No BBR plugins are specified. Running BBR with the default behavior.")
+
+		modelToHeaderPlugin, err := bodyfieldtoheader.NewBodyFieldToHeaderPlugin(modelField, bodyfieldtoheader.ModelHeader)
+		if err != nil {
+			setupLog.Error(err, "Failed to create plugin", "pluginType", bodyfieldtoheader.BodyFieldToHeaderPluginType)
+			return err
+		}
+		r.requestPlugins = append(r.requestPlugins, modelToHeaderPlugin)
+
+		// Create BaseModelToHeaderPlugin instance for extracting the "model" field into X-Gateway-Base-Model-Name
+		baseModelToHeaderPlugin, err := basemodelextractor.NewBaseModelToHeaderPlugin(func() *builder.Builder {
+			return ctrl.NewControllerManagedBy(mgr)
+		}, mgr.GetClient())
+		if err != nil {
+			setupLog.Error(err, "Failed to create plugin", "pluginType", basemodelextractor.BaseModelToHeaderPluginType)
+			return err
+		}
+
+		r.requestPlugins = append(r.requestPlugins, baseModelToHeaderPlugin)
+	} else {
+		setupLog.Info("BBR plugins are specified. Running BBR with the specified plugins.")
+
+		for _, s := range opts.PluginSpecs {
+			factory, ok := framework.Registry[s.Type]
+			if !ok {
+				setupLog.Error(err, fmt.Sprintf("unknown plugin type %q (no factory registered)\n", s.Type))
+				return err
+			}
+			instance, err := factory(s.Name, s.JSON, bbrHandle)
+			if err != nil {
+				setupLog.Error(err, fmt.Sprintf("invalid %s#%s: %v\n", s.Type, s.Name, err))
+				return err
+			}
+			if requestProcessor, ok := instance.(framework.RequestProcessor); ok {
+				r.requestPlugins = append(r.requestPlugins, requestProcessor)
+			}
+			if responseProcessor, ok := instance.(framework.ResponseProcessor); ok {
+				r.responsePlugins = append(r.responsePlugins, responseProcessor)
+			}
+		}
+	}
+
+	// Setup ExtProc Server Runner.
+	serverRunner := &runserver.ExtProcServerRunner{
+		GrpcPort:        opts.GRPCPort,
+		SecureServing:   opts.SecureServing,
+		Streaming:       opts.Streaming,
+		RequestPlugins:  r.requestPlugins,
+		ResponsePlugins: r.responsePlugins,
+	}
+
+	// Register health server.
+	if err := registerHealthServer(mgr, opts.GRPCHealthPort); err != nil {
+		return err
+	}
+
+	// Register ext-proc server.
+	if err := mgr.Add(serverRunner.AsRunnable(ctrl.Log.WithName("ext-proc"))); err != nil {
+		setupLog.Error(err, "Failed to register ext-proc gRPC server")
+		return err
+	}
+
+	// Start the manager. This blocks until a signal is received.
+	setupLog.Info("Manager starting")
+	if err := mgr.Start(ctx); err != nil {
+		setupLog.Error(err, "Error starting manager")
+		return err
+	}
+	setupLog.Info("Manager terminated")
+	return nil
+}
+
+// registerInTreePlugins registers the factory functions of all known BBR plugins
+func (r *Runner) registerInTreePlugins() {
+	framework.Register(bodyfieldtoheader.BodyFieldToHeaderPluginType, bodyfieldtoheader.BodyFieldToHeaderPluginFactory)
+	framework.Register(basemodelextractor.BaseModelToHeaderPluginType, basemodelextractor.BaseModelToHeaderPluginFactory)
+}
+
+// registerHealthServer adds the Health gRPC server as a Runnable to the given manager.
+func registerHealthServer(mgr manager.Manager, port int) error {
+	srv := grpc.NewServer()
+	healthPb.RegisterHealthServer(srv, &healthServer{})
+	if err := mgr.Add(
+		runnable.NoLeaderElection(runnable.GRPCServer("health", srv, port))); err != nil {
+		setupLog.Error(err, "Failed to register health server")
+		return err
+	}
+	return nil
+}

config/charts/body-based-routing/.helmignore

@@ -0,0 +1,23 @@
+# Patterns to ignore when building packages.
+# This supports shell glob matching, relative path matching, and
+# negation (prefixed with !). Only one pattern per line.
+.DS_Store
+# Common VCS dirs
+.git/
+.gitignore
+.bzr/
+.bzrignore
+.hg/
+.hgignore
+.svn/
+# Common backup files
+*.swp
+*.bak
+*.tmp
+*.orig
+*~
+# Various IDEs
+.project
+.idea/
+*.tmproj
+.vscode/

config/charts/body-based-routing/Chart.yaml

@@ -0,0 +1,9 @@
+apiVersion: v2
+name: body-based-routing
+description: A Helm chart for the body-based routing extension
+
+type: application
+
+version: 0.1.0
+
+appVersion: "0.2.0"