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/backup/create_scheduled_backup_test.go

@@ -148,26 +148,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) {
-	t.Helper()
+func (h *testHelper) waitForScheduledJobState(
+	t *testing.T, scheduleID jobspb.ScheduleID, state jobs.State,
+) {
 	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 status string
 		err := h.sqlDB.DB.QueryRowContext(context.Background(),
 			query, jobs.CreatedByScheduledJobs, scheduleID).Scan(&status)
 		if err != nil {
 			return err
-		} else if status != string(jobs.StateSucceeded) {
-			return errors.Newf("expected job to succeed; found %s", status)
+		} 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.StateSucceeded)
+}
+
 func (h *testHelper) waitForSuccessfulScheduledJobCount(
 	t *testing.T, scheduleID jobspb.ScheduleID, expectedCount int,
 ) {

pkg/backup/schedule_exec.go

@@ -98,6 +98,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/backup/schedule_exec_test.go

@@ -8,13 +8,15 @@ package backup
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/backup/backuppb"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"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"
@@ -58,6 +60,35 @@ func TestBackupSucceededUpdatesMetrics(t *testing.T) {
 	})
 }
 
+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.StateFailed)
+
+	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)