Szedan rh/293 factoryfunc json decoder

pkg/config/loader/configloader.go

@@ -199,7 +199,7 @@ func instantiatePlugins(configuredPlugins []configapi.PluginSpec, handle plugin.
 		if !ok {
 			return fmt.Errorf("plugin type '%s' is not registered", spec.Type)
 		}
-		plugin, err := factory(spec.Name, spec.Parameters, handle)
+		plugin, err := factory(spec.Name, plugin.StrictDecoder(spec.Parameters), handle)
 		if err != nil {
 			return fmt.Errorf("failed to create plugin '%s' (type: %s): %w", spec.Name, spec.Type, err)
 		}

pkg/config/loader/configloader_test.go

@@ -475,7 +475,7 @@ type mockProfilePicker struct{ mockPlugin }
 // compile-time type assertion
 var _ requesthandling.ProfilePicker = &mockProfilePicker{}
 
-func (m *mockProfilePicker) Pick(ctx context.Context, cycleState *plugin.CycleState, request *requesthandling.InferenceRequest,
+func (m *mockProfilePicker) Pick(ctx context.Context, request *requesthandling.InferenceRequest,
 	profiles map[string]*requesthandling.Profile) (*requesthandling.Profile, error) {
 	return nil, nil
 }
@@ -486,7 +486,7 @@ type mockRequestProcessor struct{ mockPlugin }
 // compile-time type assertion
 var _ requesthandling.RequestProcessor = &mockRequestProcessor{}
 
-func (m *mockRequestProcessor) ProcessRequest(ctx context.Context, cycleState *plugin.CycleState, request *requesthandling.InferenceRequest) error {
+func (m *mockRequestProcessor) ProcessRequest(ctx context.Context, request *requesthandling.InferenceRequest) error {
 	return nil
 }
 
@@ -496,7 +496,7 @@ type mockResponseProcessor struct{ mockPlugin }
 // compile-time type assertion
 var _ requesthandling.ResponseProcessor = &mockResponseProcessor{}
 
-func (m *mockResponseProcessor) ProcessResponse(ctx context.Context, cycleState *plugin.CycleState, request *requesthandling.InferenceResponse) error {
+func (m *mockResponseProcessor) ProcessResponse(ctx context.Context, request *requesthandling.InferenceRequest, response *requesthandling.InferenceResponse) error {
 	return nil
 }
 
@@ -506,7 +506,7 @@ type mockFilter struct{ mockPlugin }
 // compile-time type assertion
 var _ modelselector.Filter = &mockFilter{}
 
-func (m *mockFilter) Filter(_ context.Context, _ *plugin.CycleState, _ *requesthandling.InferenceRequest, models []datalayer.Model) []datalayer.Model {
+func (m *mockFilter) Filter(_ context.Context, _ *requesthandling.InferenceRequest, models []datalayer.Model) []datalayer.Model {
 	return models
 }
 
@@ -516,7 +516,7 @@ type mockScorer struct{ mockPlugin }
 // compile-time type assertion
 var _ modelselector.Scorer = &mockScorer{}
 
-func (m *mockScorer) Score(ctx context.Context, cycleState *plugin.CycleState, request *requesthandling.InferenceRequest, models []datalayer.Model) map[datalayer.Model]float64 {
+func (m *mockScorer) Score(ctx context.Context, request *requesthandling.InferenceRequest, models []datalayer.Model) map[datalayer.Model]float64 {
 	return nil
 }
 
@@ -549,7 +549,7 @@ type mockPicker struct{ mockPlugin }
 // compile-time type assertion
 var _ modelselector.Picker = &mockPicker{}
 
-func (m *mockPicker) Pick(ctx context.Context, cycleState *plugin.CycleState, scoredModels []*modelselector.ScoredModel) *modelselector.PipelineRunResult {
+func (m *mockPicker) Pick(ctx context.Context, scoredModels []*modelselector.ScoredModel) *modelselector.PipelineRunResult {
 	return nil
 }
 
@@ -558,60 +558,59 @@ func registerTestPlugins(t *testing.T) {
 
 	// Register standard test mocks.
 	plugin.Register(testPluginType,
-		func(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockPlugin{t: plugin.TypedName{Name: name, Type: testPluginType}}, nil
 		})
 
 	plugin.Register(testProfilePicker,
-		func(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockProfilePicker{mockPlugin{t: plugin.TypedName{Name: name, Type: testProfilePicker}}}, nil
 		})
 
 	plugin.Register(testRequestProcType,
-		func(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockRequestProcessor{mockPlugin{t: plugin.TypedName{Name: name, Type: testRequestProcType}}}, nil
 		})
 
 	plugin.Register(testResponseProcType,
-		func(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockResponseProcessor{mockPlugin{t: plugin.TypedName{Name: name, Type: testResponseProcType}}}, nil
 		})
 
 	plugin.Register(testPickerType,
-		func(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockPicker{mockPlugin{t: plugin.TypedName{Name: name, Type: testPickerType}}}, nil
 		})
 
-	plugin.Register(testScorerType, func(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
-		// Attempt to unmarshal to trigger errors for invalid JSON in tests.
-		if len(params) > 0 {
+	plugin.Register(testScorerType, func(name string, params *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
+		if params != nil {
 			var p struct {
 				Cost float32 `json:"cost"`
 			}
-			if err := json.Unmarshal(params, &p); err != nil {
+			if err := params.Decode(&p); err != nil {
 				return nil, err
 			}
 		}
 		return &mockScorer{mockPlugin{t: plugin.TypedName{Name: name, Type: testScorerType}}}, nil
 	})
 
 	plugin.Register(testFilterType,
-		func(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockFilter{mockPlugin{t: plugin.TypedName{Name: name, Type: testFilterType}}}, nil
 		})
 
 	plugin.Register(testExtractorType,
-		func(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockExtractor{mockPlugin{t: plugin.TypedName{Name: name, Type: testExtractorType}}}, nil
 		})
 
 	plugin.Register(testCollectorType,
-		func(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockCollector{mockPlugin{t: plugin.TypedName{Name: name, Type: testCollectorType}}}, nil
 		})
 
 	plugin.Register(testDataSourceType,
-		func(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockDataSource{mockPlugin{t: plugin.TypedName{Name: name, Type: testDataSourceType}}}, nil
 		})
 }
@@ -624,7 +623,7 @@ func registerModelSelectorPlugins(t *testing.T) {
 	plugin.Register(costaware.CostScorerType, costaware.CostScorerFactory)
 	plugin.Register(maxscore.MaxScorePickerType, maxscore.MaxScorePickerFactory)
 	plugin.Register(testFilterType,
-		func(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) {
+		func(name string, _ *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
 			return &mockFilter{mockPlugin{t: plugin.TypedName{Name: name, Type: testFilterType}}}, nil
 		})
 }

pkg/framework/interface/datalayer/datasource/types.go

@@ -46,17 +46,15 @@ const (
 
 // RequestPayload is the Payload for RequestEventType.
 type RequestPayload struct {
-	Request    *requesthandling.InferenceRequest
-	CycleState *plugin.CycleState
+	Request *requesthandling.InferenceRequest
 }
 
 // ResponsePayload is the Payload for ResponseEventType.
 type ResponsePayload struct {
-	Request    *requesthandling.InferenceRequest
-	Response   *requesthandling.InferenceResponse
-	CycleState *plugin.CycleState
-	Duration   time.Duration
-	TTFT       time.Duration
+	Request  *requesthandling.InferenceRequest
+	Response *requesthandling.InferenceResponse
+	Duration time.Duration
+	TTFT     time.Duration
 }
 
 type DatalayerProcessor interface {

pkg/framework/interface/modelselector/plugins.go

@@ -27,7 +27,7 @@ import (
 // Filter defines the interface for filtering a list of candidate models based on context.
 type Filter interface {
 	plugin.Plugin
-	Filter(ctx context.Context, cycleState *plugin.CycleState, request *requesthandling.InferenceRequest, models []datalayer.Model) []datalayer.Model
+	Filter(ctx context.Context, request *requesthandling.InferenceRequest, models []datalayer.Model) []datalayer.Model
 }
 
 // Scorer defines the interface for scoring a list of models based on context.
@@ -36,11 +36,11 @@ type Filter interface {
 // If a scorer returns value lower than 0, it will be treated as score 0.
 type Scorer interface {
 	plugin.Plugin
-	Score(ctx context.Context, cycleState *plugin.CycleState, request *requesthandling.InferenceRequest, models []datalayer.Model) map[datalayer.Model]float64
+	Score(ctx context.Context, request *requesthandling.InferenceRequest, models []datalayer.Model) map[datalayer.Model]float64
 }
 
 // Picker picks the final model(s) to send the request to.
 type Picker interface {
 	plugin.Plugin
-	Pick(ctx context.Context, cycleState *plugin.CycleState, scoredModels []*ScoredModel) *PipelineRunResult
+	Pick(ctx context.Context, scoredModels []*ScoredModel) *PipelineRunResult
 }

pkg/framework/interface/modelselector/types.go

@@ -20,7 +20,6 @@ import (
 	"context"
 
 	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/datalayer"
-	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/plugin"
 	"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/requesthandling"
 )
 
@@ -35,5 +34,5 @@ type PipelineRunResult struct {
 }
 
 type ModelSelectorPipeline interface {
-	Run(ctx context.Context, request *requesthandling.InferenceRequest, cycleState *plugin.CycleState, candidateModels []datalayer.Model) (*PipelineRunResult, error)
+	Run(ctx context.Context, request *requesthandling.InferenceRequest, candidateModels []datalayer.Model) (*PipelineRunResult, error)
 }

pkg/framework/interface/plugin/epp.go

@@ -26,3 +26,7 @@ type Plugin = eppplugin.Plugin
 type TypedName = eppplugin.TypedName
 
 type HandlePlugins = eppplugin.HandlePlugins
+
+// StrictDecoder converts raw JSON plugin parameters into a strict *json.Decoder
+// (DisallowUnknownFields), or returns nil when raw is empty.
+var StrictDecoder = eppplugin.StrictDecoder

pkg/framework/interface/plugin/registry.go

@@ -21,8 +21,11 @@ import (
 )
 
 // Factory is the definition of the factory functions that are used to instantiate plugins
-// specified in a configuration.
-type FactoryFunc func(name string, parameters json.RawMessage, handle Handle) (Plugin, error)
+// specified in a configuration. The framework provides a strict decoder
+// (DisallowUnknownFields) over the plugin's raw parameters, or nil when the plugin was
+// instantiated without parameters. Factories that ignore parameters can take the decoder
+// as `_ *json.Decoder`.
+type FactoryFunc func(name string, parameters *json.Decoder, handle Handle) (Plugin, error)
 
 // Register is a static function that can be called to register plugin factory functions.
 func Register(pluginType string, factory FactoryFunc) {

pkg/framework/interface/requesthandling/plugins.go

@@ -26,31 +26,31 @@ type ProfilePicker interface {
 	plugin.Plugin
 
 	// Pick selects the Profile to run from a list of candidate profiles, while taking into consideration the request properties.
-	Pick(ctx context.Context, cycleState *plugin.CycleState, request *InferenceRequest, profiles map[string]*Profile) (*Profile, error)
+	Pick(ctx context.Context, request *InferenceRequest, profiles map[string]*Profile) (*Profile, error)
 }
 
 type RequestProcessor interface {
 	plugin.Plugin
 	// ProcessRequest runs the RequestProcessor plugin.
 	// RequestProcessor can mutate the headers and/or the body of the request.
-	ProcessRequest(ctx context.Context, cycleState *plugin.CycleState, request *InferenceRequest) error
+	ProcessRequest(ctx context.Context, request *InferenceRequest) error
 }
 
 // ResponseProcessor processes the complete buffered response body.
 // If any plugin in a profile implements this interface, the framework buffers
 // the entire response before calling ProcessResponse on each such plugin.
 type ResponseProcessor interface {
 	plugin.Plugin
-	ProcessResponse(ctx context.Context, cycleState *plugin.CycleState, response *InferenceResponse) error
+	ProcessResponse(ctx context.Context, request *InferenceRequest, response *InferenceResponse) error
 }
 
 // ResponseHeadersProcessor processes response headers before the body arrives.
 // Plugins implementing this interface run during HandleResponseHeaders, so they
 // work for both streaming and non-streaming responses. Use this when a plugin
-// only needs CycleState and header access (not the response body).
+// only needs request attributes and header access (not the response body).
 type ResponseHeadersProcessor interface {
 	plugin.Plugin
-	ProcessResponseHeaders(ctx context.Context, cycleState *plugin.CycleState, response *InferenceResponse) error
+	ProcessResponseHeaders(ctx context.Context, request *InferenceRequest, response *InferenceResponse) error
 }
 
 // ResponseChunkProcessor processes individual response body chunks as they
@@ -59,5 +59,5 @@ type ResponseHeadersProcessor interface {
 // and mutate it via response.SetChunk().
 type ResponseChunkProcessor interface {
 	plugin.Plugin
-	ProcessResponseChunk(ctx context.Context, cycleState *plugin.CycleState, response *InferenceResponse, isFinal bool) error
+	ProcessResponseChunk(ctx context.Context, request *InferenceRequest, response *InferenceResponse, isFinal bool) error
 }