1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
// SPDX-License-Identifier: GPL-3.0-or-later
package kubernetes
import (
"context"
"sort"
"testing"
"time"
"github.com/netdata/netdata/go/go.d.plugin/agent/discovery/sd/model"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"k8s.io/client-go/tools/cache"
)
const (
startWaitTimeout = time.Second * 3
finishWaitTimeout = time.Second * 5
)
type discoverySim struct {
td *KubeDiscoverer
runAfterSync func(ctx context.Context)
sortBeforeVerify bool
wantTargetGroups []model.TargetGroup
}
func (sim discoverySim) run(t *testing.T) []model.TargetGroup {
t.Helper()
require.NotNil(t, sim.td)
require.NotEmpty(t, sim.wantTargetGroups)
in, out := make(chan []model.TargetGroup), make(chan []model.TargetGroup)
go sim.collectTargetGroups(t, in, out)
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
go sim.td.Discover(ctx, in)
select {
case <-sim.td.started:
case <-time.After(startWaitTimeout):
t.Fatalf("td %s failed to start in %s", sim.td.discoverers, startWaitTimeout)
}
synced := cache.WaitForCacheSync(ctx.Done(), sim.td.hasSynced)
require.Truef(t, synced, "td %s failed to sync", sim.td.discoverers)
if sim.runAfterSync != nil {
sim.runAfterSync(ctx)
}
groups := <-out
if sim.sortBeforeVerify {
sortTargetGroups(groups)
}
sim.verifyResult(t, groups)
return groups
}
func (sim discoverySim) collectTargetGroups(t *testing.T, in, out chan []model.TargetGroup) {
var tggs []model.TargetGroup
loop:
for {
select {
case inGroups := <-in:
if tggs = append(tggs, inGroups...); len(tggs) >= len(sim.wantTargetGroups) {
break loop
}
case <-time.After(finishWaitTimeout):
t.Logf("td %s timed out after %s, got %d groups, expected %d, some events are skipped",
sim.td.discoverers, finishWaitTimeout, len(tggs), len(sim.wantTargetGroups))
break loop
}
}
out <- tggs
}
func (sim discoverySim) verifyResult(t *testing.T, result []model.TargetGroup) {
var expected, actual any
if len(sim.wantTargetGroups) == len(result) {
expected = sim.wantTargetGroups
actual = result
} else {
want := make(map[string]model.TargetGroup)
for _, group := range sim.wantTargetGroups {
want[group.Source()] = group
}
got := make(map[string]model.TargetGroup)
for _, group := range result {
got[group.Source()] = group
}
expected, actual = want, got
}
assert.Equal(t, expected, actual)
}
type hasSynced interface {
hasSynced() bool
}
var (
_ hasSynced = &KubeDiscoverer{}
_ hasSynced = &podDiscoverer{}
_ hasSynced = &serviceDiscoverer{}
)
func (d *KubeDiscoverer) hasSynced() bool {
for _, disc := range d.discoverers {
v, ok := disc.(hasSynced)
if !ok || !v.hasSynced() {
return false
}
}
return true
}
func (p *podDiscoverer) hasSynced() bool {
return p.podInformer.HasSynced() && p.cmapInformer.HasSynced() && p.secretInformer.HasSynced()
}
func (s *serviceDiscoverer) hasSynced() bool {
return s.informer.HasSynced()
}
func sortTargetGroups(tggs []model.TargetGroup) {
if len(tggs) == 0 {
return
}
sort.Slice(tggs, func(i, j int) bool { return tggs[i].Source() < tggs[j].Source() })
}
|
