feat: add schedule health monitoring metrics to cron scheduler (#2166)

* feat: add schedule health monitoring metrics to cron scheduler

Adds Prometheus metrics for cron schedule execution health:
- cron_execution_duration_seconds: histogram of execution latency (scheduler/tenant/schedule/status labels)
- cron_executions_total: counter of completed/failed executions
- cron_lock_contention_total: counter of distributed lock skips
- cron_concurrency_rejections_total: counter for global and per-tenant semaphore rejections
- cron_last_execution_timestamp: gauge tracking last execution time per schedule
- cron_active_schedules: gauge reflecting registered schedule count after each refresh

Instruments cron.go at all execution paths: global semaphore rejection,
per-tenant semaphore rejection, lock contention skip, execution completion/failure,
and schedule refresh.

Tests cover each metric function and verify label independence.

* fix: address CodeRabbit review feedback on schedule metrics

- Extend histogram buckets from 10 to 13 (max ~409s, covers 5m default timeout)
- Add DeleteCronScheduleMetrics to remove per-schedule series on deregistration
- Call DeleteCronScheduleMetrics in refreshSchedules schedule removal path
- Fix TestRecordCronExecution_failed to use delta-based assertion

* fix: remove schedule_id from execution histogram to bound cardinality

Histogram vectors create N series per unique label combination; including
schedule_id in a multi-tenant system creates unbounded cardinality.
Labels are now scheduler/tenant_id/status (matching cronExecutionsTotal).

Per-schedule staleness tracking is retained via cronLastExecutionTimestamp
(a gauge, one series per schedule - acceptable cost).

DeleteCronScheduleMetrics signature simplified accordingly; only cleans
up the per-schedule timestamp gauge series on deregistration.

* fix: bound metric cardinality and guard per-schedule series lifecycle

- Remove tenant_id from histogram and counter labels; use scheduler/status only
  to keep series count predictable regardless of tenant or schedule growth
- Split RecordCronLastExecutionTimestamp from RecordCronExecution so call sites
  can guard against resurrecting series for deregistered schedules
- Guard timestamp write in acquireLockAndExecute: only write when schedule is
  still present in s.schedules at completion time
- Call DeleteCronScheduleMetrics on the re-register path (scheduleChanged) in
  addition to the removal path, so stale series are cleaned up when a schedule
  is reconfigured

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

shared/platform/scheduler/cron.go

@@ -286,6 +286,7 @@ func (s *CronScheduler) refreshSchedules(ctx context.Context) error {
 			s.cron.Remove(entryID)
 			delete(s.entryIDs, id)
 			delete(s.schedules, id)
+			DeleteCronScheduleMetrics(s.config.Name, id)
 			s.logger.Info("removed schedule", "schedule_id", id)
 		}
 	}
@@ -300,6 +301,7 @@ func (s *CronScheduler) refreshSchedules(ctx context.Context) error {
 			s.cron.Remove(s.entryIDs[sched.ID])
 			delete(s.entryIDs, sched.ID)
 			delete(s.schedules, sched.ID)
+			DeleteCronScheduleMetrics(s.config.Name, sched.ID)
 			s.logger.Info("schedule changed, re-registering",
 				"schedule_id", sched.ID,
 				"old_cron_expr", prev.CronExpr,
@@ -323,6 +325,8 @@ func (s *CronScheduler) refreshSchedules(ctx context.Context) error {
 			"tenant_id", sched.TenantID)
 	}
 
+	UpdateCronActiveSchedules(s.config.Name, float64(len(s.entryIDs)))
+
 	return nil
 }
 
@@ -373,6 +377,7 @@ func (s *CronScheduler) acquireGlobalSemaphore(ctx context.Context, schedule Sch
 			"schedule_id", schedule.ID,
 			"tenant_id", schedule.TenantID)
 		s.recordExecution(ctx, schedule, ExecutionStatusSkipped, nil, strPtr("concurrency limit reached"))
+		RecordCronConcurrencyRejection(s.config.Name, "global")
 		return nil, false
 	}
 }
@@ -393,6 +398,7 @@ func (s *CronScheduler) acquireTenantSemaphore(ctx context.Context, schedule Sch
 			"tenant_id", schedule.TenantID)
 		s.recordExecution(ctx, schedule, ExecutionStatusSkipped, nil,
 			strPtr(fmt.Sprintf("per-tenant concurrency limit reached for tenant %s", schedule.TenantID)))
+		RecordCronConcurrencyRejection(s.config.Name, "per_tenant")
 		return nil, false
 	}
 }
