From 48989d8996c6246a014e309ede4ef6de5d6dfeee Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 5 Sep 2018 15:44:52 +0200 Subject: [PATCH 1/8] discovery: add more tests Co-authored-by: Camille Janicki Signed-off-by: Simon Pasquier --- discovery/manager.go | 10 +- discovery/manager_test.go | 372 ++++++++++++++++++++++++++++++++------ 2 files changed, 326 insertions(+), 56 deletions(-) diff --git a/discovery/manager.go b/discovery/manager.go index bb4409fea..3b13cf1cc 100644 --- a/discovery/manager.go +++ b/discovery/manager.go @@ -93,6 +93,7 @@ func NewManager(ctx context.Context, logger log.Logger) *Manager { targets: make(map[poolKey]map[string]*targetgroup.Group), discoverCancel: []context.CancelFunc{}, ctx: ctx, + updatert: 5 * time.Second, } } @@ -111,6 +112,10 @@ type Manager struct { providers []*provider // The sync channels sends the updates in map[targetSetName] where targetSetName is the job value from the scrape config. syncCh chan map[string][]*targetgroup.Group + + // How long to wait before sending updates to the channel. The variable + // should only be modified in unit tests. + updatert time.Duration } // Run starts the background processing @@ -166,7 +171,7 @@ func (m *Manager) startProvider(ctx context.Context, p *provider) { } func (m *Manager) updater(ctx context.Context, p *provider, updates chan []*targetgroup.Group) { - ticker := time.NewTicker(5 * time.Second) + ticker := time.NewTicker(m.updatert) defer ticker.Stop() triggerUpdate := make(chan struct{}, 1) @@ -181,11 +186,10 @@ func (m *Manager) updater(ctx context.Context, p *provider, updates chan []*targ select { case m.syncCh <- m.allGroups(): // Waiting until the receiver can accept the last update. level.Debug(m.logger).Log("msg", "discoverer exited", "provider", p.name) - return case <-ctx.Done(): - return } + return } for _, s := range p.subs { m.updateGroup(poolKey{setName: s, provider: p.name}, tgs) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index d9566201f..e776de910 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -655,53 +655,69 @@ func TestTargetUpdatesOrder(t *testing.T) { }, } - for testIndex, testCase := range testCases { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - discoveryManager := NewManager(ctx, log.NewNopLogger()) + for i, tc := range testCases { + tc := tc + t.Run(tc.title, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + discoveryManager := NewManager(ctx, log.NewNopLogger()) + discoveryManager.updatert = 100 * time.Millisecond - var totalUpdatesCount int + var totalUpdatesCount int - provUpdates := make(chan []*targetgroup.Group) - for _, up := range testCase.updates { - go newMockDiscoveryProvider(up).Run(ctx, provUpdates) - if len(up) > 0 { - totalUpdatesCount = totalUpdatesCount + len(up) + provUpdates := make(chan []*targetgroup.Group) + for _, up := range tc.updates { + go newMockDiscoveryProvider(up).Run(ctx, provUpdates) + if len(up) > 0 { + totalUpdatesCount = totalUpdatesCount + len(up) + } } - } - Loop: - for x := 0; x < totalUpdatesCount; x++ { - select { - case <-time.After(10 * time.Second): - t.Errorf("%v. %q: no update arrived within the timeout limit", x, testCase.title) - break Loop - case tgs := <-provUpdates: - discoveryManager.updateGroup(poolKey{setName: strconv.Itoa(testIndex), provider: testCase.title}, tgs) - for _, received := range discoveryManager.allGroups() { - // Need to sort by the Groups source as the received order is not guaranteed. - sort.Sort(byGroupSource(received)) - if !reflect.DeepEqual(received, testCase.expectedTargets[x]) { - var receivedFormated string - for _, receivedTargets := range received { - receivedFormated = receivedFormated + receivedTargets.Source + ":" + fmt.Sprint(receivedTargets.Targets) - } - var expectedFormated string - for _, expectedTargets := range testCase.expectedTargets[x] { - expectedFormated = expectedFormated + expectedTargets.Source + ":" + fmt.Sprint(expectedTargets.Targets) - } - - t.Errorf("%v. %v: \ntargets mismatch \nreceived: %v \nexpected: %v", - x, testCase.title, - receivedFormated, - expectedFormated) + Loop: + for x := 0; x < totalUpdatesCount; x++ { + select { + case <-time.After(10 * time.Second): + t.Errorf("%d: no update arrived within the timeout limit", x) + break Loop + case tgs := <-provUpdates: + discoveryManager.updateGroup(poolKey{setName: strconv.Itoa(i), provider: tc.title}, tgs) + for _, got := range discoveryManager.allGroups() { + assertEqualGroups(t, got, tc.expectedTargets[x], func(got, expected string) string { + return fmt.Sprintf("%d: \ntargets mismatch \ngot: %v \nexpected: %v", + x, + got, + expected) + }) } } } - } + }) } } +func assertEqualGroups(t *testing.T, got, expected []*targetgroup.Group, msg func(got, expected string) string) { + t.Helper() + format := func(groups []*targetgroup.Group) string { + var s string + for i, group := range groups { + if i > 0 { + s += "," + } + s += group.Source + ":" + fmt.Sprint(group.Targets) + } + return s + } + + // Need to sort by the groups's source as the received order is not guaranteed. + sort.Sort(byGroupSource(got)) + sort.Sort(byGroupSource(expected)) + + if !reflect.DeepEqual(got, expected) { + t.Errorf(msg(format(got), format(expected))) + } + +} + func verifyPresence(t *testing.T, tSets map[poolKey]map[string]*targetgroup.Group, poolKey poolKey, label string, present bool) { if _, ok := tSets[poolKey]; !ok { t.Fatalf("'%s' should be present in Pool keys: %v", poolKey, tSets) @@ -813,6 +829,7 @@ scrape_configs: ctx, cancel := context.WithCancel(context.Background()) defer cancel() discoveryManager := NewManager(ctx, nil) + discoveryManager.updatert = 500 * time.Millisecond go discoveryManager.Run() c := make(map[string]sd_config.ServiceDiscoveryConfig) @@ -868,6 +885,251 @@ scrape_configs: } } +func TestCoordinationWithReceiver(t *testing.T) { + updateDelay := 100 * time.Millisecond + + type expect struct { + delay time.Duration + tgs map[string][]*targetgroup.Group + } + + testCases := []struct { + title string + providers map[string][]update + expected []expect + }{ + { + title: "Receiver should get updates even when the channel is blocked", + providers: map[string][]update{ + "mock1": []update{ + update{ + targetGroups: []targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + update{ + interval: 4 * updateDelay / time.Millisecond, + targetGroups: []targetgroup.Group{ + { + Source: "tg2", + Targets: []model.LabelSet{{"__instance__": "2"}}, + }, + }, + }, + }, + }, + expected: []expect{ + { + delay: 2 * updateDelay, + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + { + delay: 4 * updateDelay, + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + { + Source: "tg2", + Targets: []model.LabelSet{{"__instance__": "2"}}, + }, + }, + }, + }, + }, + }, + { + title: "The receiver gets an update when a target group is gone", + providers: map[string][]update{ + "mock1": []update{ + update{ + targetGroups: []targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + update{ + interval: 2 * updateDelay / time.Millisecond, + targetGroups: []targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{}, + }, + }, + }, + }, + }, + expected: []expect{ + { + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + { + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{}, + }, + }, + }, + }, + }, + }, + { + title: "The receiver gets merged updates", + providers: map[string][]update{ + "mock1": []update{ + // This update should never be seen by the receiver because + // it is overwritten by the next one. + update{ + targetGroups: []targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "0"}}, + }, + }, + }, + update{ + targetGroups: []targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + }, + expected: []expect{ + { + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + }, + }, + { + title: "Discovery with multiple providers", + providers: map[string][]update{ + "mock1": []update{ + // This update is available in the first receive. + update{ + targetGroups: []targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + "mock2": []update{ + // This update should only arrive after the receiver has read from the channel once. + update{ + interval: 2 * updateDelay / time.Millisecond, + targetGroups: []targetgroup.Group{ + { + Source: "tg2", + Targets: []model.LabelSet{{"__instance__": "2"}}, + }, + }, + }, + }, + }, + expected: []expect{ + { + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + { + delay: 1 * updateDelay, + tgs: map[string][]*targetgroup.Group{ + "mock1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + "mock2": []*targetgroup.Group{ + { + Source: "tg2", + Targets: []model.LabelSet{{"__instance__": "2"}}, + }, + }, + }, + }, + }, + }, + } + + for _, tc := range testCases { + tc := tc + t.Run(tc.title, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + mgr := NewManager(ctx, nil) + mgr.updatert = updateDelay + go mgr.Run() + + for name := range tc.providers { + p := newMockDiscoveryProvider(tc.providers[name]) + mgr.StartCustomProvider(ctx, name, p) + } + + for i, expected := range tc.expected { + time.Sleep(expected.delay) + tgs, ok := <-mgr.SyncCh() + if !ok { + t.Fatal("discovery manager channel is closed") + } + if len(tgs) != len(expected.tgs) { + t.Fatalf("step %d: target groups mismatch, got: %d, expected: %d\ngot: %#v\nexpected: %#v", + i, len(tgs), len(expected.tgs), tgs, expected.tgs) + } + for k := range expected.tgs { + if _, ok := tgs[k]; !ok { + t.Fatalf("step %d: target group not found: %s", i, k) + } + assertEqualGroups(t, tgs[k], expected.tgs[k], func(got, expected string) string { + return fmt.Sprintf("step %d: targets mismatch \ngot: %q \nexpected: %q", i, got, expected) + }) + } + } + }) + } +} + type update struct { targetGroups []targetgroup.Group interval time.Duration @@ -875,33 +1137,37 @@ type update struct { type mockdiscoveryProvider struct { updates []update - up chan<- []*targetgroup.Group } func newMockDiscoveryProvider(updates []update) mockdiscoveryProvider { - tp := mockdiscoveryProvider{ updates: updates, } return tp } -func (tp mockdiscoveryProvider) Run(ctx context.Context, up chan<- []*targetgroup.Group) { - tp.up = up - tp.sendUpdates() -} - -func (tp mockdiscoveryProvider) sendUpdates() { - for _, update := range tp.updates { - - time.Sleep(update.interval * time.Millisecond) - - tgs := make([]*targetgroup.Group, len(update.targetGroups)) - for i := range update.targetGroups { - tgs[i] = &update.targetGroups[i] +func (tp mockdiscoveryProvider) Run(ctx context.Context, upCh chan<- []*targetgroup.Group) { + for _, u := range tp.updates { + if u.interval > 0 { + t := time.NewTicker(u.interval * time.Millisecond) + defer t.Stop() + Loop: + for { + select { + case <-ctx.Done(): + return + case <-t.C: + break Loop + } + } } - tp.up <- tgs + tgs := make([]*targetgroup.Group, len(u.targetGroups)) + for i := range u.targetGroups { + tgs[i] = &u.targetGroups[i] + } + upCh <- tgs } + <-ctx.Done() } // byGroupSource implements sort.Interface so we can sort by the Source field. From 0798f14e02de151b5966557d879f8e08be075aac Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Tue, 11 Sep 2018 17:04:33 +0200 Subject: [PATCH 2/8] Add TestCoordinationWithEmptyProvider Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 66 ++++++++++++++++++++++++++++++--------- 1 file changed, 51 insertions(+), 15 deletions(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index e776de910..13653a6da 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -885,6 +885,31 @@ scrape_configs: } } +func TestCoordinationWithEmptyProvider(t *testing.T) { + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + mgr := NewManager(ctx, nil) + mgr.updatert = 100 * time.Millisecond + go mgr.Run() + + p := emptyProvider{} + mgr.StartCustomProvider(ctx, "empty", p) + + select { + case <-ctx.Done(): + t.Fatal("no update received in the expected timeframe") + case tgs, ok := <-mgr.SyncCh(): + if !ok { + t.Fatal("discovery manager channel is closed") + } + if len(tgs) != 0 { + t.Fatalf("target groups mismatch, got: %#v, expected: {}\n", tgs) + } + } +} + func TestCoordinationWithReceiver(t *testing.T) { updateDelay := 100 * time.Millisecond @@ -1095,7 +1120,7 @@ func TestCoordinationWithReceiver(t *testing.T) { for _, tc := range testCases { tc := tc t.Run(tc.title, func(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() mgr := NewManager(ctx, nil) @@ -1109,21 +1134,25 @@ func TestCoordinationWithReceiver(t *testing.T) { for i, expected := range tc.expected { time.Sleep(expected.delay) - tgs, ok := <-mgr.SyncCh() - if !ok { - t.Fatal("discovery manager channel is closed") - } - if len(tgs) != len(expected.tgs) { - t.Fatalf("step %d: target groups mismatch, got: %d, expected: %d\ngot: %#v\nexpected: %#v", - i, len(tgs), len(expected.tgs), tgs, expected.tgs) - } - for k := range expected.tgs { - if _, ok := tgs[k]; !ok { - t.Fatalf("step %d: target group not found: %s", i, k) + select { + case <-ctx.Done(): + t.Fatal("no update received in the expected timeframe") + case tgs, ok := <-mgr.SyncCh(): + if !ok { + t.Fatal("discovery manager channel is closed") + } + if len(tgs) != len(expected.tgs) { + t.Fatalf("step %d: target groups mismatch, got: %d, expected: %d\ngot: %#v\nexpected: %#v", + i, len(tgs), len(expected.tgs), tgs, expected.tgs) + } + for k := range expected.tgs { + if _, ok := tgs[k]; !ok { + t.Fatalf("step %d: target group not found: %s", i, k) + } + assertEqualGroups(t, tgs[k], expected.tgs[k], func(got, expected string) string { + return fmt.Sprintf("step %d: targets mismatch \ngot: %q \nexpected: %q", i, got, expected) + }) } - assertEqualGroups(t, tgs[k], expected.tgs[k], func(got, expected string) string { - return fmt.Sprintf("step %d: targets mismatch \ngot: %q \nexpected: %q", i, got, expected) - }) } } }) @@ -1176,3 +1205,10 @@ type byGroupSource []*targetgroup.Group func (a byGroupSource) Len() int { return len(a) } func (a byGroupSource) Swap(i, j int) { a[i], a[j] = a[j], a[i] } func (a byGroupSource) Less(i, j int) bool { return a[i].Source < a[j].Source } + +// emptyProvider sends no updates and closes the update channel. +type emptyProvider struct{} + +func (e emptyProvider) Run(_ context.Context, ch chan<- []*targetgroup.Group) { + close(ch) +} From 4900405d2f0328c0a6f8cecdb176ecc51fd97b95 Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 12 Sep 2018 10:15:57 +0200 Subject: [PATCH 3/8] Refactor TestCoordinationWithReceiver() to work with any Discoverer Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 143 ++++++++++++++++++++++++-------------- 1 file changed, 92 insertions(+), 51 deletions(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index 13653a6da..4b7297a2b 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -667,7 +667,7 @@ func TestTargetUpdatesOrder(t *testing.T) { provUpdates := make(chan []*targetgroup.Group) for _, up := range tc.updates { - go newMockDiscoveryProvider(up).Run(ctx, provUpdates) + go newMockDiscoveryProvider(up...).Run(ctx, provUpdates) if len(up) > 0 { totalUpdatesCount = totalUpdatesCount + len(up) } @@ -885,31 +885,6 @@ scrape_configs: } } -func TestCoordinationWithEmptyProvider(t *testing.T) { - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - mgr := NewManager(ctx, nil) - mgr.updatert = 100 * time.Millisecond - go mgr.Run() - - p := emptyProvider{} - mgr.StartCustomProvider(ctx, "empty", p) - - select { - case <-ctx.Done(): - t.Fatal("no update received in the expected timeframe") - case tgs, ok := <-mgr.SyncCh(): - if !ok { - t.Fatal("discovery manager channel is closed") - } - if len(tgs) != 0 { - t.Fatalf("target groups mismatch, got: %#v, expected: {}\n", tgs) - } - } -} - func TestCoordinationWithReceiver(t *testing.T) { updateDelay := 100 * time.Millisecond @@ -920,13 +895,76 @@ func TestCoordinationWithReceiver(t *testing.T) { testCases := []struct { title string - providers map[string][]update + providers map[string]Discoverer expected []expect }{ + { + title: "Receiver should get an empty map even when the provider sends nothing and closes the channel", + providers: map[string]Discoverer{ + "empty": &onceProvider{}, + }, + expected: []expect{ + { + tgs: map[string][]*targetgroup.Group{}, + }, + }, + }, + { + title: "Receiver should get updates even when one provider closes its channel", + providers: map[string]Discoverer{ + "once1": &onceProvider{ + tgs: []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + "mock1": newMockDiscoveryProvider( + update{ + interval: 2 * updateDelay / time.Millisecond, + targetGroups: []targetgroup.Group{ + { + Source: "tg2", + Targets: []model.LabelSet{{"__instance__": "2"}}, + }, + }, + }, + ), + }, + expected: []expect{ + { + tgs: map[string][]*targetgroup.Group{ + "once1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + }, + }, + { + tgs: map[string][]*targetgroup.Group{ + "once1": []*targetgroup.Group{ + { + Source: "tg1", + Targets: []model.LabelSet{{"__instance__": "1"}}, + }, + }, + "mock1": []*targetgroup.Group{ + { + Source: "tg2", + Targets: []model.LabelSet{{"__instance__": "2"}}, + }, + }, + }, + }, + }, + }, { title: "Receiver should get updates even when the channel is blocked", - providers: map[string][]update{ - "mock1": []update{ + providers: map[string]Discoverer{ + "mock1": newMockDiscoveryProvider( update{ targetGroups: []targetgroup.Group{ { @@ -944,7 +982,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, }, - }, + ), }, expected: []expect{ { @@ -977,8 +1015,8 @@ func TestCoordinationWithReceiver(t *testing.T) { }, { title: "The receiver gets an update when a target group is gone", - providers: map[string][]update{ - "mock1": []update{ + providers: map[string]Discoverer{ + "mock1": newMockDiscoveryProvider( update{ targetGroups: []targetgroup.Group{ { @@ -996,7 +1034,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, }, - }, + ), }, expected: []expect{ { @@ -1023,8 +1061,8 @@ func TestCoordinationWithReceiver(t *testing.T) { }, { title: "The receiver gets merged updates", - providers: map[string][]update{ - "mock1": []update{ + providers: map[string]Discoverer{ + "mock1": newMockDiscoveryProvider( // This update should never be seen by the receiver because // it is overwritten by the next one. update{ @@ -1043,7 +1081,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, }, - }, + ), }, expected: []expect{ { @@ -1060,8 +1098,8 @@ func TestCoordinationWithReceiver(t *testing.T) { }, { title: "Discovery with multiple providers", - providers: map[string][]update{ - "mock1": []update{ + providers: map[string]Discoverer{ + "mock1": newMockDiscoveryProvider( // This update is available in the first receive. update{ targetGroups: []targetgroup.Group{ @@ -1071,8 +1109,8 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, }, - }, - "mock2": []update{ + ), + "mock2": newMockDiscoveryProvider( // This update should only arrive after the receiver has read from the channel once. update{ interval: 2 * updateDelay / time.Millisecond, @@ -1082,8 +1120,7 @@ func TestCoordinationWithReceiver(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "2"}}, }, }, - }, - }, + }), }, expected: []expect{ { @@ -1127,8 +1164,7 @@ func TestCoordinationWithReceiver(t *testing.T) { mgr.updatert = updateDelay go mgr.Run() - for name := range tc.providers { - p := newMockDiscoveryProvider(tc.providers[name]) + for name, p := range tc.providers { mgr.StartCustomProvider(ctx, name, p) } @@ -1136,10 +1172,10 @@ func TestCoordinationWithReceiver(t *testing.T) { time.Sleep(expected.delay) select { case <-ctx.Done(): - t.Fatal("no update received in the expected timeframe") + t.Fatalf("step %d: no update received in the expected timeframe", i) case tgs, ok := <-mgr.SyncCh(): if !ok { - t.Fatal("discovery manager channel is closed") + t.Fatalf("step %d: discovery manager channel is closed", i) } if len(tgs) != len(expected.tgs) { t.Fatalf("step %d: target groups mismatch, got: %d, expected: %d\ngot: %#v\nexpected: %#v", @@ -1147,7 +1183,7 @@ func TestCoordinationWithReceiver(t *testing.T) { } for k := range expected.tgs { if _, ok := tgs[k]; !ok { - t.Fatalf("step %d: target group not found: %s", i, k) + t.Fatalf("step %d: target group not found: %s\ngot: %#v", i, k, tgs) } assertEqualGroups(t, tgs[k], expected.tgs[k], func(got, expected string) string { return fmt.Sprintf("step %d: targets mismatch \ngot: %q \nexpected: %q", i, got, expected) @@ -1168,7 +1204,7 @@ type mockdiscoveryProvider struct { updates []update } -func newMockDiscoveryProvider(updates []update) mockdiscoveryProvider { +func newMockDiscoveryProvider(updates ...update) mockdiscoveryProvider { tp := mockdiscoveryProvider{ updates: updates, } @@ -1206,9 +1242,14 @@ func (a byGroupSource) Len() int { return len(a) } func (a byGroupSource) Swap(i, j int) { a[i], a[j] = a[j], a[i] } func (a byGroupSource) Less(i, j int) bool { return a[i].Source < a[j].Source } -// emptyProvider sends no updates and closes the update channel. -type emptyProvider struct{} +// onceProvider sends updates once (if any) and closes the update channel. +type onceProvider struct { + tgs []*targetgroup.Group +} -func (e emptyProvider) Run(_ context.Context, ch chan<- []*targetgroup.Group) { +func (o onceProvider) Run(_ context.Context, ch chan<- []*targetgroup.Group) { + if len(o.tgs) > 0 { + ch <- o.tgs + } close(ch) } From 1cee5b5b068dc9eeca0927f97c717d6fc38d1c13 Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 12 Sep 2018 10:20:34 +0200 Subject: [PATCH 4/8] Don't multiple the interval value by 1ms in the mock Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 46 +++++++++++++++++++-------------------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index 4b7297a2b..d16a8b5e3 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -65,7 +65,7 @@ func TestTargetUpdatesOrder(t *testing.T) { "tp1": { { targetGroups: []targetgroup.Group{}, - interval: 5, + interval: 5 * time.Millisecond, }, }, }, @@ -79,19 +79,19 @@ func TestTargetUpdatesOrder(t *testing.T) { "tp1": { { targetGroups: []targetgroup.Group{}, - interval: 5, + interval: 5 * time.Millisecond, }, }, "tp2": { { targetGroups: []targetgroup.Group{}, - interval: 200, + interval: 200 * time.Millisecond, }, }, "tp3": { { targetGroups: []targetgroup.Group{}, - interval: 100, + interval: 100 * time.Millisecond, }, }, }, @@ -156,7 +156,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "3"}}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, }, }, @@ -214,7 +214,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, }, }, @@ -273,7 +273,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "1"}}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, }, }, @@ -319,7 +319,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "2"}}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, { targetGroups: []targetgroup.Group{ @@ -332,7 +332,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "4"}}, }, }, - interval: 500, + interval: 500 * time.Millisecond, }, }, "tp2": { @@ -347,7 +347,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "6"}}, }, }, - interval: 100, + interval: 100 * time.Millisecond, }, { targetGroups: []targetgroup.Group{ @@ -360,7 +360,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "8"}}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, }, }, @@ -470,7 +470,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "2"}}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, { targetGroups: []targetgroup.Group{ @@ -483,7 +483,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "4"}}, }, }, - interval: 150, + interval: 150 * time.Millisecond, }, }, "tp2": { @@ -498,7 +498,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "6"}}, }, }, - interval: 200, + interval: 200 * time.Millisecond, }, { targetGroups: []targetgroup.Group{ @@ -511,7 +511,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "8"}}, }, }, - interval: 100, + interval: 100 * time.Millisecond, }, }, }, @@ -590,7 +590,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "2"}}, }, }, - interval: 30, + interval: 30 * time.Millisecond, }, { targetGroups: []targetgroup.Group{ @@ -603,7 +603,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{}, }, }, - interval: 10, + interval: 10 * time.Millisecond, }, { targetGroups: []targetgroup.Group{ @@ -616,7 +616,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Targets: []model.LabelSet{{"__instance__": "4"}}, }, }, - interval: 300, + interval: 300 * time.Millisecond, }, }, }, @@ -922,7 +922,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, "mock1": newMockDiscoveryProvider( update{ - interval: 2 * updateDelay / time.Millisecond, + interval: 2 * updateDelay, targetGroups: []targetgroup.Group{ { Source: "tg2", @@ -974,7 +974,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, update{ - interval: 4 * updateDelay / time.Millisecond, + interval: 4 * updateDelay, targetGroups: []targetgroup.Group{ { Source: "tg2", @@ -1026,7 +1026,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, update{ - interval: 2 * updateDelay / time.Millisecond, + interval: 2 * updateDelay, targetGroups: []targetgroup.Group{ { Source: "tg1", @@ -1113,7 +1113,7 @@ func TestCoordinationWithReceiver(t *testing.T) { "mock2": newMockDiscoveryProvider( // This update should only arrive after the receiver has read from the channel once. update{ - interval: 2 * updateDelay / time.Millisecond, + interval: 2 * updateDelay, targetGroups: []targetgroup.Group{ { Source: "tg2", @@ -1214,7 +1214,7 @@ func newMockDiscoveryProvider(updates ...update) mockdiscoveryProvider { func (tp mockdiscoveryProvider) Run(ctx context.Context, upCh chan<- []*targetgroup.Group) { for _, u := range tp.updates { if u.interval > 0 { - t := time.NewTicker(u.interval * time.Millisecond) + t := time.NewTicker(u.interval) defer t.Stop() Loop: for { From 82895014200ddca3ca67e8bc1fe4eb6e5d6f3d8a Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 12 Sep 2018 16:02:15 +0200 Subject: [PATCH 5/8] Address krasi's comments Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index d16a8b5e3..8e8b2fd5e 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -899,18 +899,7 @@ func TestCoordinationWithReceiver(t *testing.T) { expected []expect }{ { - title: "Receiver should get an empty map even when the provider sends nothing and closes the channel", - providers: map[string]Discoverer{ - "empty": &onceProvider{}, - }, - expected: []expect{ - { - tgs: map[string][]*targetgroup.Group{}, - }, - }, - }, - { - title: "Receiver should get updates even when one provider closes its channel", + title: "Receiver should get all updates even when one provider closes its channel", providers: map[string]Discoverer{ "once1": &onceProvider{ tgs: []*targetgroup.Group{ @@ -962,7 +951,7 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, { - title: "Receiver should get updates even when the channel is blocked", + title: "Receiver should get all updates even when the channel is blocked", providers: map[string]Discoverer{ "mock1": newMockDiscoveryProvider( update{ From 8fd891bf3f9cdb59541324323741ecb4d05e801b Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 12 Sep 2018 16:06:31 +0200 Subject: [PATCH 6/8] Speed up tests that were still using the 5s timeout Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index 8e8b2fd5e..0a896f5d5 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -761,6 +761,7 @@ scrape_configs: ctx, cancel := context.WithCancel(context.Background()) defer cancel() discoveryManager := NewManager(ctx, log.NewNopLogger()) + discoveryManager.updatert = 100 * time.Millisecond go discoveryManager.Run() c := make(map[string]sd_config.ServiceDiscoveryConfig) @@ -829,7 +830,7 @@ scrape_configs: ctx, cancel := context.WithCancel(context.Background()) defer cancel() discoveryManager := NewManager(ctx, nil) - discoveryManager.updatert = 500 * time.Millisecond + discoveryManager.updatert = 100 * time.Millisecond go discoveryManager.Run() c := make(map[string]sd_config.ServiceDiscoveryConfig) @@ -868,6 +869,7 @@ scrape_configs: ctx, cancel := context.WithCancel(context.Background()) defer cancel() discoveryManager := NewManager(ctx, log.NewNopLogger()) + discoveryManager.updatert = 100 * time.Millisecond go discoveryManager.Run() c := make(map[string]sd_config.ServiceDiscoveryConfig) From 7dc3f113063dfa0c5303c6aafc21f42ad8b240c7 Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 12 Sep 2018 16:15:03 +0200 Subject: [PATCH 7/8] WIP discovery: refactor TestTargetUpdatesOrder Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index 0a896f5d5..8dc28b945 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -658,13 +658,13 @@ func TestTargetUpdatesOrder(t *testing.T) { for i, tc := range testCases { tc := tc t.Run(tc.title, func(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() + discoveryManager := NewManager(ctx, log.NewNopLogger()) discoveryManager.updatert = 100 * time.Millisecond var totalUpdatesCount int - provUpdates := make(chan []*targetgroup.Group) for _, up := range tc.updates { go newMockDiscoveryProvider(up...).Run(ctx, provUpdates) @@ -676,7 +676,7 @@ func TestTargetUpdatesOrder(t *testing.T) { Loop: for x := 0; x < totalUpdatesCount; x++ { select { - case <-time.After(10 * time.Second): + case <-ctx.Done(): t.Errorf("%d: no update arrived within the timeout limit", x) break Loop case tgs := <-provUpdates: From e7cee1b5ba019e6b0eb7c18f99a0d2f9118ff5db Mon Sep 17 00:00:00 2001 From: Simon Pasquier Date: Wed, 12 Sep 2018 17:56:53 +0200 Subject: [PATCH 8/8] Remove tests redundant with TestTargetUpdatesOrder Signed-off-by: Simon Pasquier --- discovery/manager_test.go | 139 -------------------------------------- 1 file changed, 139 deletions(-) diff --git a/discovery/manager_test.go b/discovery/manager_test.go index 8dc28b945..d58f1643f 100644 --- a/discovery/manager_test.go +++ b/discovery/manager_test.go @@ -1004,145 +1004,6 @@ func TestCoordinationWithReceiver(t *testing.T) { }, }, }, - { - title: "The receiver gets an update when a target group is gone", - providers: map[string]Discoverer{ - "mock1": newMockDiscoveryProvider( - update{ - targetGroups: []targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - }, - update{ - interval: 2 * updateDelay, - targetGroups: []targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{}, - }, - }, - }, - ), - }, - expected: []expect{ - { - tgs: map[string][]*targetgroup.Group{ - "mock1": []*targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - }, - }, - { - tgs: map[string][]*targetgroup.Group{ - "mock1": []*targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{}, - }, - }, - }, - }, - }, - }, - { - title: "The receiver gets merged updates", - providers: map[string]Discoverer{ - "mock1": newMockDiscoveryProvider( - // This update should never be seen by the receiver because - // it is overwritten by the next one. - update{ - targetGroups: []targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "0"}}, - }, - }, - }, - update{ - targetGroups: []targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - }, - ), - }, - expected: []expect{ - { - tgs: map[string][]*targetgroup.Group{ - "mock1": []*targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - }, - }, - }, - }, - { - title: "Discovery with multiple providers", - providers: map[string]Discoverer{ - "mock1": newMockDiscoveryProvider( - // This update is available in the first receive. - update{ - targetGroups: []targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - }, - ), - "mock2": newMockDiscoveryProvider( - // This update should only arrive after the receiver has read from the channel once. - update{ - interval: 2 * updateDelay, - targetGroups: []targetgroup.Group{ - { - Source: "tg2", - Targets: []model.LabelSet{{"__instance__": "2"}}, - }, - }, - }), - }, - expected: []expect{ - { - tgs: map[string][]*targetgroup.Group{ - "mock1": []*targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - }, - }, - { - delay: 1 * updateDelay, - tgs: map[string][]*targetgroup.Group{ - "mock1": []*targetgroup.Group{ - { - Source: "tg1", - Targets: []model.LabelSet{{"__instance__": "1"}}, - }, - }, - "mock2": []*targetgroup.Group{ - { - Source: "tg2", - Targets: []model.LabelSet{{"__instance__": "2"}}, - }, - }, - }, - }, - }, - }, } for _, tc := range testCases {