feat: Track when builders are doing work with a gauge (#13570)

pkg/bloombuild/builder/builder.go

@@ -191,6 +191,7 @@ func (b *Builder) builderLoop(c protos.PlannerForBuilder_BuilderLoopClient) erro
 			return fmt.Errorf("failed to receive task from planner: %w", err)
 		}
 
+		b.metrics.processingTask.Set(1)
 		b.metrics.taskStarted.Inc()
 		start := time.Now()
 
@@ -212,8 +213,11 @@ func (b *Builder) builderLoop(c protos.PlannerForBuilder_BuilderLoopClient) erro
 
 		b.logTaskCompleted(task, newMetas, err, start)
 		if err = b.notifyTaskCompletedToPlanner(c, task, newMetas, err); err != nil {
+			b.metrics.processingTask.Set(0)
 			return fmt.Errorf("failed to notify task completion to planner: %w", err)
 		}
+
+		b.metrics.processingTask.Set(0)
 	}
 
 	level.Debug(b.logger).Log("msg", "builder loop stopped")

pkg/bloombuild/builder/metrics.go

@@ -14,7 +14,8 @@ const (
 )
 
 type Metrics struct {
-	running prometheus.Gauge
+	running        prometheus.Gauge
+	processingTask prometheus.Gauge
 
 	taskStarted   prometheus.Counter
 	taskCompleted *prometheus.CounterVec
@@ -38,6 +39,12 @@ func NewMetrics(r prometheus.Registerer) *Metrics {
 			Name:      "running",
 			Help:      "Value will be 1 if the bloom builder is currently running on this instance",
 		}),
+		processingTask: promauto.With(r).NewGauge(prometheus.GaugeOpts{
+			Namespace: metricsNamespace,
+			Subsystem: metricsSubsystem,
+			Name:      "processing_task",
+			Help:      "Value will be 1 if the bloom builder is currently processing a task",
+		}),
 
 		taskStarted: promauto.With(r).NewCounter(prometheus.CounterOpts{
 			Namespace: metricsNamespace,