@@ -458,6 +464,7 @@ func (s *CronScheduler) acquireLockAndExecute(ctx context.Context, schedule Sche
 			s.logger.Debug("lock not acquired, skipping",
 				"schedule_id", schedule.ID)
 			s.recordExecution(ctx, schedule, ExecutionStatusSkipped, nil, strPtr("lock not acquired"))
+			RecordCronLockContention(s.config.Name)
 			return
 		}
 		defer release()
@@ -467,7 +474,26 @@ func (s *CronScheduler) acquireLockAndExecute(ctx context.Context, schedule Sche
 	now := time.Now().UTC()
 	s.recordExecutionStart(ctx, execID, schedule, now)
 
+	execStart := time.Now()
 	err := s.executor.Execute(ctx, schedule)
+	execDuration := time.Since(execStart)
+
+	execStatus := ExecutionStatusCompleted
+	if err != nil {
+		execStatus = ExecutionStatusFailed
+	}
+	RecordCronExecution(s.config.Name, execStatus, execDuration)
+
+	// Only update the per-schedule timestamp gauge when the schedule is still registered.
+	// This prevents resurrecting Prometheus series for schedules that were removed
+	// while an execution was in flight.
+	s.mu.Lock()
+	_, stillRegistered := s.schedules[schedule.ID]
+	s.mu.Unlock()
+	if stillRegistered {
+		RecordCronLastExecutionTimestamp(s.config.Name, schedule.ID)
+	}
+
 	s.recordExecutionResult(ctx, execID, schedule, err)
 
 	if err != nil {

shared/platform/scheduler/metrics.go

@@ -1,6 +1,8 @@
 package scheduler
 
 import (
+	"time"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
@@ -81,3 +83,90 @@ func RecordInFlightWork(worker string, count float64) {
 func RecordPoll(worker string) {
 	workerPollTotal.WithLabelValues(worker).Inc()
 }
+
+// Prometheus metrics for cron schedule execution health.
+var (
+	cronExecutionDurationSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "meridian",
+		Subsystem: "scheduler",
+		Name:      "cron_execution_duration_seconds",
+		Help:      "Duration of cron schedule executions in seconds",
+		Buckets:   prometheus.ExponentialBuckets(0.1, 2, 13), // 0.1s to ~409s, covers 5-minute default timeout
+	}, []string{"scheduler", "status"}) // tenant_id and schedule_id omitted: cardinality scales with tenants × schedules
+
+	cronExecutionsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "meridian",
+		Subsystem: "scheduler",
+		Name:      "cron_executions_total",
+		Help:      "Total number of cron schedule executions by status",
+	}, []string{"scheduler", "status"}) // tenant_id omitted: per-schedule detail is in the execution store
+
+	cronLockContentionTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "meridian",
+		Subsystem: "scheduler",
+		Name:      "cron_lock_contention_total",
+		Help:      "Number of times a schedule was skipped because the distributed lock was already held",
+	}, []string{"scheduler"})
+
+	cronConcurrencyRejectionsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "meridian",
+		Subsystem: "scheduler",
+		Name:      "cron_concurrency_rejections_total",
+		Help:      "Number of executions skipped due to concurrency limits",
+	}, []string{"scheduler", "limit_type"})
+
+	cronLastExecutionTimestamp = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "meridian",
+		Subsystem: "scheduler",
+		Name:      "cron_last_execution_timestamp",
+		Help:      "Unix timestamp of the most recent completed or failed execution for each schedule",
+	}, []string{"scheduler", "schedule_id"})
+
+	cronActiveSchedules = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "meridian",
+		Subsystem: "scheduler",
+		Name:      "cron_active_schedules",
+		Help:      "Number of active schedules currently registered in the cron runner",
+	}, []string{"scheduler"})
+)
+
+// RecordCronExecution records duration and count metrics for a completed or failed
+// schedule execution. Uses only bounded labels (scheduler, status) to keep cardinality
+// predictable. Per-schedule/tenant detail is available in the execution store.
+func RecordCronExecution(schedulerName string, status ExecutionStatus, duration time.Duration) {
+	statusStr := string(status)
+	cronExecutionDurationSeconds.WithLabelValues(schedulerName, statusStr).Observe(duration.Seconds())
+	cronExecutionsTotal.WithLabelValues(schedulerName, statusStr).Inc()
+}
+
+// RecordCronLastExecutionTimestamp updates the last-execution gauge for a schedule.
+// Callers should only invoke this when the schedule is confirmed to still be registered,
+// to avoid resurrecting Prometheus series for deregistered schedules.
+func RecordCronLastExecutionTimestamp(schedulerName, scheduleID string) {
+	cronLastExecutionTimestamp.WithLabelValues(schedulerName, scheduleID).SetToCurrentTime()
+}
+
+// RecordCronLockContention increments the counter for schedules skipped due to distributed lock contention.
+func RecordCronLockContention(schedulerName string) {
+	cronLockContentionTotal.WithLabelValues(schedulerName).Inc()
+}
+
+// RecordCronConcurrencyRejection increments the counter for schedules skipped due to concurrency limits.
+// limitType should be "global" or "per_tenant".
+func RecordCronConcurrencyRejection(schedulerName, limitType string) {
+	cronConcurrencyRejectionsTotal.WithLabelValues(schedulerName, limitType).Inc()
+}
+
+// UpdateCronActiveSchedules sets the gauge for the number of active schedules registered.
+func UpdateCronActiveSchedules(schedulerName string, count float64) {
+	cronActiveSchedules.WithLabelValues(schedulerName).Set(count)
+}
+
+// DeleteCronScheduleMetrics removes per-schedule Prometheus series for a schedule that
+// has been deregistered. This prevents stale series from persisting after schedules are
+// removed from the provider.
+func DeleteCronScheduleMetrics(schedulerName, scheduleID string) {
+	// The histogram is labeled by tenant_id only (not schedule_id), so no per-schedule
+	// cleanup is needed there. Only the last-execution timestamp gauge is per-schedule.
+	cronLastExecutionTimestamp.DeleteLabelValues(schedulerName, scheduleID)
+}

