fix(dyncfg): rollback non-disruptive sd update failures (#21861)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Ilya Mashchenko committed
Mar 2, 2026 at 14:06 UTC
97c7cdf0de04feacd2c6e4b07baca6f261518847
6 files changed
+78
-8
src/go/plugin/agent/discovery/sd/dyncfg.go
+1
-1
@@ -100,7 +100,7 @@ func (cb *sdCallbacks) Start(cfg sdConfig) error {
100
func (cb *sdCallbacks) Update(oldCfg, newCfg sdConfig) error {
101
pipelineCfg, err := newCfg.ToPipelineConfig(cb.sd.configDefaults)
102
if err != nil {
103
- return err
103
+ return dyncfg.MarkNonDisruptiveUpdate(err)
104
}
105
return cb.sd.mgr.Restart(cb.sd.ctx, newCfg.PipelineKey(), pipelineCfg)
106
}
src/go/plugin/agent/discovery/sd/dyncfg_test.go
+6
-5
@@ -2263,16 +2263,17 @@ func TestServiceDiscovery_DyncfgRestartErrorHandling(t *testing.T) {
2263
discovererType: testDiscovererTypeNetListeners,
2264
name: "test-job",
2265
sourceType: "dyncfg",
2266
- status: dyncfg.StatusFailed,
2266
+ status: dyncfg.StatusRunning,
2267
},
2268
},
2269
- // NOTE: When Restart fails validation (newPipeline fails), the old pipeline
2270
- // keeps running. This is the intended Restart behavior - validate before stopping.
2271
- // The status shows Failed but old pipeline continues collecting data.
2269
+ // NOTE: When Restart fails preflight validation (newPipeline fails), the
2270
+ // old pipeline stays running and handler rolls back to old running state.
2271
wantRunning: []string{"dyncfg:net_listeners:test-job"},
2272
wantDyncfgFunc: func(t *testing.T, got string) {
2273
assert.Contains(t, got, "FUNCTION_RESULT_BEGIN 3-update 200 application/json")
2275
- assert.Contains(t, got, "CONFIG test:sd:net_listeners:test-job status failed")
2274
+ line := "CONFIG test:sd:net_listeners:test-job status running"
2275
+ assert.GreaterOrEqual(t, strings.Count(got, line), 2, "expected running status to be re-notified after update failure")
2276
+ assert.NotContains(t, got, "CONFIG test:sd:net_listeners:test-job status failed")
2277
},
2278
}
2279
},
src/go/plugin/agent/discovery/sd/pipeline_manager.go
+3
-2
@@ -4,12 +4,14 @@ package sd
4
5
import (
6
"context"
7
+ "fmt"
8
"sync"
9
"time"
10
11
"github.com/netdata/netdata/go/plugins/logger"
12
"github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
13
"github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
14
+ "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
15
)
16
17
const (
@@ -97,8 +99,7 @@ func (m *PipelineManager) Restart(ctx context.Context, key string, cfg pipeline.
99
// Validate new config first by creating the pipeline (outside lock)
100
pl, err := m.newPipeline(cfg)
101
if err != nil {
100
- // New config is invalid, keep old pipeline running
101
- return err
102
+ return dyncfg.MarkNonDisruptiveUpdate(fmt.Errorf("failed to create new pipeline config: %w", err))
103
}
104
105
m.mux.Lock()
src/go/plugin/framework/dyncfg/dyncfg.go
+26
@@ -3,6 +3,7 @@
3
package dyncfg
4
5
import (
6
+ "errors"
7
"strings"
8
)
9
@@ -54,3 +55,28 @@ func JoinCommands(commands ...Command) string {
55
}
56
return strings.Join(strs, " ")
57
}
58
+
59
+// ErrNonDisruptiveUpdate marks update failures where runtime state was not changed.
60
+// Handler rollback logic uses this marker to keep old config/status authoritative.
61
+var ErrNonDisruptiveUpdate = errors.New("non-disruptive update")
62
+
63
+type nonDisruptiveUpdateError struct {
64
+ err error
65
+}
66
+
67
+func (e *nonDisruptiveUpdateError) Error() string { return e.err.Error() }
68
+func (e *nonDisruptiveUpdateError) Unwrap() error { return e.err }
69
+func (e *nonDisruptiveUpdateError) Is(target error) bool { return target == ErrNonDisruptiveUpdate }
70
+
71
+// MarkNonDisruptiveUpdate wraps err to indicate update failed before disrupting runtime.
72
+func MarkNonDisruptiveUpdate(err error) error {
73
+ if err == nil {
74
+ return nil
75
+ }
76
+
77
+ if errors.Is(err, ErrNonDisruptiveUpdate) {
78
+ return err
79
+ }
80
+
81
+ return &nonDisruptiveUpdateError{err: err}
82
+}
src/go/plugin/framework/dyncfg/handler.go
+12
@@ -646,6 +646,18 @@ func (h *Handler[C]) CmdUpdate(fn Function) {
646
}
647
648
if err != nil {
649
+ if !isConversion && errors.Is(err, ErrNonDisruptiveUpdate) {
650
+ // Update failed before runtime disruption; rollback to old cache state.
651
+ h.seen.Remove(newCfg)
652
+ h.seen.Add(oldCfg)
653
+ h.exposed.Add(entry)
654
+
655
+ h.api.SendCodef(fn, 200, "%v", err)
656
+ h.NotifyJobStatus(oldCfg, oldStatus)
657
+ // No OnStatusChange call here: effective state did not change.
658
+ return
659
+ }
660
+
661
newEntry.Status = StatusFailed
662
if isConversion {
663
h.NotifyJobCreate(newCfg, StatusFailed)
src/go/plugin/framework/dyncfg/handler_test.go
+30
@@ -935,6 +935,36 @@ func TestCmdUpdate_NonConversion_StartFails(t *testing.T) {
935
assert.Equal(t, StatusFailed, entry.Status)
936
}
937
938
+func TestCmdUpdate_NonConversion_StartFails_NonDisruptiveRollback(t *testing.T) {
939
+ cb := &mockCallbacks{}
940
+ cb.updateFn = func(_, _ testConfig) error {
941
+ return MarkNonDisruptiveUpdate(errors.New("update preflight failed"))
942
+ }
943
+ h := newTestHandler(cb)
944
+
945
+ oldCfg := testConfig{uid: "dyncfg:job1:v1", key: "job1", sourceType: "dyncfg", hash: 100}
946
+ newCfg := testConfig{uid: "dyncfg:job1:v2", key: "job1", sourceType: "dyncfg", hash: 200}
947
+ h.seen.Add(oldCfg)
948
+ h.exposed.Add(&Entry[testConfig]{Cfg: oldCfg, Status: StatusRunning})
949
+
950
+ cb.parseAndValidateFn = func(_ Function, _ string) (testConfig, error) {
951
+ return newCfg, nil
952
+ }
953
+
954
+ fn := newTestFn("test:job1", "update", "job1", []byte(`{}`))
955
+ h.CmdUpdate(fn)
956
+
957
+ entry, _ := h.exposed.LookupByKey("job1")
958
+ assert.Equal(t, StatusRunning, entry.Status)
959
+ assert.Equal(t, oldCfg.UID(), entry.Cfg.UID())
960
+
961
+ _, ok := h.seen.LookupByUID(oldCfg.UID())
962
+ assert.True(t, ok, "old config should be restored in seen cache")
963
+
964
+ _, ok = h.seen.LookupByUID(newCfg.UID())
965
+ assert.False(t, ok, "new config should be removed from seen cache on rollback")
966
+}
967
+
968
func TestCmdUpdate_Conversion_StartFails(t *testing.T) {
969
cb := &mockCallbacks{}
970
cb.startFn = func(_ testConfig) error { return errors.New("start failed") }