A development step PR for pluggable BBR framework. Does not change behavior.

Author: davidbreitgand

cmd/bbr/runner/runner.go

@@ -41,12 +41,15 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/metrics"
+	bbr "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/plugins"
+	routing "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/routing"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/server"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/common/util/logging"
 	"sigs.k8s.io/gateway-api-inference-extension/version"
 )
 
 var (
+	// Flags
 	grpcPort            = flag.Int("grpc-port", 9004, "The gRPC port used for communicating with Envoy proxy")
 	grpcHealthPort      = flag.Int("grpc-health-port", 9005, "The port used for gRPC liveness and readiness probes")
 	metricsPort         = flag.Int("metrics-port", 9090, "The metrics port")
@@ -55,7 +58,12 @@ var (
 	secureServing       = flag.Bool("secure-serving", true, "Enables secure serving.")
 	logVerbosity        = flag.Int("v", logging.DEFAULT, "number for the log level verbosity")
 
+	// Logging
 	setupLog = ctrl.Log.WithName("setup")
+
+	// Contains the BBR plugins specs specified via repeated flags:
+	//   --plugin <type>:<name>[:<json>]
+	pluginSpecs bbr.BBRPluginSpecs
 )
 
 func NewRunner() *Runner {
@@ -67,6 +75,10 @@ func NewRunner() *Runner {
 // 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.
+	bbrPluginInstances []bbr.BBRPlugin
 }
 
 // WithExecutableName sets the name of the executable containing the runner.
@@ -80,6 +92,8 @@ func (r *Runner) Run(ctx context.Context) error {
 	setupLog.Info(r.bbrExecutableName+" build", "commit-sha", version.CommitSHA, "build-ref", version.BuildRef)
 	opts := zap.Options{Development: true}
 	opts.BindFlags(flag.CommandLine)
+
+	flag.Var(&pluginSpecs, "plugin", `Repeatable. --plugin <type>:<name>[:<json>]`)
 	flag.Parse()
 	initLogging(&opts)
 
@@ -140,12 +154,46 @@ func (r *Runner) Run(ctx context.Context) error {
 		return err
 	}
 
+	// 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(pluginSpecs) == 0 {
+		setupLog.Info("No BBR plugins are specified. Running BBR with the default behavior.")
+
+		// Append a default BBRPlugin to the slice of the BBRPlugin instances using regular registered factory mechanism.
+		factory := bbr.Registry[routing.DefaultPluginType]
+		defaultPlugin, err := factory("", nil)
+		if err != nil {
+			setupLog.Error(err, "Failed to create default plugin")
+			return err
+		}
+		r.withPlugin(defaultPlugin)
+	} else {
+		setupLog.Info("BBR plugins are specified. Running BBR with the specified plugins.")
+
+		for _, s := range pluginSpecs {
+			factory, ok := bbr.Registry[s.Type]
+			if !ok {
+				fmt.Fprintf(os.Stderr, "unknown plugin type %q (no factory registered)\n", s.Type)
+				os.Exit(2)
+			}
+			instance, err := factory(s.Name, s.JSON)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "invalid %s#%s: %v\n", s.Type, s.Name, err)
+				os.Exit(2)
+			}
+			r.withPlugin(instance)
+		}
+	}
+
 	// Setup ExtProc Server Runner
 	serverRunner := &runserver.ExtProcServerRunner{
-		GrpcPort:      *grpcPort,
-		Datastore:     ds,
-		SecureServing: *secureServing,
-		Streaming:     *streaming,
+		GrpcPort:        *grpcPort,
+		Datastore:       ds,
+		SecureServing:   *secureServing,
+		Streaming:       *streaming,
+		PluginInstances: r.bbrPluginInstances,
 	}
 	if err := serverRunner.SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "Failed to setup BBR controllers")
@@ -173,6 +221,15 @@ func (r *Runner) Run(ctx context.Context) error {
 	return nil
 }
 