shared/platform/scheduler/metrics_test.go

@@ -2,6 +2,7 @@ package scheduler
 
 import (
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	io_prometheus_client "github.com/prometheus/client_model/go"
@@ -25,6 +26,38 @@ func getGaugeValue(t *testing.T, vec *prometheus.GaugeVec, label string) float64
 	return m.GetGauge().GetValue()
 }
 
+func getCounterValueMulti(t *testing.T, vec *prometheus.CounterVec, labels ...string) float64 {
+	t.Helper()
+	m := &io_prometheus_client.Metric{}
+	err := vec.WithLabelValues(labels...).(prometheus.Metric).Write(m)
+	require.NoError(t, err)
+	return m.GetCounter().GetValue()
+}
+
+func getGaugeValueMulti(t *testing.T, vec *prometheus.GaugeVec, labels ...string) float64 {
+	t.Helper()
+	m := &io_prometheus_client.Metric{}
+	err := vec.WithLabelValues(labels...).(prometheus.Metric).Write(m)
+	require.NoError(t, err)
+	return m.GetGauge().GetValue()
+}
+
+func getHistogramCount(t *testing.T, vec *prometheus.HistogramVec, labels ...string) uint64 {
+	t.Helper()
+	m := &io_prometheus_client.Metric{}
+	err := vec.WithLabelValues(labels...).(prometheus.Metric).Write(m)
+	require.NoError(t, err)
+	return m.GetHistogram().GetSampleCount()
+}
+
+func getHistogramSum(t *testing.T, vec *prometheus.HistogramVec, labels ...string) float64 {
+	t.Helper()
+	m := &io_prometheus_client.Metric{}
+	err := vec.WithLabelValues(labels...).(prometheus.Metric).Write(m)
+	require.NoError(t, err)
+	return m.GetHistogram().GetSampleSum()
+}
+
 func TestRecordWorkerStart(t *testing.T) {
 	before := getCounterValue(t, workerStartsTotal, "test-worker-start")
 	RecordWorkerStart("test-worker-start")
@@ -80,3 +113,108 @@ func TestMetrics_different_workers_are_independent(t *testing.T) {
 
 	assert.Greater(t, aVal, bVal)
 }
+
+// --- Cron execution metrics ---
+
+func TestRecordCronExecution_completed(t *testing.T) {
+	sched := "test-cron-completed"
+
+	// Histogram and counter are labeled by scheduler/status only (tenant_id and schedule_id
+	// omitted to keep cardinality bounded regardless of tenant/schedule count).
+	beforeHistogram := getHistogramCount(t, cronExecutionDurationSeconds, sched, "COMPLETED")
+	beforeCounter := getCounterValueMulti(t, cronExecutionsTotal, sched, "COMPLETED")
+
+	RecordCronExecution(sched, ExecutionStatusCompleted, 500*time.Millisecond)
+
+	assert.Equal(t, beforeHistogram+1, getHistogramCount(t, cronExecutionDurationSeconds, sched, "COMPLETED"))
+	assert.Equal(t, beforeCounter+1, getCounterValueMulti(t, cronExecutionsTotal, sched, "COMPLETED"))
+}
+
+func TestRecordCronLastExecutionTimestamp(t *testing.T) {
+	sched := "test-cron-ts"
+	sid := "sched-ts"
+
+	RecordCronLastExecutionTimestamp(sched, sid)
+	assert.Greater(t, getGaugeValueMulti(t, cronLastExecutionTimestamp, sched, sid), 0.0)
+}
+
+func TestRecordCronExecution_failed(t *testing.T) {
+	sched := "test-cron-failed"
+
+	beforeHistogram := getHistogramCount(t, cronExecutionDurationSeconds, sched, "FAILED")
+	beforeCounter := getCounterValueMulti(t, cronExecutionsTotal, sched, "FAILED")
+
+	RecordCronExecution(sched, ExecutionStatusFailed, 100*time.Millisecond)
+
+	assert.Equal(t, beforeHistogram+1, getHistogramCount(t, cronExecutionDurationSeconds, sched, "FAILED"))
+	assert.Equal(t, beforeCounter+1, getCounterValueMulti(t, cronExecutionsTotal, sched, "FAILED"))
+}
+
+func TestRecordCronExecution_duration_is_observed(t *testing.T) {
+	sched := "test-cron-duration"
+
+	RecordCronExecution(sched, ExecutionStatusCompleted, 2*time.Second)
+
+	assert.GreaterOrEqual(t, getHistogramSum(t, cronExecutionDurationSeconds, sched, "COMPLETED"), 2.0)
+}
+
+func TestRecordCronLockContention(t *testing.T) {
+	sched := "test-cron-lock"
+
+	before := getCounterValueMulti(t, cronLockContentionTotal, sched)
+	RecordCronLockContention(sched)
+	assert.Equal(t, before+1, getCounterValueMulti(t, cronLockContentionTotal, sched))
+}
+
+func TestRecordCronConcurrencyRejection_global(t *testing.T) {
+	sched := "test-cron-conc-global"
+
+	before := getCounterValueMulti(t, cronConcurrencyRejectionsTotal, sched, "global")
+	RecordCronConcurrencyRejection(sched, "global")
+	assert.Equal(t, before+1, getCounterValueMulti(t, cronConcurrencyRejectionsTotal, sched, "global"))
+}
+
+func TestRecordCronConcurrencyRejection_per_tenant(t *testing.T) {
+	sched := "test-cron-conc-tenant"
+
+	before := getCounterValueMulti(t, cronConcurrencyRejectionsTotal, sched, "per_tenant")
+	RecordCronConcurrencyRejection(sched, "per_tenant")
+	assert.Equal(t, before+1, getCounterValueMulti(t, cronConcurrencyRejectionsTotal, sched, "per_tenant"))
+}
+
+func TestUpdateCronActiveSchedules(t *testing.T) {
+	sched := "test-cron-active"
+
+	UpdateCronActiveSchedules(sched, 5)
+	assert.Equal(t, 5.0, getGaugeValueMulti(t, cronActiveSchedules, sched))
+
+	UpdateCronActiveSchedules(sched, 0)
+	assert.Equal(t, 0.0, getGaugeValueMulti(t, cronActiveSchedules, sched))
+}
+
+func TestDeleteCronScheduleMetrics_removes_series(t *testing.T) {
+	sched := "test-cron-delete"
+	sid := "sched-del"
+
+	// Record a timestamp so the per-schedule series exists
+	RecordCronLastExecutionTimestamp(sched, sid)
+	assert.Greater(t, getGaugeValueMulti(t, cronLastExecutionTimestamp, sched, sid), 0.0)
+
+	// Delete should not panic and removes the per-schedule timestamp series
+	assert.NotPanics(t, func() {
+		DeleteCronScheduleMetrics(sched, sid)
+	})
+
+	// After deletion, the timestamp series resets to zero (re-created fresh)
+	assert.Equal(t, 0.0, getGaugeValueMulti(t, cronLastExecutionTimestamp, sched, sid))
+}
+
+func TestCronMetrics_independent_schedulers(t *testing.T) {
+	RecordCronLockContention("sched-x")
+	RecordCronLockContention("sched-x")
+	RecordCronLockContention("sched-y")
+
+	xVal := getCounterValueMulti(t, cronLockContentionTotal, "sched-x")
+	yVal := getCounterValueMulti(t, cronLockContentionTotal, "sched-y")
+	assert.Greater(t, xVal, yVal)
+}