diff --git a/pkg/util/admission/BUILD.bazel b/pkg/util/admission/BUILD.bazel index 1e1f1d1148b7..daee51daf660 100644 --- a/pkg/util/admission/BUILD.bazel +++ b/pkg/util/admission/BUILD.bazel @@ -123,6 +123,7 @@ go_test( "@com_github_cockroachdb_redact//:redact", "@com_github_cockroachdb_tokenbucket//:tokenbucket", "@com_github_guptarohit_asciigraph//:asciigraph", + "@com_github_prometheus_client_model//go", "@com_github_stretchr_testify//require", ], ) diff --git a/pkg/util/admission/work_queue.go b/pkg/util/admission/work_queue.go index c672644f3ecf..8775aa8f97b1 100644 --- a/pkg/util/admission/work_queue.go +++ b/pkg/util/admission/work_queue.go @@ -528,8 +528,8 @@ func initWorkQueue( ticker := time.NewTicker(time.Second) for { select { - case <-ticker.C: - q.gcGroupsResetUsedAndUpdateEstimators() + case tickTime := <-ticker.C: + q.gcGroupsResetUsedAndUpdateEstimators(tickTime) case <-stopCh: // Channel closed. return @@ -1343,22 +1343,24 @@ func (q *WorkQueue) granted(grantChainID grantChainID) int64 { return requestedCount } +// groupGCIdleThreshold sets how long an idle non-built-in group's +// per-group metric children stay registered before Unlink, so a +// scrape (typically 10-30s) can observe them. +const groupGCIdleThreshold = time.Minute + // gcGroupsResetUsedAndUpdateEstimators does three things: // 1. It resets group.used, which is the resource count over which the // WorkQueue does fair-sharing. That is, fair-sharing is done over // intervals that are sized at the frequency with which this // function is called (as of 1/9/26, every 1s). -// 2. It GCs groupInfo entries, if a group has seen no workload over -// the interval. Built-in groups (builtinGroupConfigs) are exempt: -// they are always installed in the holder, so dropping and -// recreating them each interval is wasted churn through -// groupInfoPool. The cost is that built-ins not routed to in the -// current mode (e.g. the system tenant key in RM mode) keep -// publishing per-group metric series with values that stay at -// zero. +// 2. It GCs groupInfo entries that have been continuously idle for +// at least groupGCIdleThreshold. Built-ins are exempt; dropping +// and recreating them is wasted pool churn. The cost is that +// built-ins not routed to in the current mode keep publishing +// per-group metric series with values that stay at zero. // 3. It updates CPU time token estimators. The estimators are only used // if mode == usesCPUTimeTokens. -func (q *WorkQueue) gcGroupsResetUsedAndUpdateEstimators() { +func (q *WorkQueue) gcGroupsResetUsedAndUpdateEstimators(now time.Time) { q.mu.Lock() defer q.mu.Unlock() q.mu.defaultCPUTimeTokenEstimator.update() @@ -1366,7 +1368,11 @@ func (q *WorkQueue) gcGroupsResetUsedAndUpdateEstimators() { // longer than desired. We could break this iteration into smaller parts if // needed. for gKey, info := range q.mu.groups { - if !gKey.isBuiltin() && info.used == 0 && !isInGroupHeap(info) { + active := info.used > 0 || isInGroupHeap(info) + if active || info.idleSince.IsZero() { + // Refresh on activity; initialize on first observation. + info.idleSince = now + } else if !gKey.isBuiltin() && now.Sub(info.idleSince) >= groupGCIdleThreshold { delete(q.mu.groups, gKey) releaseGroupInfo(info) continue @@ -1846,7 +1852,14 @@ type groupInfo struct { // simply (a) do not do used--, if used is already zero, or (b) do not do // used-- if the request was canceled. This does imply some inaccuracy in // accounting -- it can be fixed if needed. - used uint64 + used uint64 + + // idleSince is the most recent GC-tick time at which this group + // was observed active, or the first tick that visited it if it + // has never been active. The group is GC'd once + // (now - idleSince) >= groupGCIdleThreshold. + idleSince time.Time + waitingWorkHeap waitingWorkHeap openEpochsHeap openEpochsHeap diff --git a/pkg/util/admission/work_queue_test.go b/pkg/util/admission/work_queue_test.go index b88a99f9b296..8b960ac5de3b 100644 --- a/pkg/util/admission/work_queue_test.go +++ b/pkg/util/admission/work_queue_test.go @@ -25,6 +25,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/util/tracing" "github.com/cockroachdb/datadriven" "github.com/cockroachdb/pebble" + prometheusgo "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" ) @@ -335,7 +336,7 @@ func TestWorkQueueBasic(t *testing.T) { return q.String() case "gc-groups-and-reset-used": - q.gcGroupsResetUsedAndUpdateEstimators() + q.gcGroupsResetUsedAndUpdateEstimators(timeSource.Now()) return q.String() default: @@ -564,7 +565,7 @@ func runCPUTimeTokenWorkQueueTest(t *testing.T, path string) { return "" case "gc-groups-and-reset-used": - q.gcGroupsResetUsedAndUpdateEstimators() + q.gcGroupsResetUsedAndUpdateEstimators(timeSource.Now()) return "" case "set-max-cpu-groups": @@ -704,7 +705,7 @@ func TestCPUTimeTokenEstimation(t *testing.T) { q.AdmittedWorkDone(resp1, 150*time.Millisecond) if i%10 == 0 { - q.gcGroupsResetUsedAndUpdateEstimators() + q.gcGroupsResetUsedAndUpdateEstimators(time.Now()) } } @@ -736,7 +737,7 @@ func TestCPUTimeTokenEstimation(t *testing.T) { q.AdmittedWorkDone(resp2, 350*time.Millisecond) if i%10 == 0 { - q.gcGroupsResetUsedAndUpdateEstimators() + q.gcGroupsResetUsedAndUpdateEstimators(time.Now()) } } @@ -759,14 +760,12 @@ func TestCPUTimeTokenEstimation(t *testing.T) { require.NoError(t, err) checkEstimation(resp3, 250*time.Millisecond) - // This is a test of GC. If a call to update happens without any - // work happening during that interval, the tenant's estimator should be - // GCed. The first call to gcGroupsResetUsedAndUpdateEstimators resets - // the interval over which activity is checked. The second call GCes the - // per-tenant estimators. So tenant 1 & tenant 2 should use the global - // estimator, just like tenant 3. - q.gcGroupsResetUsedAndUpdateEstimators() - q.gcGroupsResetUsedAndUpdateEstimators() + // GC the per-tenant estimators (two ticks separated by the idle + // threshold), forcing tenant 1 & tenant 2 to fall back to the + // global estimator like tenant 3. + t0 := time.Now() + q.gcGroupsResetUsedAndUpdateEstimators(t0) + q.gcGroupsResetUsedAndUpdateEstimators(t0.Add(groupGCIdleThreshold + time.Second)) resp1, err = q.Admit(ctx, info1) require.NoError(t, err) checkEstimation(resp1, 250*time.Millisecond) @@ -831,7 +830,7 @@ func TestSQLCPUAdmission(t *testing.T) { for i := 0; i < 100; i++ { q.AdmittedWorkDone(resp, 50*time.Millisecond) if i%10 == 0 { - q.gcGroupsResetUsedAndUpdateEstimators() + q.gcGroupsResetUsedAndUpdateEstimators(time.Now()) } } @@ -999,7 +998,7 @@ func TestWorkQueueTokenResetRace(t *testing.T) { // This hot loop with GC calls is able to trigger the previously buggy // code by squeezing in multiple times between the token grant and // cancellation. - q.gcGroupsResetUsedAndUpdateEstimators() + q.gcGroupsResetUsedAndUpdateEstimators(time.Now()) } } }() @@ -1394,7 +1393,9 @@ func TestGCExemptsBuiltins(t *testing.T) { require.NotNil(t, high) require.True(t, rgGroupKey(highResourceGroupID).isBuiltin()) withGroupLocked(q, func() { high.used = 0 }) - q.gcGroupsResetUsedAndUpdateEstimators() + t0 := time.Now() + q.gcGroupsResetUsedAndUpdateEstimators(t0) + q.gcGroupsResetUsedAndUpdateEstimators(t0.Add(groupGCIdleThreshold + time.Second)) require.NotNil(t, getGroupLocked(q, rgGroupKey(highResourceGroupID)), "built-in group must survive GC") @@ -1433,11 +1434,12 @@ func TestGCThenLazyRecreateRecoversFromHolder(t *testing.T) { require.Equal(t, uint32(99), g.weight) require.True(t, g.cpuTimeBurstBucket.maxCPU) - // Force GC eligibility: drive used to 0. + // Force GC: drive used to 0, then tick past the idle threshold. withGroupLocked(q, func() { g.used = 0 }) - q.gcGroupsResetUsedAndUpdateEstimators() - require.Nil(t, getGroupLocked(q, key), - "idle non-built-in group should be GC'd") + t0 := time.Now() + q.gcGroupsResetUsedAndUpdateEstimators(t0) + q.gcGroupsResetUsedAndUpdateEstimators(t0.Add(groupGCIdleThreshold + time.Second)) + require.Nil(t, getGroupLocked(q, key), "idle non-built-in group should be GC'd") // Next admit recreates the container via the holder. _, err = q.Admit(context.Background(), WorkInfo{ @@ -1493,3 +1495,51 @@ func TestGroupKeyForWorkInfoSelection(t *testing.T) { }) } } + +// TestGCKeepsMetricChildVisibleAcrossScrapes admits work for a +// tenant, then simulates a sequence of GC ticks and observes +// which labeled children the per-group parent counter exposes +// (i.e., what a Prometheus scrape would see). The child for the +// tenant remains visible across sub-threshold idle ticks and +// disappears only once groupGCIdleThreshold has elapsed. +func TestGCKeepsMetricChildVisibleAcrossScrapes(t *testing.T) { + defer leaktest.AfterTest(t)() + defer log.Scope(t).Close(t) + + q, _, _, cleanup := makeCPUTimeTokenWorkQueue(t) + defer cleanup() + + const tenantID = 42 + info := WorkInfo{ + TenantID: roachpb.MustMakeTenantID(tenantID), + Priority: admissionpb.NormalPri, + } + + // scrape returns the tenant_id label values currently exposed + // on the per-group admittedCount parent. + scrape := func() []string { + var tenants []string + q.perGroupAggMetrics.primary.admittedCount.Each(nil, func(m *prometheusgo.Metric) { + for _, lp := range m.Label { + if lp.GetName() == "tenant_id" { + tenants = append(tenants, lp.GetValue()) + } + } + }) + return tenants + } + + resp, err := q.Admit(context.Background(), info) + require.NoError(t, err) + q.AdmittedWorkDone(resp, 50*time.Millisecond) + require.Contains(t, scrape(), "42") + + t0 := time.Now() + for i := 0; i < 5; i++ { + q.gcGroupsResetUsedAndUpdateEstimators(t0.Add(time.Duration(i) * time.Second)) + require.Contains(t, scrape(), "42", "sub-threshold tick %d", i) + } + + q.gcGroupsResetUsedAndUpdateEstimators(t0.Add(groupGCIdleThreshold)) + require.NotContains(t, scrape(), "42") +}