Merge pull request cockroachdb#85105 from andy-kimball/backport22.1-8…

…3005

release-22.1: multitenant: exclude background job reads/writes from tenant cost con…

pkg/ccl/backupccl/backup_job.go

@@ -936,5 +936,6 @@ func init() {
 				job: job,
 			}
 		},
+		jobs.UsesTenantCostControl,
 	)
 }

pkg/ccl/backupccl/restore_job.go

@@ -2554,5 +2554,6 @@ func init() {
 				settings: settings,
 			}
 		},
+		jobs.UsesTenantCostControl,
 	)
 }

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -69,6 +69,7 @@ func init() {
 		func(job *jobs.Job, _ *cluster.Settings) jobs.Resumer {
 			return &changefeedResumer{job: job}
 		},
+		jobs.UsesTenantCostControl,
 	)
 }
 

pkg/ccl/multitenantccl/tenantcostclient/BUILD.bazel

@@ -49,6 +49,8 @@ go_test(
         "//pkg/cloud",
         "//pkg/cloud/nodelocal",
         "//pkg/cloud/nullsink",
+        "//pkg/jobs",
+        "//pkg/jobs/jobstest",
         "//pkg/keys",
         "//pkg/kv/kvclient/kvtenant",
         "//pkg/multitenant",
@@ -63,6 +65,7 @@ go_test(
         "//pkg/sql/execinfra",
         "//pkg/sql/sqlliveness",
         "//pkg/sql/sqlliveness/slinstance",
+        "//pkg/sql/stats",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/sqlutils",

pkg/ccl/multitenantccl/tenantcostclient/tenant_side_test.go

@@ -11,6 +11,7 @@ package tenantcostclient_test
 import (
 	"bytes"
 	"context"
+	gosql "database/sql"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -30,6 +31,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/cloud/nodelocal"
 	"github.com/cockroachdb/cockroach/pkg/cloud/nullsink"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobstest"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvtenant"
 	"github.com/cockroachdb/cockroach/pkg/multitenant"
@@ -43,6 +46,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness/slinstance"
+	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
@@ -968,6 +972,97 @@ func TestSQLLivenessExemption(t *testing.T) {
 	)
 }
 
+// TestScheduledJobsConsumption verifies that the scheduled jobs system itself
+// does not consume RUs, but that the jobs it runs do consume RUs.
+func TestScheduledJobsConsumption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	st := cluster.MakeTestingClusterSettings()
+	stats.AutomaticStatisticsClusterMode.Override(ctx, &st.SV, false)
+	tenantcostclient.TargetPeriodSetting.Override(ctx, &st.SV, time.Millisecond*20)
+
+	hostServer, _, _ := serverutils.StartServer(t, base.TestServerArgs{Settings: st})
+	defer hostServer.Stopper().Stop(ctx)
+
+	testProvider := newTestProvider()
+
+	env := jobstest.NewJobSchedulerTestEnv(jobstest.UseSystemTables, timeutil.Now())
+	var zeroDuration time.Duration
+	var execSchedules func() error
+	var tenantServer serverutils.TestTenantInterface
+	var tenantDB *gosql.DB
+	tenantServer, tenantDB = serverutils.StartTenant(t, hostServer, base.TestTenantArgs{
+		TenantID:                    serverutils.TestTenantID(),
+		Settings:                    st,
+		AllowSettingClusterSettings: true,
+		TestingKnobs: base.TestingKnobs{
+			TenantTestingKnobs: &sql.TenantTestingKnobs{
+				OverrideTokenBucketProvider: func(kvtenant.TokenBucketProvider) kvtenant.TokenBucketProvider {
+					return testProvider
+				},
+			},
+			TTL: &sql.TTLTestingKnobs{
+				// Don't wait until we can use a historical query.
+				AOSTDuration: &zeroDuration,
+			},
+			JobsTestingKnobs: &jobs.TestingKnobs{
+				JobSchedulerEnv: env,
+				TakeOverJobsScheduling: func(fn func(ctx context.Context, maxSchedules int64) error) {
+					execSchedules = func() error {
+						defer tenantServer.JobRegistry().(*jobs.Registry).TestingNudgeAdoptionQueue()
+						return fn(ctx, 0)
+					}
+				},
+				IntervalOverrides: jobs.TestingIntervalOverrides{
+					// Force fast adoption and resumption.
+					Adopt:             &zeroDuration,
+					RetryInitialDelay: &zeroDuration,
+					RetryMaxDelay:     &zeroDuration,
+				},
+			},
+		},
+	})
+
+	r := sqlutils.MakeSQLRunner(tenantDB)
+	// Create a table with rows that expire after a TTL. This will trigger the
+	// creation of a TTL job.
+	r.Exec(t, "CREATE TABLE t (v INT PRIMARY KEY) WITH ("+
+		"ttl_expire_after = '1 microsecond', ttl_job_cron = '* * * * ?', ttl_delete_batch_size = 1)")
+	r.Exec(t, "INSERT INTO t SELECT x FROM generate_series(1,100) g(x)")
+	before := testProvider.waitForConsumption(t)
+
+	// Ensure the job system is not consuming RUs when scanning/claiming jobs.
+	tenantServer.JobRegistry().(*jobs.Registry).TestingNudgeAdoptionQueue()
+	env.AdvanceTime(24 * time.Hour)
+	time.Sleep(100 * time.Millisecond)
+	after := testProvider.waitForConsumption(t)
+	after.Sub(&before)
+	require.Zero(t, after.WriteBatches)
+	require.Zero(t, after.WriteBytes)
+	// Expect up to 3 batches for initial auto-stats query, schema catalog fill,
+	// and anything else that happens once during server startup but might not be
+	// done by this point.
+	require.LessOrEqual(t, after.ReadBatches, uint64(3))
+
+	// Make sure that at least 100 writes (deletes) are reported. The TTL job
+	// should not be exempt from cost control.
+	testutils.SucceedsSoon(t, func() error {
+		// Run all job schedules.
+		env.AdvanceTime(time.Minute)
+		require.NoError(t, execSchedules())
+
+		// Check consumption.
+		c := testProvider.waitForConsumption(t)
+		c.Sub(&before)
+		if c.WriteRequests < 100 {
+			return errors.New("no write requests reported")
+		}
+		return nil
+	})
+}
+
 // TestConsumption verifies consumption reporting from a tenant server process.
 func TestConsumptionChangefeeds(t *testing.T) {
 	defer leaktest.AfterTest(t)()

pkg/ccl/streamingccl/streamingest/stream_ingestion_job.go

@@ -190,5 +190,7 @@ func init() {
 		func(job *jobs.Job,
 			settings *cluster.Settings) jobs.Resumer {
 			return &streamIngestionResumer{job: job}
-		})
+		},
+		jobs.UsesTenantCostControl,
+	)
 }

