memoryChunks counter fix, metics updated, small cleanup

Signed-off-by: Ganesh Vernekar <[email protected]>

Author: Ganesh Vernekar

pkg/ingester/ingester.go

@@ -48,7 +48,7 @@ type ingesterMetrics struct {
 	queriedSamples      prometheus.Histogram
 	queriedSeries       prometheus.Histogram
 	queriedChunks       prometheus.Histogram
-	walReplayDuration   prometheus.Summary
+	walReplayDuration   prometheus.Gauge
 }
 
 func newIngesterMetrics(r prometheus.Registerer) *ingesterMetrics {
@@ -87,10 +87,9 @@ func newIngesterMetrics(r prometheus.Registerer) *ingesterMetrics {
 			// A small number of chunks per series - 10*(8^(7-1)) = 2.6m.
 			Buckets: prometheus.ExponentialBuckets(10, 8, 7),
 		}),
-		walReplayDuration: prometheus.NewSummary(prometheus.SummaryOpts{
-			Name:       "cortex_ingester_wal_replay_duration_seconds",
-			Help:       "Time taken to replay the checkpoint and the WAL.",
-			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		walReplayDuration: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "cortex_ingester_wal_replay_duration_seconds",
+			Help: "Time taken to replay the checkpoint and the WAL.",
 		}),
 	}
 
@@ -103,6 +102,7 @@ func newIngesterMetrics(r prometheus.Registerer) *ingesterMetrics {
 			m.queriedSamples,
 			m.queriedSeries,
 			m.queriedChunks,
+			m.walReplayDuration,
 		)
 	}
 
@@ -247,11 +247,12 @@ func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, c
 		level.Info(util.Logger).Log("msg", "recovering from WAL")
 		start := time.Now()
 		if err := recoverFromWAL(i); err != nil {
+			level.Error(util.Logger).Log("msg", "failed to recover from WAL", "time", time.Since(start).String())
 			return nil, err
 		}
 		elapsed := time.Since(start)
 		level.Info(util.Logger).Log("msg", "recovery from WAL completed", "time", elapsed.String())
-		i.metrics.walReplayDuration.Observe(elapsed.Seconds())
+		i.metrics.walReplayDuration.Set(elapsed.Seconds())
 	}
 
 	// If the WAL recover happened, then the userStates would already be set.

pkg/ingester/series.go

@@ -204,8 +204,6 @@ func (s *memorySeries) setChunks(descs []*desc) error {
 	if len(descs) > 0 {
 		s.lastTime = descs[len(descs)-1].LastTime
 	}
-	memoryChunks.Add(float64(len(descs)))
-
 	return nil
 }
 

pkg/ingester/wal.go

@@ -72,11 +72,10 @@ type walWrapper struct {
 	checkpointDeleteTotal   prometheus.Counter
 	checkpointCreationFail  prometheus.Counter
 	checkpointCreationTotal prometheus.Counter
+	checkpointDuration      prometheus.Summary
 }
 
-// newWAL creates a WAL object.
-// * If the WAL is disabled, then the returned WAL is a no-op WAL.
-// * If WAL recovery is enabled, then the userStates is always set for ingester.
+// newWAL creates a WAL object. If the WAL is disabled, then the returned WAL is a no-op WAL.
 func newWAL(cfg WALConfig, userStatesFunc func() map[string]*userState) (WAL, error) {
 	if !cfg.walEnabled {
 		return &noopWAL{}, nil
@@ -114,12 +113,18 @@ func newWAL(cfg WALConfig, userStatesFunc func() map[string]*userState) (WAL, er
 		Name: "ingester_checkpoint_creations_total",
 		Help: "Total number of checkpoint creations attempted.",
 	})
+	w.checkpointDuration = prometheus.NewSummary(prometheus.SummaryOpts{
+		Name:       "ingester_checkpoint_duration_seconds",
+		Help:       "Time taken to create a checkpoint.",
+		Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+	})
 	if cfg.metricsRegisterer != nil {
 		cfg.metricsRegisterer.MustRegister(
 			w.checkpointDeleteFail,
 			w.checkpointDeleteTotal,
 			w.checkpointCreationFail,
 			w.checkpointCreationTotal,
+			w.checkpointDuration,
 		)
 	}
 
@@ -131,7 +136,6 @@ func newWAL(cfg WALConfig, userStatesFunc func() map[string]*userState) (WAL, er
 func (w *walWrapper) Stop() {
 	close(w.quit)
 	w.wait.Wait()
-
 	w.wal.Close()
 }
 
@@ -161,7 +165,7 @@ func (w *walWrapper) run() {
 	ticker := time.NewTicker(w.cfg.checkpointDuration)
 	defer ticker.Stop()
 
-	for !w.isStopped() {
+	for {
 		select {
 		case <-ticker.C:
 			start := time.Now()
@@ -172,7 +176,9 @@ func (w *walWrapper) run() {
 			}
 			elapsed := time.Since(start)
 			level.Info(util.Logger).Log("msg", "checkpoint done", "time", elapsed.String())
+			w.checkpointDuration.Observe(elapsed.Seconds())
 		case <-w.quit:
+			level.Info(util.Logger).Log("msg", "creating checkpoint before shutdown")
 			if err := w.performCheckpoint(); err != nil {
 				level.Error(util.Logger).Log("msg", "error checkpointing series during shutdown", "err", err)
 			}
@@ -181,15 +187,6 @@ func (w *walWrapper) run() {
 	}
 }
 
-func (w *walWrapper) isStopped() bool {
-	select {
-	case <-w.quit:
-		return true
-	default:
-		return false
-	}
-}
-
 const checkpointPrefix = "checkpoint."
 
 func (w *walWrapper) performCheckpoint() (err error) {
@@ -511,11 +508,8 @@ Loop:
 	case capturedErr = <-errChan:
 		return capturedErr
 	default:
-		if err := reader.Err(); err != nil {
-			return err
-		}
+		return reader.Err()
 	}
-	return nil
 }
 
 func copyLabelAdapters(las []client.LabelAdapter) []client.LabelAdapter {
@@ -563,6 +557,7 @@ func processCheckpointRecord(userStates *userStates, seriesPool *sync.Pool, stat
 			errChan <- err
 			return
 		}
+		memoryChunks.Add(float64(len(descs)))
 
 		seriesCache[s.UserId][s.Fingerprint] = series
 		seriesPool.Put(s)
@@ -705,16 +700,8 @@ Loop:
 	case capturedErr = <-errChan:
 		return capturedErr
 	default:
-		if err := reader.Err(); err != nil {
-			return err
-		}
+		return reader.Err()
 	}
-
-	if err != nil {
-		return err
-	}
-
-	return nil
 }
 
 func processWALSamples(userStates *userStates, stateCache map[string]*userState, seriesCache map[string]map[uint64]*memorySeries,

pkg/ring/lifecycler.go

@@ -114,11 +114,12 @@ type Lifecycler struct {
 	actorChan chan func()
 
 	// These values are initialised at startup, and never change
+	ID       string
+	Addr     string
+	RingName string
+	RingKey  string
 
-	ID              string
-	Addr            string
-	RingName        string
-	RingKey         string
+	// Whether to flush if transfer fails on shutdown.
 	flushOnShutdown bool
 
 	// We need to remember the ingester state just in case consul goes away and comes