From 8df03fc0cdb1a3f7667d2627755b5712c27db2ef Mon Sep 17 00:00:00 2001 From: twpsfirmament Date: Sat, 31 Oct 2026 19:13:01 +0800 Subject: [PATCH 1/2] Add runs.active gauge and dashboard panel for RUNNING runs Signed-off-by: twpsfirmament --- monitoring/dashboards/flyte-execution.json | 73 ++++++++++++++++++++++ runs/repository/impl/action.go | 40 ++++++++++++ runs/repository/impl/action_test.go | 48 ++++++++++++++ 3 files changed, 161 insertions(+) diff --git a/monitoring/dashboards/flyte-execution.json b/monitoring/dashboards/flyte-execution.json index a698efddfa..7dbaa257d6 100644 --- a/monitoring/dashboards/flyte-execution.json +++ b/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" + } + ] } ] } diff --git a/runs/repository/impl/action.go b/runs/repository/impl/action.go index ee57b97363..fc1cc4c7ba 100644 --- a/runs/repository/impl/action.go +++ b/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,40 @@ func newNotificationMetrics(provider metric.MeterProvider) (*notificationMetrics return ¬ificationMetrics{notifyDeletedCount: notifyDeletedCount}, 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() + + 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 fmt.Errorf("count active runs: %w", 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 +190,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, diff --git a/runs/repository/impl/action_test.go b/runs/repository/impl/action_test.go index 78244d0f03..a3ff1ba298 100644 --- a/runs/repository/impl/action_test.go +++ b/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, From 84250d1b6dffc3afe6cf41c00a3812c10e14be34 Mon Sep 17 00:00:00 2001 From: twpsfirmament Date: Sun, 1 Nov 2026 22:25:18 +0800 Subject: [PATCH 2/2] Report active runs to Sentry every 60s Signed-off-by: twpsfirmament --- runs/repository/impl/action.go | 19 ++++++--- runs/service/active_runs_reporter.go | 49 +++++++++++++++++++++++ runs/service/active_runs_reporter_test.go | 40 ++++++++++++++++++ runs/setup.go | 7 ++++ 4 files changed, 110 insertions(+), 5 deletions(-) create mode 100644 runs/service/active_runs_reporter.go create mode 100644 runs/service/active_runs_reporter_test.go diff --git a/runs/repository/impl/action.go b/runs/repository/impl/action.go index fc1cc4c7ba..e708dd0e82 100644 --- a/runs/repository/impl/action.go +++ b/runs/repository/impl/action.go @@ -81,6 +81,17 @@ func newNotificationMetrics(provider metric.MeterProvider) (*notificationMetrics return ¬ificationMetrics{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 @@ -103,11 +114,9 @@ func registerActiveRunsGauge(provider metric.MeterProvider, db *sqlx.DB) error { ctx, cancel := context.WithTimeout(ctx, activeRunsQueryTimeout) defer cancel() - 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 fmt.Errorf("count active runs: %w", err) + count, err := CountRunningRuns(ctx, db) + if err != nil { + return err } o.ObserveInt64(active, count) return nil diff --git a/runs/service/active_runs_reporter.go b/runs/service/active_runs_reporter.go new file mode 100644 index 0000000000..6c54147e7e --- /dev/null +++ b/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)) +} diff --git a/runs/service/active_runs_reporter_test.go b/runs/service/active_runs_reporter_test.go new file mode 100644 index 0000000000..431b980af0 --- /dev/null +++ b/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) +} diff --git a/runs/setup.go b/runs/setup.go index a6f3bc2a20..91e324ca46 100644 --- a/runs/setup.go +++ b/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(