feat(scheduling): add endpoint attribute weight scorer

Refs #2201

Signed-off-by: nicole-lihui <nicole.li@daocloud.io>

Author: nicole-lihui

cmd/epp/runner/runner.go

@@ -68,6 +68,7 @@ import (
 	attrtopology "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/topology"
 	discoveryfile "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/discovery/file"
 	extdcgm "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/dcgm"
+	labelproducer "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/label"
 	extractormetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/metrics"
 	extmodels "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/models"
 	exttopology "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/topology"
@@ -118,6 +119,7 @@ import (
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/profilehandler/disagg"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/profilehandler/single"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/activerequest"
+	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/attributeweight"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/contextlengthaware"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/endpointattribute"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/scorer/kvcacheutilization"
@@ -565,8 +567,10 @@ func (r *Runner) registerInTreePlugins() {
 	fwkplugin.Register(loadaware.LoadAwareType, loadaware.Factory)
 	fwkplugin.Register(sessionaffinity.SessionAffinityType, sessionaffinity.Factory)
 	fwkplugin.Register(contextlengthaware.ContextLengthAwareType, contextlengthaware.Factory)
+	fwkplugin.Register(attributeweight.EndpointAttributeWeightScorerType, attributeweight.Factory)
 
 	// data layer models source/extractor
+	fwkplugin.Register(labelproducer.LabelProducerType, labelproducer.Factory)
 	fwkplugin.Register(srcmodels.ModelsDataSourceType, srcmodels.ModelDataSourceFactory)
 	fwkplugin.Register(attrmodels.ModelsExtractorType, extmodels.ModelServerExtractorFactory)
 	fwkplugin.Register(attrtopology.TopologyExtractorType, exttopology.Factory)

pkg/epp/framework/plugins/datalayer/attribute/string/data_types.go

@@ -0,0 +1,33 @@
+/*
+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 stringattribute declares string-valued endpoint attributes.
+package stringattribute
+
+import fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
+
+// Value is a string-valued endpoint attribute.
+type Value string
+
+// Clone returns the value unchanged.
+func (v Value) Clone() fwkdl.Cloneable {
+	return v
+}
+
+// ReadValue retrieves a string value from an endpoint attribute map.
+func ReadValue(attrs fwkdl.AttributeMap, key string) (Value, bool) {
+	return fwkdl.ReadAttribute[Value](attrs, key)
+}

pkg/epp/framework/plugins/datalayer/extractor/label/producer.go

@@ -0,0 +1,148 @@
+/*
+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 label provides an endpoint extractor that copies a Pod label into
+// an endpoint attribute.
+package label
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+
+	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
+	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
+	attrmetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/metrics"
+	attrstring "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/string"
+	sourcenotifications "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/notifications"
+)
+
+const (
+	// LabelProducerType identifies the label producer plugin.
+	LabelProducerType = "label-producer"
+
+	valueTypeString = "string"
+	valueTypeNumber = "number"
+)
+
+var (
+	_ fwkplugin.Plugin        = (*Producer)(nil)
+	_ fwkdl.Registrant        = (*Producer)(nil)
+	_ fwkdl.EndpointExtractor = (*Producer)(nil)
+)
+
+type parameters struct {
+	// Label is the Pod label to copy.
+	Label string `json:"label"`
+	// AttributeKey is the endpoint attribute key to populate.
+	AttributeKey string `json:"attributeKey"`
+	// ValueType controls whether the value is stored as a string or number.
+	ValueType string `json:"valueType"`
+}
+
+// Factory creates a label producer.
+func Factory(name string, decoder *json.Decoder, _ fwkplugin.Handle) (fwkplugin.Plugin, error) {
+	var params parameters
+	if decoder != nil {
+		if err := decoder.Decode(&params); err != nil {
+			return nil, fmt.Errorf("failed to parse the parameters of the '%s' plugin: %w", LabelProducerType, err)
+		}
+	}
+	return NewProducer(name, params)
+}
+
+// NewProducer validates parameters and creates a label producer.
+func NewProducer(name string, params parameters) (*Producer, error) {
+	if name == "" {
+		name = LabelProducerType
+	}
+	if params.Label == "" {
+		return nil, fmt.Errorf("'%s' requires a non-empty 'label'", LabelProducerType)
+	}
+	if params.AttributeKey == "" {
+		return nil, fmt.Errorf("'%s' requires a non-empty 'attributeKey'", LabelProducerType)
+	}
+	if params.ValueType == "" {
+		params.ValueType = valueTypeString
+	}
+	if params.ValueType != valueTypeString && params.ValueType != valueTypeNumber {
+		return nil, fmt.Errorf("'%s' valueType must be %q or %q, got %q", LabelProducerType, valueTypeString, valueTypeNumber, params.ValueType)
+	}
+
+	return &Producer{
+		typedName:    fwkplugin.TypedName{Type: LabelProducerType, Name: name},
+		label:        params.Label,
+		attributeKey: params.AttributeKey,
+		valueType:    params.ValueType,
+	}, nil
+}
+
+// Producer copies a Pod label into an endpoint attribute.
+type Producer struct {
+	typedName    fwkplugin.TypedName
+	label        string
+	attributeKey string
+	valueType    string
+}
+
+// TypedName returns the type and name of the producer.
+func (p *Producer) TypedName() fwkplugin.TypedName {
+	return p.typedName
+}
+
+// RegisterDependencies wires the producer to endpoint lifecycle events.
+func (p *Producer) RegisterDependencies(r fwkdl.Registrar) error {
+	return r.Register(fwkdl.PendingRegistration{
+		Owner:      p.typedName,
+		SourceType: sourcenotifications.EndpointNotificationSourceType,
+		Extractor:  p,
+		DefaultSource: sourcenotifications.NewEndpointDataSource(
+			sourcenotifications.EndpointNotificationSourceType,
+			sourcenotifications.EndpointNotificationSourceType,
+		),
+	})
+}
+
+// Extract copies the configured label into the endpoint attribute.
+func (p *Producer) Extract(_ context.Context, event fwkdl.EndpointEvent) error {
+	if event.Type == fwkdl.EventDelete || event.Endpoint == nil {
+		return nil
+	}
+
+	meta := event.Endpoint.GetMetadata()
+	if meta == nil {
+		return nil
+	}
+	value := meta.Labels[p.label]
+
+	if p.valueType == valueTypeString {
+		event.Endpoint.GetAttributes().Put(p.attributeKey, attrstring.Value(value))
+		return nil
+	}
+
+	if value == "" {
+		event.Endpoint.GetAttributes().Put(p.attributeKey, attrmetrics.ScalarMetricValue(0))
+		return nil
+	}
+	number, err := strconv.ParseFloat(value, 64)
+	if err != nil {
+		event.Endpoint.GetAttributes().Put(p.attributeKey, attrmetrics.ScalarMetricValue(0))
+		return fmt.Errorf("label %q on endpoint %s has non-numeric value %q: %w", p.label, meta.NamespacedName, value, err)
+	}
+	event.Endpoint.GetAttributes().Put(p.attributeKey, attrmetrics.ScalarMetricValue(number))
+	return nil
+}

pkg/epp/framework/plugins/datalayer/extractor/label/producer_test.go

@@ -0,0 +1,149 @@
+/*
+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 label
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer"
+	fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
+	attrmetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/metrics"
+	attrstring "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/string"
+)
+
+const testLabel = "nvidia.com/gpu.product"
+
+type captureRegistrar struct {
+	registration fwkdl.PendingRegistration
+}
+
+func (r *captureRegistrar) Register(reg fwkdl.PendingRegistration) error {
+	r.registration = reg
+	return nil
+}
+
+func newEndpoint(labels map[string]string) fwkdl.Endpoint {
+	return fwkdl.NewEndpoint(&fwkdl.EndpointMetadata{Labels: labels}, nil)
+}
+
+func TestFactory(t *testing.T) {
+	producer, err := Factory("gpu-product", fwkplugin.StrictDecoder(json.RawMessage(`{
+		"label": "nvidia.com/gpu.product",
+		"attributeKey": "gpu.product",
+		"valueType": "string"
+	}`)), nil)
+
+	require.NoError(t, err)
+	assert.Equal(t, LabelProducerType, producer.TypedName().Type)
+	assert.Equal(t, "gpu-product", producer.TypedName().Name)
+}
+
+func TestFactory_Validation(t *testing.T) {
+	tests := []struct {
+		name       string
+		parameters string
+		wantErr    string
+	}{
+		{name: "missing label", parameters: `{"attributeKey":"gpu.product"}`, wantErr: "label"},
+		{name: "missing attribute key", parameters: `{"label":"gpu"}`, wantErr: "attributeKey"},
+		{name: "invalid value type", parameters: `{"label":"gpu","attributeKey":"gpu.product","valueType":"bool"}`, wantErr: "valueType"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			producer, err := Factory("test", fwkplugin.StrictDecoder(json.RawMessage(test.parameters)), nil)
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), test.wantErr)
+			assert.Nil(t, producer)
+		})
+	}
+}
+
+func TestExtract_StringValue(t *testing.T) {
+	producer, err := NewProducer("gpu-product", parameters{
+		Label:        testLabel,
+		AttributeKey: "gpu.product",
+		ValueType:    valueTypeString,
+	})
+	require.NoError(t, err)
+
+	endpoint := newEndpoint(map[string]string{testLabel: "H100"})
+	require.NoError(t, producer.Extract(context.Background(), fwkdl.EndpointEvent{
+		Type:     fwkdl.EventAddOrUpdate,
+		Endpoint: endpoint,
+	}))
+
+	value, ok := attrstring.ReadValue(endpoint.GetAttributes(), "gpu.product")
+	require.True(t, ok)
+	assert.Equal(t, attrstring.Value("H100"), value)
+}
+
+func TestExtract_NumberValue(t *testing.T) {
+	producer, err := NewProducer("gpu-score", parameters{
+		Label:        "gpu.score",
+		AttributeKey: "gpu.score",
+		ValueType:    valueTypeNumber,
+	})
+	require.NoError(t, err)
+
+	endpoint := newEndpoint(map[string]string{"gpu.score": "2.5"})
+	require.NoError(t, producer.Extract(context.Background(), fwkdl.EndpointEvent{
+		Type:     fwkdl.EventAddOrUpdate,
+		Endpoint: endpoint,
+	}))
+
+	value, ok := attrmetrics.ReadScalarMetricValue(endpoint.GetAttributes(), "gpu.score")
+	require.True(t, ok)
+	assert.Equal(t, attrmetrics.ScalarMetricValue(2.5), value)
+}
+
+func TestExtract_MissingLabelWritesFallbackValue(t *testing.T) {
+	producer, err := NewProducer("gpu-product", parameters{
+		Label:        testLabel,
+		AttributeKey: "gpu.product",
+		ValueType:    valueTypeString,
+	})
+	require.NoError(t, err)
+
+	endpoint := newEndpoint(nil)
+	require.NoError(t, producer.Extract(context.Background(), fwkdl.EndpointEvent{
+		Type:     fwkdl.EventAddOrUpdate,
+		Endpoint: endpoint,
+	}))
+
+	value, ok := attrstring.ReadValue(endpoint.GetAttributes(), "gpu.product")
+	require.True(t, ok)
+	assert.Empty(t, value)
+}
+
+func TestRegisterDependencies(t *testing.T) {
+	producer, err := NewProducer("gpu-product", parameters{
+		Label:        testLabel,
+		AttributeKey: "gpu.product",
+	})
+	require.NoError(t, err)
+
+	var registrar captureRegistrar
+	require.NoError(t, producer.RegisterDependencies(&registrar))
+	assert.Equal(t, "endpoint-notification-source", registrar.registration.SourceType)
+	assert.Same(t, producer, registrar.registration.Extractor)
+}