master
go 128 lines 3.11 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 package pipeline
4
5 import (
6 "context"
7 "sort"
8 "testing"
9 "time"
10
11 "github.com/netdata/netdata/go/plugins/logger"
12 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/model"
13 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
14
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/require"
17 "gopkg.in/yaml.v2"
18 )
19
20 type discoverySim struct {
21 config string
22 discoverers []model.Discoverer
23 wantClassifyCalls int
24 wantComposeCalls int
25 wantConfGroups []*confgroup.Group
26
27 // New: when true (or when cfg.Services is non-empty), run with services engine only.
28 useServices bool
29 }
30
31 func (sim discoverySim) run(t *testing.T) {
32 t.Helper()
33
34 var cfg Config
35 err := yaml.Unmarshal([]byte(sim.config), &cfg)
36 require.Nilf(t, err, "cfg unmarshal")
37
38 accum := newAccumulator()
39 accum.sendEvery = time.Second * 2
40
41 pl := &Pipeline{
42 Logger: logger.New(),
43 discoverers: sim.discoverers,
44 accum: accum,
45 configs: make(map[string]map[uint64][]confgroup.Config),
46 }
47 pl.accum.Logger = pl.Logger
48
49 // Prefer services when either explicitly requested or present in config.
50 if sim.useServices || len(cfg.Services) > 0 {
51 // --- services-only path ---
52 svr, err := newServiceEngine(cfg.Services)
53 require.Nil(t, err, "newServiceEngine")
54
55 mockSvr := &mockComposer{cmr: svr} // reuse mock to count compose()
56 pl.svr = mockSvr // set services engine
57 svr.Logger = pl.Logger
58
59 groups := sim.collectGroups(t, pl)
60 sortConfigGroups(groups)
61 sortConfigGroups(sim.wantConfGroups)
62
63 assert.Equal(t, sim.wantConfGroups, groups)
64 // When services is used, classify is not called.
65 assert.Equalf(t, 0, sim.wantClassifyCalls, "classify calls should be zero in services mode")
66 assert.Equalf(t, sim.wantComposeCalls, mockSvr.calls, "compose (services) calls")
67 return
68 }
69
70 groups := sim.collectGroups(t, pl)
71
72 sortConfigGroups(groups)
73 sortConfigGroups(sim.wantConfGroups)
74
75 assert.Equal(t, sim.wantConfGroups, groups)
76 }
77
78 func (sim discoverySim) collectGroups(t *testing.T, pl *Pipeline) []*confgroup.Group {
79 ctx, cancel := context.WithCancel(context.Background())
80 defer cancel()
81
82 in := make(chan []*confgroup.Group)
83 done := make(chan struct{})
84
85 go func() { defer close(done); pl.Run(ctx, in) }()
86
87 timeout := time.Second * 10
88 var groups []*confgroup.Group
89
90 func() {
91 for {
92 select {
93 case inGroups := <-in:
94 groups = append(groups, inGroups...)
95 case <-done:
96 return
97 case <-time.After(timeout):
98 t.Logf("discovery timed out after %s, got %d groups, expected %d, some events are skipped",
99 timeout, len(groups), len(sim.wantConfGroups))
100 return
101 }
102 }
103 }()
104
105 return groups
106 }
107
108 type mockComposer struct {
109 calls int
110 cmr composer
111 }
112
113 func (m *mockComposer) compose(tgt model.Target) []confgroup.Config {
114 m.calls++
115 return m.cmr.compose(tgt)
116 }
117
118 func sortConfigGroups(groups []*confgroup.Group) {
119 sort.Slice(groups, func(i, j int) bool {
120 return groups[i].Source < groups[j].Source
121 })
122
123 for _, g := range groups {
124 sort.Slice(g.Configs, func(i, j int) bool {
125 return g.Configs[i].Name() < g.Configs[j].Name()
126 })
127 }
128 }