Skip to content

Commit 043d710

Browse files
bwplotkabwplotka
committed
alternative
Signed-off-by: bwplotka <[email protected]>
1 parent 0f38319 commit 043d710

2 files changed

Lines changed: 33 additions & 59 deletions

File tree

‎discovery/manager.go‎

Lines changed: 18 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"sync"
2323
"time"
2424

25+
"github.com/cenkalti/backoff/v5"
2526
"github.com/prometheus/client_golang/prometheus"
2627
"github.com/prometheus/common/config"
2728
"github.com/prometheus/common/promslog"
@@ -95,7 +96,7 @@ func NewManager(ctx context.Context, logger *slog.Logger, registerer prometheus.
9596
targets: make(map[poolKey]map[string]*targetgroup.Group),
9697
ctx: ctx,
9798
updatert: 5 * time.Second,
98-
triggerSend: make(chan struct{}, 1),
99+
triggerSend: make(chan struct{}, 1), // At least one element to ensure we can do a delayed read.
99100
registerer: registerer,
100101
sdMetrics: sdMetrics,
101102
}
@@ -158,17 +159,6 @@ func FeatureRegistry(fr features.Collector) func(*Manager) {
158159
}
159160
}
160161

161-
// SkipStartupWait configures the manager to skip the initial wait on startup.
162-
// This is useful for Prometheus in agent mode or serverless flavours of OTel's prometheusreceiver
163-
// which are sensitive to startup latencies.
164-
func SkipStartupWait() func(*Manager) {
165-
return func(m *Manager) {
166-
m.mtx.Lock()
167-
defer m.mtx.Unlock()
168-
m.skipStartupWait = true
169-
}
170-
}
171-
172162
// Manager maintains a set of discovery providers and sends each update to a map channel.
173163
// Targets are grouped by the target set name.
174164
type Manager struct {
@@ -206,11 +196,6 @@ type Manager struct {
206196

207197
// featureRegistry is used to track which service discovery providers are configured.
208198
featureRegistry features.Collector
209-
210-
// skipStartupWait allows the discovery manager to skip the initial wait before sending updates
211-
// to the channel. This is useful for Prometheus in agent mode or serverless flavours of OTel's prometheusreceiver
212-
// which are sensitive to startup latencies.
213-
skipStartupWait bool
214199
}
215200

216201
// Providers returns the currently configured SD providers.
@@ -253,8 +238,6 @@ func (m *Manager) ApplyConfig(cfg map[string]Configs) error {
253238
var (
254239
wg sync.WaitGroup
255240
newProviders []*Provider
256-
// triggerSync shows if we should trigger send to notify downstream of changes.
257-
triggerSync bool
258241
)
259242
for _, prov := range m.providers {
260243
// Cancel obsolete providers if it has no new subs and it has a cancel function.
@@ -269,7 +252,6 @@ func (m *Manager) ApplyConfig(cfg map[string]Configs) error {
269252

270253
prov.cancel()
271254
prov.mu.RUnlock()
272-
triggerSync = true // Trigger send to notify downstream of dropped targets
273255
continue
274256
}
275257
prov.mu.RUnlock()
@@ -281,7 +263,6 @@ func (m *Manager) ApplyConfig(cfg map[string]Configs) error {
281263

282264
m.targetsMtx.Lock()
283265
for s := range prov.subs {
284-
triggerSync = true // Trigger send because this is an existing provider (reload)
285266
refTargets = m.targets[poolKey{s, prov.name}]
286267
// Remove obsolete subs' targets.
287268
if _, ok := prov.newSubs[s]; !ok {
@@ -314,7 +295,7 @@ func (m *Manager) ApplyConfig(cfg map[string]Configs) error {
314295
// See https://github.com/prometheus/prometheus/pull/8639 for details.
315296
// This also helps making the downstream managers drop stale targets as soon as possible.
316297
// See https://github.com/prometheus/prometheus/pull/13147 for details.
317-
if triggerSync {
298+
if len(m.providers) > 0 {
318299
select {
319300
case m.triggerSend <- struct{}{}:
320301
default:
@@ -402,59 +383,47 @@ func (m *Manager) updater(ctx context.Context, p *Provider, updates chan []*targ
402383
}
403384
}
404385

405-
func (m *Manager) flushUpdates(ctx context.Context, timeout <-chan time.Time) {
406-
m.metrics.SentUpdates.Inc()
407-
select {
408-
case m.syncCh <- m.allGroups():
409-
case <-timeout:
410-
m.metrics.DelayedUpdates.Inc()
411-
m.logger.Debug("Discovery receiver's channel was full so will retry the next cycle")
412-
select {
413-
case m.triggerSend <- struct{}{}:
414-
case <-ctx.Done():
415-
return
416-
default:
417-
}
418-
}
419-
}
420-
421386
func (m *Manager) sender() {
422-
ticker := time.NewTicker(m.updatert)
423387
defer func() {
424-
ticker.Stop()
425388
close(m.syncCh)
426389
}()
427390

428-
if m.skipStartupWait {
429-
select {
430-
case <-m.triggerSend:
431-
m.flushUpdates(m.ctx, ticker.C)
432-
case <-m.ctx.Done():
433-
return
434-
}
435-
ticker.Reset(m.updatert)
391+
// Some discoverers send updates too often, so we throttle these with a backoff interval that
392+
// increased the interval up to m.updatert delay.
393+
lastSent := time.Now().Add(-1 * m.updatert)
394+
b := &backoff.ExponentialBackOff{
395+
InitialInterval: 100 * time.Millisecond,
396+
RandomizationFactor: backoff.DefaultRandomizationFactor,
397+
Multiplier: backoff.DefaultMultiplier,
398+
MaxInterval: m.updatert,
436399
}
437400

438401
for {
439402
select {
440403
case <-m.ctx.Done():
441404
return
442-
case <-ticker.C: // Some discoverers send updates too often, so we throttle these with the ticker.
405+
case <-time.After(b.NextBackOff()):
443406
select {
444407
case <-m.triggerSend:
445408
m.metrics.SentUpdates.Inc()
446409
select {
447410
case m.syncCh <- m.allGroups():
411+
lastSent = time.Now()
448412
default:
449413
m.metrics.DelayedUpdates.Inc()
450414
m.logger.Debug("Discovery receiver's channel was full so will retry the next cycle")
415+
// Ensure we don't miss this update.
451416
select {
452417
case m.triggerSend <- struct{}{}:
453418
default:
454419
}
420+
455421
}
456422
default:
457423
}
424+
if time.Since(lastSent) > m.updatert {
425+
b.Reset() // Nothing happened for a while, start again from low interval for prompt updates.
426+
}
458427
}
459428
}
460429
}

‎discovery/manager_test.go‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -799,14 +799,14 @@ func TestTargetSetTargetGroupsPresentOnStartup(t *testing.T) {
799799
{
800800
name: "startup wait with short interval succeeds",
801801
updatert: 10 * time.Millisecond,
802-
readTimeout: 100 * time.Millisecond,
802+
readTimeout: 300 * time.Millisecond,
803803
expectedTargets: 1,
804804
},
805805
{
806806
name: "skip startup wait",
807807
skipInitialWait: true,
808808
updatert: 100 * time.Hour,
809-
readTimeout: 100 * time.Millisecond,
809+
readTimeout: 300 * time.Millisecond,
810810
expectedTargets: 1,
811811
},
812812
}
@@ -839,20 +839,25 @@ func TestTargetSetTargetGroupsPresentOnStartup(t *testing.T) {
839839

840840
synctest.Wait()
841841

842-
var syncedTargets map[string][]*targetgroup.Group
843-
select {
844-
case syncedTargets = <-discoveryManager.SyncCh():
845-
case <-time.After(tc.readTimeout):
842+
timeout := time.After(tc.readTimeout)
843+
var lastSyncedTargets map[string][]*targetgroup.Group
844+
testFor:
845+
for {
846+
select {
847+
case <-timeout:
848+
break testFor
849+
case lastSyncedTargets = <-discoveryManager.SyncCh():
850+
}
846851
}
847852

848853
if tc.expectedTargets == 0 {
849-
require.Nil(t, syncedTargets)
854+
require.Nil(t, lastSyncedTargets)
850855
return
851856
}
852857

853-
require.Len(t, syncedTargets, 1)
854-
require.Len(t, syncedTargets["prometheus"], tc.expectedTargets)
855-
verifySyncedPresence(t, syncedTargets, "prometheus", "{__address__=\"foo:9090\"}", true)
858+
require.Len(t, lastSyncedTargets, 1)
859+
require.Len(t, lastSyncedTargets["prometheus"], tc.expectedTargets)
860+
verifySyncedPresence(t, lastSyncedTargets, "prometheus", "{__address__=\"foo:9090\"}", true)
856861

857862
p := pk("static", "prometheus", 0)
858863
verifyPresence(t, discoveryManager.targets, p, "{__address__=\"foo:9090\"}", true)

0 commit comments

Comments
 (0)