Add active runs metric (OTel gauge, Grafana panel, Sentry)

monitoring/dashboards/flyte-execution.json

@@ -1647,6 +1647,79 @@
           "legendFormat": "run"
         }
       ]
+    },
+    {
+      "id": 29,
+      "type": "row",
+      "title": "Runs service",
+      "collapsed": false,
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus"
+      },
+      "gridPos": {
+        "x": 0,
+        "y": 71,
+        "w": 24,
+        "h": 1
+      },
+      "panels": []
+    },
+    {
+      "id": 30,
+      "type": "timeseries",
+      "title": "Active runs (RUNNING)",
+      "description": "Custom gauge runs.active: number of runs (root actions) currently in the RUNNING phase, read from the runs database. max() because every runs-service replica reports the same database-wide count.",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "prometheus"
+      },
+      "gridPos": {
+        "x": 0,
+        "y": 72,
+        "w": 12,
+        "h": 8
+      },
+      "fieldConfig": {
+        "defaults": {
+          "unit": "short",
+          "custom": {
+            "drawStyle": "line",
+            "lineWidth": 1,
+            "fillOpacity": 10,
+            "showPoints": "never",
+            "stacking": {
+              "mode": "none"
+            }
+          }
+        },
+        "overrides": []
+      },
+      "options": {
+        "legend": {
+          "displayMode": "table",
+          "placement": "bottom",
+          "calcs": [
+            "lastNotNull",
+            "max"
+          ]
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "desc"
+        }
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "prometheus"
+          },
+          "expr": "max(runs_active{service_name=\"runs-service\"})",
+          "legendFormat": "running",
+          "refId": "A"
+        }
+      ]
     }
   ]
 }

runs/repository/impl/action.go

@@ -36,6 +36,8 @@ const (
 	notificationDeletedMetricName                    = "runs.notification.evictions"
 	notificationBufferAction      notificationBuffer = "action"
 	notificationBufferRun         notificationBuffer = "run"
+	activeRunsMetricName                             = "runs.active"
+	activeRunsQueryTimeout                           = 5 * time.Second
 )
 
 // Defaults for the pq.Listener behind LISTEN/NOTIFY.
@@ -79,6 +81,49 @@ func newNotificationMetrics(provider metric.MeterProvider) (*notificationMetrics
 	return &notificationMetrics{notifyDeletedCount: notifyDeletedCount}, nil
 }
 
