fix(go.d/dyncfg): remove wait-decision timeout and make handoff non-droppable (#22201)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Ilya Mashchenko committed
Apr 14, 2026 at 00:39 UTC
30e29e578d7b91c5f790e031ebe8c1d3f9d4c592
14 files changed
+313
-293
src/go/plugin/agent/discovery/sd/dyncfg_handoff.go
+11
-22
@@ -8,33 +8,22 @@ import (
8
"github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
9
)
10
11
-const sdBusyMsg = "Service discovery is busy, try again later."
11
+const sdShuttingDownMsg = "Service discovery is shutting down."
12
13
+// enqueueDyncfgFunction blocks until the function is accepted by the run loop
14
+// or service discovery shuts down. We deliberately do NOT honor a per-function
15
+// timeout here: dropping an awaited enable/disable would wedge the wait gate
16
+// (since waitDecisionTimeout was removed) because the caller would keep
17
+// waiting for a decision that can never be produced. Back-pressure flows
18
+// upstream to netdata via the OS pipe.
19
func (d *ServiceDiscovery) enqueueDyncfgFunction(fn dyncfg.Function) {
14
- handoffCtx, cancel := d.dyncfgHandoffContext(fn)
15
- defer cancel()
16
-
17
- switch dyncfg.BoundedSend(handoffCtx, d.dyncfgCh, fn, dyncfg.DefaultDownstreamHandoffCap) {
18
- case dyncfg.BoundedSendOK:
19
- return
20
- case dyncfg.BoundedSendContextDone:
21
- if d.ctx != nil && d.ctx.Err() != nil {
22
- d.dyncfgApi.SendCodef(fn, 503, "Service discovery is shutting down.")
23
- return
24
- }
25
- d.dyncfgApi.SendCodef(fn, 503, sdBusyMsg)
26
- case dyncfg.BoundedSendTimeout:
27
- d.dyncfgApi.SendCodef(fn, 503, sdBusyMsg)
28
- }
29
-}
30
-
31
-func (d *ServiceDiscovery) dyncfgHandoffContext(fn dyncfg.Function) (context.Context, context.CancelFunc) {
20
ctx := d.ctx
21
if ctx == nil {
22
ctx = context.Background()
23
}
36
- if timeout := fn.Fn().Timeout; timeout > 0 {
37
- return context.WithTimeout(ctx, timeout)
24
+ select {
25
+ case d.dyncfgCh <- fn:
26
+ case <-ctx.Done():
27
+ d.dyncfgApi.SendCodef(fn, 503, sdShuttingDownMsg)
28
}
39
- return ctx, func() {}
29
}
src/go/plugin/agent/discovery/sd/sd.go
-12
@@ -8,7 +8,6 @@ import (
8
"io"
9
"log/slog"
10
"sync"
11
- "time"
11
12
"github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
13
"github.com/netdata/netdata/go/plugins/plugin/agent/policy"
@@ -21,8 +20,6 @@ import (
20
"github.com/netdata/netdata/go/plugins/pkg/netdataapi"
21
)
22
24
-const waitDecisionTimeout = 5 * time.Second
25
-
23
type Config struct {
24
ConfigDefaults confgroup.Registry
25
PluginName string
@@ -76,7 +73,6 @@ func NewServiceDiscovery(cfg Config) (*ServiceDiscovery, error) {
73
WaitKey: func(cfg sdConfig) string {
74
return cfg.PipelineKey()
75
},
79
- WaitTimeout: waitDecisionTimeout,
76
77
Path: fmt.Sprintf(dyncfgSDPath, cfg.PluginName),
78
EnableFailCode: 422,
@@ -184,14 +180,6 @@ func (d *ServiceDiscovery) run(ctx context.Context) {
180
d.dyncfgSeqExec(step.Command)
181
continue
182
}
187
- if step.TimedOut {
188
- d.Errorf(
189
- "dyncfg: timed out waiting for enable/disable decision for '%s' (elapsed=%s threshold=%s); keeping status 'accepted' and continuing",
190
- step.Timeout.Key,
191
- step.Timeout.Elapsed,
192
- step.Timeout.Threshold,
193
- )
194
- }
183
} else {
184
select {
185
case <-ctx.Done():
src/go/plugin/agent/discovery/sd/wait_decision_test.go
+31
-49
@@ -24,28 +24,9 @@ import (
24
25
func TestServiceDiscovery_Run_WaitDecision(t *testing.T) {
26
tests := map[string]struct {
27
- waitTimeout time.Duration
28
- run func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func())
27
+ run func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func())
28
}{
30
- "timeout clears wait gate and keeps accepted state": {
31
- waitTimeout: 40 * time.Millisecond,
32
- run: func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func()) {
33
- cfg := prepareConfigFile("/etc/netdata/sd.d/job1.conf", "job1")
34
- confCh <- cfg
35
-
36
- require.Eventually(t, sd.handler.WaitingForDecision, time.Second, 10*time.Millisecond)
37
- require.Eventually(t, func() bool { return !sd.handler.WaitingForDecision() }, time.Second, 10*time.Millisecond)
38
-
39
- stop()
40
-
41
- entry, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job1")
42
- require.True(t, ok, "expected discovered config to stay exposed after wait timeout")
43
- assert.Equal(t, dyncfg.StatusAccepted, entry.Status)
44
- assert.False(t, sd.mgr.IsRunning(pipelineKeyFromSource(cfg.source)))
45
- },
46
- },
47
- "enable command before timeout clears wait and starts pipeline": {
48
- waitTimeout: 750 * time.Millisecond,
29
+ "enable command clears wait and starts pipeline": {
30
run: func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func()) {
31
cfg := prepareConfigFile("/etc/netdata/sd.d/job1.conf", "job1")
32
confCh <- cfg
@@ -70,9 +51,13 @@ func TestServiceDiscovery_Run_WaitDecision(t *testing.T) {
51
assert.Equal(t, dyncfg.StatusRunning, entry.Status)
52
},
53
},
73
- "timeout unblocks and next config is processed": {
74
- waitTimeout: 40 * time.Millisecond,
54
+ "second config blocks while wait gate is open and proceeds after decision": {
55
run: func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func()) {
56
+ // Without a wait-decision timeout, the run loop must stay in
57
+ // WaitingForDecision until an explicit enable/disable for cfg1
58
+ // arrives. Sending cfg2 in the meantime must block until the
59
+ // gate clears. This guards against accidental gate-clearing or
60
+ // a regression that lets new configs interleave.
61
cfg1 := prepareConfigFile("/etc/netdata/sd.d/job1.conf", "job1")
62
cfg2 := prepareConfigFile("/etc/netdata/sd.d/job2.conf", "job2")
63
@@ -85,43 +70,41 @@ func TestServiceDiscovery_Run_WaitDecision(t *testing.T) {
70
close(secondSent)
71
}()
72
73
+ // Give the goroutine a chance to either block (expected) or
74
+ // race ahead. We can't use require.Eventually for negative
75
+ // "still blocked" assertions, but a short window is enough to
76
+ // catch a regression that lets the second config flow through
77
+ // while the wait gate is still open.
78
select {
79
case <-secondSent:
90
- t.Fatalf("second config should block while wait gate is active")
91
- case <-time.After(20 * time.Millisecond):
80
+ t.Fatal("second config was processed while wait gate was open")
81
+ case <-time.After(100 * time.Millisecond):
82
}
83
+ require.True(t, sd.handler.WaitingForDecision(), "wait gate should still be open before decision")
84
94
- require.Eventually(t, func() bool { return !sd.handler.WaitingForDecision() }, time.Second, 10*time.Millisecond)
95
- require.Eventually(t, func() bool {
96
- select {
97
- case <-secondSent:
98
- return true
99
- default:
100
- return false
101
- }
102
- }, time.Second, 10*time.Millisecond)
85
+ // Send the matching enable for cfg1 — this clears the wait gate.
86
+ sd.dyncfgCh <- dyncfg.NewFunction(functions.Function{
87
+ UID: "enable-job1",
88
+ Args: []string{sd.dyncfgJobID(testDiscovererTypeNetListeners, "job1"), "enable"},
89
+ })
90
+
91
+ select {
92
+ case <-secondSent:
93
+ case <-time.After(2 * time.Second):
94
+ t.Fatal("second config did not proceed after wait gate cleared")
95
+ }
96
97
require.Eventually(t, func() bool {
105
- ok1 := exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job1")
106
- ok2 := exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job2")
107
- return ok1 && ok2
98
+ return exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job1") &&
99
+ exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job2")
100
}, time.Second, 10*time.Millisecond)
109
-
110
- stop()
111
-
112
- entry1, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job1")
113
- require.True(t, ok)
114
- assert.Equal(t, dyncfg.StatusAccepted, entry1.Status)
115
- entry2, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job2")
116
- require.True(t, ok)
117
- assert.Equal(t, dyncfg.StatusAccepted, entry2.Status)
101
},
102
},
103
}
104
105
for name, tc := range tests {
106
t.Run(name, func(t *testing.T) {
124
- sd, confCh, cancel, done := newWaitTestServiceDiscovery(t, tc.waitTimeout)
107
+ sd, confCh, cancel, done := newWaitTestServiceDiscovery(t)
108
stopped := false
109
stop := func() {
110
if stopped {
@@ -136,7 +119,7 @@ func TestServiceDiscovery_Run_WaitDecision(t *testing.T) {
119
}
120
}
121
139
-func newWaitTestServiceDiscovery(t *testing.T, waitTimeout time.Duration) (*ServiceDiscovery, chan confFile, context.CancelFunc, <-chan struct{}) {
122
+func newWaitTestServiceDiscovery(t *testing.T) (*ServiceDiscovery, chan confFile, context.CancelFunc, <-chan struct{}) {
123
t.Helper()
124
125
var out bytes.Buffer
@@ -166,7 +149,6 @@ func newWaitTestServiceDiscovery(t *testing.T, waitTimeout time.Duration) (*Serv
149
WaitKey: func(cfg sdConfig) string {
150
return cfg.PipelineKey()
151
},
169
- WaitTimeout: waitTimeout,
152
153
Path: fmt.Sprintf(dyncfgSDPath, testPluginName),
154
EnableFailCode: 422,
src/go/plugin/agent/jobmgr/dyncfg_collector_test.go
-1
@@ -816,7 +816,6 @@ func newCollectorTestHandler(mgr *Manager, cb dyncfg.Callbacks[confgroup.Config]
816
WaitKey: func(cfg confgroup.Config) string {
817
return cfg.FullName()
818
},
819
- WaitTimeout: waitDecisionTimeout,
819
Path: "/collectors/test/Jobs",
820
EnableFailCode: 200,
821
RemoveStockOnEnableFail: true,
src/go/plugin/agent/jobmgr/dyncfg_handoff.go
+13
-28
@@ -3,38 +3,23 @@
3
package jobmgr
4
5
import (
6
- "context"
7
-
6
"github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
7
)
8
11
-const (
12
- dyncfgBusyMsg = "Job manager is busy, try again later."
13
- dyncfgShuttingDownMsg = "Job manager is shutting down."
14
-)
9
+const dyncfgShuttingDownMsg = "Job manager is shutting down."
10
11
+// enqueueDyncfgFunction blocks until the function is accepted by the run loop
12
+// or the manager shuts down. We deliberately do NOT honor a per-function
13
+// timeout here: dropping an awaited enable/disable would wedge jobmgr's wait
14
+// gate (since waitDecisionTimeout was removed). Back-pressure flows upstream:
15
+// dyncfgCh full -> framework worker blocks here -> scheduler fills ->
16
+// dispatchInvocation blocks -> stdin reader pauses -> netdata's pipe write
17
+// blocks. This is intentional so awaited state transitions preserve ordering
18
+// and eventually slow the producer instead of being dropped.
19
func (m *Manager) enqueueDyncfgFunction(fn dyncfg.Function) {
17
- handoffCtx, cancel := m.dyncfgHandoffContext(fn)
18
- defer cancel()
19
-
20
- switch dyncfg.BoundedSend(handoffCtx, m.dyncfgCh, fn, dyncfg.DefaultDownstreamHandoffCap) {
21
- case dyncfg.BoundedSendOK:
22
- return
23
- case dyncfg.BoundedSendContextDone:
24
- if m.baseContext().Err() != nil {
25
- m.dyncfgResponder.SendCodef(fn, 503, dyncfgShuttingDownMsg)
26
- return
27
- }
28
- m.dyncfgResponder.SendCodef(fn, 503, dyncfgBusyMsg)
29
- case dyncfg.BoundedSendTimeout:
30
- m.dyncfgResponder.SendCodef(fn, 503, dyncfgBusyMsg)
31
- }
32
-}
33
-
34
-func (m *Manager) dyncfgHandoffContext(fn dyncfg.Function) (context.Context, context.CancelFunc) {
35
- ctx := m.baseContext()
36
- if timeout := fn.Fn().Timeout; timeout > 0 {
37
- return context.WithTimeout(ctx, timeout)
20
+ select {
21
+ case m.dyncfgCh <- fn:
22
+ case <-m.baseContext().Done():
23
+ m.dyncfgResponder.SendCodef(fn, 503, dyncfgShuttingDownMsg)
24
}
39
- return ctx, func() {}
25
}
src/go/plugin/agent/jobmgr/manager.go
-10
@@ -50,7 +50,6 @@ type Config struct {
50
}
51
52
const (
53
- waitDecisionTimeout = 5 * time.Second
53
cmdTestWorkerCap = 4
54
cmdTestDefaultTimeout = 60 * time.Second
55
cmdTestWorkerDrainWait = 5 * time.Second
@@ -136,7 +135,6 @@ func New(cfg Config) *Manager {
135
WaitKey: func(cfg confgroup.Config) string {
136
return cfg.FullName()
137
},
139
- WaitTimeout: waitDecisionTimeout,
138
139
Path: fmt.Sprintf(dyncfgCollectorPath, cfg.PluginName),
140
EnableFailCode: 200,
@@ -334,14 +332,6 @@ func (m *Manager) run() {
332
m.dyncfgSeqExec(step.Command)
333
continue
334
}
337
- if step.TimedOut {
338
- m.Errorf(
339
- "dyncfg: timed out waiting for enable/disable decision for '%s' (elapsed=%s threshold=%s); keeping status 'accepted' and continuing",
340
- step.Timeout.Key,
341
- step.Timeout.Elapsed,
342
- step.Timeout.Threshold,
343
- )
344
- }
335
} else {
336
select {
337
case <-m.ctx.Done():
src/go/plugin/agent/jobmgr/manager_process_test.go
-52
@@ -17,7 +17,6 @@ import (
17
"github.com/netdata/netdata/go/plugins/plugin/agent/secrets/secretstore"
18
"github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
19
"github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
20
- "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
20
"github.com/netdata/netdata/go/plugins/plugin/framework/functions"
21
"github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
22
)
@@ -69,57 +68,6 @@ func TestRunProcessConfGroups_ChannelCloseDoesNotSpin(t *testing.T) {
68
}
69
}
70
72
-func TestRun_WaitTimeoutClearsGateAndKeepsAccepted(t *testing.T) {
73
- mgr := New(Config{PluginName: testPluginName})
74
- mgr.modules = prepareMockRegistry()
75
-
76
- ctx, cancel := context.WithCancel(context.Background())
77
- defer cancel()
78
- mgr.ctx = ctx
79
-
80
- done := make(chan struct{})
81
- go func() {
82
- mgr.run()
83
- close(done)
84
- }()
85
- defer func() {
86
- cancel()
87
- select {
88
- case <-done:
89
- case <-time.After(2 * time.Second):
90
- t.Fatal("run did not stop after cancel")
91
- }
92
- }()
93
-
94
- cfg1 := prepareStockCfg("success", "wait1")
95
- cfg2 := prepareStockCfg("success", "wait2")
96
-
97
- mgr.addCh <- cfg1
98
- require.Eventually(t, mgr.collectorHandler.WaitingForDecision, time.Second, 10*time.Millisecond)
99
-
100
- secondSent := make(chan struct{})
101
- go func() {
102
- mgr.addCh <- cfg2
103
- close(secondSent)
104
- }()
105
-
106
- select {
107
- case <-secondSent:
108
- t.Fatal("second add was processed before wait timeout")
109
- case <-time.After(500 * time.Millisecond):
110
- }
111
-
112
- select {
113
- case <-secondSent:
114
- case <-time.After(7 * time.Second):
115
- t.Fatal("second add did not progress after wait timeout")
116
- }
117
-
118
- entry1, ok := mgr.collectorExposed.LookupByKey(cfg1.ExposedKey())
119
- require.True(t, ok, "first config must stay exposed after timeout")
120
- assert.Equal(t, dyncfg.StatusAccepted, entry1.Status)
121
-}
122
-
71
func TestRunNotifyRunningJobs_TickOutsideLock(t *testing.T) {
72
mgr := New(Config{PluginName: testPluginName})
73
src/go/plugin/framework/functions/README.md
+2
-4
@@ -57,7 +57,7 @@ Admission checks:
57
- manager stopping -> reject `503`
58
- unknown/nil handler -> reject `501`
59
- duplicate active/tombstoned UID -> ignore duplicate input (debug/warn log, no terminal output)
60
-- queue full -> reject `503`
60
+- queue full -> blocks on `scheduler.enqueue` until space frees (back-pressures stdin reader -> netdata via OS pipe). The only errors returned from this path are manager-stopping (`503`) on shutdown and invalid-request (`500`) for malformed input.
61
62
### Keyed scheduler + worker pool
63
@@ -123,7 +123,6 @@ Pathology-focused metrics currently exposed:
123
- `netdata.go.plugin.framework.functions.manager.invocations_awaiting_result`
124
- `netdata.go.plugin.framework.functions.manager.scheduler_pending`
125
- counters:
126
- - `netdata.go.plugin.framework.functions.manager.queue_full_total`
126
- `netdata.go.plugin.framework.functions.manager.cancel_fallback_total`
127
- `netdata.go.plugin.framework.functions.manager.late_terminal_dropped_total`
128
- `netdata.go.plugin.framework.functions.manager.duplicate_uid_ignored_total`
@@ -188,8 +187,7 @@ flowchart TD
187
C1 -->|stopping| R503["respf 503"]
188
C1 -->|unregistered/nil handler| R501["respf 501"]
189
C1 -->|duplicate/tombstoned UID| DUP["ignore duplicate + log"]
191
- C1 -->|queue full| R503
192
- C1 -->|accepted| Q["scheduler.enqueue by route key + state=queued"]
190
+ C1 -->|accepted| Q["scheduler.enqueue by route key + state=queued (blocks if queue full)"]
191
Q --> S["keyScheduler"]
192
S -->|same key busy| SQ["lane queue (serialized)"]
193
S -->|key free| W["Worker"]
src/go/plugin/framework/functions/manager.go
+103
-20
@@ -33,8 +33,18 @@ const (
33
// queue/worker logic.
34
// TODO: establish and document a MethodHandler goroutine-safety contract
35
// before increasing this default.
36
- defaultWorkerCount = 1
37
- defaultQueueSize = 64
36
+ defaultWorkerCount = 1
37
+ // defaultQueueSize is intentionally 1 (not removed: the keyed scheduler is
38
+ // still the dispatch primitive, we just don't want it to absorb bursts).
39
+ // Rationale: every downstream stage is single-threaded today (1 worker, 1
40
+ // jobmgr loop, serial Check()), so admitting more than 1 extra request
41
+ // only buys wedge surface (a queued request that is later cancelled by
42
+ // netdata cannot reach jobmgr, so the dyncfg wait gate stays in
43
+ // 'accepted' forever). With queue=1 the stdin reader back-pressures
44
+ // earlier through the OS pipe instead of
45
+ // piling up admitted-but-unprocessed work in stateQueued. If/when we add
46
+ // real downstream concurrency, raise this again.
47
+ defaultQueueSize = 1
48
defaultCancelFallbackDelay = 5 * time.Second
49
defaultShutdownDrainTimeout = 8 * time.Second
50
defaultTombstoneTTL = 60 * time.Second
@@ -151,6 +161,22 @@ func (m *Manager) run(ctx context.Context, quitCh chan struct{}) {
161
162
m.startWorkers(&workersWG)
163
164
+ // Wake any enqueue() blocked on a full scheduler when the parent context
165
+ // is canceled. Necessary because the reader loop below may itself be
166
+ // blocked inside dispatchInvocation -> scheduler.enqueue waiting for
167
+ // space, and would otherwise miss the ctx.Done signal.
168
+ stopWatcher := make(chan struct{})
169
+ defer close(stopWatcher)
170
+ go func() {
171
+ select {
172
+ case <-ctx.Done():
173
+ if m.scheduler != nil {
174
+ m.scheduler.stop()
175
+ }
176
+ case <-stopWatcher:
177
+ }
178
+ }()
179
+
180
for {
181
select {
182
case <-ctx.Done():
@@ -311,18 +337,24 @@ func (m *Manager) dispatchInvocation(parentCtx context.Context, fn *Function) {
337
return
338
}
339
340
+ // scheduler.enqueue blocks when the queue is full; it only returns an
341
+ // error for invalid inputs or when the scheduler is stopping (shutdown).
342
+ // The blocking is intentional: by stalling here, the stdin reader slows
343
+ // down as well, which propagates back-pressure to netdata through the OS
344
+ // pipe instead of allowing unbounded buffering or dropping requests.
345
if err := m.scheduler.enqueue(req); err != nil {
346
cancel()
347
switch {
317
- case errors.Is(err, errSchedulerQueueFull):
318
- m.respf(fn, 503, "function queue is full")
319
- m.observeQueueFull()
348
case errors.Is(err, errSchedulerStopping):
349
m.respf(fn, 503, "functions manager is stopping")
350
case errors.Is(err, errSchedulerInvalid):
351
m.respf(fn, 500, "invalid scheduler request")
352
default:
325
- m.respf(fn, 503, "function queue is full")
353
+ // Should be unreachable: enqueue's other return points block.
354
+ // If a new error variant is added in the future, surface it
355
+ // loudly instead of swallowing it as a generic 5xx.
356
+ m.Warningf("unexpected scheduler enqueue error for '%s': %v", fn.Name, err)
357
+ m.respf(fn, 500, "unexpected scheduler error: %v", err)
358
}
359
return
360
}
@@ -419,14 +451,15 @@ func (m *Manager) handleCancelEvent(event inputEvent) {
451
return
452
}
453
422
- state, ok := m.requestCancellation(uid)
423
- if !ok {
454
+ if _, ok := m.requestCancellation(uid); !ok {
455
m.Debugf("ignoring cancel for unknown transaction id: %s", uid)
456
return
457
}
427
- if state == stateQueued {
428
- m.respUID(uid, 499, "request canceled")
429
- }
458
+ // No immediate terminal response. requestCancellation handled state
459
+ // transitions: for queued functions the cancel is intentionally ignored
460
+ // (function will run to completion); for running/awaiting functions a
461
+ // fallback timer was armed and will emit 499 + tombstone after
462
+ // cancelFallbackDelay.
463
}
464
465
func (m *Manager) requestCancellation(uid string) (invocationState, bool) {
@@ -438,6 +471,22 @@ func (m *Manager) requestCancellation(uid string) (invocationState, bool) {
471
return 0, false
472
}
473
474
+ // For stateQueued we deliberately ignore the cancel entirely: no
475
+ // cancelRequested flag, no ctx cancellation, no fallback timer, no
476
+ // tombstone. Reason: dyncfg commands (enable/disable/update/restart)
477
+ // carry side-effects that must reach jobmgr, otherwise the wait gate
478
+ // stays in 'accepted' forever (since the wait-decision timeout was
479
+ // removed). Setting cancelRequested would make startInvocation skip the
480
+ // handler when the worker pulls; the fallback timer's tryFinalize would
481
+ // also tombstone+remove from invState before the worker gets there. So
482
+ // either path wedges the function. We let it run to completion as if
483
+ // nothing happened; netdata already considers the transaction done (it
484
+ // 504'd before sending CANCEL), so the eventual terminal response just
485
+ // produces a benign "transaction not found" log on the netdata side.
486
+ if rec.state == stateQueued {
487
+ return rec.state, true
488
+ }
489
+
490
if rec.cancelRequested {
491
return rec.state, true
492
}
@@ -446,14 +495,7 @@ func (m *Manager) requestCancellation(uid string) (invocationState, bool) {
495
rec.cancel()
496
}
497
449
- if rec.state == stateQueued && m.scheduler != nil {
450
- m.scheduler.cancelQueued(rec.scheduleKey, uid)
451
- m.observeSchedulerPending()
452
- }
453
-
454
- if rec.state == stateRunning || rec.state == stateAwaitingResult {
455
- m.startCancelFallbackTimerLocked(uid, rec)
456
- }
498
+ m.startCancelFallbackTimerLocked(uid, rec)
499
500
return rec.state, true
501
}
@@ -509,6 +551,42 @@ func (m *Manager) forceFinalizeAll(code int, message string) {
551
}
552
}
553
554
+// markCancelled performs the same bookkeeping as tryFinalize (tombstone +
555
+// scheduler.complete + remove from invState) but does NOT emit anything to
556
+// netdata. Used by the cancel fallback path: by the time CANCEL is sent,
557
+// netdata has already timed out the transaction and removed its inflight
558
+// entry, so any response we emit just produces a "transaction not found"
559
+// log on the netdata side. We still need the bookkeeping so the lane
560
+// advances and any later terminal response from the handler is dropped.
561
+func (m *Manager) markCancelled(uid string) {
562
+ if uid == "" {
563
+ return
564
+ }
565
+
566
+ m.invStateMux.Lock()
567
+ now := time.Now()
568
+ m.pruneExpiredTombstonesLocked(now)
569
+ if _, ok := m.tombstones[uid]; ok {
570
+ m.invStateMux.Unlock()
571
+ return
572
+ }
573
+
574
+ var scheduleKey string
575
+ if rec, ok := m.invState[uid]; ok && rec != nil {
576
+ m.stopTimersLocked(rec)
577
+ scheduleKey = rec.scheduleKey
578
+ }
579
+ delete(m.invState, uid)
580
+ m.tombstones[uid] = now.Add(m.tombstoneTTL)
581
+ m.observeInvocationsLocked()
582
+ m.invStateMux.Unlock()
583
+
584
+ if scheduleKey != "" && m.scheduler != nil {
585
+ m.scheduler.complete(scheduleKey, uid)
586
+ m.observeSchedulerPending()
587
+ }
588
+}
589
+
590
// tryFinalize emits a terminal response once per transaction UID.
591
// Later terminal attempts for the same UID are dropped while tombstone is active.
592
func (m *Manager) tryFinalize(uid, source string, emit func()) bool {
@@ -587,7 +665,12 @@ func (m *Manager) startCancelFallbackTimerLocked(uid string, rec *invocationReco
665
uidCopy := uid
666
rec.fallbackTimer = time.AfterFunc(m.cancelFallbackDelay, func() {
667
m.observeCancelFallback()
590
- m.respUID(uidCopy, 499, "request canceled")
668
+ // Don't emit a 499: by the time CANCEL was sent, netdata has already
669
+ // 504'd the transaction and removed its inflight entry, so any
670
+ // response we send just produces a "transaction not found" log.
671
+ // markCancelled does the bookkeeping (tombstone + lane advance) so
672
+ // the late real response from the handler is dropped.
673
+ m.markCancelled(uidCopy)
674
})
675
}
676
src/go/plugin/framework/functions/manager_flow_test.go
+52
-40
@@ -114,14 +114,22 @@ func TestManager_FlowScenarios(t *testing.T) {
114
tests := map[string]struct {
115
run func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer)
116
}{
117
- "queued cancel emits 499 and skips execution": {
117
+ "queued cancel is ignored; function still executes": {
118
run: func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer) {
119
+ // New semantics: a CANCEL for a queued function is a no-op at
120
+ // the framework level. The function stays in the scheduler,
121
+ // runs to completion when its turn comes, and any terminal
122
+ // response goes through normally. No 499 is emitted by the
123
+ // framework (netdata already considers the transaction done
124
+ // via its own per-transaction timer, so an extra 499 would
125
+ // just be noise).
126
var (
127
mu sync.Mutex
128
executed []string
129
)
130
started := make(chan struct{}, 1)
131
release := make(chan struct{})
132
+ tx2Done := make(chan struct{})
133
134
mgr.Register("fn", func(fn Function) {
135
mu.Lock()
@@ -130,6 +138,14 @@ func TestManager_FlowScenarios(t *testing.T) {
138
if fn.UID == "tx1" {
139
started <- struct{}{}
140
<-release
141
+ // Emit terminal response so the per-key lane advances
142
+ // to tx2; otherwise the lane stays "owned" by tx1 and
143
+ // tx2 never gets picked up.
144
+ mgr.respUID(fn.UID, 200, "ok")
145
+ }
146
+ if fn.UID == "tx2" {
147
+ mgr.respUID(fn.UID, 200, "ok")
148
+ close(tx2Done)
149
}
150
})
151
@@ -140,18 +156,38 @@ func TestManager_FlowScenarios(t *testing.T) {
156
<-started
157
in.ch <- functionLine("tx2", "fn")
158
in.ch <- "FUNCTION_CANCEL tx2"
159
+
160
+ // tx2 must NOT receive a 499 from the framework. We can't
161
+ // assert "no 499 ever" without waiting forever, but we can at
162
+ // least verify it isn't there immediately after CANCEL.
163
+ time.Sleep(50 * time.Millisecond)
164
+ assert.NotContains(t, out.String(), "FUNCTION_RESULT_BEGIN tx2 499")
165
+
166
close(release)
167
+ select {
168
+ case <-tx2Done:
169
+ case <-time.After(time.Second):
170
+ t.Fatal("tx2 did not execute after release")
171
+ }
172
+
173
close(in.ch)
174
waitForDone(t, done)
175
147
- waitForSubstring(t, out.String, "FUNCTION_RESULT_BEGIN tx2 499", time.Second)
176
+ // Both tx1 and tx2 should have executed (in order).
177
mu.Lock()
178
defer mu.Unlock()
150
- assert.Equal(t, []string{"tx1"}, executed)
179
+ assert.Equal(t, []string{"tx1", "tx2"}, executed)
180
+ // tx2's normal 200 response went through (no tombstone).
181
+ assert.Contains(t, out.String(), "FUNCTION_RESULT_BEGIN tx2 200")
182
},
183
},
153
- "running cancel fallback emits 499 once": {
184
+ "running cancel fallback tombstones silently (no emit)": {
185
run: func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer) {
186
+ // New semantics: the cancel fallback timer no longer emits a
187
+ // 499 to netdata. By the time CANCEL arrives, netdata has
188
+ // already 504'd and removed the inflight transaction, so any
189
+ // response would just produce a "transaction not found" log.
190
+ // We only do the bookkeeping (tombstone + lane advance).
191
mgr.cancelFallbackDelay = 50 * time.Millisecond
192
started := make(chan struct{}, 1)
193
release := make(chan struct{})
@@ -169,16 +205,18 @@ func TestManager_FlowScenarios(t *testing.T) {
205
in.ch <- functionLine("tx1", "fn")
206
<-started
207
in.ch <- "FUNCTION_CANCEL tx1"
172
- waitForSubstring(t, out.String, "FUNCTION_RESULT_BEGIN tx1 499", time.Second)
208
+ // Wait long enough for the fallback timer to fire.
209
+ time.Sleep(150 * time.Millisecond)
210
211
close(release)
212
close(in.ch)
213
waitForDone(t, done)
214
178
- assert.Equal(t, 1, strings.Count(out.String(), "FUNCTION_RESULT_BEGIN tx1 499"))
215
+ // No 499 should be emitted, ever.
216
+ assert.Equal(t, 0, strings.Count(out.String(), "FUNCTION_RESULT_BEGIN tx1 499"))
217
},
218
},
181
- "repeated cancel for same uid still emits one terminal response": {
219
+ "repeated cancel for same uid is idempotent and silent": {
220
run: func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer) {
221
mgr.cancelFallbackDelay = 50 * time.Millisecond
222
started := make(chan struct{}, 1)
@@ -198,13 +236,13 @@ func TestManager_FlowScenarios(t *testing.T) {
236
<-started
237
in.ch <- "FUNCTION_CANCEL tx1"
238
in.ch <- "FUNCTION_CANCEL tx1"
201
- waitForSubstring(t, out.String, "FUNCTION_RESULT_BEGIN tx1 499", time.Second)
239
+ time.Sleep(150 * time.Millisecond)
240
241
close(release)
242
close(in.ch)
243
waitForDone(t, done)
244
207
- assert.Equal(t, 1, strings.Count(out.String(), "FUNCTION_RESULT_BEGIN tx1 499"))
245
+ assert.Equal(t, 0, strings.Count(out.String(), "FUNCTION_RESULT_BEGIN tx1 499"))
246
},
247
},
248
"running cancel drops late terminal response": {
@@ -227,14 +265,17 @@ func TestManager_FlowScenarios(t *testing.T) {
265
in.ch <- functionLine("tx1", "fn")
266
<-started
267
in.ch <- "FUNCTION_CANCEL tx1"
230
- waitForSubstring(t, out.String, "FUNCTION_RESULT_BEGIN tx1 499", time.Second)
268
+ // Let the fallback timer fire (tombstones the UID).
269
+ time.Sleep(150 * time.Millisecond)
270
271
close(release)
272
close(in.ch)
273
waitForDone(t, done)
274
275
got := out.String()
237
- assert.Equal(t, 1, strings.Count(got, "FUNCTION_RESULT_BEGIN tx1 499"))
276
+ // No 499 emitted by the fallback path.
277
+ assert.Equal(t, 0, strings.Count(got, "FUNCTION_RESULT_BEGIN tx1 499"))
278
+ // Late 200 from the handler is dropped by the tombstone.
279
assert.Equal(t, 0, strings.Count(got, "FUNCTION_RESULT_BEGIN tx1 200"))
280
},
281
},
@@ -342,35 +383,6 @@ func TestManager_FlowScenarios(t *testing.T) {
383
assert.EqualValues(t, 1, calls.Load())
384
},
385
},
345
- "queue full is rejected with 503": {
346
- run: func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer) {
347
- mgr.queueSize = 1
348
- mgr.workerCount = 1
349
- started := make(chan struct{}, 1)
350
- release := make(chan struct{})
351
-
352
- mgr.Register("fn", func(fn Function) {
353
- if fn.UID == "tx1" {
354
- started <- struct{}{}
355
- <-release
356
- }
357
- mgr.respUID(fn.UID, 200, "ok")
358
- })
359
-
360
- cancel, done := startFlowManager(t, mgr)
361
- defer cancel()
362
-
363
- in.ch <- functionLine("tx1", "fn")
364
- <-started
365
- in.ch <- functionLine("tx2", "fn")
366
- in.ch <- functionLine("tx3", "fn")
367
- waitForSubstring(t, out.String, "FUNCTION_RESULT_BEGIN tx3 503", time.Second)
368
-
369
- close(release)
370
- close(in.ch)
371
- waitForDone(t, done)
372
- },
373
- },
386
"panic in handler emits 500": {
387
run: func(t *testing.T, mgr *Manager, in *chanInput, out *safeBuffer) {
388
mgr.Register("fn", func(Function) { panic("boom") })
src/go/plugin/framework/functions/runtime_metrics.go
-14
@@ -12,7 +12,6 @@ type managerRuntimeMetrics struct {
12
schedulerPending metrix.StatefulGauge
13
14
functionCallsTotal metrix.StatefulCounter
15
- queueFullTotal metrix.StatefulCounter
15
cancelFallbackTotal metrix.StatefulCounter
16
lateTerminalDropped metrix.StatefulCounter
17
duplicateUIDIgnored metrix.StatefulCounter
@@ -49,12 +48,6 @@ func newManagerRuntimeMetrics(store metrix.RuntimeStore) *managerRuntimeMetrics
48
metrix.WithChartFamily("Framework/Functions/Calls"),
49
metrix.WithUnit("calls"),
50
),
52
- queueFullTotal: metrix.SeededCounter(meter,
53
- "queue_full_total",
54
- metrix.WithDescription("Total number of function requests rejected due to queue full"),
55
- metrix.WithChartFamily("Framework/Functions/Failures"),
56
- metrix.WithUnit("requests"),
57
- ),
51
cancelFallbackTotal: metrix.SeededCounter(meter,
52
"cancel_fallback_total",
53
metrix.WithDescription("Total number of function requests finalized by cancel fallback timer"),
@@ -102,13 +95,6 @@ func (m *Manager) observeSchedulerPending() {
95
m.runtimeMetrics.schedulerPending.Set(float64(m.scheduler.pendingCount()))
96
}
97
105
-func (m *Manager) observeQueueFull() {
106
- if m == nil || m.runtimeMetrics == nil {
107
- return
108
- }
109
- m.runtimeMetrics.queueFullTotal.Add(1)
110
-}
111
-
98
func (m *Manager) observeFunctionCall() {
99
if m == nil || m.runtimeMetrics == nil {
100
return
src/go/plugin/framework/functions/runtime_metrics_test.go
+4
-3
@@ -107,7 +107,9 @@ func TestManager_RuntimeMetricsScenarios(t *testing.T) {
107
in.ch <- functionLine("tx1", "fn")
108
<-started
109
in.ch <- functionLine("tx2", "fn")
110
- in.ch <- functionLine("tx3", "fn") // queue-full
110
+ // tx3 (former queue-full step) removed: the scheduler now blocks
111
+ // on full instead of rejecting, so the reader would deadlock here
112
+ // and the FUNCTION_CANCEL below would never be processed.
113
in.ch <- functionLine("tx1", "fn") // duplicate-active
114
in.ch <- "FUNCTION_CANCEL tx1" // fallback->499
115
@@ -135,11 +137,10 @@ func TestManager_RuntimeMetricsScenarios(t *testing.T) {
137
return ok && pending == 0
138
}, "runtime gauges settle after shutdown")
139
138
- assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".queue_full_total", nil), float64(1))
140
assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".cancel_fallback_total", nil), float64(1))
141
assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".late_terminal_dropped_total", nil), float64(1))
142
assert.GreaterOrEqual(t, runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".duplicate_uid_ignored_total", nil), float64(2))
142
- assert.Equal(t, float64(5), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".calls_total", nil))
143
+ assert.Equal(t, float64(4), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".calls_total", nil))
144
145
assert.Equal(t, float64(0), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".invocations_active", nil))
146
assert.Equal(t, float64(0), runtimeMetricValue(t, mgr.runtimeStore, functionsRuntimeMetricPrefix+".invocations_awaiting_result", nil))
src/go/plugin/framework/functions/scheduler.go
+39
-21
@@ -8,9 +8,8 @@ import (
8
)
9
10
var (
11
- errSchedulerQueueFull = errors.New("scheduler queue is full")
12
- errSchedulerStopping = errors.New("scheduler is stopping")
13
- errSchedulerInvalid = errors.New("scheduler invalid request")
11
+ errSchedulerStopping = errors.New("scheduler is stopping")
12
+ errSchedulerInvalid = errors.New("scheduler invalid request")
13
)
14
15
type scheduleLane struct {
@@ -32,6 +31,11 @@ type keyScheduler struct {
31
pending int
32
accepting bool
33
stopping bool
34
+
35
+ // enqueueWaiters counts goroutines currently blocked inside enqueue()
36
+ // waiting for space. Used by tests to synchronize deterministically
37
+ // instead of sleeping.
38
+ enqueueWaiters int
39
}
40
41
func newKeyScheduler(maxPending int) *keyScheduler {
@@ -52,11 +56,22 @@ func (s *keyScheduler) enqueue(req *invocationRequest) error {
56
s.mux.Lock()
57
defer s.mux.Unlock()
58
55
- if s.stopping || !s.accepting {
56
- return errSchedulerStopping
57
- }
58
- if s.maxPending > 0 && s.pending >= s.maxPending {
59
- return errSchedulerQueueFull
59
+ // Block until there is space, or the scheduler is stopped. We must not
60
+ // drop dyncfg commands silently: an awaited enable/disable that is dropped
61
+ // here would wedge jobmgr's wait gate by leaving it waiting for a
62
+ // completion that will never arrive. Back-pressure flows upstream: the
63
+ // manager run-loop stops draining stdin, and netdata's write blocks on
64
+ // the OS pipe.
65
+ for {
66
+ if s.stopping || !s.accepting {
67
+ return errSchedulerStopping
68
+ }
69
+ if s.maxPending <= 0 || s.pending < s.maxPending {
70
+ break
71
+ }
72
+ s.enqueueWaiters++
73
+ s.cond.Wait()
74
+ s.enqueueWaiters--
75
}
76
77
lane := s.lanes[req.scheduleKey]
@@ -97,9 +112,8 @@ func (s *keyScheduler) next() (*invocationRequest, bool) {
112
if s.pending > 0 {
113
s.pending--
114
}
100
- if s.drainedLocked() {
101
- s.cond.Broadcast()
102
- }
115
+ // Wake any enqueue() waiters blocked on a full queue.
116
+ s.cond.Broadcast()
117
return req, true
118
}
119
@@ -130,9 +144,8 @@ func (s *keyScheduler) cancelQueued(scheduleKey, uid string) bool {
144
if lane.ownerUID == "" && len(lane.queue) == 0 {
145
delete(s.lanes, scheduleKey)
146
}
133
- if s.drainedLocked() {
134
- s.cond.Broadcast()
135
- }
147
+ // Wake any enqueue() waiters blocked on a full queue.
148
+ s.cond.Broadcast()
149
return true
150
}
151
return false
@@ -157,9 +170,7 @@ func (s *keyScheduler) complete(scheduleKey, uid string) {
170
171
if s.stopping {
172
delete(s.lanes, scheduleKey)
160
- if s.drainedLocked() {
161
- s.cond.Broadcast()
162
- }
173
+ s.cond.Broadcast()
174
return
175
}
176
@@ -175,7 +186,8 @@ func (s *keyScheduler) complete(scheduleKey, uid string) {
186
187
lane.ownerUID = next.fn.UID
188
s.ready = append(s.ready, next)
178
- s.cond.Signal()
189
+ // Broadcast: wakes both next() consumers and enqueue() producers.
190
+ s.cond.Broadcast()
191
return
192
}
193
@@ -183,9 +195,7 @@ func (s *keyScheduler) complete(scheduleKey, uid string) {
195
if len(lane.queue) == 0 {
196
delete(s.lanes, scheduleKey)
197
}
186
- if s.drainedLocked() {
187
- s.cond.Broadcast()
188
- }
198
+ s.cond.Broadcast()
199
}
200
201
func (s *keyScheduler) stopAccepting() {
@@ -226,3 +236,11 @@ func (s *keyScheduler) pendingCount() int {
236
defer s.mux.Unlock()
237
return s.pending
238
}
239
+
240
+// enqueueWaiterCount reports how many goroutines are currently blocked
241
+// inside enqueue() waiting for queue space. Intended for tests.
242
+func (s *keyScheduler) enqueueWaiterCount() int {
243
+ s.mux.Lock()
244
+ defer s.mux.Unlock()
245
+ return s.enqueueWaiters
246
+}
src/go/plugin/framework/functions/scheduler_test.go
+58
-17
@@ -117,16 +117,6 @@ func TestKeyScheduler_EnqueueValidation(t *testing.T) {
117
adjust: func(s *keyScheduler) { s.stopAccepting() },
118
want: errSchedulerStopping,
119
},
120
- "full scheduler returns queue full error": {
121
- req: &invocationRequest{
122
- fn: &Function{UID: "tx1"},
123
- scheduleKey: "k",
124
- },
125
- adjust: func(s *keyScheduler) {
126
- s.pending = 1
127
- },
128
- want: errSchedulerQueueFull,
129
- },
120
"valid request is admitted": {
121
req: &invocationRequest{
122
fn: &Function{UID: "tx1"},
@@ -196,32 +186,83 @@ func TestKeyScheduler_StopPaths(t *testing.T) {
186
}
187
}
188
199
-func TestKeyScheduler_QueueFullRecovery(t *testing.T) {
189
+func TestKeyScheduler_EnqueueBlocksUntilSpace(t *testing.T) {
190
tests := map[string]struct {
191
run func(t *testing.T)
192
}{
203
- "full queue recovers after dequeue and completion": {
193
+ "enqueue blocks until next() frees space": {
194
run: func(t *testing.T) {
195
s := newKeyScheduler(1)
196
req1 := &invocationRequest{
197
fn: &Function{UID: "tx1"},
208
- scheduleKey: "k",
198
+ scheduleKey: "k1",
199
}
200
req2 := &invocationRequest{
201
fn: &Function{UID: "tx2"},
212
- scheduleKey: "k",
202
+ scheduleKey: "k2",
203
}
204
205
require.NoError(t, s.enqueue(req1))
216
- require.ErrorIs(t, s.enqueue(req2), errSchedulerQueueFull)
206
+
207
+ enqueued := make(chan error, 1)
208
+ go func() {
209
+ enqueued <- s.enqueue(req2)
210
+ }()
211
+
212
+ require.Eventually(t, func() bool {
213
+ return s.enqueueWaiterCount() == 1
214
+ }, time.Second, time.Millisecond, "second enqueue never reached blocking wait")
215
+
216
+ select {
217
+ case <-enqueued:
218
+ t.Fatal("second enqueue should still be blocked while queue is full")
219
+ default:
220
+ }
221
222
got, ok := s.next()
223
require.True(t, ok)
224
require.NotNil(t, got)
225
require.Equal(t, "tx1", got.fn.UID)
226
223
- s.complete("k", "tx1")
224
- require.NoError(t, s.enqueue(req2))
227
+ select {
228
+ case err := <-enqueued:
229
+ require.NoError(t, err)
230
+ case <-time.After(time.Second):
231
+ t.Fatal("second enqueue did not unblock after space freed")
232
+ }
233
+ },
234
+ },
235
+ "stop unblocks waiting enqueue with stopping error": {
236
+ run: func(t *testing.T) {
237
+ s := newKeyScheduler(1)
238
+ req1 := &invocationRequest{
239
+ fn: &Function{UID: "tx1"},
240
+ scheduleKey: "k1",
241
+ }
242
+ req2 := &invocationRequest{
243
+ fn: &Function{UID: "tx2"},
244
+ scheduleKey: "k2",
245
+ }
246
+
247
+ require.NoError(t, s.enqueue(req1))
248
+
249
+ enqueued := make(chan error, 1)
250
+ go func() {
251
+ enqueued <- s.enqueue(req2)
252
+ }()
253
+
254
+ require.Eventually(t, func() bool {
255
+ return s.enqueueWaiterCount() == 1
256
+ }, time.Second, time.Millisecond, "second enqueue never reached blocking wait")
257
+
258
+ s.stop()
259
+
260
+ select {
261
+ case err := <-enqueued:
262
+ require.ErrorIs(t, err, errSchedulerStopping)
263
+ case <-time.After(time.Second):
264
+ t.Fatal("blocked enqueue did not return after stop()")
265
+ }
266
},
267
},
268
}