From d082db2a730d52cb7c8612183a6dbc8cddb727bf Mon Sep 17 00:00:00 2001 From: aviavissar Date: Tue, 21 Jul 2026 12:08:46 +0300 Subject: [PATCH] feat: add byfieldattribute filter Signed-off-by: aviavissar --- cmd/runner/runner.go | 3 + .../interface/datalayer/attributemap.go | 51 ++-- .../filter/byfieldattribute/README.md | 74 +++++ .../filter/byfieldattribute/filter.go | 167 +++++++++++ .../filter/byfieldattribute/filter_test.go | 270 ++++++++++++++++++ 5 files changed, 531 insertions(+), 34 deletions(-) create mode 100644 pkg/framework/plugins/modelselector/filter/byfieldattribute/README.md create mode 100644 pkg/framework/plugins/modelselector/filter/byfieldattribute/filter.go create mode 100644 pkg/framework/plugins/modelselector/filter/byfieldattribute/filter_test.go diff --git a/cmd/runner/runner.go b/cmd/runner/runner.go index 782627f6..54a15226 100644 --- a/cmd/runner/runner.go +++ b/cmd/runner/runner.go @@ -50,6 +50,7 @@ import ( modelconfigcollector "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/datalayer/modelconfigcollector" requestcostmetadata "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/datalayer/requestcostmetadata" requestmetadata "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/datalayer/requestmetadata" + "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/filter/byfieldattribute" "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/filter/modelgroup" "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/picker/maxscore" "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/picker/random" @@ -299,6 +300,8 @@ func (r *Runner) registerInTreePlugins() { plugin.Register(modelconfigcollector.PluginType, modelconfigcollector.DatasourceFactory) // register model selector plugins plugin.Register(modelgroup.ModelGroupFilterType, modelgroup.ModelGroupFilterFactory) + plugin.Register(byfieldattribute.ModelNameFilterType, byfieldattribute.ModelNameFilterFactory) + plugin.Register(byfieldattribute.ByFieldFilterType, byfieldattribute.ByFieldFilterFactory) plugin.Register(random.RandomPickerType, random.RandomPickerFactory) plugin.Register(maxscore.MaxScorePickerType, maxscore.MaxScorePickerFactory) plugin.Register(weightedrandom.WeightedRandomPickerType, weightedrandom.WeightedRandomPickerFactory) diff --git a/pkg/framework/interface/datalayer/attributemap.go b/pkg/framework/interface/datalayer/attributemap.go index 9e1a6428..020bf045 100644 --- a/pkg/framework/interface/datalayer/attributemap.go +++ b/pkg/framework/interface/datalayer/attributemap.go @@ -21,47 +21,41 @@ import ( "sync" ) -// Cloneable types support cloning of the value. -// All values stored in AttributeMap must implement this interface -// to ensure data isolation and prevent unintended mutations. +// Cloneable types support cloning, required for all values stored in an +// AttributeMap to prevent unintended mutations. type Cloneable interface { Clone() Cloneable } -// AttributeMap is used to store flexible metadata or traits -// across different aspects of a model. +// StringAttribute is a Cloneable wrapper around a plain string. +type StringAttribute string + +func (s StringAttribute) Clone() Cloneable { + return s +} + +// AttributeMap stores flexible, goroutine-safe metadata about a model. // Stored values must be Cloneable. -// -// All operations are goroutine-safe. type AttributeMap interface { - // Put stores or updates an attribute. - // Empty keys and nil values are ignored (no-op). + // Put stores or updates an attribute. Empty keys and nil values are no-ops. Put(key string, value Cloneable) - // Get retrieves a cloned copy of the attribute value. - // Returns (value, true) if found, (nil, false) if not found. - // The returned value is a clone to prevent unintended mutations. + // Get returns a cloned copy of the attribute value, if found. Get(key string) (Cloneable, bool) - // Delete removes an attribute by key. - // No-op if key doesn't exist. Delete(key string) - // Keys returns all attribute keys as a string slice. - // Order is not guaranteed. + // Keys returns all attribute keys, in no particular order. Keys() []string - // Clone creates a deep copy of the entire attribute map. Clone() AttributeMap } -// Attributes provides a goroutine-safe implementation of AttributeMap. -// Uses sync.Map for concurrent access without explicit locking. +// Attributes is a goroutine-safe AttributeMap backed by sync.Map. type Attributes struct { - data sync.Map // key: attribute name (string), value: attribute value (Cloneable) + data sync.Map // key: attribute name, value: Cloneable } -// NewAttributes creates a new AttributeMap instance. func NewAttributes() AttributeMap { return &Attributes{} } @@ -116,16 +110,8 @@ func (a *Attributes) Clone() AttributeMap { return clone } -// ReadAttributeKey reads an attribute by key and asserts the value to type T. -// Returns an error if the key is not found or the type assertion fails. -// -// This is a convenience function for type-safe attribute retrieval. -// The returned value is a clone (as per AttributeMap.Get behavior). -// -// Type Parameter T: -// - Can be a value type (e.g., MyType) or pointer type (e.g., *MyType) -// - Must match the concrete type returned by the stored value's Clone() method -// - For types implementing Cloneable, use the same type as Clone() returns +// ReadAttributeKey reads an attribute by key and asserts it to type T, which +// must match the concrete type returned by the stored value's Clone(). func ReadAttributeKey[T any](attrs AttributeMap, key string) (T, error) { var zero T @@ -134,9 +120,6 @@ func ReadAttributeKey[T any](attrs AttributeMap, key string) (T, error) { return zero, fmt.Errorf("attribute %q: not found", key) } - // Attempt direct type assertion to T - // This works for both value types and pointer types because - // Get() returns the result of Clone(), which returns the concrete type val, ok := raw.(T) if !ok { return zero, fmt.Errorf("unexpected type for key %q: got %T, want %T", key, raw, zero) diff --git a/pkg/framework/plugins/modelselector/filter/byfieldattribute/README.md b/pkg/framework/plugins/modelselector/filter/byfieldattribute/README.md new file mode 100644 index 00000000..3d11e3e4 --- /dev/null +++ b/pkg/framework/plugins/modelselector/filter/byfieldattribute/README.md @@ -0,0 +1,74 @@ +# By-Field Filter (and Model Name Filter) + +Restricts the candidate models based on a configurable field in the request body, compared +against either the candidate model's name or one of its attributes. + +The filter matches the requested value against the candidate models the pipeline hands it from +the datalayer. "Available" below means "present in that candidate list". + +## `by-field-filter` + +The generic filter. Registered as type `by-field-filter` and runs as a modelselector filter. + +## What it does + +1. Reads the configured `fieldName` field from the request body. +2. Compares that value against each candidate's comparison value: + - By default, the comparison value is the candidate's name (`Model.GetName()`). + - When `byField` is set, the value compares with the `byField` value attribute for + the model, which is defined in the candidate's `AttributeMap`. +3. Returns every candidate whose comparison value matches. Candidates that don't carry the + `byField` attribute (when configured) are skipped, not treated as an error. +4. If the field is absent or an empty string, all incoming candidates are kept. +5. If no candidate matches, or the field is malformed (not a string), the filter returns no + candidates and the pipeline rejects the request with HTTP 429. + +## Configuration + +- `fieldName` (required) — the request-body field to read the comparison value from. +- `byField` (optional) — the name of the candidate attribute to compare against. + When empty or omitted, defaults to the candidate's name (`Model.GetName()`). + +## Inputs consumed + +- The `fieldName` field of the request body. +- The candidate model list passed in by the pipeline, including each candidate's attributes (when + `byField` is configured). + +## Example configuration: filter by model name + +Equivalent to the built-in `model-name-filter` (see below): + +```yaml +plugins: +- type: by-field-filter + params: + fieldName: model +``` + +## Example configuration: filter by a custom field/attribute + +For example, filtering Anthropic-on-Bedrock requests by their `anthropic_version` field, matched +against a candidate attribute named `anthropicVersion` (populated by another plugin, e.g. the +`model-config-datasource` plugin, onto each candidate's `AttributeMap`): + +```yaml +plugins: +- type: by-field-filter + params: + fieldName: anthropic_version + byField: anthropicVersion +``` + +## `model-name-filter` + +Kept as a separate, backward-compatible plugin type with the original, fixed behavior: it always +reads the `model` field and compares by the candidate's name, ignoring any parameters passed to it. + +It is registered as type `model-name-filter` and behaves exactly like `by-field-filter` configured +with `fieldName: model` and no `byField` override. + +```yaml +plugins: +- type: model-name-filter +``` diff --git a/pkg/framework/plugins/modelselector/filter/byfieldattribute/filter.go b/pkg/framework/plugins/modelselector/filter/byfieldattribute/filter.go new file mode 100644 index 00000000..995d025a --- /dev/null +++ b/pkg/framework/plugins/modelselector/filter/byfieldattribute/filter.go @@ -0,0 +1,167 @@ +/* +Copyright 2026 The llm-d 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 byfieldattribute implements a modelselector filter that matches +// candidates by name or attribute. See the package README for details. +package byfieldattribute + +import ( + "context" + "encoding/json" + "fmt" + + "sigs.k8s.io/controller-runtime/pkg/log" + + logutil "github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/logging" + "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/modelselector" + "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" +) + +type StringAttribute = datalayer.StringAttribute + +const ( + // ModelNameFilterType is the registered plugin type for model-name-filter. + ModelNameFilterType = "model-name-filter" + // ByFieldFilterType is the registered plugin type for by-field-filter. + ByFieldFilterType = "by-field-filter" + + requestModelField = "model" + defaultByFieldValue = "" +) + +var _ modelselector.Filter = &ByFieldFilter{} +var _ datalayer.Cloneable = StringAttribute("") + +// ByFieldFilterConfig is the JSON parameter shape for ByFieldFilter. +type ByFieldFilterConfig struct { + FieldName string `json:"fieldName"` + ByField string `json:"byField,omitempty"` +} + +// ModelNameFilterFactory ignores its parameters: fieldName is always "model" +// and byField always defaults to Model.GetName(). +func ModelNameFilterFactory(name string, _ json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) { + return newByFieldFilter(ModelNameFilterType, requestModelField, defaultByFieldValue).WithName(name), nil +} + +// ByFieldFilterFactory parses fieldName (required) and byField (optional) +// from params and returns a configured ByFieldFilter. +func ByFieldFilterFactory(name string, params json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) { + var cfg ByFieldFilterConfig + if len(params) > 0 { + if err := json.Unmarshal(params, &cfg); err != nil { + return nil, fmt.Errorf("failed to parse parameters of '%s': %w", ByFieldFilterType, err) + } + } + if cfg.FieldName == "" { + return nil, fmt.Errorf("'%s' requires a non-empty fieldName parameter", ByFieldFilterType) + } + + return newByFieldFilter(ByFieldFilterType, cfg.FieldName, cfg.ByField).WithName(name), nil +} + +// NewModelNameFilter initializes a ByFieldFilter with model-name-filter's +// fixed settings (fieldName "model", default byField). +func NewModelNameFilter() *ByFieldFilter { + return newByFieldFilter(ModelNameFilterType, requestModelField, defaultByFieldValue) +} + +// NewByFieldFilter initializes a by-field-filter with the given fieldName +// and byField (empty byField means Model.GetName()). +func NewByFieldFilter(fieldName, byField string) *ByFieldFilter { + return newByFieldFilter(ByFieldFilterType, fieldName, byField) +} + +func newByFieldFilter(typ, fieldName, byField string) *ByFieldFilter { + if typ == ModelNameFilterType { + log.Log.Info("'" + ModelNameFilterType + "' is deprecated and will be removed in ipp v0.3.0; " + + "use '" + ByFieldFilterType + "' with fieldName: model instead") + } + return &ByFieldFilter{ + typedName: plugin.TypedName{Type: typ, Name: typ}, + fieldName: fieldName, + byField: byField, + } +} + +// ByFieldFilter restricts candidates to those whose comparison value (the +// name, or a configured attribute) matches a configurable request body field. +type ByFieldFilter struct { + typedName plugin.TypedName + fieldName string + byField string +} + +func (f *ByFieldFilter) TypedName() plugin.TypedName { + return f.typedName +} + +func (f *ByFieldFilter) WithName(name string) *ByFieldFilter { + f.typedName.Name = name + return f +} + +// Filter keeps candidates whose comparison value matches the request field. +// An absent or empty field keeps all candidates; a non-string field value is +// malformed and yields none (the pipeline rejects the request). +func (f *ByFieldFilter) Filter(ctx context.Context, _ *plugin.CycleState, request *requesthandling.InferenceRequest, models []datalayer.Model) []datalayer.Model { + logger := log.FromContext(ctx) + + raw := request.Body[f.fieldName] + requested, ok := raw.(string) + if !ok && raw != nil { + logger.V(logutil.VERBOSE).Info("malformed request field, no available model candidates", "field", f.fieldName) + return []datalayer.Model{} + } + if requested == "" { + logger.V(logutil.VERBOSE).Info("no value in request field. All available models are considered as candidates", "field", f.fieldName) + return models + } + + result := make([]datalayer.Model, 0, len(models)) + for _, model := range models { + value, found := f.candidateValue(model) + if !found { + continue + } + if value == requested { + result = append(result, model) + } + } + + if len(result) == 0 { + logger.V(logutil.VERBOSE).Info("request field value is not configured", "requested", requested) + return []datalayer.Model{} + } + + logger.V(logutil.DEBUG).Info("by-field filter applied", "requested", requested, "candidates", len(result)) + return result +} + +// candidateValue returns the model's name by default, or its byField +// attribute when configured; false if that attribute is missing. +func (f *ByFieldFilter) candidateValue(model datalayer.Model) (string, bool) { + if f.byField == defaultByFieldValue { + return model.GetName(), true + } + value, err := datalayer.ReadAttributeKey[StringAttribute](model.GetAttributes(), f.byField) + if err != nil { + return "", false + } + return string(value), true +} diff --git a/pkg/framework/plugins/modelselector/filter/byfieldattribute/filter_test.go b/pkg/framework/plugins/modelselector/filter/byfieldattribute/filter_test.go new file mode 100644 index 00000000..21241d8f --- /dev/null +++ b/pkg/framework/plugins/modelselector/filter/byfieldattribute/filter_test.go @@ -0,0 +1,270 @@ +/* +Copyright 2026 The llm-d 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 byfieldattribute + +import ( + "context" + "encoding/json" + "sort" + "testing" + + "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/requesthandling" +) + +// candidateModels builds datalayer models; attrs optionally maps a model name +// to a byField attribute name/value pair to set on it. +func candidateModels(attrs map[string][2]string, names ...string) []datalayer.Model { + models := make([]datalayer.Model, len(names)) + for idx, name := range names { + m := datalayer.NewModel(name) + if kv, ok := attrs[name]; ok { + m.GetAttributes().Put(kv[0], StringAttribute(kv[1])) + } + models[idx] = m + } + return models +} + +func modelNames(models []datalayer.Model) []string { + out := make([]string, len(models)) + for idx, model := range models { + out[idx] = model.GetName() + } + sort.Strings(out) + return out +} + +// requestWithField builds a request whose body holds value under field; +// nil leaves the field absent. +func requestWithField(field string, value any) *requesthandling.InferenceRequest { + r := requesthandling.NewInferenceRequest() + if value != nil { + r.Body[field] = value + } + return r +} + +func TestModelNameFilterFactory(t *testing.T) { + t.Run("ignores parameters", func(t *testing.T) { + p, err := ModelNameFilterFactory("my-filter", json.RawMessage(`{"fieldName":"anthropic_version","byField":"id"}`), nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + f := p.(*ByFieldFilter) + if got := f.TypedName().Name; got != "my-filter" { + t.Errorf("Name = %s, want my-filter", got) + } + if got := f.TypedName().Type; got != ModelNameFilterType { + t.Errorf("Type = %s, want %s", got, ModelNameFilterType) + } + if f.fieldName != requestModelField { + t.Errorf("fieldName = %s, want %s", f.fieldName, requestModelField) + } + if f.byField != defaultByFieldValue { + t.Errorf("byField = %q, want default", f.byField) + } + }) + + t.Run("nil parameters", func(t *testing.T) { + p, err := ModelNameFilterFactory("my-filter", nil, nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := p.(*ByFieldFilter).TypedName().Type; got != ModelNameFilterType { + t.Errorf("Type = %s, want %s", got, ModelNameFilterType) + } + }) +} + +func TestByFieldFilterFactory(t *testing.T) { + tests := []struct { + name string + params json.RawMessage + wantErr bool + wantField string + wantByAttr string + }{ + { + name: "fieldName only", + params: json.RawMessage(`{"fieldName":"anthropic_version"}`), + wantField: "anthropic_version", + }, + { + name: "fieldName with byField override", + params: json.RawMessage(`{"fieldName":"anthropic_version","byField":"apiVersion"}`), + wantField: "anthropic_version", + wantByAttr: "apiVersion", + }, + { + name: "fieldName with empty byField", + params: json.RawMessage(`{"fieldName":"model","byField":""}`), + wantField: "model", + wantByAttr: "", + }, + { + name: "missing fieldName", + params: json.RawMessage(`{}`), + wantErr: true, + }, + { + name: "nil parameters", + params: nil, + wantErr: true, + }, + { + name: "malformed JSON", + params: json.RawMessage(`{"fieldName":`), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p, err := ByFieldFilterFactory("my-filter", tt.params, nil) + if tt.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + f := p.(*ByFieldFilter) + if got := f.TypedName().Type; got != ByFieldFilterType { + t.Errorf("Type = %s, want %s", got, ByFieldFilterType) + } + if f.fieldName != tt.wantField { + t.Errorf("fieldName = %s, want %s", f.fieldName, tt.wantField) + } + if f.byField != tt.wantByAttr { + t.Errorf("byField = %q, want %q", f.byField, tt.wantByAttr) + } + }) + } +} + +func TestModelNameFilter_Filter(t *testing.T) { + all := []string{"qwen3-8b", "qwen3-32b", "llama3-8b"} + + tests := []struct { + name string + modelBody any + want []string + }{ + {name: "missing model field passes all through", modelBody: nil, want: all}, + {name: "empty string passes all through", modelBody: "", want: all}, + {name: "matching model name returns single candidate", modelBody: "qwen3-8b", want: []string{"qwen3-8b"}}, + {name: "unregistered model name yields empty", modelBody: "gpt-4", want: []string{}}, + {name: "non-string model field yields empty (malformed)", modelBody: 42, want: []string{}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := NewModelNameFilter() + req := requestWithField(requestModelField, tt.modelBody) + + got := modelNames(f.Filter(context.Background(), nil, req, candidateModels(nil, all...))) + want := append([]string{}, tt.want...) + sort.Strings(want) + + if len(got) != len(want) { + t.Fatalf("Filter() = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("Filter() = %v, want %v", got, want) + break + } + } + }) + } +} + +func TestByFieldFilter_Filter_ByName(t *testing.T) { + f := NewByFieldFilter("model", "") + candidates := candidateModels(nil, "qwen3-8b", "qwen3-32b") + + req := requestWithField("model", "qwen3-32b") + got := modelNames(f.Filter(context.Background(), nil, req, candidates)) + want := []string{"qwen3-32b"} + + if len(got) != len(want) || got[0] != want[0] { + t.Errorf("Filter() = %v, want %v", got, want) + } +} + +func TestByFieldFilter_Filter_ByAttribute(t *testing.T) { + attrs := map[string][2]string{ + "claude-bedrock-a": {"anthropicVersion", "bedrock-2023-05-31"}, + "claude-bedrock-b": {"anthropicVersion", "bedrock-2023-05-31"}, + "claude-legacy": {"anthropicVersion", "bedrock-2022-11-01"}, + // "gpt-oss" has no anthropicVersion attribute. + } + candidates := candidateModels(attrs, "claude-bedrock-a", "claude-bedrock-b", "claude-legacy", "gpt-oss") + + f := NewByFieldFilter("anthropic_version", "anthropicVersion") + + t.Run("matches all candidates sharing the requested version", func(t *testing.T) { + req := requestWithField("anthropic_version", "bedrock-2023-05-31") + got := modelNames(f.Filter(context.Background(), nil, req, candidates)) + want := []string{"claude-bedrock-a", "claude-bedrock-b"} + sort.Strings(want) + if len(got) != len(want) { + t.Fatalf("Filter() = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("Filter() = %v, want %v", got, want) + } + } + }) + + t.Run("candidate missing the attribute is skipped, not erroring", func(t *testing.T) { + req := requestWithField("anthropic_version", "bedrock-2022-11-01") + got := modelNames(f.Filter(context.Background(), nil, req, candidates)) + want := []string{"claude-legacy"} + if len(got) != len(want) || got[0] != want[0] { + t.Errorf("Filter() = %v, want %v", got, want) + } + }) + + t.Run("no requested field keeps all candidates including those missing the attribute", func(t *testing.T) { + req := requestWithField("anthropic_version", nil) + got := modelNames(f.Filter(context.Background(), nil, req, candidates)) + if len(got) != len(candidates) { + t.Errorf("Filter() = %v, want all %d candidates", got, len(candidates)) + } + }) + + t.Run("unmatched version yields empty", func(t *testing.T) { + req := requestWithField("anthropic_version", "bedrock-1999-01-01") + got := modelNames(f.Filter(context.Background(), nil, req, candidates)) + if len(got) != 0 { + t.Errorf("Filter() = %v, want empty", got) + } + }) + + t.Run("non-string request field yields empty (malformed)", func(t *testing.T) { + req := requestWithField("anthropic_version", 42) + got := modelNames(f.Filter(context.Background(), nil, req, candidates)) + if len(got) != 0 { + t.Errorf("Filter() = %v, want empty", got) + } + }) +}