diff --git a/CHANGELOG.md b/CHANGELOG.md index e4f4a43765b2..9293fde765d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,7 +33,7 @@ for specific instructions. using `-config.expand-env`. (@rfratto) - [BUGFIX] The directory of the logs positions file will now properly be created - on startup for all instances. + on startup for all instances. (@rfratto) - [BUGFIX] The Linux system packages will now configure the grafana-agent user to be a member of the adm and systemd-journal groups. This will allow logs to @@ -56,6 +56,7 @@ for specific instructions. - [BUGFIX] Fix yaml marshalling tag for cert_file in kafka exporter agent config. (@rgeyer) - [BUGFIX] Register missing metric for configstore consul request duration. + (@rfratto) - [BUGFIX] Logs should contain a caller field with file and line numbers again (@kgeckhart) @@ -65,6 +66,9 @@ for specific instructions. - [BUGFIX] In scraping service mode, scraping service can deadlock when reloading during join. (@mattdurham) +- [BUGFIX] Scraping service: prevent more than one refresh from being queued at + a time. (@rfratto) + - [CHANGE] Breaking change: reduced verbosity of tracing autologging by not logging `STATUS_CODE_UNSET` status codes. (@mapno) diff --git a/pkg/metrics/cluster/cluster.go b/pkg/metrics/cluster/cluster.go index c1c09d6a6c5f..fa3a80ad9261 100644 --- a/pkg/metrics/cluster/cluster.go +++ b/pkg/metrics/cluster/cluster.go @@ -107,14 +107,11 @@ func (c *Cluster) storeValidate(cfg *instance.Config) error { // Reshard implements agentproto.ScrapingServiceServer, and syncs the state of // configs with the configstore. func (c *Cluster) Reshard(ctx context.Context, _ *agentproto.ReshardRequest) (*empty.Empty, error) { - go func() { - c.mut.RLock() - defer c.mut.RUnlock() - err := c.watcher.Refresh(context.Background()) - if err != nil { - level.Error(c.log).Log("msg", "failed to perform local reshard", "err", err) - } - }() + c.mut.RLock() + defer c.mut.RUnlock() + + level.Info(c.log).Log("msg", "received reshard notification, requesting refresh") + c.watcher.RequestRefresh() return &empty.Empty{}, nil } @@ -144,20 +141,8 @@ func (c *Cluster) ApplyConfig( c.cfg = cfg // Force a refresh so all the configs get updated with new defaults. - level.Info(c.log).Log("msg", "cluster config changed, refreshing from configstore in background") - go func() { - ctx := context.Background() - if c.cfg.ReshardTimeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, c.cfg.ReshardTimeout) - defer cancel() - } - err := c.watcher.Refresh(ctx) - if err != nil { - level.Error(c.log).Log("msg", "failed to perform local reshard", "err", err) - } - }() - + level.Info(c.log).Log("msg", "cluster config changed, queueing refresh") + c.watcher.RequestRefresh() return nil } diff --git a/pkg/metrics/cluster/config_watcher.go b/pkg/metrics/cluster/config_watcher.go index 18881e4bfa31..16da7a7e8c91 100644 --- a/pkg/metrics/cluster/config_watcher.go +++ b/pkg/metrics/cluster/config_watcher.go @@ -37,7 +37,7 @@ type configWatcher struct { owns OwnershipFunc validate ValidationFunc - refreshMut sync.Mutex + refreshCh chan struct{} instanceMut sync.Mutex instances map[string]struct{} } @@ -63,6 +63,7 @@ func newConfigWatcher(log log.Logger, cfg Config, store configstore.Store, im in owns: owns, validate: validate, + refreshCh: make(chan struct{}, 1), instances: make(map[string]struct{}), } if err := w.ApplyConfig(cfg); err != nil { @@ -97,12 +98,16 @@ func (w *configWatcher) run(ctx context.Context) { select { case <-ctx.Done(): return - case <-time.After(nextPoll): - err := w.Refresh(ctx) + case <-w.refreshCh: + err := w.refresh(ctx) if err != nil { - level.Error(w.log).Log("msg", "failed polling refresh", "err", err) + level.Error(w.log).Log("msg", "refresh failed", "err", err) } + case <-time.After(nextPoll): + level.Info(w.log).Log("msg", "reshard timer ticked, scheduling refresh") + w.RequestRefresh() case ev := <-w.store.Watch(): + level.Debug(w.log).Log("msg", "handling event from config store") if err := w.handleEvent(ev); err != nil { level.Error(w.log).Log("msg", "failed to handle changed or deleted config", "key", ev.Key, "err", err) } @@ -110,9 +115,20 @@ func (w *configWatcher) run(ctx context.Context) { } } -// Refresh reloads all configs from the configstore. Deleted configs will be -// removed. -func (w *configWatcher) Refresh(ctx context.Context) (err error) { +// RequestRefresh will queue a refresh. No more than one refresh can be queued at a time. +func (w *configWatcher) RequestRefresh() { + select { + case w.refreshCh <- struct{}{}: + level.Debug(w.log).Log("msg", "successfully scheduled a refresh") + default: + level.Debug(w.log).Log("msg", "ignoring request refresh: refresh already scheduled") + } +} + +// refresh reloads all configs from the configstore. Deleted configs will be +// removed. refresh may not be called concurrently and must only be invoked from run. +// Call RequestRefresh to queue a call to refresh. +func (w *configWatcher) refresh(ctx context.Context) (err error) { w.mut.Lock() enabled := w.cfg.Enabled refreshTimeout := w.cfg.ReshardTimeout @@ -122,9 +138,7 @@ func (w *configWatcher) Refresh(ctx context.Context) (err error) { level.Debug(w.log).Log("msg", "refresh skipped because clustering is disabled") return nil } - - w.refreshMut.Lock() - defer w.refreshMut.Unlock() + level.Info(w.log).Log("msg", "starting refresh") if refreshTimeout > 0 { var cancel context.CancelFunc @@ -138,7 +152,9 @@ func (w *configWatcher) Refresh(ctx context.Context) (err error) { if err != nil { success = "0" } - reshardDuration.WithLabelValues(success).Observe(time.Since(start).Seconds()) + duration := time.Since(start) + level.Info(w.log).Log("msg", "refresh finished", "duration", duration, "success", success, "err", err) + reshardDuration.WithLabelValues(success).Observe(duration.Seconds()) }() // This is used to determine if the context was already exceeded before calling the kv provider diff --git a/pkg/metrics/cluster/config_watcher_test.go b/pkg/metrics/cluster/config_watcher_test.go index 48ca2da0cb3d..2d9eef47f51c 100644 --- a/pkg/metrics/cluster/config_watcher_test.go +++ b/pkg/metrics/cluster/config_watcher_test.go @@ -48,7 +48,7 @@ func Test_configWatcher_Refresh(t *testing.T) { return ch, nil } - err = w.Refresh(context.Background()) + err = w.refresh(context.Background()) require.NoError(t, err) // Then: return a "new" config. @@ -61,7 +61,7 @@ func Test_configWatcher_Refresh(t *testing.T) { return ch, nil } - err = w.Refresh(context.Background()) + err = w.refresh(context.Background()) require.NoError(t, err) // "hello" and "new" should've been applied, and "hello" should've been deleted