pusher: simplify metrics (#1106)

pkg/pusher/metrics.go

@@ -10,45 +10,57 @@ import (
 )
 
 type metrics struct {
-	// all metrics fields must be exported
-	// to be able to return them by Metrics()
-	// using reflection
-
-	TotalChunksToBeSentCounter prometheus.Counter
-	TotalChunksSynced          prometheus.Counter
-	ErrorSettingChunkToSynced  prometheus.Counter
-	MarkAndSweepTimer          prometheus.Histogram
+	TotalToPush      prometheus.Counter
+	TotalSynced      prometheus.Counter
+	TotalErrors      prometheus.Counter
+	MarkAndSweepTime prometheus.Histogram
+	SyncTime         prometheus.Histogram
+	ErrorTime        prometheus.Histogram
 }
 
 func newMetrics() metrics {
-	subsystem := "pushsync"
+	subsystem := "pusher"
 
 	return metrics{
-		TotalChunksToBeSentCounter: prometheus.NewCounter(prometheus.CounterOpts{
+		TotalToPush: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: m.Namespace,
 			Subsystem: subsystem,
-			Name:      "total_chunk_to_be_sent",
-			Help:      "Total chunks to be sent.",
+			Name:      "total_to_push",
+			Help:      "Total chunks to push (chunks may be repeated).",
 		}),
-		TotalChunksSynced: prometheus.NewCounter(prometheus.CounterOpts{
+		TotalSynced: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: m.Namespace,
 			Subsystem: subsystem,
-			Name:      "total_chunk_synced",
+			Name:      "total_synced",
 			Help:      "Total chunks synced successfully with valid receipts.",
 		}),
-		ErrorSettingChunkToSynced: prometheus.NewCounter(prometheus.CounterOpts{
+		TotalErrors: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: m.Namespace,
 			Subsystem: subsystem,
-			Name:      "cannot_set_chunk_sync_in_db",
-			Help:      "Total no of times the chunk cannot be synced in DB.",
+			Name:      "total_errors",
+			Help:      "Total errors encountered.",
 		}),
-		MarkAndSweepTimer: prometheus.NewHistogram(prometheus.HistogramOpts{
+		MarkAndSweepTime: prometheus.NewHistogram(prometheus.HistogramOpts{
 			Namespace: m.Namespace,
 			Subsystem: subsystem,
-			Name:      "mark_and_sweep_time_histogram",
+			Name:      "mark_and_sweep_time",
 			Help:      "Histogram of time spent in mark and sweep.",
 			Buckets:   []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 60},
 		}),
+		SyncTime: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "sync_time",
+			Help:      "Histogram of time spent to sync a chunk.",
+			Buckets:   []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 60},
+		}),
+		ErrorTime: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "error_time",
+			Help:      "Histogram of time spent before giving up on syncing a chunk.",
+			Buckets:   []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 60},
+		}),
 	}
 }
 

pkg/pusher/pusher.go

@@ -96,7 +96,8 @@ LOOP:
 			// postpone a retry only after we've finished processing everything in index
 			timer.Reset(retryInterval)
 			chunksInBatch++
-			s.metrics.TotalChunksToBeSentCounter.Inc()
+			s.metrics.TotalToPush.Inc()
+
 			select {
 			case sem <- struct{}{}:
 			case <-s.quit:
@@ -120,11 +121,19 @@ LOOP:
 			mtx.Unlock()
 
 			go func(ctx context.Context, ch swarm.Chunk) {
-				var err error
+				var (
+					err       error
+					startTime = time.Now()
+				)
 				defer func() {
 					if err == nil {
+						s.metrics.TotalSynced.Inc()
+						s.metrics.SyncTime.Observe(time.Since(startTime).Seconds())
 						// only print this if there was no error while sending the chunk
 						s.logger.Tracef("pusher pushed chunk %s", ch.Address().String())
+					} else {
+						s.metrics.TotalErrors.Inc()
+						s.metrics.ErrorTime.Observe(time.Since(startTime).Seconds())
 					}
 					mtx.Lock()
 					delete(inflight, ch.Address().String())
@@ -145,7 +154,6 @@ LOOP:
 					s.logger.Debugf("pusher: error setting chunk as synced: %v", err)
 					return
 				}
-
 			}(ctx, ch)
 		case <-timer.C:
 			// initially timer is set to go off as well as every time we hit the end of push index
@@ -163,7 +171,7 @@ LOOP:
 
 			// reset timer to go off after retryInterval
 			timer.Reset(retryInterval)
-			s.metrics.MarkAndSweepTimer.Observe(time.Since(startTime).Seconds())
+			s.metrics.MarkAndSweepTime.Observe(time.Since(startTime).Seconds())
 
 			if span != nil {
 				span.Finish()
@@ -201,8 +209,8 @@ LOOP:
 func (s *Service) setChunkAsSynced(ctx context.Context, ch swarm.Chunk) error {
 	if err := s.storer.Set(ctx, storage.ModeSetSync, ch.Address()); err != nil {
 		s.logger.Errorf("pusher: error setting chunk as synced: %v", err)
-		s.metrics.ErrorSettingChunkToSynced.Inc()
 	}
+
 	t, err := s.tagg.Get(ch.TagID())
 	if err == nil && t != nil {
 		err = t.Inc(tags.StateSynced)