Move datasource/extractor validations to datalayer runtime.

- Extractors declare their expected input type (unchanged)
- DataSource declare their output type and expected Extractor type (new)
- Compatibility validation utility functions and tests in plugins/datalayer/source (new, moved)
- Datalayer WithConfig() validates compatbility before calling AddExtractor (new)
- DataSource can optionaly implement validating interface method to perform specific compatibility checks (new)
- Current DataSource checks on AddExtractor and their tests are not changed to minimize the code changes.

Signed-off-by: Etai Lev Ran <elevran@gmail.com>

Author: elevran

pkg/epp/datalayer/config.go

@@ -18,8 +18,10 @@ package datalayer
 
 import (
 	"fmt"
+	"reflect"
 
 	fwkdl "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/datalayer/source"
 )
 
 // Config defines the configuration of EPP data layer, as the set of DataSources
@@ -54,6 +56,24 @@ func WithConfig(cfg *Config, disallowedExtractorType string) error {
 				return fmt.Errorf("disallowed Extractor %s is configured for source %s",
 					extractor.TypedName().String(), srcCfg.Plugin.TypedName().String())
 			}
+			// Validate extractor input type is compatible with datasource output type
+			if err := source.ValidateInputTypeCompatible(srcCfg.Plugin.OutputType(), extractor.ExpectedInputType()); err != nil {
+				return fmt.Errorf("extractor %s input type incompatible with datasource %s: %w",
+					extractor.TypedName(), srcCfg.Plugin.TypedName(), err)
+			}
+			// Validate extractor type is compatible with datasource expected extractor type
+			extractorType := reflect.TypeOf(extractor)
+			if err := source.ValidateExtractorCompatible(extractorType, srcCfg.Plugin.ExtractorType()); err != nil {
+				return fmt.Errorf("extractor %s type incompatible with datasource %s: %w",
+					extractor.TypedName(), srcCfg.Plugin.TypedName(), err)
+			}
+			// Allow datasource to perform additional custom validation
+			if validator, ok := srcCfg.Plugin.(fwkdl.ValidatingDataSource); ok {
+				if err := validator.ValidateExtractor(extractor); err != nil {
+					return fmt.Errorf("extractor %s failed custom validation for datasource %s: %w",
+						extractor.TypedName(), srcCfg.Plugin.TypedName(), err)
+				}
+			}
 			if err := srcCfg.Plugin.AddExtractor(extractor); err != nil {
 				return fmt.Errorf("failed to add Extractor %s to DataSource %s: %w", extractor.TypedName(),
 					srcCfg.Plugin.TypedName(), err)

pkg/epp/datalayer/fake.go

@@ -18,6 +18,7 @@ package datalayer
 
 import (
 	"context"
+	"reflect"
 	"sync/atomic"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -46,6 +47,15 @@ func (fds *FakeDataSource) TypedName() plugin.TypedName {
 		Name: fakeSource,
 	}
 }
+
+func (fds *FakeDataSource) OutputType() reflect.Type {
+	return reflect.TypeOf(fwkdl.Metrics{})
+}
+
+func (fds *FakeDataSource) ExtractorType() reflect.Type {
+	return reflect.TypeOf((*fwkdl.Extractor)(nil)).Elem()
+}
+
 func (fds *FakeDataSource) Extractors() []string                 { return []string{} }
 func (fds *FakeDataSource) AddExtractor(_ fwkdl.Extractor) error { return nil }
 
@@ -69,6 +79,14 @@ func (m *FakeNotificationSource) TypedName() plugin.TypedName {
 	return m.typedName
 }
 
+func (m *FakeNotificationSource) OutputType() reflect.Type {
+	return reflect.TypeOf(fwkdl.NotificationEvent{})
+}
+
+func (m *FakeNotificationSource) ExtractorType() reflect.Type {
+	return reflect.TypeOf((*fwkdl.NotificationExtractor)(nil)).Elem()
+}
+
 func (m *FakeNotificationSource) GVK() schema.GroupVersionKind {
 	return m.gvk
 }

pkg/epp/framework/interface/datalayer/plugin.go

@@ -28,6 +28,13 @@ type DataSource interface {
 	plugin.Plugin
 	// Extractors returns a list of registered Extractor names.
 	Extractors() []string
+	// OutputType returns the type of data this DataSource produces.
+	// Used for validating extractor compatibility.
+	OutputType() reflect.Type
+	// ExtractorType returns the type of Extractor this DataSource expects.
+	// For poll-based sources, this is the base Extractor interface.
+	// For notification sources, this is the NotificationExtractor interface.
+	ExtractorType() reflect.Type
 	// AddExtractor adds an extractor to the data source. Multiple
 	// Extractors can be registered.
 	// The extractor will be called whenever the DataSource might
@@ -50,3 +57,11 @@ type Extractor interface {
 	// attribute, stored on the given endpoint.
 	Extract(ctx context.Context, data any, ep Endpoint) error
 }
+
+// ValidatingDataSource is an optional interface that DataSources can implement
+// to perform additional custom validation when adding extractors.
+type ValidatingDataSource interface {
+	// ValidateExtractor allows the DataSource to perform additional validation
+	// beyond the standard type compatibility checks. Return an error if validation fails.
+	ValidateExtractor(extractor Extractor) error
+}

pkg/epp/framework/plugins/datalayer/source/http/datasource.go

@@ -26,9 +26,9 @@ import (
 	"reflect"
 	"sync"
 
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datalayer"
 	fwkdl "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
 	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/datalayer/source"
 )
 
 // HTTPDataSource is a data source that receives its data using HTTP client.
@@ -77,6 +77,16 @@ func (dataSrc *HTTPDataSource) TypedName() fwkplugin.TypedName {
 	return dataSrc.typedName
 }
 
+// OutputType returns the type of data this DataSource produces.
+func (dataSrc *HTTPDataSource) OutputType() reflect.Type {
+	return dataSrc.outputType
+}
+
+// ExtractorType returns the type of Extractor this DataSource expects.
+func (dataSrc *HTTPDataSource) ExtractorType() reflect.Type {
+	return source.ExtractorType
+}
+
 // Extractors returns a list of registered Extractor names.
 func (dataSrc *HTTPDataSource) Extractors() []string {
 	extractors := []string{}
@@ -92,7 +102,7 @@ func (dataSrc *HTTPDataSource) Extractors() []string {
 // AddExtractor adds an extractor to the data source, validating it can process
 // the data source output type.
 func (dataSrc *HTTPDataSource) AddExtractor(extractor fwkdl.Extractor) error {
-	if err := datalayer.ValidateExtractorType(dataSrc.outputType, extractor.ExpectedInputType()); err != nil {
+	if err := source.ValidateInputTypeCompatible(dataSrc.OutputType(), extractor.ExpectedInputType()); err != nil {
 		return err
 	}
 	if _, loaded := dataSrc.extractors.LoadOrStore(extractor.TypedName().Name, extractor); loaded {

pkg/epp/framework/plugins/datalayer/source/notifications/k8s_datasource.go

@@ -20,13 +20,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
 	"sync"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	fwkdl "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
 	fwkplugin "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/plugin"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/plugins/datalayer/source"
 )
 
 var (
@@ -61,6 +63,16 @@ func (s *K8sNotificationSource) GVK() schema.GroupVersionKind {
 	return s.gvk
 }
 
+// OutputType returns the type of data this DataSource produces (NotificationEvent).
+func (s *K8sNotificationSource) OutputType() reflect.Type {
+	return source.NotificationEventType
+}
+
+// ExtractorType returns the type of Extractor this DataSource expects (NotificationExtractor).
+func (s *K8sNotificationSource) ExtractorType() reflect.Type {
+	return source.NotificationExtractorType
+}
+
 // Extractors returns names of registered extractors.
 func (s *K8sNotificationSource) Extractors() []string {
 	var names []string
@@ -79,10 +91,12 @@ func (s *K8sNotificationSource) AddExtractor(ext fwkdl.Extractor) error {
 	if ext == nil {
 		return errors.New("cannot add nil extractor")
 	}
-	notifyExt, ok := ext.(fwkdl.NotificationExtractor)
-	if !ok {
-		return fmt.Errorf("extractor %s does not implement NotificationExtractor", ext.TypedName())
+	extractorType := reflect.TypeOf(ext)
+	expectedType := reflect.TypeOf((*fwkdl.NotificationExtractor)(nil)).Elem()
+	if err := source.ValidateExtractorCompatible(extractorType, expectedType); err != nil {
+		return err
 	}
+	notifyExt := ext.(fwkdl.NotificationExtractor)
 	if _, loaded := s.extractors.LoadOrStore(notifyExt.TypedName().Name, notifyExt); loaded {
 		return fmt.Errorf("duplicate extractor %s on notification source %s",
 			notifyExt.TypedName(), s.TypedName())

pkg/epp/framework/plugins/datalayer/source/validate.go

@@ -0,0 +1,63 @@
+/*
+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 source
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/framework/interface/datalayer"
+)
+
+var (
+	ExtractorType             = reflect.TypeOf((*datalayer.Extractor)(nil)).Elem()
+	NotificationExtractorType = reflect.TypeOf((*datalayer.NotificationExtractor)(nil)).Elem()
+	NotificationEventType     = reflect.TypeOf(datalayer.NotificationEvent{})
+)
+
+// ValidateInputTypeCompatible checks if the extractor's expected input type is
+// compatible with the DataSource's output type.
+func ValidateInputTypeCompatible(dataSourceOutput, extractorInput reflect.Type) error {
+	if dataSourceOutput == nil || extractorInput == nil {
+		return errors.New("data source output type or extractor input type can't be nil")
+	}
+	if dataSourceOutput == extractorInput ||
+		(extractorInput.Kind() == reflect.Interface && extractorInput.NumMethod() == 0) ||
+		(extractorInput.Kind() == reflect.Interface && dataSourceOutput.Implements(extractorInput)) {
+		return nil
+	}
+	return fmt.Errorf("extractor input type %v is not compatible with data source output type %v",
+		extractorInput, dataSourceOutput)
+}
+
+// ValidateExtractorCompatible checks if the extractor type is compatible with the
+// expected extractor interface type. For notification DataSources, this validates
+// that the extractor implements NotificationExtractor.
+func ValidateExtractorCompatible(extractorType reflect.Type, expectedInterfaceType reflect.Type) error {
+	if extractorType == nil || expectedInterfaceType == nil {
+		return errors.New("extractor type or expected interface type can't be nil")
+	}
+	if expectedInterfaceType.Kind() != reflect.Interface {
+		return fmt.Errorf("expected type must be an interface, got %v", expectedInterfaceType.Kind())
+	}
+	if !extractorType.Implements(expectedInterfaceType) {
+		return fmt.Errorf("extractor type %v does not implement interface %v",
+			extractorType, expectedInterfaceType)
+	}
+	return nil
+}

pkg/epp/framework/plugins/datalayer/source/validate_test.go

@@ -0,0 +1,119 @@
+/*
+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 source
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateInputTypeCompatible(t *testing.T) {
+	type rawStruct struct{}
+	type iface interface{ Foo() }
+
+	tests := []struct {
+		name   string
+		output reflect.Type
+		input  reflect.Type
+		valid  bool
+	}{
+		{"exact match", typeOf(rawStruct{}), typeOf(rawStruct{}), true},
+		{"input is interface{}", typeOf(rawStruct{}), typeOf((*any)(nil)), true},
+		{"nil types are not allowed", typeOf(rawStruct{}), typeOf(nil), false},
+		{"output does not implement input", typeOf(rawStruct{}), typeOf((*iface)(nil)), false},
+	}
+
+	for _, tt := range tests {
+		err := ValidateInputTypeCompatible(tt.output, tt.input)
+		if tt.valid {
+			assert.NoError(t, err, "%s: expected valid extractor type", tt.name)
+		} else {
+			assert.Error(t, err, "%s: expected invalid extractor type", tt.name)
+		}
+	}
+}
+
+func TestValidateExtractorCompatible(t *testing.T) {
+	type notExtractor struct{}
+
+	tests := []struct {
+		name         string
+		extType      reflect.Type
+		expectedType reflect.Type
+		valid        bool
+		errContains  string
+	}{
+		{
+			name:         "nil extractor type",
+			extType:      nil,
+			expectedType: ExtractorType,
+			valid:        false,
+			errContains:  "can't be nil",
+		},
+		{
+			name:         "nil expected type",
+			extType:      reflect.TypeOf(&notExtractor{}),
+			expectedType: nil,
+			valid:        false,
+			errContains:  "can't be nil",
+		},
+		{
+			name:         "expected type not interface",
+			extType:      reflect.TypeOf(&notExtractor{}),
+			expectedType: reflect.TypeOf("string"),
+			valid:        false,
+			errContains:  "must be an interface",
+		},
+		{
+			name:         "does not implement interface",
+			extType:      reflect.TypeOf(&notExtractor{}),
+			expectedType: ExtractorType,
+			valid:        false,
+			errContains:  "does not implement interface",
+		},
+	}
+
+	for _, tt := range tests {
+		err := ValidateExtractorCompatible(tt.extType, tt.expectedType)
+		if tt.valid {
+			assert.NoError(t, err, "%s: expected valid", tt.name)
+		} else {
+			assert.Error(t, err, "%s: expected error", tt.name)
+			if tt.errContains != "" {
+				assert.Contains(t, err.Error(), tt.errContains, "%s: error should contain", tt.name)
+			}
+		}
+	}
+}
+
+func TestTypeConstants(t *testing.T) {
+	assert.True(t, ExtractorType.Kind() == reflect.Interface, "ExtractorType should be an interface")
+	assert.True(t, NotificationExtractorType.Kind() == reflect.Interface, "NotificationExtractorType should be an interface")
+}
+
+func typeOf(v any) reflect.Type {
+	t := reflect.TypeOf(v)
+	if t == nil {
+		return nil
+	}
+	if t.Kind() == reflect.Ptr {
+		return t.Elem()
+	}
+	return t
+}