feat(datalayer): move interval to plugin parameters

noalimoy · noalimoy · commit 97fcf9e8797c · 2026-07-29T14:41:35.000+03:00
Signed-off-by: noalimoy &lt;nlimoy@redhat.com&gt;
diff --git a/apix/config/v1alpha1/endpointpickerconfig_types.go b/apix/config/v1alpha1/endpointpickerconfig_types.go
@@ -278,13 +278,6 @@ type DataLayerSource struct {
 	// this Source. The entries are references to the names of entries of the Plugins
 	// defined in the configuration's Plugins section
 	Extractors []DataLayerExtractor `json:"extractors"`
-
-	// +optional
-	// Interval is the scrape period for this polling source. It must be a
-	// positive multiple of --refresh-metrics-interval (default 50ms). When
-	// omitted, the source runs on every base tick. Ignored for
-	// notification/endpoint sources.
-	Interval *metav1.Duration `json:"interval,omitempty"`
 }
 
 func (dls DataLayerSource) String() string {
@@ -293,9 +286,6 @@ func (dls DataLayerSource) String() string {
 	if len(dls.Extractors) > 0 {
 		parts = append(parts, fmt.Sprintf("Extractors: %v", dls.Extractors))
 	}
-	if dls.Interval != nil {
-		parts = append(parts, "Interval: "+dls.Interval.Duration.String())
-	}
 	return "{" + strings.Join(parts, ", ") + "}"
 }
 
diff --git a/apix/config/v1alpha1/zz_generated.deepcopy.go b/apix/config/v1alpha1/zz_generated.deepcopy.go
diff --git a/deploy/config/sim-epp-gpu-config.yaml b/deploy/config/sim-epp-gpu-config.yaml
@@ -7,6 +7,7 @@ plugins:
   name: dcgm-source
   parameters:
     port: 9400
+    interval: "1s"
 - type: dcgm-extractor
   name: dcgm-extractor
 - type: endpoint-attribute-filter
@@ -37,7 +38,6 @@ plugins:
 dataLayer:
   sources:
   - pluginRef: dcgm-source
-    interval: 1s
     extractors:
     - pluginRef: dcgm-extractor
 schedulingProfiles:
diff --git a/docs/architecture.md b/docs/architecture.md
@@ -225,10 +225,10 @@ The data layer follows a Source -> Extract -> Attribute lifecycle:
 - Scoring can rely on numerical metrics or metadata (model ID, adapter tags)
 
 Polling sources share one Collector goroutine per endpoint. The base tick is
-`--refresh-metrics-interval` (default 50ms). A polling entry under `dataLayer.sources`
-may set `interval` to a positive multiple of that base tick; when omitted, the source
-runs on every base tick. Example: `interval: 1s` with the default base tick scrapes
-that source once per second.
+`--refresh-metrics-interval` (default 50ms). Each polling source plugin accepts an
+`interval` parameter (e.g. `"1s"`) that must be a positive multiple of the base tick;
+when omitted, the source runs on every base tick. The runtime converts each source's
+interval to base-tick multiples and schedules dispatches accordingly.
 
 See the upstream [Data Layer](https://github.com/llm-d/llm-d/blob/main/docs/architecture/core/router/epp/datalayer.md) doc for the canonical model.
 
diff --git a/pkg/epp/config/loader/configloader.go b/pkg/epp/config/loader/configloader.go
@@ -438,9 +438,6 @@ func buildDataLayerConfig(rawDataConfig *configapi.DataLayerConfig, handle fwkpl
 				Plugin:     sourcePlugin,
 				Extractors: []fwkplugin.Plugin{},
 			}
-			if source.Interval != nil {
-				sourceConfig.Interval = source.Interval.Duration
-			}
 			for _, extractor := range source.Extractors {
 				extractorPlugin := handle.Plugin(extractor.PluginRef)
 				if extractorPlugin == nil {
diff --git a/pkg/epp/datalayer/collector_test.go b/pkg/epp/datalayer/collector_test.go
@@ -80,6 +80,7 @@ func (e *errSource) Dispatch(_ context.Context, _ fwkdl.Endpoint) error {
 	return e.err
 }
 func (e *errSource) AppendExtractor(_ fwkplugin.Plugin) error { return nil }
+func (e *errSource) Interval() time.Duration                  { return 0 }
 
 type dataSource struct {
 	kind      string
@@ -115,6 +116,7 @@ func (d *dataSource) AppendExtractor(p fwkplugin.Plugin) error {
 	d.mu.Unlock()
 	return nil
 }
+func (d *dataSource) Interval() time.Duration { return 0 }
 
 type stubExtractor struct {
 	kind string
diff --git a/pkg/epp/datalayer/config.go b/pkg/epp/datalayer/config.go
@@ -18,7 +18,6 @@ package datalayer
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
 )
@@ -50,9 +49,6 @@ func (c *Config) String() string {
 type DataSourceConfig struct {
 	Plugin     plugin.Plugin   // the source plugin instance (DataSource or PollingDispatcher)
 	Extractors []plugin.Plugin // extractors defined for the data source
-	// Interval is the scrape period for polling sources. Zero means every
-	// Runtime base tick. Ignored for notification/endpoint sources.
-	Interval time.Duration
 }
 
 func (dsc DataSourceConfig) String() string {
diff --git a/pkg/epp/datalayer/runtime.go b/pkg/epp/datalayer/runtime.go
@@ -121,9 +121,9 @@ func (r *Runtime) Configure(cfg *Config, logger logr.Logger) error {
 			}
 
 			periodTicks := 1
-			if _, ok := src.(fwkdl.PollingDispatcher); ok {
+			if disp, ok := src.(fwkdl.PollingDispatcher); ok {
 				var err error
-				periodTicks, err = PeriodTicks(srcCfg.Interval, r.pollingInterval)
+				periodTicks, err = PeriodTicks(disp.Interval(), r.pollingInterval)
 				if err != nil {
 					return fmt.Errorf("source %s: %w", srcName, err)
 				}
diff --git a/pkg/epp/datalayer/runtime_validate_test.go b/pkg/epp/datalayer/runtime_validate_test.go
@@ -123,6 +123,7 @@ func (m *multiVariantSource) TypedName() fwkplugin.TypedName
 func (m *multiVariantSource) GVK() schema.GroupVersionKind                       { return m.gvk }
 func (m *multiVariantSource) Dispatch(_ context.Context, _ fwkdl.Endpoint) error { return nil }
 func (m *multiVariantSource) AppendExtractor(_ fwkplugin.Plugin) error           { return nil }
+func (m *multiVariantSource) Interval() time.Duration                            { return 0 }
 func (m *multiVariantSource) Notify(_ context.Context, event fwkdl.NotificationEvent) (*fwkdl.NotificationEvent, error) {
 	// Body never runs in this test. Configure rejects the source for
 	// implementing multiple variants before any Notify call. Echo the
@@ -148,14 +149,21 @@ func TestRuntimeConfigure_SourceImplementingMultipleVariants_Rejected(t *testing
 	assert.Contains(t, err.Error(), "multiple variant")
 }
 
+// intervalMockSource wraps MetricsDataSource with a configurable Interval.
+type intervalMockSource struct {
+	mocks.MetricsDataSource
+	interval time.Duration
+}
+
+func (s *intervalMockSource) Interval() time.Duration { return s.interval }
+
 func TestRuntimeConfigure_InvalidSourceInterval(t *testing.T) {
 	logger := newTestLogger(t)
 	r := NewRuntime(50 * time.Millisecond)
 
 	cfg := &Config{
 		Sources: []DataSourceConfig{{
-			Plugin:   &mocks.MetricsDataSource{},
-			Interval: 75 * time.Millisecond,
+			Plugin: &intervalMockSource{interval: 75 * time.Millisecond},
 		}},
 	}
 
@@ -170,8 +178,7 @@ func TestRuntimeConfigure_SourceIntervalStoredAsPeriodTicks(t *testing.T) {
 
 	cfg := &Config{
 		Sources: []DataSourceConfig{{
-			Plugin:   &mocks.MetricsDataSource{},
-			Interval: time.Second,
+			Plugin: &intervalMockSource{interval: time.Second},
 		}},
 	}
 
diff --git a/pkg/epp/framework/interface/datalayer/plugin.go b/pkg/epp/framework/interface/datalayer/plugin.go
@@ -18,6 +18,7 @@ package datalayer
 
 import (
 	"context"
+	"time"
 
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -74,10 +75,12 @@ type NotificationExtractor interface {
 //     DataLayerExtractErrorsTotal and do NOT surface as the return error.
 //   - AppendExtractor is a pure append; duplicate-Type detection is the caller's
 //     responsibility (see runtime.Configure).
+//   - Interval returns the desired scrape period. Zero means every base tick.
 type PollingDispatcher interface {
 	plugin.Plugin
 	Dispatch(ctx context.Context, ep Endpoint) error
 	AppendExtractor(ext plugin.Plugin) error
+	Interval() time.Duration
 }
 
 // EventType identifies the type of mutation that triggered a notification.
diff --git a/pkg/epp/framework/plugins/datalayer/source/dcgm/README.md b/pkg/epp/framework/plugins/datalayer/source/dcgm/README.md
@@ -30,6 +30,9 @@ server port via `HTTPDataSource` options.
 - `insecureSkipVerify` (bool, optional, default: `true`): Skip TLS certificate verification.
 - `useNodeAddress` (bool, optional, default: `false`): When true, scrape
   `NodeAddress:port` (DaemonSet). When false, scrape `PodIP:port` (sidecar).
+- `interval` (string, optional): Scrape period (e.g. `"1s"`). Must be a positive
+  multiple of `--refresh-metrics-interval` (default 50ms). Omit to scrape every
+  base tick.
 
 DaemonSet deployments require `DCGM_EXPORTER_KUBERNETES=true` on the exporter
 so metrics include a `pod` label (GPU Operator sets this by default).
@@ -60,15 +63,13 @@ plugins:
   parameters:
     port: 9400
     useNodeAddress: true
+    interval: "1s"
 - type: dcgm-extractor
   name: dcgm-extractor
 # ... other plugins (filters, scorers, profile handler, picker) ...
 dataLayer:
   sources:
   - pluginRef: dcgm-source
-    # Scrape period for this source. Must be a positive multiple of
-    # --refresh-metrics-interval (default 50ms). Omit to scrape every base tick.
-    interval: 1s
     extractors:
     - pluginRef: dcgm-extractor
 ```
diff --git a/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource.go b/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource.go
@@ -45,6 +45,9 @@ type dcgmDatasourceParams struct {
 	InsecureSkipVerify bool   `json:"insecureSkipVerify"`
 	// UseNodeAddress scrapes the node IP (DaemonSet) instead of the pod IP (sidecar).
 	UseNodeAddress bool `json:"useNodeAddress"`
+	// Interval is the scrape period (e.g. "1s"). Must be a positive multiple of
+	// --refresh-metrics-interval. Empty or omitted means every base tick.
+	Interval string `json:"interval"`
 }
 
 // NewHTTPDCGMDataSource constructs a DCGMDataSource with the given parameters.
@@ -71,6 +74,11 @@ func DCGMDataSourceFactory(name string, parameters *json.Decoder, _ plugin.Handl
 	if cfg.UseNodeAddress {
 		opts = append(opts, http.WithUseNodeAddress())
 	}
+	if intervalOpt, err := http.ParseIntervalOption(cfg.Interval); err != nil {
+		return nil, err
+	} else if intervalOpt != nil {
+		opts = append(opts, intervalOpt)
+	}
 
 	ds, err := http.NewHTTPDataSource(cfg.Scheme, cfg.Path, http.TLSOptions{SkipVerify: cfg.InsecureSkipVerify},
 		DCGMDataSourceType, name, parsePrometheus, opts...)
diff --git a/pkg/epp/framework/plugins/datalayer/source/http/datasource.go b/pkg/epp/framework/plugins/datalayer/source/http/datasource.go
@@ -62,6 +62,8 @@ type HTTPDataSource[T any] struct {
 	// useNodeAddress, when true, scrapes NodeAddress:portOverride instead
 	// of the pod IP. Used for node-level exporters (e.g. DCGM DaemonSet).
 	useNodeAddress bool
+	// interval is the desired scrape period; zero means every base tick.
+	interval time.Duration
 
 	client Client
 	// parser converts the response body to T. MUST NOT return (zero, nil) for nilable T;
@@ -92,6 +94,7 @@ type Option func(*options)
 type options struct {
 	portOverride   int
 	useNodeAddress bool
+	interval       time.Duration
 }
 
 // WithPortOverride makes the source scrape podIP:port instead of the
@@ -108,6 +111,27 @@ func WithUseNodeAddress() Option {
 	return func(o *options) { o.useNodeAddress = true }
 }
 
+// WithInterval sets the desired scrape period for this source. The datalayer
+// runtime converts it to base-tick multiples at configure time. Zero means
+// every base tick (the default).
+func WithInterval(d time.Duration) Option {
+	return func(o *options) { o.interval = d }
+}
+
+// ParseIntervalOption parses a duration string from plugin parameters and
+// returns a WithInterval Option. Returns (nil, nil) when raw is empty so
+// callers can simply append the non-nil result to their option slice.
+func ParseIntervalOption(raw string) (Option, error) {
+	if raw == "" {
+		return nil, nil
+	}
+	d, err := time.ParseDuration(raw)
+	if err != nil {
+		return nil, fmt.Errorf("invalid interval %q: %w", raw, err)
+	}
+	return WithInterval(d), nil
+}
+
 // NewHTTPDataSource constructs a typed polling dispatcher. For https, tlsOpts configures
 // server verification (CACertPath) and optional mTLS (ClientCertPath/ClientKeyPath).
 func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions,
@@ -124,6 +148,9 @@ func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions,
 	if cfg.useNodeAddress && cfg.portOverride == 0 {
 		return nil, errors.New("WithUseNodeAddress requires a non-zero WithPortOverride")
 	}
+	if cfg.interval < 0 {
+		return nil, fmt.Errorf("interval must be non-negative, got %s", cfg.interval)
+	}
 
 	cl := &client{
 		Client: http.Client{
@@ -146,6 +173,7 @@ func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions,
 		path:           path,
 		portOverride:   cfg.portOverride,
 		useNodeAddress: cfg.useNodeAddress,
+		interval:       cfg.interval,
 		client:         cl,
 		parser:         parser,
 	}, nil
@@ -192,6 +220,7 @@ func tlsClientConfig(opts TLSOptions) (*tls.Config, error) {
 }
 
 func (s *HTTPDataSource[T]) TypedName() fwkplugin.TypedName { return s.typedName }
+func (s *HTTPDataSource[T]) Interval() time.Duration        { return s.interval }
 
 // Poll fetches and parses one tick. Exposed for tests; runtime uses Dispatch.
 func (s *HTTPDataSource[T]) Poll(ctx context.Context, ep fwkdl.Endpoint) (T, error) {
diff --git a/pkg/epp/framework/plugins/datalayer/source/metrics/datasource.go b/pkg/epp/framework/plugins/datalayer/source/metrics/datasource.go
@@ -51,6 +51,9 @@ type metricsDatasourceParams struct {
 	// target authenticates this scraper by certificate instead of a bearer token. Both set together.
 	ClientCertPath string `json:"clientCertPath"`
 	ClientKeyPath  string `json:"clientKeyPath"`
+	// Interval is the scrape period (e.g. "1s"). Must be a positive multiple of
+	// --refresh-metrics-interval. Empty or omitted means every base tick.
+	Interval string `json:"interval"`
 }
 
 // NewHTTPMetricsDataSource constructs a MetricsDataSource with the given scheme and path.
@@ -72,14 +75,21 @@ func MetricsDataSourceFactory(name string, parameters *json.Decoder, handle fwkp
 		}
 	}
 
+	var opts []http.Option
+	if intervalOpt, err := http.ParseIntervalOption(cfg.Interval); err != nil {
+		return nil, err
+	} else if intervalOpt != nil {
+		opts = append(opts, intervalOpt)
+	}
+
 	return http.NewHTTPDataSource(cfg.Scheme, cfg.Path,
 		http.TLSOptions{
 			SkipVerify:     cfg.InsecureSkipVerify,
 			CACertPath:     cfg.CACertPath,
 			ClientCertPath: cfg.ClientCertPath,
 			ClientKeyPath:  cfg.ClientKeyPath,
 		},
-		MetricsDataSourceType, name, parseMetrics)
+		MetricsDataSourceType, name, parseMetrics, opts...)
 }
 
 func defaultDataSourceConfigParams() *metricsDatasourceParams {
diff --git a/pkg/epp/framework/plugins/datalayer/source/mocks/data_source_mock.go b/pkg/epp/framework/plugins/datalayer/source/mocks/data_source_mock.go
@@ -90,6 +90,7 @@ func (fds *MetricsDataSource) Dispatch(_ context.Context, ep fwkdl.Endpoint) err
 // AppendExtractor is a no-op: this mock sets endpoint metrics directly in
 // Dispatch instead of running extractors.
 func (fds *MetricsDataSource) AppendExtractor(_ plugin.Plugin) error { return nil }
+func (fds *MetricsDataSource) Interval() time.Duration               { return 0 }
 
 // NotificationSource implements DataSource + NotificationSource for testing.
 type NotificationSource struct {
diff --git a/pkg/epp/framework/plugins/datalayer/source/models/README.md b/pkg/epp/framework/plugins/datalayer/source/models/README.md
@@ -22,6 +22,9 @@ The Models Data Source polls inference server pods for model information and pas
 - `insecureSkipVerify` (bool, optional, default: `true`): Skip TLS certificate verification.
 - `caCertPath` (string, optional): PEM CA bundle to verify the target's server cert.
 - `clientCertPath` / `clientKeyPath` (string, optional): client certificate for mTLS. Set both together.
+- `interval` (string, optional): Scrape period (e.g. `"5s"`). Must be a positive
+  multiple of `--refresh-metrics-interval` (default 50ms). Omit to scrape every
+  base tick.
 
 ```yaml
 - type: models-data-source
@@ -55,15 +58,13 @@ plugins:
   parameters:
     scheme: "https"
     insecureSkipVerify: false
+    interval: "5s"
 - type: models-data-extractor
   name: vllm-models-extractor
 # ... other plugins (filters, scorers, profile handler, picker) ...
 dataLayer:
   sources:
   - pluginRef: vllm-models-source
-    # Scrape period for this source. Must be a positive multiple of
-    # --refresh-metrics-interval (default 50ms). Omit to scrape every base tick.
-    interval: 5s
     extractors:
     - pluginRef: vllm-models-extractor
 ```
diff --git a/pkg/epp/framework/plugins/datalayer/source/models/datasource.go b/pkg/epp/framework/plugins/datalayer/source/models/datasource.go

Original file line number	Diff line number	Diff line change
`@@ -80,6 +80,7 @@ func (e *errSource) Dispatch(_ context.Context, _ fwkdl.Endpoint) error {`
`80`	`80`	`return e.err`
`81`	`81`	`}`
`82`	`82`	`func (e *errSource) AppendExtractor(_ fwkplugin.Plugin) error { return nil }`
	`83`	`+func (e *errSource) Interval() time.Duration { return 0 }`
`83`	`84`
`84`	`85`	`type dataSource struct {`
`85`	`86`	`kind string`
`@@ -115,6 +116,7 @@ func (d *dataSource) AppendExtractor(p fwkplugin.Plugin) error {`
`115`	`116`	`d.mu.Unlock()`
`116`	`117`	`return nil`
`117`	`118`	`}`
	`119`	`+func (d *dataSource) Interval() time.Duration { return 0 }`
`118`	`120`
`119`	`121`	`type stubExtractor struct {`
`120`	`122`	`kind string`
Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,6 @@ package datalayer`
`18`	`18`
`19`	`19`	`import (`
`20`	`20`	`"fmt"`
`21`		`- "time"`
`22`	`21`
`23`	`22`	`"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"`
`24`	`23`	`)`
`@@ -50,9 +49,6 @@ func (c *Config) String() string {`
`50`	`49`	`type DataSourceConfig struct {`
`51`	`50`	`Plugin plugin.Plugin // the source plugin instance (DataSource or PollingDispatcher)`
`52`	`51`	`Extractors []plugin.Plugin // extractors defined for the data source`
`53`		`- // Interval is the scrape period for polling sources. Zero means every`
`54`		`- // Runtime base tick. Ignored for notification/endpoint sources.`
`55`		`- Interval time.Duration`
`56`	`52`	`}`
`57`	`53`
`58`	`54`	`func (dsc DataSourceConfig) String() string {`
Original file line number	Diff line number	Diff line change
`@@ -121,9 +121,9 @@ func (r Runtime) Configure(cfg Config, logger logr.Logger) error {`
`121`	`121`	`}`
`122`	`122`
`123`	`123`	`periodTicks := 1`
`124`		`- if _, ok := src.(fwkdl.PollingDispatcher); ok {`
	`124`	`+ if disp, ok := src.(fwkdl.PollingDispatcher); ok {`
`125`	`125`	`var err error`
`126`		`- periodTicks, err = PeriodTicks(srcCfg.Interval, r.pollingInterval)`
	`126`	`+ periodTicks, err = PeriodTicks(disp.Interval(), r.pollingInterval)`
`127`	`127`	`if err != nil {`
`128`	`128`	`return fmt.Errorf("source %s: %w", srcName, err)`
`129`	`129`	`}`