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

…trol

Previously, reads and writes issued to the Storage layer by the job system
and by background jobs were included in tenant accounting. This is not ideal,
since we make a best effort to exclude operations from costing if they are
not under user control. This commit excludes the activities of the job system
from cost control, as well as the following jobs:

  - Long-running migration job
  - SQL stats compaction
  - Span config reconciler

None of these jobs are triggered by user actions and none can be disabled by
users.

NOTE: A cost control exemption does not exclude CPU or Egress costs from
accounting, since those cannot be attributed to individual jobs.

Release note: None

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 {