backup: updates metric schedule options now also sets updates metric backup option

We currently set two different options on backup that control whether
two different metrics are updated:

1. `updates_last_backup_metric` schedule option
2. `updates_cluster_monitoring_metrics` backup option

These two should ideally be consolidated into one, but for now we will
ensure that setting the former will also set the latter when executing a
scheduled backup.

Epic: None

Release note: None

Author: kev-cao

pkg/ccl/backupccl/BUILD.bazel

@@ -198,6 +198,7 @@ go_test(
         "restore_span_covering_test.go",
         "restore_test.go",
         "revision_reader_test.go",
+        "schedule_exec_test.go",
         "schedule_pts_chaining_test.go",
         "show_test.go",
         "system_schema_test.go",
@@ -316,6 +317,7 @@ go_test(
         "//pkg/util/log",
         "//pkg/util/log/eventpb",
         "//pkg/util/log/logpb",
+        "//pkg/util/metric",
         "//pkg/util/mon",
         "//pkg/util/protoutil",
         "//pkg/util/randutil",

pkg/ccl/backupccl/create_scheduled_backup_test.go

@@ -143,19 +143,31 @@ func (h *testHelper) clearSchedules(t *testing.T) {
 	h.sqlDB.Exec(t, "DELETE FROM system.scheduled_jobs WHERE true")
 }
 
-func (h *testHelper) waitForSuccessfulScheduledJob(t *testing.T, scheduleID jobspb.ScheduleID) {
-	query := "SELECT id FROM " + h.env.SystemJobsTableName() +
-		" WHERE status=$1 AND created_by_type=$2 AND created_by_id=$3"
+func (h *testHelper) waitForScheduledJobState(
+	t *testing.T, scheduleID jobspb.ScheduleID, state jobs.Status,
+) {
+	query := "SELECT status FROM " + h.env.SystemJobsTableName() +
+		" WHERE created_by_type=$1 AND created_by_id=$2 ORDER BY created DESC LIMIT 1"
 
 	testutils.SucceedsSoon(t, func() error {
-		// Force newly created job to be adopted and verify it succeeds.
 		h.server.JobRegistry().(*jobs.Registry).TestingNudgeAdoptionQueue()
-		var unused int64
-		return h.sqlDB.DB.QueryRowContext(context.Background(),
-			query, jobs.StatusSucceeded, jobs.CreatedByScheduledJobs, scheduleID).Scan(&unused)
+		var status string
+		err := h.sqlDB.DB.QueryRowContext(context.Background(),
+			query, jobs.CreatedByScheduledJobs, scheduleID).Scan(&status)
+		if err != nil {
+			return err
+		} else if status != string(state) {
+			return errors.Newf("expected job in %s; found %s", state, status)
+		}
+		return nil
 	})
 }
 
+func (h *testHelper) waitForSuccessfulScheduledJob(t *testing.T, scheduleID jobspb.ScheduleID) {
+	t.Helper()
+	h.waitForScheduledJobState(t, scheduleID, jobs.StatusSucceeded)
+}
+
 func (h *testHelper) waitForSuccessfulScheduledJobCount(
 	t *testing.T, scheduleID jobspb.ScheduleID, expectedCount int,
 ) {

pkg/ccl/backupccl/schedule_exec.go

@@ -97,6 +97,21 @@ func (e *scheduledBackupExecutor) executeBackup(
 	}
 	backupStmt.AsOf = tree.AsOfClause{Expr: endTime}
 
+	// We currently set two different options that control whether two different
+	// backup metrics are updated:
+	// 1. updates_last_backup_metric on the schedule
+	// 2. updates_cluster_monitoring_metrics on the backup statement
+	// We should probably consolidate these two options int one, but for now we
+	// will ensure that setting updates_last_backup_metric on the schedule also
+	// sets updates_cluster_monitoring_metrics on the backup statement.
+	args := &backuppb.ScheduledBackupExecutionArgs{}
+	if err := pbtypes.UnmarshalAny(sj.ExecutionArgs().Args, args); err != nil {
+		return errors.Wrap(err, "un-marshaling args")
+	}
+	if args.UpdatesLastBackupMetric {
+		backupStmt.Options.UpdatesClusterMonitoringMetrics = tree.DBoolTrue
+	}
+
 	// Invoke backup plan hook.
 	hook, cleanup := cfg.PlanHookMaker(ctx, "exec-backup", txn.KV(), sj.Owner())
 	defer cleanup()

pkg/ccl/backupccl/schedule_exec_test.go

@@ -0,0 +1,81 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package backupccl
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/ccl/backupccl/backuppb"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
+	pbtypes "github.com/gogo/protobuf/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBackupSucceededUpdatesMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	ctx := context.Background()
+	executor := &scheduledBackupExecutor{
+		metrics: backupMetrics{
+			RpoMetric: metric.NewGauge(metric.Metadata{}),
+		},
+	}
+
+	schedule := createSchedule(t, true)
+	endTime := hlc.Timestamp{WallTime: hlc.UnixNano()}
+	details := jobspb.BackupDetails{EndTime: endTime}
+
+	err := executor.backupSucceeded(ctx, nil, schedule, details, nil)
+	require.NoError(t, err)
+	require.Equal(t, endTime.GoTime().Unix(), executor.metrics.RpoMetric.Value())
+}
+
+func TestBackupFailedUpdatesMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	th, cleanup := newTestHelper(t)
+	defer cleanup()
+
+	th.setOverrideAsOfClauseKnob(t)
+	schedules, err := th.createBackupSchedule(
+		t,
+		`CREATE SCHEDULE FOR
+		BACKUP INTO 'nodelocal://1/backup' WITH kms = 'aws-kms:///not-a-real-kms-jpeg?AUTH=implicit&REGION=us-east-1'
+		RECURRING '@hourly' FULL BACKUP ALWAYS
+		WITH SCHEDULE OPTIONS updates_cluster_last_backup_time_metric`,
+	)
+	require.NoError(t, err)
+	require.Len(t, schedules, 1)
+	schedule := schedules[0]
+
+	th.env.SetTime(schedule.NextRun().Add(1 * time.Second))
+	require.NoError(t, th.executeSchedules())
+	th.waitForScheduledJobState(t, schedule.ScheduleID(), jobs.StatusFailed)
+
+	metrics, ok := th.server.JobRegistry().(*jobs.Registry).MetricsStruct().Backup.(*BackupMetrics)
+	require.True(t, ok)
+
+	require.Greater(t, metrics.LastKMSInaccessibleErrorTime.Value(), int64(0))
+}
+
+func createSchedule(t *testing.T, updatesLastBackupMetric bool) *jobs.ScheduledJob {
+	schedule := jobs.NewScheduledJob(nil)
+
+	args := &backuppb.ScheduledBackupExecutionArgs{
+		UpdatesLastBackupMetric: updatesLastBackupMetric,
+	}
+	any, err := pbtypes.MarshalAny(args)
+	require.NoError(t, err)
+	schedule.SetExecutionDetails(schedule.ExecutorType(), jobspb.ExecutionArguments{Args: any})
+	return schedule
+}