[Recovery] Metrics

pkg/foundation/metrics/measure/measure.go

@@ -28,6 +28,16 @@ var (
 	PipelinesGauge = metrics.NewLabeledGauge("conduit_pipelines",
 		"Number of pipelines by status.",
 		[]string{"status"})
+	PipelineStatusGauge = metrics.NewLabeledGauge(
+		"conduit_pipeline_status",
+		"Pipeline statuses.",
+		[]string{"pipeline_name", "status"},
+	)
+	PipelineRecoveringCount = metrics.NewLabeledCounter(
+		"pipeline_recovering_count",
+		"Number of times pipelines have been recovering (by by pipeline name)",
+		[]string{"pipeline_name"},
+	)
 	ConnectorsGauge = metrics.NewLabeledGauge("conduit_connectors",
 		"Number of connectors by type.",
 		[]string{"type"})

pkg/foundation/metrics/metrics.go

@@ -68,7 +68,12 @@ type Gauge interface {
 
 // LabeledGauge describes a gauge that must have values populated before use.
 type LabeledGauge interface {
+	// WithValues returns the Gauge for the given slice of label
+	// values (same order as the label names used when creating this LabeledGauge).
+	// If that combination of label values is accessed for the first time,
+	// a new Gauge is created.
 	WithValues(labels ...string) Gauge
+	DeleteLabels(labels ...string) bool
 }
 
 // Timer is a metric that allows collecting the duration of an action in
@@ -318,6 +323,15 @@ func (mt *labeledGauge) WithValues(vs ...string) Gauge {
 	return g
 }
 
+func (mt *labeledGauge) DeleteLabels(labels ...string) bool {
+	deleted := len(labels) > 0
+	for _, m := range mt.metrics {
+		deleted = deleted && m.DeleteLabels(labels...)
+	}
+
+	return deleted
+}
+
 type timer struct {
 	spec
 	metrics []Timer

pkg/foundation/metrics/noop/metrics.go

@@ -40,6 +40,7 @@ func (Gauge) Set(float64)    {}
 type LabeledGauge struct{}
 
 func (LabeledGauge) WithValues(...string) metrics.Gauge { return Gauge{} }
+func (LabeledGauge) DeleteLabels(...string) bool        { return true }
 
 type Timer struct{}
 

pkg/foundation/metrics/prometheus/gauge.go

@@ -27,6 +27,10 @@ func (lg *labeledGauge) WithValues(labels ...string) metrics.Gauge {
 	return &gauge{pg: lg.pg.WithLabelValues(labels...)}
 }
 
+func (lg *labeledGauge) DeleteLabels(labels ...string) bool {
+	return lg.pg.DeleteLabelValues(labels...)
+}
+
 func (lg *labeledGauge) Describe(c chan<- *prometheus.Desc) {
 	lg.pg.Describe(c)
 }
@@ -44,7 +48,6 @@ func (g *gauge) Inc(vs ...float64) {
 		g.pg.Inc()
 		return
 	}
-
 	g.pg.Add(sumFloat64(vs...))
 }
 

pkg/pipeline/lifecycle.go

@@ -593,10 +593,12 @@ func (s *Service) runPipeline(ctx context.Context, pl *Instance) error {
 		})
 	}
 
-	measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Dec()
+	s.updateOldStatusMetrics(pl)
+
 	pl.SetStatus(StatusRunning)
 	pl.Error = ""
-	measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Inc()
+
+	s.updateNewStatusMetrics(pl)
 
 	err := s.store.Set(ctx, pl.ID, pl)
 	if err != nil {
@@ -613,7 +615,7 @@ func (s *Service) runPipeline(ctx context.Context, pl *Instance) error {
 		nodesWg.Wait()
 		err := pl.t.Err()
 
-		measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Dec()
+		s.updateOldStatusMetrics(pl)
 
 		switch err {
 		case tomb.ErrStillAlive:
@@ -640,7 +642,7 @@ func (s *Service) runPipeline(ctx context.Context, pl *Instance) error {
 		s.notify(pl.ID, err)
 		// It's important to update the metrics before we handle the error from s.Store.Set() (if any),
 		// since the source of the truth is the actual pipeline (stored in memory).
-		measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Inc()
+		s.updateNewStatusMetrics(pl)
 
 		storeErr := s.store.Set(ctx, pl.ID, pl)
 		if storeErr != nil {

pkg/pipeline/service.go

@@ -85,7 +85,8 @@ func (s *Service) Init(ctx context.Context) error {
 			// change status to "systemStopped" to mark which pipeline was running
 			instance.SetStatus(StatusSystemStopped)
 		}
-		measure.PipelinesGauge.WithValues(strings.ToLower(instance.GetStatus().String())).Inc()
+
+		s.updateNewStatusMetrics(instance)
 	}
 
 	s.logger.Info(ctx).Int("count", len(s.instances)).Msg("pipelines initialized")
@@ -143,7 +144,8 @@ func (s *Service) Create(ctx context.Context, id string, cfg Config, p Provision
 
 	s.instances[pl.ID] = pl
 	s.instanceNames[cfg.Name] = true
-	measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Inc()
+
+	s.updateNewStatusMetrics(pl)
 
 	return pl, nil
 }
@@ -308,7 +310,8 @@ func (s *Service) Delete(ctx context.Context, pipelineID string) error {
 
 	delete(s.instances, pl.ID)
 	delete(s.instanceNames, pl.Config.Name)
-	measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Dec()
+
+	s.updateOldStatusMetrics(pl)
 
 	return nil
 }
@@ -363,3 +366,13 @@ func (s *Service) validatePipeline(cfg Config, id string) error {
 
 	return cerrors.Join(errs...)
 }
+
+func (s *Service) updateOldStatusMetrics(pl *Instance) {
+	measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Dec()
+	measure.PipelineStatusGauge.DeleteLabels(pl.Config.Name, strings.ToLower(pl.GetStatus().String()))
+}
+
+func (s *Service) updateNewStatusMetrics(pl *Instance) {
+	measure.PipelinesGauge.WithValues(strings.ToLower(pl.GetStatus().String())).Inc()
+	measure.PipelineStatusGauge.WithValues(pl.Config.Name, strings.ToLower(pl.GetStatus().String())).Set(1)
+}