pkg/ccl/streamingccl/streamingest/stream_ingestion_planning.go

@@ -206,5 +206,6 @@ func init() {
 				job: job,
 			}
 		},
+		jobs.UsesTenantCostControl,
 	)
 }

pkg/ccl/streamingccl/streamproducer/producer_job.go

@@ -120,5 +120,6 @@ func init() {
 				timer:      ts.NewTimer(),
 			}
 		},
+		jobs.UsesTenantCostControl,
 	)
 }

pkg/ccl/streamingccl/streamproducer/producer_job_test.go

@@ -131,7 +131,7 @@ func TestStreamReplicationProducerJob(t *testing.T) {
 				resumer:           r,
 				revertingFinished: waitJobFinishReverting,
 			}
-		})
+		}, jobs.UsesTenantCostControl)
 		return mt,
 			func() {
 				in <- struct{}{} // Signals the timer that a new time is assigned to time source

pkg/cli/debug_job_trace_test.go

@@ -96,6 +96,7 @@ func TestDebugJobTrace(t *testing.T) {
 				recordedSpanCh:    recordedSpanCh,
 			}
 		},
+		jobs.UsesTenantCostControl,
 	)
 
 	// Create a "backup job" but we have overridden the resumer constructor above

pkg/jobs/BUILD.bazel

@@ -30,6 +30,7 @@ go_library(
         "//pkg/base",
         "//pkg/jobs/jobspb",
         "//pkg/kv",
+        "//pkg/multitenant",
         "//pkg/roachpb",
         "//pkg/scheduledjobs",
         "//pkg/security",

pkg/jobs/adopt.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/multitenant"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -318,6 +319,12 @@ func (r *Registry) resumeJob(ctx context.Context, jobID jobspb.JobID, s sqlliven
 	}
 	resumeCtx, cancel := r.makeCtx()
 
+	// If the job's type was registered to disable tenant cost control, then
+	// exclude the job's costs from tenant accounting.
+	if opts, ok := options[payload.Type()]; ok && opts.disableTenantCostControl {
+		resumeCtx = multitenant.WithTenantCostControlExemption(resumeCtx)
+	}
+
 	if alreadyAdopted := r.addAdoptedJob(jobID, s, cancel); alreadyAdopted {
 		return nil
 	}

pkg/jobs/delegate_control_test.go

@@ -176,7 +176,7 @@ func TestJobsControlForSchedules(t *testing.T) {
 				return nil
 			},
 		}
-	})
+	}, UsesTenantCostControl)
 
 	record := Record{
 		Description: "fake job",
@@ -289,7 +289,7 @@ func TestFilterJobsControlForSchedules(t *testing.T) {
 				return nil
 			},
 		}
-	})
+	}, UsesTenantCostControl)
 
 	record := Record{
 		Description: "fake job",
@@ -394,7 +394,7 @@ func TestJobControlByType(t *testing.T) {
 					return nil
 				},
 			}
-		})
+		}, UsesTenantCostControl)
 	}
 
 	for _, jobType := range allJobTypes {

pkg/jobs/job_scheduler.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/multitenant"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
 	"github.com/cockroachdb/cockroach/pkg/security"
@@ -547,6 +548,11 @@ func StartJobSchedulerDaemon(
 	cfg *scheduledjobs.JobExecutionConfig,
 	env scheduledjobs.JobSchedulerEnv,
 ) {
+	// Since the job scheduler system is not under user control, exclude it from
+	// from cost accounting and control. Individual jobs are not part of this
+	// exclusion.
+	ctx = multitenant.WithTenantCostControlExemption(ctx)
+
 	schedulerEnv := env
 	var daemonKnobs *TestingKnobs
 	if jobsKnobs, ok := cfg.TestingKnobs.(*TestingKnobs); ok {