feat(datalayer): allow per-source scrape intervals

Polling sources shared one Collector tick (~50ms), so slow-changing
sources (DCGM, /v1/models) were scraped far more often than useful.
Schedule each dataLayer.sources entry by a tick multiple of
--refresh-metrics-interval; omit keeps every-tick behavior.

Signed-off-by: noalimoy <nlimoy@redhat.com>

Author: noalimoy

apix/config/v1alpha1/endpointpickerconfig_types.go

@@ -278,6 +278,13 @@ 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 {
@@ -286,6 +293,9 @@ 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, ", ") + "}"
 }
 

apix/config/v1alpha1/zz_generated.deepcopy.go

@@ -37,6 +37,7 @@ plugins:
 dataLayer:
   sources:
   - pluginRef: dcgm-source
+    interval: 1s
     extractors:
     - pluginRef: dcgm-extractor
 schedulingProfiles:

deploy/config/sim-epp-gpu-config.yaml

@@ -224,6 +224,12 @@ The data layer follows a Source -> Extract -> Attribute lifecycle:
 - Extractors populate per-endpoint attributes in the shared datastore for scorers
 - 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.
+
 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.
 
 ---

docs/architecture.md

@@ -438,6 +438,9 @@ 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 {

pkg/epp/config/loader/configloader.go

@@ -76,15 +76,19 @@ func NewCollector() *Collector {
 }
 
 // Start launches the collection goroutine.
-// Each PollingDispatcher owns its extractors; the Collector calls Dispatch per tick.
-func (c *Collector) Start(ctx context.Context, ticker Ticker, ep fwkdl.Endpoint, dispatchers []fwkdl.PollingDispatcher) error {
-	if len(dispatchers) == 0 {
+// Each ScheduledDispatcher owns its extractors; the Collector invokes
+// Dispatch on a source only when its period elapses (measured in base ticks).
+func (c *Collector) Start(ctx context.Context, ticker Ticker, ep fwkdl.Endpoint, scheduled []ScheduledDispatcher) error {
+	if len(scheduled) == 0 {
 		return errors.New("cannot start collector with empty dispatchers")
 	}
-	for _, d := range dispatchers {
-		if d == nil {
+	for _, s := range scheduled {
+		if s.Dispatcher == nil {
 			return errors.New("cannot add nil dispatcher")
 		}
+		if s.PeriodTicks < 1 {
+			return errors.New("periodTicks must be >= 1")
+		}
 	}
 	if err := ctx.Err(); err != nil {
 		return err
@@ -97,7 +101,7 @@ func (c *Collector) Start(ctx context.Context, ticker Ticker, ep fwkdl.Endpoint,
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	c.cancel = cancel
-	go c.run(ctx, ticker, ep, dispatchers)
+	go c.run(ctx, ticker, ep, scheduled)
 	return nil
 }
 
@@ -112,30 +116,40 @@ func (c *Collector) Stop() {
 	}
 }
 
-func (c *Collector) run(ctx context.Context, ticker Ticker, ep fwkdl.Endpoint, dispatchers []fwkdl.PollingDispatcher) {
+func (c *Collector) run(ctx context.Context, ticker Ticker, ep fwkdl.Endpoint, scheduled []ScheduledDispatcher) {
 	defer func() {
 		close(c.done)
 		ticker.Stop()
 	}()
 	logger := log.FromContext(ctx).WithValues("endpoint", ep.GetMetadata().GetIPAddress())
 
+	// nextDue[i] is the tick index when scheduled[i] should next Dispatch.
+	// Zero means fire on the first tick (tick == 0).
+	nextDue := make([]int, len(scheduled))
+	tick := 0
+
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.Channel():
-			for _, disp := range dispatchers {
+			for i, s := range scheduled {
 				if ctx.Err() != nil {
 					return
 				}
+				if tick != nextDue[i] {
+					continue
+				}
 				dispCtx, cancel := context.WithTimeout(ctx, defaultCollectionTimeout)
-				if err := disp.Dispatch(dispCtx, ep); err != nil {
-					tn := disp.TypedName()
+				if err := s.Dispatcher.Dispatch(dispCtx, ep); err != nil {
+					tn := s.Dispatcher.TypedName()
 					metrics.RecordDataLayerPollError(tn.Type)
 					logger.V(logging.DEBUG).Info("dispatch failed", "source", tn, "err", err)
 				}
 				cancel()
+				nextDue[i] = tick + s.PeriodTicks
 			}
+			tick++
 		}
 	}
 }

pkg/epp/datalayer/collector.go

@@ -49,9 +49,18 @@ func defaultEndpoint() fwkdl.Endpoint {
 
 var (
 	endpoint = defaultEndpoint()
-	sources  = []fwkdl.PollingDispatcher{&datasourcemocks.MetricsDataSource{}}
+	sources  = scheduleEveryTick([]fwkdl.PollingDispatcher{&datasourcemocks.MetricsDataSource{}})
 )
 
+// scheduleEveryTick wraps each dispatcher with PeriodTicks=1 for collector tests.
+func scheduleEveryTick(dispatchers []fwkdl.PollingDispatcher) []ScheduledDispatcher {
+	out := make([]ScheduledDispatcher, len(dispatchers))
+	for i, d := range dispatchers {
+		out[i] = ScheduledDispatcher{Dispatcher: d, PeriodTicks: 1}
+	}
+	return out
+}
+
 // Mock PollingDispatchers for collector tests. Each tracks its own
 // invocation count and (for dataSource) runs bound extractors with metric
 // instrumentation. Same behavior the framework collector did before the
@@ -121,13 +130,14 @@ func TestCollectorStartInputs(t *testing.T) {
 	tests := []struct {
 		name        string
 		ctxCanceled bool
-		sources     []fwkdl.PollingDispatcher
+		sources     []ScheduledDispatcher
 		wantErr     bool
 		wantErrIs   error
 	}{
 		{name: "valid sources, live ctx", sources: sources},
-		{name: "empty sources", sources: []fwkdl.PollingDispatcher{}, wantErr: true},
-		{name: "nil source", sources: []fwkdl.PollingDispatcher{nil}, wantErr: true},
+		{name: "empty sources", sources: []ScheduledDispatcher{}, wantErr: true},
+		{name: "nil source", sources: []ScheduledDispatcher{{Dispatcher: nil, PeriodTicks: 1}}, wantErr: true},
+		{name: "periodTicks zero", sources: []ScheduledDispatcher{{Dispatcher: &datasourcemocks.MetricsDataSource{}, PeriodTicks: 0}}, wantErr: true},
 		{name: "cancelled parent ctx", ctxCanceled: true, sources: sources, wantErr: true, wantErrIs: context.Canceled},
 	}
 
@@ -182,7 +192,7 @@ func TestCollectorStop(t *testing.T) {
 			setup: func(t *testing.T) *Collector {
 				c := NewCollector()
 				ticker := mocks.NewTicker()
-				_ = c.Start(context.Background(), ticker, endpoint, []fwkdl.PollingDispatcher{})
+				_ = c.Start(context.Background(), ticker, endpoint, []ScheduledDispatcher{})
 				return c
 			},
 		},
@@ -213,7 +223,7 @@ func TestCollectorCollectsOnTicks(t *testing.T) {
 	c := NewCollector()
 	ticker := mocks.NewTicker()
 
-	require.NoError(t, c.Start(context.Background(), ticker, endpoint, []fwkdl.PollingDispatcher{source}))
+	require.NoError(t, c.Start(context.Background(), ticker, endpoint, scheduleEveryTick([]fwkdl.PollingDispatcher{source})))
 	defer c.Stop()
 
 	ticker.Tick()
@@ -224,6 +234,30 @@ func TestCollectorCollectsOnTicks(t *testing.T) {
 	}, 1*time.Second, 2*time.Millisecond, "expected 2 collections")
 }
 
+// TestCollectorRespectsPeriodTicks confirms slower sources skip ticks.
+func TestCollectorRespectsPeriodTicks(t *testing.T) {
+	fast := &errSource{kind: "fast"}
+	slow := &errSource{kind: "slow"}
+
+	c := NewCollector()
+	ticker := mocks.NewTicker()
+	require.NoError(t, c.Start(context.Background(), ticker, endpoint, []ScheduledDispatcher{
+		{Dispatcher: fast, PeriodTicks: 1},
+		{Dispatcher: slow, PeriodTicks: 4},
+	}))
+	defer c.Stop()
+
+	for i := 0; i < 8; i++ {
+		ticker.Tick()
+	}
+
+	require.Eventually(t, func() bool {
+		return atomic.LoadInt64(&fast.CallCount) == 8 && atomic.LoadInt64(&slow.CallCount) == 2
+	}, 1*time.Second, 2*time.Millisecond,
+		"fast=%d slow=%d want fast=8 slow=2",
+		atomic.LoadInt64(&fast.CallCount), atomic.LoadInt64(&slow.CallCount))
+}
+
 // TestCollectorErrorMetrics confirms Poll/Extract errors increment per-event
 // counters (no transition dedup) and successes do not.
 func TestCollectorErrorMetrics(t *testing.T) {
@@ -285,7 +319,7 @@ func TestCollectorErrorMetrics(t *testing.T) {
 
 			c := NewCollector()
 			ticker := mocks.NewTicker()
-			require.NoError(t, c.Start(context.Background(), ticker, endpoint, []fwkdl.PollingDispatcher{src}))
+			require.NoError(t, c.Start(context.Background(), ticker, endpoint, scheduleEveryTick([]fwkdl.PollingDispatcher{src})))
 			defer c.Stop()
 
 			for i := 0; i < tt.ticks; i++ {
@@ -324,7 +358,7 @@ func TestCollectorRapidStartStopRaceFree(t *testing.T) {
 	for i := 0; i < 100; i++ {
 		c := NewCollector()
 		ticker := mocks.NewTicker()
-		require.NoError(t, c.Start(context.Background(), ticker, endpoint, []fwkdl.PollingDispatcher{src}))
+		require.NoError(t, c.Start(context.Background(), ticker, endpoint, scheduleEveryTick([]fwkdl.PollingDispatcher{src})))
 		ticker.Tick()
 		c.Stop()
 	}

pkg/epp/datalayer/collector_test.go

@@ -18,6 +18,7 @@ package datalayer
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
 )
@@ -49,6 +50,9 @@ 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 {

pkg/epp/datalayer/config.go

@@ -128,46 +128,66 @@ func (m *variantSourceMap[T]) findFirst(matches func(fwkplugin.Plugin) bool) sou
 	return found
 }
 
-// pollingDispatchers stores PollingDispatchers keyed by source name. Each
-// dispatcher owns its own extractors internally; the framework treats them
-// as opaque dispatch units.
+// pollingDispatchers stores scheduled PollingDispatchers keyed by source name.
+// Each dispatcher owns its own extractors; PeriodTicks is how often the
+// Collector should invoke it (in base ticks).
 type pollingDispatchers struct {
 	mu sync.RWMutex
-	m  map[string]fwkdl.PollingDispatcher
+	m  map[string]ScheduledDispatcher
 }
 
 func newPollingDispatchers() *pollingDispatchers {
-	return &pollingDispatchers{m: make(map[string]fwkdl.PollingDispatcher)}
+	return &pollingDispatchers{m: make(map[string]ScheduledDispatcher)}
 }
 
-// Register installs disp under its TypedName.Name. Duplicate names fail loudly
+// Register installs s under its TypedName.Name. Duplicate names fail loudly
 // so a config error surfaces at startup instead of silently shadowing telemetry.
-func (p *pollingDispatchers) Register(disp fwkdl.PollingDispatcher) error {
+func (p *pollingDispatchers) Register(s ScheduledDispatcher) error {
+	if s.Dispatcher == nil {
+		return fmt.Errorf("cannot register nil %s dispatcher", variantPolling)
+	}
+	if s.PeriodTicks < 1 {
+		return fmt.Errorf("periodTicks must be >= 1 for source %q", s.Dispatcher.TypedName().Name)
+	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	name := disp.TypedName().Name
+	name := s.Dispatcher.TypedName().Name
 	if _, exists := p.m[name]; exists {
 		return fmt.Errorf("duplicate %s source name %q", variantPolling, name)
 	}
-	p.m[name] = disp
+	p.m[name] = s
 	return nil
 }
 
 // Get returns the dispatcher registered under name, if any.
 func (p *pollingDispatchers) Get(name string) (fwkdl.PollingDispatcher, bool) {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
-	d, ok := p.m[name]
-	return d, ok
+	s, ok := p.m[name]
+	if !ok {
+		return nil, false
+	}
+	return s.Dispatcher, true
 }
 
-// Dispatchers returns a snapshot of all dispatchers.
+// Dispatchers returns a snapshot of all dispatchers (without schedule metadata).
 func (p *pollingDispatchers) Dispatchers() map[string]fwkdl.PollingDispatcher {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 	out := make(map[string]fwkdl.PollingDispatcher, len(p.m))
 	for k, v := range p.m {
-		out[k] = v
+		out[k] = v.Dispatcher
+	}
+	return out
+}
+
+// Scheduled returns a snapshot of all scheduled dispatchers.
+func (p *pollingDispatchers) Scheduled() []ScheduledDispatcher {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	out := make([]ScheduledDispatcher, 0, len(p.m))
+	for _, v := range p.m {
+		out = append(out, v)
 	}
 	return out
 }