From d9ef4e673e0c62b486c5534114dfed1dfad62297 Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Thu, 2 Sep 2021 15:24:28 -0400 Subject: [PATCH 1/7] refresh queue, remove mut --- pkg/metrics/cluster/cluster.go | 28 +++++---------------- pkg/metrics/cluster/config_watcher.go | 29 +++++++++++++++------- pkg/metrics/cluster/config_watcher_test.go | 4 +-- 3 files changed, 28 insertions(+), 33 deletions(-) diff --git a/pkg/metrics/cluster/cluster.go b/pkg/metrics/cluster/cluster.go index c1c09d6a6c5f..6b2242cc46b5 100644 --- a/pkg/metrics/cluster/cluster.go +++ b/pkg/metrics/cluster/cluster.go @@ -107,14 +107,10 @@ 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() + + c.watcher.RequestRefresh() return &empty.Empty{}, nil } @@ -144,20 +140,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..42a0799399a7 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,11 +98,13 @@ 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) } + case <-time.After(nextPoll): + w.RequestRefresh() case ev := <-w.store.Watch(): 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 +113,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{}{}: + // no-op: refresh has been scheduled + 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 @@ -123,9 +137,6 @@ func (w *configWatcher) Refresh(ctx context.Context) (err error) { return nil } - w.refreshMut.Lock() - defer w.refreshMut.Unlock() - if refreshTimeout > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, refreshTimeout) 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 From 3d681c0e8de604d3886737f018d1a7e888edd08b Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Thu, 2 Sep 2021 15:27:51 -0400 Subject: [PATCH 2/7] fix logs --- pkg/metrics/cluster/config_watcher.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/metrics/cluster/config_watcher.go b/pkg/metrics/cluster/config_watcher.go index 42a0799399a7..fdfba078c11d 100644 --- a/pkg/metrics/cluster/config_watcher.go +++ b/pkg/metrics/cluster/config_watcher.go @@ -101,9 +101,10 @@ func (w *configWatcher) run(ctx context.Context) { 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(): if err := w.handleEvent(ev); err != nil { From b1a5ad3b3f040b9a257e212161d83c571a3837a0 Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Thu, 2 Sep 2021 15:29:13 -0400 Subject: [PATCH 3/7] more logs! --- pkg/metrics/cluster/config_watcher.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/metrics/cluster/config_watcher.go b/pkg/metrics/cluster/config_watcher.go index fdfba078c11d..67e1c960687c 100644 --- a/pkg/metrics/cluster/config_watcher.go +++ b/pkg/metrics/cluster/config_watcher.go @@ -118,7 +118,7 @@ func (w *configWatcher) run(ctx context.Context) { func (w *configWatcher) RequestRefresh() { select { case w.refreshCh <- struct{}{}: - // no-op: refresh has been scheduled + level.Debug(w.log).Log("msg", "successfully scheduled a refresh") default: level.Debug(w.log).Log("msg", "ignoring request refresh: refresh already scheduled") } From 3ec13916815af9afb0ca31dbcf1d047b78a1e97f Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Thu, 2 Sep 2021 15:33:13 -0400 Subject: [PATCH 4/7] how many logs can i add? --- pkg/metrics/cluster/config_watcher.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/metrics/cluster/config_watcher.go b/pkg/metrics/cluster/config_watcher.go index 67e1c960687c..24ce37615b10 100644 --- a/pkg/metrics/cluster/config_watcher.go +++ b/pkg/metrics/cluster/config_watcher.go @@ -137,6 +137,7 @@ func (w *configWatcher) refresh(ctx context.Context) (err error) { level.Debug(w.log).Log("msg", "refresh skipped because clustering is disabled") return nil } + level.Debug(w.log).Log("msg", "starting refresh") if refreshTimeout > 0 { var cancel context.CancelFunc From 87d23d57a81431bef0377085eb91508dedd75351 Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Thu, 2 Sep 2021 15:34:56 -0400 Subject: [PATCH 5/7] how many logs can i change? --- pkg/metrics/cluster/config_watcher.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/metrics/cluster/config_watcher.go b/pkg/metrics/cluster/config_watcher.go index 24ce37615b10..d0cd222be416 100644 --- a/pkg/metrics/cluster/config_watcher.go +++ b/pkg/metrics/cluster/config_watcher.go @@ -104,9 +104,10 @@ func (w *configWatcher) run(ctx context.Context) { 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") + level.Debug(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) } From 9d6a15cebfd2093461d21fe80fcf8ef4c4f5c135 Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Tue, 7 Sep 2021 09:18:28 -0400 Subject: [PATCH 6/7] more logging --- pkg/metrics/cluster/cluster.go | 1 + pkg/metrics/cluster/config_watcher.go | 8 +++++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/pkg/metrics/cluster/cluster.go b/pkg/metrics/cluster/cluster.go index 6b2242cc46b5..fa3a80ad9261 100644 --- a/pkg/metrics/cluster/cluster.go +++ b/pkg/metrics/cluster/cluster.go @@ -110,6 +110,7 @@ func (c *Cluster) Reshard(ctx context.Context, _ *agentproto.ReshardRequest) (*e 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 } diff --git a/pkg/metrics/cluster/config_watcher.go b/pkg/metrics/cluster/config_watcher.go index d0cd222be416..16da7a7e8c91 100644 --- a/pkg/metrics/cluster/config_watcher.go +++ b/pkg/metrics/cluster/config_watcher.go @@ -104,7 +104,7 @@ func (w *configWatcher) run(ctx context.Context) { level.Error(w.log).Log("msg", "refresh failed", "err", err) } case <-time.After(nextPoll): - level.Debug(w.log).Log("msg", "reshard timer ticked, scheduling refresh") + 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") @@ -138,7 +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 } - level.Debug(w.log).Log("msg", "starting refresh") + level.Info(w.log).Log("msg", "starting refresh") if refreshTimeout > 0 { var cancel context.CancelFunc @@ -152,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 From 549be3104bda89518b875c4756cade942ca742e0 Mon Sep 17 00:00:00 2001 From: Robert Fratto Date: Tue, 7 Sep 2021 09:19:46 -0400 Subject: [PATCH 7/7] changelog --- CHANGELOG.md | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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)