@cryptotaxi247 / netdata-1 / commits / 483bda2d0

chore(go.d): add file persister (#19716)

Ilya Mashchenko committed Feb 26, 2025 at 15:23 UTC 483bda2d02674f93373d9b4418c435a0ad3e4854
12 files changed +385 -487
src/go/plugin/go.d/agent/agent.go
+1 -17
@@ -18,7 +18,6 @@ import (
18 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
19 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/discovery"
21 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/filestatus"
21 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
22 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/jobmgr"
23 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
@@ -193,6 +192,7 @@ func (a *Agent) run(ctx context.Context) {
192 jobMgr := jobmgr.New()
193 jobMgr.PluginName = a.Name
194 jobMgr.Out = a.Out
195 + jobMgr.StateFile = a.StateFile
196 jobMgr.Modules = enabledModules
197 jobMgr.ConfigDefaults = discCfg.Registry
198 jobMgr.FnReg = fnMgr
@@ -201,17 +201,6 @@ func (a *Agent) run(ctx context.Context) {
201 jobMgr.Vnodes = reg
202 }
203
204 - var fsMgr *filestatus.Manager
205 - if !isTerminal && a.StateFile != "" {
206 - fsMgr = filestatus.NewManager(a.StateFile)
207 - jobMgr.FileStatus = fsMgr
208 - if store, err := filestatus.LoadStore(a.StateFile); err != nil {
209 - a.Warningf("couldn't load state file: %v", err)
210 - } else {
211 - jobMgr.FileStatusStore = store
212 - }
213 - }
214 -
204 in := make(chan []*confgroup.Group)
205 var wg sync.WaitGroup
206
@@ -224,11 +213,6 @@ func (a *Agent) run(ctx context.Context) {
213 wg.Add(1)
214 go func() { defer wg.Done(); discMgr.Run(ctx, in) }()
215
227 - if fsMgr != nil {
228 - wg.Add(1)
229 - go func() { defer wg.Done(); fsMgr.Run(ctx) }()
230 - }
231 -
216 wg.Wait()
217 <-ctx.Done()
218 }
src/go/plugin/go.d/agent/filepersister/persister.go new
+89
@@ -0,0 +1,89 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package filepersister
4 +
5 +import (
6 + "context"
7 + "log/slog"
8 + "os"
9 + "time"
10 +
11 + "github.com/netdata/netdata/go/plugins/logger"
12 +)
13 +
14 +type Data interface {
15 + Bytes() ([]byte, error)
16 + Updated() <-chan struct{}
17 +}
18 +
19 +func New(path string) *Persister {
20 + return &Persister{
21 + Logger: logger.New().With(
22 + slog.String("component", "file persister"),
23 + slog.String("file", path),
24 + ),
25 + filepath: path,
26 + flushEvery: time.Second * 5,
27 + flushCh: make(chan struct{}, 1),
28 + }
29 +}
30 +
31 +type Persister struct {
32 + *logger.Logger
33 +
34 + data Data
35 + filepath string
36 + flushEvery time.Duration
37 + flushCh chan struct{}
38 +}
39 +
40 +func (p *Persister) Run(ctx context.Context, data Data) {
41 + p.Info("instance is started")
42 + defer func() { p.Info("instance is stopped") }()
43 +
44 + p.data = data
45 +
46 + tk := time.NewTicker(p.flushEvery)
47 + defer tk.Stop()
48 + defer p.flush()
49 +
50 + for {
51 + select {
52 + case <-ctx.Done():
53 + return
54 + case <-p.data.Updated():
55 + p.triggerFlush()
56 + case <-tk.C:
57 + p.tryFlush()
58 + }
59 + }
60 +}
61 +
62 +func (p *Persister) triggerFlush() {
63 + select {
64 + case p.flushCh <- struct{}{}:
65 + default:
66 + // already has a pending flush
67 + }
68 +}
69 +
70 +func (p *Persister) tryFlush() {
71 + select {
72 + case <-p.flushCh:
73 + p.flush()
74 + default:
75 + // no pending flush
76 + }
77 +}
78 +
79 +func (p *Persister) flush() {
80 + bs, err := p.data.Bytes()
81 + if err != nil {
82 + p.Debugf("failed to marshal data: %v", err)
83 + return
84 + }
85 +
86 + _ = os.WriteFile(p.filepath, bs, 0644)
87 +
88 + p.Debug("file persisted successfully")
89 +}
src/go/plugin/go.d/agent/filepersister/persister_test.go new
+133
@@ -0,0 +1,133 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package filepersister
4 +
5 +import (
6 + "context"
7 + "errors"
8 + "os"
9 + "path"
10 + "strings"
11 + "testing"
12 + "time"
13 +
14 + "github.com/stretchr/testify/assert"
15 + "github.com/stretchr/testify/require"
16 +)
17 +
18 +func TestNew(t *testing.T) {
19 + tests := map[string]struct {
20 + path string
21 + wantErr bool
22 + }{
23 + "empty filepath": {
24 + wantErr: true,
25 + path: "",
26 + },
27 + "not empty filepath": {
28 + wantErr: false,
29 + path: "testdata/test.json",
30 + },
31 + }
32 +
33 + for name, test := range tests {
34 + t.Run(name, func(t *testing.T) {
35 + p := New(test.path)
36 + require.NotNil(t, p)
37 + })
38 + }
39 +}
40 +
41 +func TestPersister_Run(t *testing.T) {
42 + tests := map[string]struct {
43 + wantErr bool
44 + wantFile string
45 + }{
46 + "no save because data bytes error": {
47 + wantErr: true,
48 + },
49 + "successful save": {
50 + wantErr: false,
51 + wantFile: `
52 +{
53 + "module1": {
54 + "name1:17896517344060997937": "ok"
55 + },
56 + "module2": {
57 + "name2:14519194242031159283": "ok"
58 + }
59 +}
60 +`,
61 + },
62 + }
63 +
64 + for name, test := range tests {
65 + t.Run(name, func(t *testing.T) {
66 + dir, err := os.MkdirTemp(os.TempDir(), "netdata-go.d-test-filepersister-run")
67 + require.NoError(t, err)
68 + defer func() { assert.NoError(t, os.RemoveAll(dir)) }()
69 +
70 + filename := path.Join(dir, "filestatus")
71 +
72 + p := New(filename)
73 +
74 + data := newMockData(test.wantFile)
75 + data.wantError = test.wantErr
76 +
77 + ctx, cancel := context.WithCancel(context.Background())
78 + done := make(chan struct{})
79 + go func() {
80 + defer close(done)
81 + p.Run(ctx, data)
82 + }()
83 +
84 + cancel()
85 +
86 + timeout := time.Second * 5
87 + tk := time.NewTimer(timeout)
88 + defer tk.Stop()
89 +
90 + select {
91 + case <-done:
92 + case <-tk.C:
93 + t.Errorf("timed out after %s", timeout)
94 + }
95 +
96 + bs, err := os.ReadFile(filename)
97 +
98 + if test.wantErr {
99 + require.Error(t, err)
100 + } else {
101 + require.NoError(t, err)
102 + assert.Equal(t, strings.TrimSpace(test.wantFile), strings.TrimSpace(string(bs)))
103 + }
104 + })
105 + }
106 +}
107 +
108 +func newMockData(s string) *mockData {
109 + m := &mockData{
110 + data: s,
111 + ch: make(chan struct{}, 1),
112 + }
113 + m.ch <- struct{}{}
114 +
115 + return m
116 +}
117 +
118 +type mockData struct {
119 + data string
120 + ch chan struct{}
121 + wantError bool
122 +}
123 +
124 +func (m *mockData) Bytes() ([]byte, error) {
125 + if m.wantError {
126 + return nil, errors.New("mockData.Bytes() mock error")
127 + }
128 + return []byte(m.data), nil
129 +}
130 +
131 +func (m *mockData) Updated() <-chan struct{} {
132 + return m.ch
133 +}
src/go/plugin/go.d/agent/filestatus/manager.go deleted
-91
@@ -1,91 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package filestatus
4 -
5 -import (
6 - "context"
7 - "log/slog"
8 - "os"
9 - "time"
10 -
11 - "github.com/netdata/netdata/go/plugins/logger"
12 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
13 -)
14 -
15 -func NewManager(path string) *Manager {
16 - return &Manager{
17 - Logger: logger.New().With(
18 - slog.String("component", "filestatus manager"),
19 - ),
20 - path: path,
21 - store: &Store{},
22 - flushEvery: time.Second * 5,
23 - flushCh: make(chan struct{}, 1),
24 - }
25 -}
26 -
27 -type Manager struct {
28 - *logger.Logger
29 -
30 - path string
31 -
32 - store *Store
33 -
34 - flushEvery time.Duration
35 - flushCh chan struct{}
36 -}
37 -
38 -func (m *Manager) Run(ctx context.Context) {
39 - m.Info("instance is started")
40 - defer func() { m.Info("instance is stopped") }()
41 -
42 - tk := time.NewTicker(m.flushEvery)
43 - defer tk.Stop()
44 - defer m.flush()
45 -
46 - for {
47 - select {
48 - case <-ctx.Done():
49 - return
50 - case <-tk.C:
51 - m.tryFlush()
52 - }
53 - }
54 -}
55 -
56 -func (m *Manager) Save(cfg confgroup.Config, status string) {
57 - if v, ok := m.store.lookup(cfg); !ok || status != v {
58 - m.store.add(cfg, status)
59 - m.triggerFlush()
60 - }
61 -}
62 -
63 -func (m *Manager) Remove(cfg confgroup.Config) {
64 - if _, ok := m.store.lookup(cfg); ok {
65 - m.store.remove(cfg)
66 - m.triggerFlush()
67 - }
68 -}
69 -
70 -func (m *Manager) triggerFlush() {
71 - select {
72 - case m.flushCh <- struct{}{}:
73 - default:
74 - }
75 -}
76 -
77 -func (m *Manager) tryFlush() {
78 - select {
79 - case <-m.flushCh:
80 - m.flush()
81 - default:
82 - }
83 -}
84 -
85 -func (m *Manager) flush() {
86 - bs, err := m.store.bytes()
87 - if err != nil {
88 - return
89 - }
90 - _ = os.WriteFile(m.path, bs, 0644)
91 -}
src/go/plugin/go.d/agent/filestatus/manager_test.go deleted
-122
@@ -1,122 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package filestatus
4 -
5 -import (
6 - "context"
7 - "os"
8 - "path"
9 - "strings"
10 - "testing"
11 - "time"
12 -
13 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
14 -
15 - "github.com/stretchr/testify/assert"
16 - "github.com/stretchr/testify/require"
17 -)
18 -
19 -func TestNewManager(t *testing.T) {
20 - mgr := NewManager("")
21 - assert.NotNil(t, mgr.store)
22 -}
23 -
24 -func TestManager_Run(t *testing.T) {
25 - type testAction struct {
26 - name string
27 - cfg confgroup.Config
28 - status string
29 - }
30 - tests := map[string]struct {
31 - actions []testAction
32 - wantFile string
33 - }{
34 - "save": {
35 - actions: []testAction{
36 - {
37 - name: "save", status: "ok",
38 - cfg: prepareConfig("module", "module1", "name", "name1"),
39 - },
40 - {
41 - name: "save", status: "ok",
42 - cfg: prepareConfig("module", "module2", "name", "name2"),
43 - },
44 - },
45 - wantFile: `
46 -{
47 - "module1": {
48 - "name1:17896517344060997937": "ok"
49 - },
50 - "module2": {
51 - "name2:14519194242031159283": "ok"
52 - }
53 -}
54 -`,
55 - },
56 - "remove": {
57 - actions: []testAction{
58 - {
59 - name: "save", status: "ok",
60 - cfg: prepareConfig("module", "module1", "name", "name1"),
61 - },
62 - {
63 - name: "save", status: "ok",
64 - cfg: prepareConfig("module", "module2", "name", "name2"),
65 - },
66 - {
67 - name: "remove",
68 - cfg: prepareConfig("module", "module2", "name", "name2"),
69 - },
70 - },
71 - wantFile: `
72 -{
73 - "module1": {
74 - "name1:17896517344060997937": "ok"
75 - }
76 -}
77 -`,
78 - },
79 - }
80 -
81 - for name, test := range tests {
82 - t.Run(name, func(t *testing.T) {
83 - dir, err := os.MkdirTemp(os.TempDir(), "netdata-go-test-filestatus-run")
84 - require.NoError(t, err)
85 - defer func() { assert.NoError(t, os.RemoveAll(dir)) }()
86 -
87 - filename := path.Join(dir, "filestatus")
88 -
89 - mgr := NewManager(filename)
90 -
91 - ctx, cancel := context.WithCancel(context.Background())
92 - done := make(chan struct{})
93 - go func() { defer close(done); mgr.Run(ctx) }()
94 -
95 - for _, v := range test.actions {
96 - switch v.name {
97 - case "save":
98 - mgr.Save(v.cfg, v.status)
99 - case "remove":
100 - mgr.Remove(v.cfg)
101 - }
102 - }
103 -
104 - cancel()
105 -
106 - timeout := time.Second * 5
107 - tk := time.NewTimer(timeout)
108 - defer tk.Stop()
109 -
110 - select {
111 - case <-done:
112 - case <-tk.C:
113 - t.Errorf("timed out after %s", timeout)
114 - }
115 -
116 - bs, err := os.ReadFile(filename)
117 - require.NoError(t, err)
118 -
119 - assert.Equal(t, strings.TrimSpace(test.wantFile), string(bs))
120 - })
121 - }
122 -}
src/go/plugin/go.d/agent/filestatus/store.go deleted
-90
@@ -1,90 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package filestatus
4 -
5 -import (
6 - "encoding/json"
7 - "fmt"
8 - "os"
9 - "slices"
10 - "sync"
11 -
12 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
13 -)
14 -
15 -func LoadStore(path string) (*Store, error) {
16 - var s Store
17 -
18 - f, err := os.Open(path)
19 - if err != nil {
20 - return nil, err
21 - }
22 - defer func() { _ = f.Close() }()
23 -
24 - return &s, json.NewDecoder(f).Decode(&s.items)
25 -}
26 -
27 -type Store struct {
28 - mux sync.Mutex
29 - items map[string]map[string]string // [module][name:hash]status
30 -}
31 -
32 -func (s *Store) Contains(cfg confgroup.Config, statuses ...string) bool {
33 - status, ok := s.lookup(cfg)
34 - if !ok {
35 - return false
36 - }
37 -
38 - return slices.Contains(statuses, status)
39 -}
40 -
41 -func (s *Store) lookup(cfg confgroup.Config) (string, bool) {
42 - s.mux.Lock()
43 - defer s.mux.Unlock()
44 -
45 - jobs, ok := s.items[cfg.Module()]
46 - if !ok {
47 - return "", false
48 - }
49 -
50 - status, ok := jobs[storeJobKey(cfg)]
51 -
52 - return status, ok
53 -}
54 -
55 -func (s *Store) add(cfg confgroup.Config, status string) {
56 - s.mux.Lock()
57 - defer s.mux.Unlock()
58 -
59 - if s.items == nil {
60 - s.items = make(map[string]map[string]string)
61 - }
62 -
63 - if s.items[cfg.Module()] == nil {
64 - s.items[cfg.Module()] = make(map[string]string)
65 - }
66 -
67 - s.items[cfg.Module()][storeJobKey(cfg)] = status
68 -}
69 -
70 -func (s *Store) remove(cfg confgroup.Config) {
71 - s.mux.Lock()
72 - defer s.mux.Unlock()
73 -
74 - delete(s.items[cfg.Module()], storeJobKey(cfg))
75 -
76 - if len(s.items[cfg.Module()]) == 0 {
77 - delete(s.items, cfg.Module())
78 - }
79 -}
80 -
81 -func (s *Store) bytes() ([]byte, error) {
82 - s.mux.Lock()
83 - defer s.mux.Unlock()
84 -
85 - return json.MarshalIndent(s.items, "", " ")
86 -}
87 -
88 -func storeJobKey(cfg confgroup.Config) string {
89 - return fmt.Sprintf("%s:%d", cfg.Name(), cfg.Hash())
90 -}
src/go/plugin/go.d/agent/filestatus/store_test.go deleted
-138
@@ -1,138 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package filestatus
4 -
5 -import (
6 - "testing"
7 -
8 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
9 -
10 - "github.com/stretchr/testify/assert"
11 -)
12 -
13 -// TODO: tech debt
14 -func TestLoadStore(t *testing.T) {
15 -
16 -}
17 -
18 -// TODO: tech debt
19 -func TestStore_Contains(t *testing.T) {
20 -
21 -}
22 -
23 -func TestStore_add(t *testing.T) {
24 - tests := map[string]struct {
25 - prepare func() *Store
26 - input confgroup.Config
27 - wantItemsNum int
28 - }{
29 - "add cfg to the empty store": {
30 - prepare: func() *Store {
31 - return &Store{}
32 - },
33 - input: prepareConfig(
34 - "module", "modName",
35 - "name", "jobName",
36 - ),
37 - wantItemsNum: 1,
38 - },
39 - "add cfg that already in the store": {
40 - prepare: func() *Store {
41 - return &Store{
42 - items: map[string]map[string]string{
43 - "modName": {"jobName:14246646561040761746": "state"},
44 - },
45 - }
46 - },
47 - input: prepareConfig(
48 - "module", "modName",
49 - "name", "jobName",
50 - ),
51 - wantItemsNum: 1,
52 - },
53 - "add cfg with same module, same name, but specific options": {
54 - prepare: func() *Store {
55 - return &Store{
56 - items: map[string]map[string]string{
57 - "modName": {"jobName:18299273693089411682": "state"},
58 - },
59 - }
60 - },
61 - input: prepareConfig(
62 - "module", "modName",
63 - "name", "jobName",
64 - "opt", "val",
65 - ),
66 - wantItemsNum: 2,
67 - },
68 - }
69 -
70 - for name, test := range tests {
71 - t.Run(name, func(t *testing.T) {
72 - s := test.prepare()
73 - s.add(test.input, "state")
74 - assert.Equal(t, test.wantItemsNum, calcStoreItems(s))
75 - })
76 - }
77 -}
78 -
79 -func TestStore_remove(t *testing.T) {
80 - tests := map[string]struct {
81 - prepare func() *Store
82 - input confgroup.Config
83 - wantItemsNum int
84 - }{
85 - "remove cfg from the empty store": {
86 - prepare: func() *Store {
87 - return &Store{}
88 - },
89 - input: prepareConfig(
90 - "module", "modName",
91 - "name", "jobName",
92 - ),
93 - wantItemsNum: 0,
94 - },
95 - "remove cfg from the store": {
96 - prepare: func() *Store {
97 - return &Store{
98 - items: map[string]map[string]string{
99 - "modName": {
100 - "jobName:14246646561040761746": "state",
101 - "jobName:14246646561040761747": "state",
102 - },
103 - },
104 - }
105 - },
106 - input: prepareConfig(
107 - "module", "modName",
108 - "name", "jobName",
109 - ),
110 - wantItemsNum: 1,
111 - },
112 - }
113 -
114 - for name, test := range tests {
115 - t.Run(name, func(t *testing.T) {
116 - s := test.prepare()
117 - s.remove(test.input)
118 - assert.Equal(t, test.wantItemsNum, calcStoreItems(s))
119 - })
120 - }
121 -}
122 -
123 -func calcStoreItems(s *Store) (num int) {
124 - for _, v := range s.items {
125 - for range v {
126 - num++
127 - }
128 - }
129 - return num
130 -}
131 -
132 -func prepareConfig(values ...string) confgroup.Config {
133 - cfg := confgroup.Config{}
134 - for i := 1; i < len(values); i += 2 {
135 - cfg[values[i-1]] = values[i]
136 - }
137 - return cfg
138 -}
src/go/plugin/go.d/agent/jobmgr/di.go
-10
@@ -4,20 +4,10 @@ package jobmgr
4
5 import (
6 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
7 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
7 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
8 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
9 )
10
12 -type FileStatus interface {
13 - Save(cfg confgroup.Config, state string)
14 - Remove(cfg confgroup.Config)
15 -}
16 -
17 -type FileStatusStore interface {
18 - Contains(cfg confgroup.Config, states ...string) bool
19 -}
20 -
11 type Vnodes interface {
12 Lookup(key string) (*vnodes.VirtualNode, bool)
13 }
src/go/plugin/go.d/agent/jobmgr/dyncfg_collector.go
+5 -5
@@ -347,7 +347,7 @@ func (m *Manager) dyncfgConfigRestart(fn functions.Function) {
347 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
348 return
349 case dyncfgRunning:
350 - m.FileStatus.Remove(ecfg.cfg)
350 + m.fileStatus.remove(ecfg.cfg)
351 m.stopRunningJob(ecfg.cfg.FullName())
352 default:
353 }
@@ -365,7 +365,7 @@ func (m *Manager) dyncfgConfigRestart(fn functions.Function) {
365 ecfg.status = dyncfgRunning
366
367 if isDyncfg(ecfg.cfg) {
368 - m.FileStatus.Save(ecfg.cfg, ecfg.status.String())
368 + m.fileStatus.add(ecfg.cfg, ecfg.status.String())
369 }
370 m.startRunningJob(job)
371 m.dyncfgRespf(fn, 200, "")
@@ -445,7 +445,7 @@ func (m *Manager) dyncfgConfigEnable(fn functions.Function) {
445 ecfg.status = dyncfgRunning
446
447 if isDyncfg(ecfg.cfg) {
448 - m.FileStatus.Save(ecfg.cfg, ecfg.status.String())
448 + m.fileStatus.add(ecfg.cfg, ecfg.status.String())
449 }
450
451 m.startRunningJob(job)
@@ -482,7 +482,7 @@ func (m *Manager) dyncfgConfigDisable(fn functions.Function) {
482 case dyncfgRunning:
483 m.stopRunningJob(ecfg.cfg.FullName())
484 if isDyncfg(ecfg.cfg) {
485 - m.FileStatus.Remove(ecfg.cfg)
485 + m.fileStatus.remove(ecfg.cfg)
486 }
487 default:
488 }
@@ -583,7 +583,7 @@ func (m *Manager) dyncfgConfigRemove(fn functions.Function) {
583 m.seenConfigs.remove(ecfg.cfg)
584 m.exposedConfigs.remove(ecfg.cfg)
585 m.stopRunningJob(ecfg.cfg.FullName())
586 - m.FileStatus.Remove(ecfg.cfg)
586 + m.fileStatus.remove(ecfg.cfg)
587
588 m.dyncfgRespf(fn, 200, "")
589 m.dyncfgJobRemove(ecfg.cfg)
src/go/plugin/go.d/agent/jobmgr/filestatus.go new
+144
@@ -0,0 +1,144 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobmgr
4 +
5 +import (
6 + "encoding/json"
7 + "fmt"
8 + "os"
9 + "slices"
10 + "sync"
11 +
12 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
13 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/filepersister"
14 +)
15 +
16 +func (m *Manager) loadFileStatus() {
17 + m.fileStatus = newFileStatus()
18 +
19 + if isTerminal || m.StateFile == "" {
20 + return
21 + }
22 +
23 + s, err := loadFileStatus(m.StateFile)
24 + if err != nil {
25 + m.Warningf("failed to load state file: %v", err)
26 + return
27 + }
28 + m.fileStatus = s
29 +}
30 +
31 +func (m *Manager) runFileStatusPersistence() {
32 + if m.StateFile == "" {
33 + return
34 + }
35 + p := filepersister.New(m.StateFile)
36 + p.Run(m.ctx, m.fileStatus)
37 +}
38 +
39 +func loadFileStatus(path string) (*fileStatus, error) {
40 + f, err := os.Open(path)
41 + if err != nil {
42 + return nil, err
43 + }
44 + defer func() { _ = f.Close() }()
45 +
46 + s := newFileStatus()
47 +
48 + return s, json.NewDecoder(f).Decode(&s.items)
49 +}
50 +
51 +func newFileStatus() *fileStatus {
52 + return &fileStatus{
53 + items: make(map[string]map[string]string),
54 + ch: make(chan struct{}, 1),
55 + }
56 +}
57 +
58 +type fileStatus struct {
59 + mux sync.Mutex
60 + items map[string]map[string]string // [module][name:hash]status
61 + ch chan struct{}
62 +}
63 +
64 +func (s *fileStatus) Bytes() ([]byte, error) {
65 + s.mux.Lock()
66 + defer s.mux.Unlock()
67 +
68 + return json.MarshalIndent(s.items, "", " ")
69 +}
70 +
71 +func (s *fileStatus) Updated() <-chan struct{} {
72 + return s.ch
73 +}
74 +
75 +func (s *fileStatus) contains(cfg confgroup.Config, statuses ...string) bool {
76 + s.mux.Lock()
77 + defer s.mux.Unlock()
78 +
79 + status, ok := s.lookup(cfg)
80 + if !ok {
81 + return false
82 + }
83 +
84 + return slices.Contains(statuses, status)
85 +}
86 +
87 +func (s *fileStatus) lookup(cfg confgroup.Config) (string, bool) {
88 + s.mux.Lock()
89 + defer s.mux.Unlock()
90 +
91 + jobs, ok := s.items[cfg.Module()]
92 + if !ok {
93 + return "", false
94 + }
95 +
96 + status, ok := jobs[s.jobKey(cfg)]
97 +
98 + return status, ok
99 +}
100 +
101 +func (s *fileStatus) add(cfg confgroup.Config, status string) {
102 + s.mux.Lock()
103 + defer s.mux.Unlock()
104 +
105 + defer s.setUpdated()
106 +
107 + if s.items == nil {
108 + s.items = make(map[string]map[string]string)
109 + }
110 +
111 + if s.items[cfg.Module()] == nil {
112 + s.items[cfg.Module()] = make(map[string]string)
113 + }
114 +
115 + s.items[cfg.Module()][s.jobKey(cfg)] = status
116 +
117 + select {
118 + case s.ch <- struct{}{}:
119 + default:
120 + }
121 +}
122 +
123 +func (s *fileStatus) remove(cfg confgroup.Config) {
124 + s.mux.Lock()
125 + defer s.mux.Unlock()
126 +
127 + defer s.setUpdated()
128 +
129 + delete(s.items[cfg.Module()], s.jobKey(cfg))
130 +
131 + if len(s.items[cfg.Module()]) == 0 {
132 + delete(s.items, cfg.Module())
133 + }
134 +}
135 +func (s *fileStatus) setUpdated() {
136 + select {
137 + case s.ch <- struct{}{}:
138 + default:
139 + }
140 +}
141 +
142 +func (s *fileStatus) jobKey(cfg confgroup.Config) string {
143 + return fmt.Sprintf("%s:%d", cfg.Name(), cfg.Hash())
144 +}
src/go/plugin/go.d/agent/jobmgr/manager.go
+13 -10
@@ -32,10 +32,8 @@ func New() *Manager {
32 Logger: logger.New().With(
33 slog.String("component", "job manager"),
34 ),
35 - Out: io.Discard,
36 - FileStatus: noop{},
37 - FileStatusStore: noop{},
38 - FnReg: noop{},
35 + Out: io.Discard,
36 + FnReg: noop{},
37
38 Vnodes: make(map[string]*vnodes.VirtualNode),
39
@@ -62,11 +60,11 @@ type Manager struct {
60 Out io.Writer
61 Modules module.Registry
62 ConfigDefaults confgroup.Registry
63 + StateFile string
64 + FnReg FunctionRegistry
65 + Vnodes map[string]*vnodes.VirtualNode
66
66 - FileStatus FileStatus
67 - FileStatusStore FileStatusStore
68 - FnReg FunctionRegistry
69 - Vnodes map[string]*vnodes.VirtualNode
67 + fileStatus *fileStatus
68
69 discoveredConfigs *discoveredConfigs
70 seenConfigs *seenConfigs
@@ -101,8 +99,13 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
99 m.dyncfgCollectorModuleCreate(name)
100 }
101
102 + m.loadFileStatus()
103 +
104 var wg sync.WaitGroup
105
106 + wg.Add(1)
107 + go func() { defer wg.Done(); m.runFileStatusPersistence() }()
108 +
109 wg.Add(1)
110 go func() { defer wg.Done(); m.runProcessConfGroups(in) }()
111
@@ -191,7 +194,7 @@ func (m *Manager) addConfig(cfg confgroup.Config) {
194 }
195 if ecfg.status == dyncfgRunning {
196 m.stopRunningJob(ecfg.cfg.FullName())
194 - m.FileStatus.Remove(ecfg.cfg)
197 + m.fileStatus.remove(ecfg.cfg)
198 }
199 scfg.status = dyncfgAccepted
200 m.exposedConfigs.add(scfg) // replace existing exposed
@@ -223,7 +226,7 @@ func (m *Manager) removeConfig(cfg confgroup.Config) {
226
227 m.exposedConfigs.remove(cfg)
228 m.stopRunningJob(cfg.FullName())
226 - m.FileStatus.Remove(cfg)
229 + m.fileStatus.remove(cfg)
230
231 if !isStock(cfg) || ecfg.status == dyncfgRunning {
232 m.dyncfgJobRemove(cfg)
src/go/plugin/go.d/agent/jobmgr/noop.go
-4
@@ -5,7 +5,6 @@ package jobmgr
5 import (
6 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
7
8 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
8 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
9 )
10
@@ -14,9 +13,6 @@ type noop struct{}
13 func (n noop) Lock(string) (bool, error) { return true, nil }
14 func (n noop) Unlock(string) {}
15 func (n noop) UnlockAll() {}
17 -func (n noop) Save(confgroup.Config, string) {}
18 -func (n noop) Remove(confgroup.Config) {}
19 -func (n noop) Contains(confgroup.Config, ...string) bool { return false }
16 func (n noop) Lookup(string) (*vnodes.VirtualNode, bool) { return nil, false }
17 func (n noop) Register(name string, reg func(functions.Function)) {}
18 func (n noop) Unregister(name string) {}