Merge pull request #1320 from DigitalVeer/per-rule-invoke-interval

Add per-rule custom plugin intervals

docs/custom_plugin_monitor.md

@@ -11,6 +11,20 @@
 * `enable_message_change_based_condition_update`: Flag controls whether message change should result in a condition update.
 * `skip_initial_status`: Flag controls whether condition will be emitted during plugin initialization.
 
+### Rule Config
+
+* `timeout`: The execution timeout for one rule. A rule `timeout` greater than the global `timeout` fails validation at startup.
+* `invoke_interval`: The invoke interval for one rule. The rule uses the global `invoke_interval` when this field is absent. Rules with different effective intervals run on independent schedules. Their plugins can run at the same time, up to the global `concurrency` limit. A rule never runs concurrently with itself. The monitor runs at most one catch-up batch when a batch overruns its interval.
+
+For example, this rule runs every seven seconds:
+
+```
+{
+  "path": "./config/plugin/check_ntp.sh",
+  "invoke_interval": "7s"
+}
+```
+
 ### Annotated Plugin Configuration Example
 
 ```
@@ -41,6 +55,7 @@
       "reason": "NTPIsDown",            // This is the reason shown for this event
                                         // and the message shown comes from stdout.
       "path": "./config/plugin/check_ntp.sh",
+      "invoke_interval": "7s",
       "timeout": "3s"
     },
     {

pkg/custompluginmonitor/plugin/plugin.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"k8s.io/klog/v2"
+	"k8s.io/utils/clock"
 
 	cpmtypes "k8s.io/node-problem-detector/pkg/custompluginmonitor/types"
 	"k8s.io/node-problem-detector/pkg/util"
@@ -42,17 +43,28 @@ type Plugin struct {
 	syncChan   chan struct{}
 	resultChan chan cpmtypes.Result
 	tomb       *tomb.Tomb
+	clock      clock.WithTicker
+	runFunc    func(cpmtypes.CustomRule) (cpmtypes.Status, string)
 	sync.WaitGroup
 }
 
+type intervalGroup struct {
+	interval time.Duration
+	rules    []*cpmtypes.CustomRule
+	ticker   clock.Ticker
+}
+
 func NewPlugin(config cpmtypes.CustomPluginConfig) *Plugin {
-	return &Plugin{
+	p := &Plugin{
 		config:   config,
 		syncChan: make(chan struct{}, *config.PluginGlobalConfig.Concurrency),
 		// A 1000 size channel should be big enough.
 		resultChan: make(chan cpmtypes.Result, 1000),
 		tomb:       tomb.NewTomb(),
+		clock:      clock.RealClock{},
 	}
+	p.runFunc = p.run
+	return p
 }
 
 func (p *Plugin) GetResultChan() <-chan cpmtypes.Result {
@@ -66,44 +78,107 @@ func (p *Plugin) Run() {
 		p.tomb.Done()
 	}()
 
-	runTicker := time.NewTicker(*p.config.PluginGlobalConfig.InvokeInterval)
-	defer runTicker.Stop()
-
-	// on boot run once
-	select {
-	case <-p.tomb.Stopping():
+	groups := p.intervalGroups()
+	if len(groups) == 0 {
+		<-p.tomb.Stopping()
 		return
-	default:
-		p.runRules()
 	}
 
-	// run every InvokeInterval
+	for i := range groups {
+		groups[i].ticker = p.clock.NewTicker(groups[i].interval)
+	}
+	defer func() {
+		for i := range groups {
+			groups[i].ticker.Stop()
+		}
+	}()
+
+	// On boot, run every rule in one batch.
+	if !p.runRules(p.config.Rules) {
+		return
+	}
+
+	for i := range groups {
+		p.Add(1)
+		go p.runGroup(&groups[i])
+	}
+	p.Wait()
+}
+
+func (p *Plugin) intervalGroups() []intervalGroup {
+	groups := []intervalGroup{}
+	groupIndexes := make(map[time.Duration]int)
+	for _, rule := range p.config.Rules {
+		interval := p.effectiveInterval(rule)
+		groupIndex, ok := groupIndexes[interval]
+		if !ok {
+			groupIndex = len(groups)
+			groupIndexes[interval] = groupIndex
+			groups = append(groups, intervalGroup{interval: interval})
+		}
+		groups[groupIndex].rules = append(groups[groupIndex].rules, rule)
+	}
+	return groups
+}
+
+func (p *Plugin) effectiveInterval(rule *cpmtypes.CustomRule) time.Duration {
+	if rule.InvokeInterval != nil {
+		return *rule.InvokeInterval
+	}
+	return *p.config.PluginGlobalConfig.InvokeInterval
+}
+
+func (p *Plugin) runGroup(group *intervalGroup) {
+	defer p.Done()
 	for {
 		select {
-		case <-runTicker.C:
-			p.runRules()
+		case <-group.ticker.C():
+			if !p.runRules(group.rules) {
+				return
+			}
 		case <-p.tomb.Stopping():
 			return
 		}
 	}
 }
 
-// run each rule in parallel and wait for them to complete
-func (p *Plugin) runRules() {
+// runRules runs each rule in parallel and waits for the batch to complete.
+func (p *Plugin) runRules(rules []*cpmtypes.CustomRule) bool {
 	klog.V(3).Info("Start to run custom plugins")
+	var workers sync.WaitGroup
 
-	for _, rule := range p.config.Rules {
+	for _, rule := range rules {
 		// syncChan limits concurrent goroutines to configured PluginGlobalConfig.Concurrency value
-		p.syncChan <- struct{}{}
-		p.Add(1)
+		select {
+		case p.syncChan <- struct{}{}:
+		case <-p.tomb.Stopping():
+			workers.Wait()
+			return false
+		}
+
+		select {
+		case <-p.tomb.Stopping():
+			<-p.syncChan
+			workers.Wait()
+			return false
+		default:
+		}
+
+		workers.Add(1)
 		go func(rule *cpmtypes.CustomRule) {
-			defer p.Done()
+			defer workers.Done()
 			defer func() {
 				<-p.syncChan
 			}()
 
+			select {
+			case <-p.tomb.Stopping():
+				return
+			default:
+			}
+
 			start := time.Now()
-			exitStatus, message := p.run(*rule)
+			exitStatus, message := p.runFunc(*rule)
 			level := klog.Level(3)
 			if exitStatus != 0 {
 				level = klog.Level(2)
@@ -118,15 +193,20 @@ func (p *Plugin) runRules() {
 			}
 
 			// pipes result into resultChan which customPluginMonitor instance generates status from
-			p.resultChan <- result
+			select {
+			case p.resultChan <- result:
+			case <-p.tomb.Stopping():
+				return
+			}
 
 			// Let the result be logged at a higher verbosity level. If there is a change in status it is logged later.
 			klog.V(level).Infof("Add check result %+v for rule %+v", result, rule)
 		}(rule)
 	}
 
-	p.Wait()
+	workers.Wait()
 	klog.V(3).Info("Finish running custom plugins")
+	return true
 }
 
 // readFromReader reads the maxBytes from the reader and drains the rest.

pkg/custompluginmonitor/plugin/plugin_scheduler_test.go

@@ -0,0 +1,651 @@
+/*
+Copyright 2026 The Kubernetes Authors All rights reserved.
+
+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 plugin
+
+import (
+	"reflect"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	"k8s.io/utils/clock"
+	testclock "k8s.io/utils/clock/testing"
+
+	cpmtypes "k8s.io/node-problem-detector/pkg/custompluginmonitor/types"
+)
+
+const schedulerTestTimeout = 2 * time.Second
+
+type recordingClock struct {
+	*testclock.FakeClock
+	mu      sync.Mutex
+	tickers []clock.Ticker
+}
+
+func newRecordingClock() *recordingClock {
+	return &recordingClock{FakeClock: testclock.NewFakeClock(time.Unix(0, 0))}
+}
+
+func (c *recordingClock) NewTicker(interval time.Duration) clock.Ticker {
+	ticker := c.FakeClock.NewTicker(interval)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.tickers = append(c.tickers, ticker)
+	return ticker
+}
+
+func (c *recordingClock) tickerCount() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return len(c.tickers)
+}
+
+func (c *recordingClock) ticker(index int) clock.Ticker {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.tickers[index]
+}
+
+type invocation struct {
+	rule  string
+	count int
+}
+
+type invocationKey struct {
+	rule  string
+	count int
+}
+
+type executionRecorder struct {
+	mu          sync.Mutex
+	counts      map[string]int
+	active      map[string]int
+	maxActive   map[string]int
+	activeTotal int
+	highWater   int
+	blockers    map[invocationKey]<-chan struct{}
+	started     chan invocation
+	beforeRun   func(string, int)
+}
+
+func newExecutionRecorder() *executionRecorder {
+	return &executionRecorder{
+		counts:    make(map[string]int),
+		active:    make(map[string]int),
+		maxActive: make(map[string]int),
+		blockers:  make(map[invocationKey]<-chan struct{}),
+		started:   make(chan invocation, 100),
+	}
+}
+
+func (r *executionRecorder) block(rule string, count int, release <-chan struct{}) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.blockers[invocationKey{rule: rule, count: count}] = release
+}
+
+func (r *executionRecorder) run(rule cpmtypes.CustomRule) (cpmtypes.Status, string) {
+	r.mu.Lock()
+	r.counts[rule.Path]++
+	count := r.counts[rule.Path]
+	r.active[rule.Path]++
+	if r.active[rule.Path] > r.maxActive[rule.Path] {
+		r.maxActive[rule.Path] = r.active[rule.Path]
+	}
+	r.activeTotal++
+	if r.activeTotal > r.highWater {
+		r.highWater = r.activeTotal
+	}
+	blocker := r.blockers[invocationKey{rule: rule.Path, count: count}]
+	beforeRun := r.beforeRun
+	r.mu.Unlock()
+
+	if beforeRun != nil {
+		beforeRun(rule.Path, count)
+	}
+	r.started <- invocation{rule: rule.Path, count: count}
+	if blocker != nil {
+		<-blocker
+	}
+
+	r.mu.Lock()
+	r.active[rule.Path]--
+	r.activeTotal--
+	r.mu.Unlock()
+	return cpmtypes.OK, rule.Path
+}
+
+type executionSnapshot struct {
+	counts      map[string]int
+	maxActive   map[string]int
+	activeTotal int
+	highWater   int
+}
+
+func (r *executionRecorder) snapshot() executionSnapshot {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	counts := make(map[string]int, len(r.counts))
+	for rule, count := range r.counts {
+		counts[rule] = count
+	}
+	maxActive := make(map[string]int, len(r.maxActive))
+	for rule, count := range r.maxActive {
+		maxActive[rule] = count
+	}
+	return executionSnapshot{
+		counts:      counts,
+		maxActive:   maxActive,
+		activeTotal: r.activeTotal,
+		highWater:   r.highWater,
+	}
+}
+
+func schedulerRule(name string, interval *time.Duration) *cpmtypes.CustomRule {
+	return &cpmtypes.CustomRule{Path: name, InvokeInterval: interval}
+}
+
+func newSchedulerPlugin(t *testing.T, rules []*cpmtypes.CustomRule, globalInterval time.Duration, concurrency int) (*Plugin, *recordingClock, *executionRecorder) {
+	t.Helper()
+	globalIntervalString := globalInterval.String()
+	config := cpmtypes.CustomPluginConfig{Rules: rules}
+	config.PluginGlobalConfig.InvokeIntervalString = &globalIntervalString
+	config.PluginGlobalConfig.Concurrency = &concurrency
+	if err := config.ApplyConfiguration(); err != nil {
+		t.Fatalf("ApplyConfiguration() failed: %v", err)
+	}
+	p := NewPlugin(config)
+	fakeClock := newRecordingClock()
+	recorder := newExecutionRecorder()
+	p.clock = fakeClock
+	p.runFunc = recorder.run
+	return p, fakeClock, recorder
+}
+
+func startPlugin(t *testing.T, p *Plugin, fakeClock *recordingClock, tickerCount int) {
+	t.Helper()
+	go p.Run()
+	waitFor(t, "tickers to be armed", func() bool {
+		return fakeClock.HasWaiters() == (tickerCount > 0) && fakeClock.tickerCount() == tickerCount
+	})
+}
+
+func stopPlugin(t *testing.T, p *Plugin) {
+	t.Helper()
+	stopped := make(chan struct{})
+	go func() {
+		p.Stop()
+		close(stopped)
+	}()
+	waitChannel(t, "plugin to stop", stopped)
+}
+
+func waitFor(t *testing.T, description string, condition func() bool) {
+	t.Helper()
+	timer := time.NewTimer(schedulerTestTimeout)
+	defer timer.Stop()
+	for !condition() {
+		select {
+		case <-timer.C:
+			t.Fatalf("Timed out waiting for %s", description)
+		default:
+			runtime.Gosched()
+		}
+	}
+}
+
+func waitChannel(t *testing.T, description string, channel <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-channel:
+	case <-time.After(schedulerTestTimeout):
+		t.Fatalf("Timed out waiting for %s", description)
+	}
+}
+
+func waitInvocations(t *testing.T, recorder *executionRecorder, count int) []invocation {
+	t.Helper()
+	invocations := make([]invocation, 0, count)
+	for len(invocations) < count {
+		select {
+		case invocation := <-recorder.started:
+			invocations = append(invocations, invocation)
+		case <-time.After(schedulerTestTimeout):
+			t.Fatalf("Timed out after %d of %d invocation starts", len(invocations), count)
+		}
+	}
+	return invocations
+}
+
+func waitResults(t *testing.T, p *Plugin, count int) []cpmtypes.Result {
+	t.Helper()
+	results := make([]cpmtypes.Result, 0, count)
+	for len(results) < count {
+		select {
+		case result, ok := <-p.resultChan:
+			if !ok {
+				t.Fatalf("Result channel closed after %d of %d results", len(results), count)
+			}
+			results = append(results, result)
+		case <-time.After(schedulerTestTimeout):
+			t.Fatalf("Timed out after %d of %d results", len(results), count)
+		}
+	}
+	return results
+}
+
+func stepClock(t *testing.T, fakeClock *recordingClock, duration time.Duration) {
+	t.Helper()
+	if !fakeClock.HasWaiters() {
+		t.Fatal("Fake clock has no armed ticker")
+	}
+	fakeClock.Step(duration)
+}
+
+func assertCounts(t *testing.T, recorder *executionRecorder, wanted map[string]int) {
+	t.Helper()
+	got := recorder.snapshot().counts
+	if !reflect.DeepEqual(got, wanted) {
+		t.Fatalf("Invocation counts differ: got %v, wanted %v", got, wanted)
+	}
+}
+
+func TestPluginSchedulerBootRunsOneCombinedBatch(t *testing.T) {
+	interval5 := 5 * time.Second
+	interval7 := 7 * time.Second
+	interval11 := 11 * time.Second
+	rules := []*cpmtypes.CustomRule{
+		schedulerRule("five", &interval5),
+		schedulerRule("seven", &interval7),
+		schedulerRule("eleven", &interval11),
+	}
+	p, fakeClock, recorder := newSchedulerPlugin(t, rules, 30*time.Second, 3)
+	release := make(chan struct{})
+	for _, rule := range rules {
+		recorder.block(rule.Path, 1, release)
+	}
+	recorder.beforeRun = func(_ string, _ int) {
+		if fakeClock.Waiters() != 3 {
+			t.Errorf("Boot execution started with %d tickers; wanted 3", fakeClock.Waiters())
+		}
+	}
+
+	startPlugin(t, p, fakeClock, 3)
+	waitInvocations(t, recorder, 3)
+	assertCounts(t, recorder, map[string]int{"five": 1, "seven": 1, "eleven": 1})
+	close(release)
+	waitResults(t, p, 3)
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerDefaultParityAndSameGroupCoupling(t *testing.T) {
+	rules := []*cpmtypes.CustomRule{
+		schedulerRule("one", nil),
+		schedulerRule("two", nil),
+		schedulerRule("three", nil),
+	}
+	p, fakeClock, recorder := newSchedulerPlugin(t, rules, 10*time.Second, 3)
+	release := make(chan struct{})
+	recorder.block("one", 2, release)
+
+	startPlugin(t, p, fakeClock, 1)
+	waitInvocations(t, recorder, 3)
+	waitResults(t, p, 3)
+	assertCounts(t, recorder, map[string]int{"one": 1, "two": 1, "three": 1})
+
+	stepClock(t, fakeClock, 10*time.Second)
+	waitInvocations(t, recorder, 3)
+	waitResults(t, p, 2)
+	assertCounts(t, recorder, map[string]int{"one": 2, "two": 2, "three": 2})
+
+	stepClock(t, fakeClock, 10*time.Second)
+	if len(fakeClock.ticker(0).C()) != 1 {
+		t.Fatalf("Pending same-group tick count is %d; wanted 1", len(fakeClock.ticker(0).C()))
+	}
+	assertCounts(t, recorder, map[string]int{"one": 2, "two": 2, "three": 2})
+
+	close(release)
+	waitResults(t, p, 1)
+	waitInvocations(t, recorder, 3)
+	waitResults(t, p, 3)
+	assertCounts(t, recorder, map[string]int{"one": 3, "two": 3, "three": 3})
+
+	waitFor(t, "the parity group to consume its pending tick", func() bool {
+		return len(fakeClock.ticker(0).C()) == 0
+	})
+	stepClock(t, fakeClock, 10*time.Second)
+	waitInvocations(t, recorder, 3)
+	waitResults(t, p, 3)
+	assertCounts(t, recorder, map[string]int{"one": 4, "two": 4, "three": 4})
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerMixedCadences(t *testing.T) {
+	interval7 := 7 * time.Second
+	rules := []*cpmtypes.CustomRule{
+		schedulerRule("short", &interval7),
+		schedulerRule("global", nil),
+	}
+	p, fakeClock, recorder := newSchedulerPlugin(t, rules, 30*time.Second, 2)
+	startPlugin(t, p, fakeClock, 2)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 2)
+
+	steps := []struct {
+		advance       time.Duration
+		newExecutions int
+		shortCount    int
+		globalCount   int
+	}{
+		{7 * time.Second, 1, 2, 1},
+		{7 * time.Second, 1, 3, 1},
+		{7 * time.Second, 1, 4, 1},
+		{7 * time.Second, 1, 5, 1},
+		{2 * time.Second, 1, 5, 2},
+		{5 * time.Second, 1, 6, 2},
+		{7 * time.Second, 1, 7, 2},
+		{7 * time.Second, 1, 8, 2},
+		{7 * time.Second, 1, 9, 2},
+		{4 * time.Second, 1, 9, 3},
+		{3 * time.Second, 1, 10, 3},
+		{7 * time.Second, 1, 11, 3},
+		{7 * time.Second, 1, 12, 3},
+		{7 * time.Second, 1, 13, 3},
+		{6 * time.Second, 1, 13, 4},
+	}
+	for _, step := range steps {
+		stepClock(t, fakeClock, step.advance)
+		waitInvocations(t, recorder, step.newExecutions)
+		waitResults(t, p, step.newExecutions)
+		assertCounts(t, recorder, map[string]int{"short": step.shortCount, "global": step.globalCount})
+	}
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerEqualParsedIntervalsShareGroup(t *testing.T) {
+	explicitIntervalString := "30000ms"
+	rules := []*cpmtypes.CustomRule{
+		{Path: "unset"},
+		{Path: "explicit", InvokeIntervalString: &explicitIntervalString},
+	}
+	p, fakeClock, recorder := newSchedulerPlugin(t, rules, 30*time.Second, 2)
+	groups := p.intervalGroups()
+	if len(groups) != 1 || len(groups[0].rules) != 2 {
+		t.Fatalf("Parsed-equal rules formed groups %+v; wanted one two-rule group", groups)
+	}
+
+	startPlugin(t, p, fakeClock, 1)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 2)
+	stepClock(t, fakeClock, 30*time.Second)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 2)
+	assertCounts(t, recorder, map[string]int{"unset": 2, "explicit": 2})
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerCrossGroupIndependence(t *testing.T) {
+	interval5 := 5 * time.Second
+	interval7 := 7 * time.Second
+	p, fakeClock, recorder := newSchedulerPlugin(t, []*cpmtypes.CustomRule{
+		schedulerRule("blocked", &interval5),
+		schedulerRule("independent", &interval7),
+	}, 30*time.Second, 2)
+	release := make(chan struct{})
+	recorder.block("blocked", 2, release)
+
+	startPlugin(t, p, fakeClock, 2)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 2)
+	stepClock(t, fakeClock, 5*time.Second)
+	waitInvocations(t, recorder, 1)
+	stepClock(t, fakeClock, 2*time.Second)
+	invocations := waitInvocations(t, recorder, 1)
+	if invocations[0] != (invocation{rule: "independent", count: 2}) {
+		t.Fatalf("Independent invocation is %+v", invocations[0])
+	}
+	waitResults(t, p, 1)
+	assertCounts(t, recorder, map[string]int{"blocked": 2, "independent": 2})
+	close(release)
+	waitResults(t, p, 1)
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerConcurrencyReachesLimit(t *testing.T) {
+	rules := []*cpmtypes.CustomRule{
+		schedulerRule("one", nil),
+		schedulerRule("two", nil),
+		schedulerRule("three", nil),
+		schedulerRule("four", nil),
+	}
+	p, fakeClock, recorder := newSchedulerPlugin(t, rules, 10*time.Second, 2)
+	startPlugin(t, p, fakeClock, 1)
+	waitInvocations(t, recorder, 4)
+	waitResults(t, p, 4)
+
+	release := make(chan struct{})
+	for _, rule := range rules {
+		recorder.block(rule.Path, 2, release)
+	}
+	stepClock(t, fakeClock, 10*time.Second)
+	waitInvocations(t, recorder, 2)
+	snapshot := recorder.snapshot()
+	if snapshot.activeTotal != 2 || snapshot.highWater != 2 {
+		t.Fatalf("Concurrency state is active=%d high-water=%d; wanted 2 and 2", snapshot.activeTotal, snapshot.highWater)
+	}
+	close(release)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 4)
+	if highWater := recorder.snapshot().highWater; highWater != 2 {
+		t.Fatalf("Concurrency high-water is %d; wanted 2", highWater)
+	}
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerRuleNeverOverlapsAndOverrunCatchesUpOnce(t *testing.T) {
+	interval := 5 * time.Second
+	p, fakeClock, recorder := newSchedulerPlugin(t, []*cpmtypes.CustomRule{
+		schedulerRule("rule", &interval),
+	}, 30*time.Second, 2)
+	release := make(chan struct{})
+	recorder.block("rule", 2, release)
+
+	startPlugin(t, p, fakeClock, 1)
+	waitInvocations(t, recorder, 1)
+	waitResults(t, p, 1)
+	stepClock(t, fakeClock, interval)
+	waitInvocations(t, recorder, 1)
+	for i := 0; i < 3; i++ {
+		stepClock(t, fakeClock, interval)
+	}
+	assertCounts(t, recorder, map[string]int{"rule": 2})
+	close(release)
+	waitResults(t, p, 1)
+	waitInvocations(t, recorder, 1)
+	waitResults(t, p, 1)
+	assertCounts(t, recorder, map[string]int{"rule": 3})
+	maxActive := recorder.snapshot().maxActive
+	if maxActive["rule"] != 1 {
+		t.Fatalf("Rule concurrency high-water is %d; wanted 1", maxActive["rule"])
+	}
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerConcurrencyOneDoesNotStarveGroups(t *testing.T) {
+	interval7 := 7 * time.Second
+	p, fakeClock, recorder := newSchedulerPlugin(t, []*cpmtypes.CustomRule{
+		schedulerRule("short", &interval7),
+		schedulerRule("long", nil),
+	}, 30*time.Second, 1)
+	startPlugin(t, p, fakeClock, 2)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 2)
+
+	elapsed := time.Duration(0)
+	nextShort := interval7
+	nextLong := 30 * time.Second
+	shortCount := 1
+	longCount := 1
+	for elapsed < 210*time.Second {
+		nextBoundary := nextShort
+		if nextLong < nextBoundary {
+			nextBoundary = nextLong
+		}
+		due := 0
+		if nextShort == nextBoundary {
+			shortCount++
+			nextShort += interval7
+			due++
+		}
+		if nextLong == nextBoundary {
+			longCount++
+			nextLong += 30 * time.Second
+			due++
+		}
+		stepClock(t, fakeClock, nextBoundary-elapsed)
+		elapsed = nextBoundary
+		waitInvocations(t, recorder, due)
+		waitResults(t, p, due)
+		assertCounts(t, recorder, map[string]int{"short": shortCount, "long": longCount})
+	}
+	assertCounts(t, recorder, map[string]int{"short": 31, "long": 8})
+	stopPlugin(t, p)
+}
+
+func TestPluginSchedulerZeroRulesWaitsForStop(t *testing.T) {
+	p, fakeClock, recorder := newSchedulerPlugin(t, nil, 30*time.Second, 1)
+	started := make(chan struct{})
+	finished := make(chan struct{})
+	go func() {
+		close(started)
+		p.Run()
+		close(finished)
+	}()
+	<-started
+	for i := 0; i < 100; i++ {
+		runtime.Gosched()
+	}
+	if len(recorder.started) != 0 || fakeClock.HasWaiters() {
+		t.Fatalf("Zero-rule scheduler started executions or tickers")
+	}
+	select {
+	case <-finished:
+		t.Fatal("Zero-rule scheduler returned before stop")
+	default:
+	}
+	stopPlugin(t, p)
+	waitChannel(t, "zero-rule Run to return", finished)
+}
+
+func TestPluginSchedulerStopWhileIdle(t *testing.T) {
+	p, fakeClock, recorder := newSchedulerPlugin(t, []*cpmtypes.CustomRule{
+		schedulerRule("idle", nil),
+	}, 30*time.Second, 1)
+	startPlugin(t, p, fakeClock, 1)
+	waitInvocations(t, recorder, 1)
+	waitResults(t, p, 1)
+	stopPlugin(t, p)
+	if _, ok := <-p.resultChan; ok {
+		t.Fatal("Result channel remained open after idle stop")
+	}
+}
+
+func TestPluginSchedulerStopWhileSemaphoreAcquireBlocked(t *testing.T) {
+	interval5 := 5 * time.Second
+	interval7 := 7 * time.Second
+	p, fakeClock, recorder := newSchedulerPlugin(t, []*cpmtypes.CustomRule{
+		schedulerRule("holder", &interval5),
+		schedulerRule("waiter", &interval7),
+	}, 30*time.Second, 1)
+	release := make(chan struct{})
+	recorder.block("holder", 2, release)
+	startPlugin(t, p, fakeClock, 2)
+	waitInvocations(t, recorder, 2)
+	waitResults(t, p, 2)
+
+	stepClock(t, fakeClock, 5*time.Second)
+	waitInvocations(t, recorder, 1)
+	stepClock(t, fakeClock, 2*time.Second)
+	waitFor(t, "waiter group to consume its tick", func() bool {
+		return len(fakeClock.ticker(1).C()) == 0
+	})
+
+	stopped := make(chan struct{})
+	go func() {
+		p.Stop()
+		close(stopped)
+	}()
+	waitFor(t, "stop signal while semaphore acquire is blocked", func() bool {
+		select {
+		case <-p.tomb.Stopping():
+			return true
+		default:
+			return false
+		}
+	})
+	close(release)
+	waitChannel(t, "stop with blocked semaphore acquire", stopped)
+	assertCounts(t, recorder, map[string]int{"holder": 2, "waiter": 1})
+}
+
+func TestPluginSchedulerStopWithExecutionInFlight(t *testing.T) {
+	p, fakeClock, recorder := newSchedulerPlugin(t, []*cpmtypes.CustomRule{
+		schedulerRule("in-flight", nil),
+	}, 5*time.Second, 1)
+	release := make(chan struct{})
+	recorder.block("in-flight", 1, release)
+	startPlugin(t, p, fakeClock, 1)
+	waitInvocations(t, recorder, 1)
+
+	stopped := make(chan struct{})
+	go func() {
+		p.Stop()
+		close(stopped)
+	}()
+	waitFor(t, "stop signal with execution in flight", func() bool {
+		select {
+		case <-p.tomb.Stopping():
+			return true
+		default:
+			return false
+		}
+	})
+	select {
+	case <-stopped:
+		t.Fatal("Stop returned before the in-flight execution finished")
+	default:
+	}
+	close(release)
+	waitChannel(t, "stop with execution in flight", stopped)
+
+	stepClock(t, fakeClock, 5*time.Second)
+	assertCounts(t, recorder, map[string]int{"in-flight": 1})
+	for {
+		select {
+		case _, ok := <-p.resultChan:
+			if !ok {
+				if _, open := <-p.resultChan; open {
+					t.Fatal("Result channel reopened after close")
+				}
+				return
+			}
+		case <-time.After(schedulerTestTimeout):
+			t.Fatal("Result channel did not close after in-flight stop")
+		}
+	}
+}

pkg/custompluginmonitor/types/config.go

@@ -120,6 +120,13 @@ func (cpc *CustomPluginConfig) ApplyConfiguration() error {
 			}
 			rule.Timeout = &timeout
 		}
+		if rule.InvokeIntervalString != nil {
+			invokeInterval, err := time.ParseDuration(*rule.InvokeIntervalString)
+			if err != nil {
+				return fmt.Errorf("error in parsing rule invoke interval %+v: %v", rule, err)
+			}
+			rule.InvokeInterval = &invokeInterval
+		}
 	}
 
 	if cpc.EnableMetricsReporting == nil {
@@ -134,8 +141,14 @@ func (cpc CustomPluginConfig) Validate() error {
 	if cpc.Plugin != customPluginName {
 		return fmt.Errorf("NPD does not support %q plugin for now. Only support \"custom\"", cpc.Plugin)
 	}
+	if *cpc.PluginGlobalConfig.InvokeInterval <= 0 {
+		return fmt.Errorf("global invoke interval must be greater than zero: %v", *cpc.PluginGlobalConfig.InvokeInterval)
+	}
 
 	for _, rule := range cpc.Rules {
+		if rule.InvokeInterval != nil && *rule.InvokeInterval <= 0 {
+			return fmt.Errorf("rule invoke interval must be greater than zero. Rule: %+v", rule)
+		}
 		if rule.Timeout != nil && *rule.Timeout > *cpc.PluginGlobalConfig.Timeout {
 			return fmt.Errorf("plugin timeout is greater than global timeout. "+
 				"Rule: %+v. Global timeout: %v", rule, cpc.PluginGlobalConfig.Timeout)

pkg/custompluginmonitor/types/config_test.go

@@ -17,14 +17,22 @@ limitations under the License.
 package types
 
 import (
+	"fmt"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
 	"k8s.io/node-problem-detector/pkg/types"
 )
 
-func TestCustomPluginConfigApplyConfiguration(t *testing.T) {
+type applyConfigurationTestCase struct {
+	Orig              CustomPluginConfig
+	Wanted            CustomPluginConfig
+	ErrorMessageStart string
+}
+
+func applyConfigurationTestCases() map[string]applyConfigurationTestCase {
 	globalTimeout := 6 * time.Second
 	globalTimeoutString := globalTimeout.String()
 	invokeInterval := 31 * time.Second
@@ -37,11 +45,11 @@ func TestCustomPluginConfigApplyConfiguration(t *testing.T) {
 
 	ruleTimeout := 1 * time.Second
 	ruleTimeoutString := ruleTimeout.String()
+	ruleInvokeInterval := 7 * time.Second
+	ruleInvokeIntervalString := ruleInvokeInterval.String()
+	invalidRuleInvokeIntervalString := "invalid"
 
-	utMetas := map[string]struct {
-		Orig   CustomPluginConfig
-		Wanted CustomPluginConfig
-	}{
+	return map[string]applyConfigurationTestCase{
 		"global default settings": {
 			Orig: CustomPluginConfig{
 				Rules: []*CustomRule{
@@ -68,16 +76,61 @@ func TestCustomPluginConfigApplyConfiguration(t *testing.T) {
 				EnableMetricsReporting: &defaultEnableMetricsReporting,
 				Rules: []*CustomRule{
 					{
-						Path: "../plugin/test-data/ok.sh",
+						Path:                 "../plugin/test-data/ok.sh",
+						InvokeIntervalString: nil,
+						InvokeInterval:       nil,
 					},
 					{
-						Path:          "../plugin/test-data/warning.sh",
-						Timeout:       &ruleTimeout,
-						TimeoutString: &ruleTimeoutString,
+						Path:                 "../plugin/test-data/warning.sh",
+						Timeout:              &ruleTimeout,
+						TimeoutString:        &ruleTimeoutString,
+						InvokeIntervalString: nil,
+						InvokeInterval:       nil,
 					},
 				},
 			},
 		},
+		"custom rule invoke interval": {
+			Orig: CustomPluginConfig{
+				Rules: []*CustomRule{
+					{
+						Path:                 "../plugin/test-data/ok.sh",
+						InvokeIntervalString: &ruleInvokeIntervalString,
+					},
+				},
+			},
+			Wanted: CustomPluginConfig{
+				PluginGlobalConfig: pluginGlobalConfig{
+					InvokeIntervalString:                    &defaultInvokeIntervalString,
+					InvokeInterval:                          &defaultInvokeInterval,
+					TimeoutString:                           &defaultGlobalTimeoutString,
+					Timeout:                                 &defaultGlobalTimeout,
+					MaxOutputLength:                         &defaultMaxOutputLength,
+					Concurrency:                             &defaultConcurrency,
+					EnableMessageChangeBasedConditionUpdate: &defaultMessageChangeBasedConditionUpdate,
+					SkipInitialStatus:                       &defaultSkipInitialStatus,
+				},
+				EnableMetricsReporting: &defaultEnableMetricsReporting,
+				Rules: []*CustomRule{
+					{
+						Path:                 "../plugin/test-data/ok.sh",
+						InvokeIntervalString: &ruleInvokeIntervalString,
+						InvokeInterval:       &ruleInvokeInterval,
+					},
+				},
+			},
+		},
+		"invalid rule invoke interval": {
+			Orig: CustomPluginConfig{
+				Rules: []*CustomRule{
+					{
+						Path:                 "../plugin/test-data/ok.sh",
+						InvokeIntervalString: &invalidRuleInvokeIntervalString,
+					},
+				},
+			},
+			ErrorMessageStart: "error in parsing rule invoke interval",
+		},
 		"custom invoke interval": {
 			Orig: CustomPluginConfig{
 				PluginGlobalConfig: pluginGlobalConfig{
@@ -217,9 +270,25 @@ func TestCustomPluginConfigApplyConfiguration(t *testing.T) {
 			},
 		},
 	}
+}
 
-	for desp, utMeta := range utMetas {
-		if err := (&utMeta.Orig).ApplyConfiguration(); err != nil {
+func TestCustomPluginConfigApplyConfiguration(t *testing.T) {
+	for desp, utMeta := range applyConfigurationTestCases() {
+		err := (&utMeta.Orig).ApplyConfiguration()
+		if utMeta.ErrorMessageStart != "" {
+			if err == nil {
+				t.Errorf("Error in apply configuration for %q: wanted an error got nil", desp)
+				continue
+			}
+			if !strings.HasPrefix(err.Error(), utMeta.ErrorMessageStart) {
+				t.Errorf("Error in apply configuration for %q: wanted prefix %q, got %q", desp, utMeta.ErrorMessageStart, err)
+			}
+			if !strings.Contains(err.Error(), fmt.Sprintf("%+v", utMeta.Orig.Rules[0])) {
+				t.Errorf("Error in apply configuration for %q does not include rule %+v: %v", desp, utMeta.Orig.Rules[0], err)
+			}
+			continue
+		}
+		if err != nil {
 			t.Errorf("Error in apply configuration for %q: %v", desp, err)
 		}
 		if !reflect.DeepEqual(utMeta.Orig, utMeta.Wanted) {
@@ -232,10 +301,14 @@ func TestCustomPluginConfigApplyConfiguration(t *testing.T) {
 func TestCustomPluginConfigValidate(t *testing.T) {
 	normalRuleTimeout := defaultGlobalTimeout - 1*time.Second
 	exceededRuleTimeout := defaultGlobalTimeout + 1*time.Second
+	zeroInvokeInterval := time.Duration(0)
+	negativeInvokeInterval := -1 * time.Second
 
 	utMetas := map[string]struct {
-		Conf    CustomPluginConfig
-		IsError bool
+		Conf              CustomPluginConfig
+		IsError           bool
+		ErrorContains     string
+		ErrorIncludesRule bool
 	}{
 		"normal": {
 			Conf: CustomPluginConfig{
@@ -311,6 +384,59 @@ func TestCustomPluginConfigValidate(t *testing.T) {
 			},
 			IsError: true,
 		},
+		"zero rule invoke interval": {
+			Conf: CustomPluginConfig{
+				Plugin: customPluginName,
+				PluginGlobalConfig: pluginGlobalConfig{
+					InvokeInterval:  &defaultInvokeInterval,
+					Timeout:         &defaultGlobalTimeout,
+					MaxOutputLength: &defaultMaxOutputLength,
+					Concurrency:     &defaultConcurrency,
+				},
+				Rules: []*CustomRule{
+					{
+						Path:           "../plugin/test-data/ok.sh",
+						InvokeInterval: &zeroInvokeInterval,
+					},
+				},
+			},
+			IsError:           true,
+			ErrorContains:     "Rule:",
+			ErrorIncludesRule: true,
+		},
+		"negative rule invoke interval": {
+			Conf: CustomPluginConfig{
+				Plugin: customPluginName,
+				PluginGlobalConfig: pluginGlobalConfig{
+					InvokeInterval:  &defaultInvokeInterval,
+					Timeout:         &defaultGlobalTimeout,
+					MaxOutputLength: &defaultMaxOutputLength,
+					Concurrency:     &defaultConcurrency,
+				},
+				Rules: []*CustomRule{
+					{
+						Path:           "../plugin/test-data/ok.sh",
+						InvokeInterval: &negativeInvokeInterval,
+					},
+				},
+			},
+			IsError:           true,
+			ErrorContains:     "Rule:",
+			ErrorIncludesRule: true,
+		},
+		"zero global invoke interval": {
+			Conf: CustomPluginConfig{
+				Plugin: customPluginName,
+				PluginGlobalConfig: pluginGlobalConfig{
+					InvokeInterval:  &zeroInvokeInterval,
+					Timeout:         &defaultGlobalTimeout,
+					MaxOutputLength: &defaultMaxOutputLength,
+					Concurrency:     &defaultConcurrency,
+				},
+			},
+			IsError:       true,
+			ErrorContains: "global invoke interval",
+		},
 		"permanent problem has preset default condition": {
 			Conf: CustomPluginConfig{
 				Plugin: customPluginName,
@@ -372,5 +498,11 @@ func TestCustomPluginConfigValidate(t *testing.T) {
 			t.Error(desp)
 			t.Errorf("Error in validating custom plugin configuration %+v. Wanted an error got nil", utMeta)
 		}
+		if err != nil && utMeta.ErrorContains != "" && !strings.Contains(err.Error(), utMeta.ErrorContains) {
+			t.Errorf("Error in validating %q: wanted error containing %q, got %q", desp, utMeta.ErrorContains, err)
+		}
+		if err != nil && utMeta.ErrorIncludesRule && !strings.Contains(err.Error(), fmt.Sprintf("%+v", utMeta.Conf.Rules[0])) {
+			t.Errorf("Error in validating %q does not include rule %+v: %v", desp, utMeta.Conf.Rules[0], err)
+		}
 	}
 }

pkg/custompluginmonitor/types/types.go

@@ -55,5 +55,8 @@ type CustomRule struct {
 	TimeoutString *string `json:"timeout"`
 	// Timeout is the timeout for the custom plugin to execute.
 	Timeout *time.Duration `json:"-"`
-	// TODO(andyxning) Add support for per-rule interval.
+	// InvokeIntervalString is the interval string at which the plugin will be invoked.
+	InvokeIntervalString *string `json:"invoke_interval,omitempty"`
+	// InvokeInterval is the interval at which the plugin will be invoked.
+	InvokeInterval *time.Duration `json:"-"`
 }