+// registerInTreePlugins registers the factory functions of all known BBR plugins
+func (r *Runner) registerInTreePlugins() {
+	bbr.Register(routing.DefaultPluginType, routing.DefaultPluginFactory)
+}
+
+func (r *Runner) withPlugin(p bbr.BBRPlugin) {
+	r.bbrPluginInstances = append(r.bbrPluginInstances, p)
+}
+
 // registerHealthServer adds the Health gRPC server as a Runnable to the given manager.
 func registerHealthServer(mgr manager.Manager, port int) error {
 	srv := grpc.NewServer()

config/charts/body-based-routing/templates/bbr.yaml

@@ -13,7 +13,7 @@ spec:
       labels:
         app: {{ .Values.bbr.name }}
     spec:
-      serviceAccountName: {{ .Values.bbr.name}}-{{ .Release.Name }}
+      serviceAccountName: {{ .Values.bbr.name }}-{{ .Release.Name }}
       containers:
       - name: bbr
         image: {{ .Values.bbr.image.hub }}/{{ .Values.bbr.image.name }}:{{ .Values.bbr.image.tag }}
@@ -24,6 +24,14 @@ spec:
         {{- range $key, $value := .Values.bbr.flags }}
         - --{{ $key }}
         - "{{ $value }}"
+        {{- end }}      
+        {{- range .Values.bbr.plugins }}
+        - --plugin
+        {{- if .json }}
+        - {{ printf "%s:%s:%s" .type .name (toJson .json) | quote }}
+        {{- else }}
+        - {{ printf "%s:%s" .type .name | quote }}
+        {{- end }}
         {{- end }}
         {{- if not .Values.bbr.multiNamespace }}
         env:

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

@@ -14,6 +14,12 @@ bbr:
     # Log verbosity
     v: 3
 
+  # Plugin configuration (transitional - currently runs no-op default plugin)
+  plugins:
+    - type: default-bbr
+      name: default-bbr
+      json: {"v":0, "no-op":true}
+
 provider:
   name: none
 

pkg/bbr/handlers/request.go

@@ -19,6 +19,7 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 
 	basepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	eppb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
@@ -43,12 +44,31 @@ func (s *Server) HandleRequestBody(ctx context.Context, requestBodyBytes []byte)
 	logger := log.FromContext(ctx)
 	var ret []*eppb.ProcessingResponse
 
+	// Executing BBR plugins in the order they were registered.
+	// This change is a transitional development step to test the BBR plugins.
+	// At the moment, the loop runs a no-op plugin.
+	// Once the Default BBR plugin is fully implemented and integrated, the loop will run actual pluggable logic.
+	for _, plugin := range s.pluginInstances {
+		var headers map[string][]string
+		var err error
+		logger.Info("Executing plugin", "plugin", plugin.TypedName())
+
+		requestBodyBytes, headers, err = plugin.Execute(requestBodyBytes)
+		if err != nil {
+			logger.Error(err, "Plugin execution failed", "plugin", plugin.TypedName())
+			return nil, fmt.Errorf("plugin %s failed: %w", plugin.TypedName(), err)
+		}
+		_ = headers // TODO: Handle headers returned by plugins
+	}
+
 	var requestBody RequestBody
 	if err := json.Unmarshal(requestBodyBytes, &requestBody); err != nil {
 		metrics.RecordModelNotParsedCounter()
 		return nil, err
 	}
 
+	logger.Info("Parsed model name", "model", requestBody.Model)
+
 	if requestBody.Model == "" {
 		metrics.RecordModelNotInBodyCounter()
 		logger.V(logutil.DEFAULT).Info("Request body does not contain model parameter")
@@ -73,6 +93,8 @@ func (s *Server) HandleRequestBody(ctx context.Context, requestBodyBytes []byte)
 	metrics.RecordSuccessCounter()
 	baseModel := s.ds.GetBaseModel(requestBody.Model)
 
+	logger.Info("Base model from datastore", "baseModel", baseModel)
+
 	if s.streaming {
 		ret = append(ret, &eppb.ProcessingResponse{
 			Response: &eppb.ProcessingResponse_RequestHeaders{

pkg/bbr/handlers/request_test.go

@@ -267,7 +267,7 @@ func TestHandleRequestBody(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			server := NewServer(test.streaming, &fakeDatastore{})
+			server := NewServer(test.streaming, &fakeDatastore{}, fakeBBRPlugins{})
 			bodyBytes, _ := json.Marshal(test.body)
 			resp, err := server.HandleRequestBody(ctx, bodyBytes)
 			if err != nil {

pkg/bbr/handlers/server.go

@@ -29,24 +29,28 @@ import (
 
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/common/util/logging"
 	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
+
+	bbr "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/plugins"
 )
 
 type Datastore interface {
 	GetBaseModel(modelName string) string
 }
 
-func NewServer(streaming bool, ds Datastore) *Server {
+func NewServer(streaming bool, ds Datastore, bbrPluginInstances []bbr.BBRPlugin) *Server {
 	return &Server{
-		streaming: streaming,
-		ds:        ds,
+		streaming:       streaming,
+		ds:              ds,
+		pluginInstances: bbrPluginInstances,
 	}
 }
 
 // Server implements the Envoy external processing server.
 // https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/ext_proc/v3/external_processor.proto
 type Server struct {
-	streaming bool
-	ds        Datastore
+	streaming       bool
+	ds              Datastore
+	pluginInstances []bbr.BBRPlugin
 }
 
 func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {

pkg/bbr/handlers/server_test.go

@@ -27,6 +27,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/common/util/logging"
+
+	bbr "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/plugins"
+	plugins "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
 )
 
 func TestProcessRequestBody(t *testing.T) {
@@ -139,7 +142,8 @@ func TestProcessRequestBody(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.desc, func(t *testing.T) {
-			srv := NewServer(tc.streaming, &fakeDatastore{})
+			fakePlugins := fakeBBRPlugins{}
+			srv := NewServer(tc.streaming, &fakeDatastore{}, fakePlugins)
 			streamedBody := &streamedBody{}
 			for i, body := range tc.bodys {
 				got, err := srv.processRequestBody(context.Background(), body, streamedBody, log.FromContext(ctx))
@@ -162,3 +166,18 @@ type fakeDatastore struct{}
 func (ds *fakeDatastore) GetBaseModel(modelName string) string {
 	return ""
 }
+
+type fakeBBRPlugin struct{}
+
+type fakeBBRPlugins []bbr.BBRPlugin
+
+func (p *fakeBBRPlugin) Execute(requestBodyBytes []byte) (mutatedBodyBytes []byte, headers map[string][]string, err error) {
+	return []byte{}, map[string][]string{}, nil
+}
+
+func (p *fakeBBRPlugin) TypedName() plugins.TypedName {
+	return plugins.TypedName{
+		Type: "fake",
+		Name: "test-plugin",
+	}
+}

pkg/bbr/plugins/plugins.go

@@ -23,9 +23,10 @@ import (
 type BBRPlugin interface {
 	plugin.Plugin
 
-	// Execute runs the plugin logic on the request body and a map of headers.
-	// A plugin's imnplementation logic CAN mutate the body of the message.
+	// Execute runs the plugin's logic on the request body.
+	// A plugin's implementation logic CAN mutate the body of the message.
 	// A plugin's implementation MUST return a map of headers.
 	// If no headers are set by the implementation, the return headers map is nil.
-	Execute(requestBodyBytes []byte, requestHeaders map[string][]string) (mutatedBodyBytes []byte, headers map[string][]string, err error)
+	// In the future, a headers map can be added to the plugin interface to allow different plugins on a chain to share information on the headers.
+	Execute(requestBodyBytes []byte) (mutatedBodyBytes []byte, headers map[string][]string, err error)
 }

pkg/bbr/routing/default.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2026 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 routing
+
+import (
+	"encoding/json"
+
+	bbr "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/plugins"
+	epp "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
+)
+
+const (
+	DefaultPluginType = "default-bbr"
+)
+
+// compile-time type validation
+var _ bbr.BBRPlugin = &DefaultPlugin{}
+
+type DefaultPlugin struct {
+	typedName epp.TypedName
+}
+
+// DefaultPluginFactory defines the factory function for DefaultPlugin.
+// The name and rawParameters are ignored as the plugin uses the default configuration.
+func DefaultPluginFactory(_ string, _ json.RawMessage) (bbr.BBRPlugin, error) {
+	return NewDefaultPlugin(), nil
+}
+
+// / NewDefaultPlugin returns a concrete *DefaultPlugin.
+func NewDefaultPlugin() *DefaultPlugin {
+	return &DefaultPlugin{
+		typedName: epp.TypedName{Type: DefaultPluginType, Name: DefaultPluginType},
+	}
+}
+
+func (p *DefaultPlugin) Execute(requestBodyBytes []byte) ([]byte, map[string][]string, error) {
+	// No-op BBR plugin to be replaced by the actual logic currently running in the
+	return requestBodyBytes, nil, nil
+}
+
+// TypedName returns the type and name tuple of this plugin instance.
+func (p *DefaultPlugin) TypedName() epp.TypedName {
+	return p.typedName
+}
+
+// WithName sets the name of the default BBR plugin
+func (p *DefaultPlugin) WithName(name string) *DefaultPlugin {
+	p.typedName.Name = name
+	return p
+}

pkg/bbr/server/runserver.go

@@ -33,14 +33,16 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/controller"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/handlers"
+	bbr "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/plugins"
 )
 
 // ExtProcServerRunner provides methods to manage an external process server.
 type ExtProcServerRunner struct {
-	GrpcPort      int
-	Datastore     datastore.Datastore
-	SecureServing bool
-	Streaming     bool
+	GrpcPort        int
+	Datastore       datastore.Datastore
+	SecureServing   bool
+	Streaming       bool
+	PluginInstances []bbr.BBRPlugin
 }
 
 func NewDefaultExtProcServerRunner(port int, streaming bool) *ExtProcServerRunner {
@@ -81,7 +83,7 @@ func (r *ExtProcServerRunner) AsRunnable(logger logr.Logger) manager.Runnable {
 			srv = grpc.NewServer()
 		}
 
-		extProcPb.RegisterExternalProcessorServer(srv, handlers.NewServer(r.Streaming, r.Datastore))
+		extProcPb.RegisterExternalProcessorServer(srv, handlers.NewServer(r.Streaming, r.Datastore, r.PluginInstances))
 
 		// Forward to the gRPC runnable.
 		return runnable.GRPCServer("ext-proc", srv, r.GrpcPort).Start(ctx)