+// CountRunningRuns returns the number of runs currently in the RUNNING phase.
+func CountRunningRuns(ctx context.Context, db *sqlx.DB) (int64, error) {
+	var count int64
+	if err := sqlx.GetContext(ctx, db, &count,
+		"SELECT COUNT(*) FROM actions WHERE parent_action_name IS NULL AND phase = $1",
+		int32(common.ActionPhase_ACTION_PHASE_RUNNING)); err != nil {
+		return 0, fmt.Errorf("count running runs: %w", err)
+	}
+	return count, nil
+}
+
+func registerActiveRunsGauge(provider metric.MeterProvider, db *sqlx.DB) error {
+	if provider == nil {
+		return nil
+	}
+	if _, ok := provider.(metricnoop.MeterProvider); ok {
+		return nil
+	}
+
+	meter := provider.Meter(notificationMeterName)
+
+	active, err := meter.Int64ObservableGauge(
+		activeRunsMetricName,
+		metric.WithDescription("Number of runs currently in the RUNNING phase"),
+	)
+	if err != nil {
+		return err
+	}
+
+	_, err = meter.RegisterCallback(func(ctx context.Context, o metric.Observer) error {
+		ctx, cancel := context.WithTimeout(ctx, activeRunsQueryTimeout)
+		defer cancel()
+
+		count, err := CountRunningRuns(ctx, db)
+		if err != nil {
+			return err
+		}
+		o.ObserveInt64(active, count)
+		return nil
+	}, active)
+	return err
+}
+
 func (m *notificationMetrics) recordDeleted(ctx context.Context, bufferFrom notificationBuffer, count int64) {
 	if m == nil || m.notifyDeletedCount == nil || count == 0 {
 		return
@@ -154,6 +199,10 @@ func NewActionRepo(
 		return nil, fmt.Errorf("register notification metrics: %w", err)
 	}
 
+	if err := registerActiveRunsGauge(meterProvider, db); err != nil {
+		return nil, fmt.Errorf("register active runs gauge: %w", err)
+	}
+
 	dsn := database.GetPostgresDsn(context.Background(), dbConfig.Postgres)
 	repo := &actionRepo{
 		db:                  db,

runs/repository/impl/action_test.go

@@ -1092,6 +1092,54 @@ func collectNotificationMetric(t *testing.T, reader *sdkmetric.ManualReader, nam
 	return metricdata.Metrics{}
 }
 
+func TestRegisterActiveRunsGaugeNoop(t *testing.T) {
+	require.NoError(t, registerActiveRunsGauge(nil, nil))
+	require.NoError(t, registerActiveRunsGauge(metricnoop.NewMeterProvider(), nil))
+}
+
+func TestActiveRunsGauge(t *testing.T) {
+	db := setupActionDB(t)
+	defer func() { db.Exec("DELETE FROM actions") }()
+
+	reader := sdkmetric.NewManualReader()
+	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	actionRepo, err := NewActionRepo(db, testDbConfig, testNotificationConfig, provider)
+	require.NoError(t, err)
+	ctx := context.Background()
+
+	createRun := func(name string, phase common.ActionPhase) {
+		_, err := actionRepo.CreateAction(ctx, &models.Run{
+			Project: "proj1",
+			Domain:  "domain1",
+			RunName: name,
+			Name:    rootActionName,
+			Phase:   int32(phase),
+		}, false)
+		require.NoError(t, err)
+	}
+	createRun("running-1", common.ActionPhase_ACTION_PHASE_RUNNING)
+	createRun("running-2", common.ActionPhase_ACTION_PHASE_RUNNING)
+	createRun("queued", common.ActionPhase_ACTION_PHASE_QUEUED)
+	createRun("succeeded", common.ActionPhase_ACTION_PHASE_SUCCEEDED)
+
+	// A RUNNING child action is not a run and must not be counted. It hangs off an
+	// already-RUNNING run, because creating a child promotes a QUEUED parent to RUNNING.
+	_, err = actionRepo.CreateAction(ctx, &models.Action{
+		Project:          "proj1",
+		Domain:           "domain1",
+		RunName:          "running-1",
+		Name:             "a1",
+		ParentActionName: sql.NullString{String: rootActionName, Valid: true},
+		Phase:            int32(common.ActionPhase_ACTION_PHASE_RUNNING),
+	}, false)
+	require.NoError(t, err)
+
+	gauge, ok := collectNotificationMetric(t, reader, activeRunsMetricName).Data.(metricdata.Gauge[int64])
+	require.True(t, ok)
+	require.Len(t, gauge.DataPoints, 1)
+	assert.Equal(t, int64(2), gauge.DataPoints[0].Value)
+}
+
 func TestNotificationBufferLimitConfig(t *testing.T) {
 	r := newNotifyTestRepo(NewNotificationConfig(
 		defaultNotificationBufferSize,

runs/service/active_runs_reporter.go

@@ -0,0 +1,49 @@
+package service
+
+import (
+	"context"
+	"time"
+
+	"github.com/flyteorg/flyte/v2/flytestdlib/logger"
+	"github.com/flyteorg/flyte/v2/flytestdlib/sentryutils"
+)
+
+const activeRunsSentryMetric = "flyte.active_runs"
+
+// ActiveRunsReporter periodically reports the number of RUNNING runs to Sentry.
+type ActiveRunsReporter struct {
+	count    func(ctx context.Context) (int64, error)
+	interval time.Duration
+	report   func(ctx context.Context, name string, value float64)
+}
+
+// NewActiveRunsReporter creates a reporter that calls count once per interval.
+func NewActiveRunsReporter(count func(ctx context.Context) (int64, error), interval time.Duration) *ActiveRunsReporter {
+	return &ActiveRunsReporter{count: count, interval: interval, report: sentryutils.Gauge}
+}
+
+// Run reports once per interval until ctx is cancelled.
+func (r *ActiveRunsReporter) Run(ctx context.Context) error {
+	ticker := time.NewTicker(r.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			r.reportOnce(ctx)
+		}
+	}
+}
+
+func (r *ActiveRunsReporter) reportOnce(ctx context.Context) {
+	ctx, cancel := context.WithTimeout(ctx, r.interval)
+	defer cancel()
+
+	n, err := r.count(ctx)
+	if err != nil {
+		logger.Warnf(ctx, "active runs reporter: %v", err)
+		return
+	}
+	r.report(ctx, activeRunsSentryMetric, float64(n))
+}

runs/service/active_runs_reporter_test.go

@@ -0,0 +1,40 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestActiveRunsReporter_ReportsRunningCount(t *testing.T) {
+	reporter := NewActiveRunsReporter(func(context.Context) (int64, error) {
+		return 3, nil
+	}, time.Minute)
+
+	var gotName string
+	var gotValue float64
+	reporter.report = func(_ context.Context, name string, value float64) {
+		gotName, gotValue = name, value
+	}
+
+	reporter.reportOnce(context.Background())
+
+	assert.Equal(t, "flyte.active_runs", gotName)
+	assert.Equal(t, float64(3), gotValue)
+}
+
+func TestActiveRunsReporter_SkipsOnError(t *testing.T) {
+	reporter := NewActiveRunsReporter(func(context.Context) (int64, error) {
+		return 0, errors.New("db down")
+	}, time.Minute)
+
+	called := false
+	reporter.report = func(context.Context, string, float64) { called = true }
+
+	reporter.reportOnce(context.Background())
+
+	assert.False(t, called)
+}

runs/setup.go

@@ -91,6 +91,13 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 			sentryutils.Flush()
 			return nil
 		})
+
+		activeRunsReporter := service.NewActiveRunsReporter(func(ctx context.Context) (int64, error) {
+			return impl.CountRunningRuns(ctx, sc.DB)
+		}, 60*time.Second)
+		sc.AddWorker("active-runs-reporter", func(ctx context.Context) error {
+			return activeRunsReporter.Run(ctx)
+		})
 	}
 
 	repo, err := repository.NewRepository(