diff --git a/docs/generated/settings/settings.html b/docs/generated/settings/settings.html index 38bfb632934d..f620b9a0894c 100644 --- a/docs/generated/settings/settings.html +++ b/docs/generated/settings/settings.html @@ -140,6 +140,7 @@
kv.transaction.write_pipelining.max_batch_size
(alias: kv.transaction.write_pipelining_max_batch_size)
integer128if non-zero, defines that maximum size batch that will be pipelined through Raft consensusBasic/Standard/Advanced/Self-Hosted
kvadmission.store.provisioned_bandwidth
byte size0 Bif set to a non-zero value, this is used as the provisioned bandwidth (in bytes/s), for each store. It can be overridden on a per-store basis using the --store flag. Note that setting the provisioned bandwidth to a positive value may enable disk bandwidth based admission control, since admission.disk_bandwidth_tokens.elastic.enabled defaults to trueAdvanced/Self-Hosted
kvadmission.store.snapshot_ingest_bandwidth_control.enabled
booleantrueif set to true, snapshot ingests will be subject to disk write control in ACAdvanced/Self-Hosted +
kvadmission.store.snapshot_ingest_bandwidth_control.min_rate.enabled
booleantrueif set to true, snapshot ingests will be admitted at a minimum rate when kvadmission.store.provisioned_bandwidth is set to a non-zero value. Disabling this setting can lead to snapshots being starved out by foreground traffic.Advanced/Self-Hosted
log.channel_compatibility_mode.enabled
booleanfalsewhen true, logs will to log to their legacy (pre 26.1) logging channels; when false, logs will be logged to new logging channelsBasic/Standard/Advanced/Self-Hosted
obs.tablemetadata.automatic_updates.enabled
booleanfalseenables automatic updates of the table metadata cache system.table_metadataBasic/Standard/Advanced/Self-Hosted
obs.tablemetadata.data_valid_duration
duration20m0sthe duration for which the data in system.table_metadata is considered validBasic/Standard/Advanced/Self-Hosted diff --git a/pkg/kv/kvserver/kv_snapshot_strategy.go b/pkg/kv/kvserver/kv_snapshot_strategy.go index d02679457b19..48265cd41474 100644 --- a/pkg/kv/kvserver/kv_snapshot_strategy.go +++ b/pkg/kv/kvserver/kv_snapshot_strategy.go @@ -18,6 +18,7 @@ import ( "github.com/cockroachdb/cockroach/pkg/storage/fs" "github.com/cockroachdb/cockroach/pkg/util/admission" "github.com/cockroachdb/cockroach/pkg/util/log" + "github.com/cockroachdb/cockroach/pkg/util/timeutil" "github.com/cockroachdb/cockroach/pkg/util/tracing" "github.com/cockroachdb/cockroach/pkg/util/uuid" "github.com/cockroachdb/errors" @@ -142,7 +143,37 @@ func (kvSS *kvBatchSnapshotStrategy) Receive( // Using a nil pacer is effectively a noop if snapshot control is disabled. var pacer *admission.SnapshotPacer = nil if admission.DiskBandwidthForSnapshotIngest.Get(&s.cfg.Settings.SV) && snapshotQ != nil { - pacer = admission.NewSnapshotPacer(snapshotQ) + minRate := int64(0) + if admission.DiskBandwidthForSnapshotIngestMinRateEnabled.Get(&s.cfg.Settings.SV) { + minFractionOfTimeoutForApplyingSnapshot := 1 - snapshotReservationQueueTimeoutFraction.Get(&s.cfg.Settings.SV) + // Use a slowdown factor of half the permittedRangeScanSlowdown + // factor, which means we allow snapshots to ingest atleast as fast + // as would be necessary to complete within half of a snapshots + // timeout duration. + snapshotApplySlowdownFactor := (permittedRangeScanSlowdown / 2) * minFractionOfTimeoutForApplyingSnapshot + if snapshotApplySlowdownFactor < 1 { + // Avoid division by 0. A snapshotApplySlowdownFactor between 0 + // and 1 would cause the minRate to be greater than + // rebalanceSnapshotRate, which is the max speed snapshots can + // be sent at, so we don't need to ingest snapshots at a faster + // rate than that. + snapshotApplySlowdownFactor = 1 + } + + minRate = int64(float64(rebalanceSnapshotRate.Get(&s.cfg.Settings.SV)) / snapshotApplySlowdownFactor) + storeBW := s.cfg.KVAdmissionController.GetProvisionedBandwidth(s.StoreID()) + if storeBW > 0 { + if minRate > int64(float64(storeBW)*0.25) { + log.KvDistribution.Warningf(ctx, + "snapshot ingest minRate is greater than 25%% of the store's provisioned bandwidth, minrate: %d, storeBW: %d", + minRate, storeBW, + ) + } + } + } + + timer := &timeutil.Timer{} + pacer = admission.NewSnapshotPacer(snapshotQ, minRate, timer.AsTimerI()) } for { diff --git a/pkg/kv/kvserver/kvadmission/kvadmission.go b/pkg/kv/kvserver/kvadmission/kvadmission.go index 6cb00e6bf2f8..9884ae9c9008 100644 --- a/pkg/kv/kvserver/kvadmission/kvadmission.go +++ b/pkg/kv/kvserver/kvadmission/kvadmission.go @@ -193,6 +193,10 @@ type Controller interface { // GetSnapshotQueue returns the SnapshotQueue which is used for ingesting raft // snapshots. GetSnapshotQueue(roachpb.StoreID) *admission.SnapshotQueue + // GetProvisionedBandwidth returns the provisioned disk bandwidth for the + // given store in bytes/second, or 0 if the store is not found or bandwidth + // is not configured. + GetProvisionedBandwidth(roachpb.StoreID) int64 } // TenantWeightProvider can be periodically asked to provide the tenant @@ -602,6 +606,10 @@ func (n *controllerImpl) GetSnapshotQueue(storeID roachpb.StoreID) *admission.Sn return sq.(*admission.SnapshotQueue) } +func (n *controllerImpl) GetProvisionedBandwidth(storeID roachpb.StoreID) int64 { + return n.storeGrantCoords.TryGetProvisionedBandwidthForStore(storeID) +} + // FollowerStoreWriteBytes captures stats about writes done to a store by a // replica that is not the leaseholder. These are used for admission control. type FollowerStoreWriteBytes struct { diff --git a/pkg/util/admission/disk_bandwidth.go b/pkg/util/admission/disk_bandwidth.go index fc68ba042485..6f6262eb78d3 100644 --- a/pkg/util/admission/disk_bandwidth.go +++ b/pkg/util/admission/disk_bandwidth.go @@ -110,6 +110,11 @@ func newDiskBandwidthLimiter() *diskBandwidthLimiter { } } +// getProvisionedBandwidth returns the provisioned disk bandwidth in bytes/second. +func (d *diskBandwidthLimiter) getProvisionedBandwidth() int64 { + return d.state.diskLoad.intProvisionedDiskBytes / adjustmentInterval +} + // diskTokens tokens represent actual bytes and IO on physical disks. Currently, // these are used to impose disk bandwidth limits on elastic traffic, but // regular traffic will also deduct from these buckets. diff --git a/pkg/util/admission/snapshot_queue.go b/pkg/util/admission/snapshot_queue.go index 9e6c8bac58de..7bf3959d4443 100644 --- a/pkg/util/admission/snapshot_queue.go +++ b/pkg/util/admission/snapshot_queue.go @@ -63,6 +63,16 @@ var DiskBandwidthForSnapshotIngest = settings.RegisterBoolSetting( settings.WithPublic, ) +var DiskBandwidthForSnapshotIngestMinRateEnabled = settings.RegisterBoolSetting( + settings.SystemOnly, + "kvadmission.store.snapshot_ingest_bandwidth_control.min_rate.enabled", + "if set to true, snapshot ingests will be admitted at a minimum rate when "+ + "kvadmission.store.provisioned_bandwidth is set to a non-zero value. Disabling this "+ + "setting can lead to snapshots being starved out by foreground traffic.", + true, + settings.WithPublic, +) + var snapshotWaitDur = metric.Metadata{ Name: "admission.wait_durations.snapshot_ingest", Help: "Wait time for snapshot ingest requests that waited", @@ -71,7 +81,8 @@ var snapshotWaitDur = metric.Metadata{ } type SnapshotMetrics struct { - WaitDurations metric.IHistogram + WaitDurations metric.IHistogram + AdmittedSnapshotBytes metric.Counter } func makeSnapshotQueueMetrics(registry *metric.Registry) *SnapshotMetrics { @@ -82,6 +93,12 @@ func makeSnapshotQueueMetrics(registry *metric.Registry) *SnapshotMetrics { Duration: base.DefaultHistogramWindowInterval(), BucketConfig: metric.IOLatencyBuckets, }), + AdmittedSnapshotBytes: *metric.NewCounter(metric.Metadata{ + Name: "admission.admitted_snapshot_bytes", + Help: "Number of bytes admitted for snapshot ingests when provisioned bandwidth AC is enabled", + Measurement: "Bytes", + Unit: metric.Unit_BYTES, + }), } registry.AddMetricStruct(m) return m @@ -89,7 +106,7 @@ func makeSnapshotQueueMetrics(registry *metric.Registry) *SnapshotMetrics { // snapshotRequester is a wrapper used for test purposes. type snapshotRequester interface { - Admit(ctx context.Context, count int64) error + Admit(ctx context.Context, count int64, minRate int64, timerForMinRate timeutil.TimerI) error } // SnapshotQueue implements the requester interface. It is used to request @@ -165,7 +182,14 @@ func (s *SnapshotQueue) close() { // Admit is called whenever a snapshot ingest request needs to update the number // of byte tokens it is using. Note that it accepts negative values, in which // case it will return the tokens back to the granter. -func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error { +func (s *SnapshotQueue) Admit( + ctx context.Context, count int64, minRate int64, timerForMinRate timeutil.TimerI, +) (err error) { + defer func() { + if err == nil && count > 0 { + s.metrics.AdmittedSnapshotBytes.Inc(count) + } + }() if count == 0 { return nil } @@ -176,6 +200,8 @@ func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error { if s.snapshotGranter.tryGet(canBurst /*arbitrary*/, count) { return nil } + // INVARIANT: count > 0. + // We were unable to get tokens for admission, so we queue. // // Reminder: there is a race here where a call to hasWaitingRequests after @@ -198,6 +224,11 @@ func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error { s.addLocked(item) }() + if minRate != 0 { + maxWaitDuration := (time.Second * time.Duration(count)) / time.Duration(minRate) + timerForMinRate.Reset(maxWaitDuration) + defer timerForMinRate.Stop() + } // Start waiting for admission. select { case <-ctx.Done(): @@ -236,6 +267,28 @@ func (s *SnapshotQueue) Admit(ctx context.Context, count int64) error { waitDur := timeutil.Since(item.enqueueingTime).Nanoseconds() s.metrics.WaitDurations.RecordValue(waitDur) return nil + case t := <-timerForMinRate.Ch(): + waitDur := t.Sub(item.enqueueingTime).Nanoseconds() + // INVARIANT: tokensToSubtract >= 0, since item.count >= 0. + tokensToSubtract := item.count + func() { + s.mu.Lock() + defer s.mu.Unlock() + if item.mu.granted { + // NB: we must call snapshotGranter.tookWithoutPermission after + // releasing the mutex. + tokensToSubtract = 0 + } + // NB: See the ctx.Done() case above for why we mark the item as + // cancelled instead of removing the item from the queue. + item.mu.cancelled = true + }() + if tokensToSubtract != 0 { + s.snapshotGranter.tookWithoutPermission(tokensToSubtract) + } + shouldRelease = false + s.metrics.WaitDurations.RecordValue(waitDur) + return nil } } @@ -288,14 +341,20 @@ func newSnapshotWorkItem(count int64) *snapshotWorkItem { } type SnapshotPacer struct { - snapshotQ snapshotRequester - intWriteBytes int64 + snapshotQ snapshotRequester + intWriteBytes int64 + minRate int64 + timerForMinRate timeutil.TimerI } -func NewSnapshotPacer(q snapshotRequester) *SnapshotPacer { +func NewSnapshotPacer( + q snapshotRequester, minRate int64, timerForMinRate timeutil.TimerI, +) *SnapshotPacer { return &SnapshotPacer{ - snapshotQ: q, - intWriteBytes: 0, + snapshotQ: q, + intWriteBytes: 0, + minRate: minRate, + timerForMinRate: timerForMinRate, } } @@ -308,7 +367,7 @@ func (p *SnapshotPacer) Pace(ctx context.Context, writeBytes int64, final bool) if p.intWriteBytes <= SnapshotBurstSize && !final { return nil } - if err := p.snapshotQ.Admit(ctx, p.intWriteBytes); err != nil { + if err := p.snapshotQ.Admit(ctx, p.intWriteBytes, p.minRate, p.timerForMinRate); err != nil { return errors.Wrapf(err, "snapshot admission queue") } p.intWriteBytes = 0 diff --git a/pkg/util/admission/snapshot_queue_test.go b/pkg/util/admission/snapshot_queue_test.go index 552c7277789e..85636bf69506 100644 --- a/pkg/util/admission/snapshot_queue_test.go +++ b/pkg/util/admission/snapshot_queue_test.go @@ -37,9 +37,11 @@ func TestSnapshotQueue(t *testing.T) { var tg *testGranter var buf builderWithMu var wrkMap workMap + var minRate int64 initialTime := timeutil.FromUnixMicros(int64(0)) registry := metric.NewRegistry() metrics := makeSnapshotQueueMetrics(registry) + ts := timeutil.NewManualTime(time.Unix(0, 0)) datadriven.RunTest(t, datapathutils.TestDataPath(t, "snapshot_queue"), func(t *testing.T, d *datadriven.TestData) string { @@ -66,8 +68,10 @@ func TestSnapshotQueue(t *testing.T) { q.ts.(*timeutil.ManualTime).AdvanceTo(timeutil.FromUnixNanos(int64(createTime) * time.Millisecond.Nanoseconds())) ctx, cancel := context.WithCancel(context.Background()) wrkMap.set(id, &testWork{cancel: cancel}) - go func(ctx context.Context, id int, count int) { - err := q.Admit(ctx, int64(count)) + // Create a fresh timer for each admit call so minRate timing works correctly. + timer := ts.NewTimer() + go func(ctx context.Context, id int, count int, minRate int64) { + err := q.Admit(ctx, int64(count), minRate, timer) if err != nil { buf.printf("id %d: admit failed", id) wrkMap.delete(id) @@ -75,7 +79,7 @@ func TestSnapshotQueue(t *testing.T) { buf.printf("id %d: admit succeeded", id) wrkMap.setAdmitted(id, StoreWorkHandle{}) } - }(ctx, id, count) + }(ctx, id, count, minRate) // Need deterministic output, and this is racing with the goroutine // which is trying to get admitted. Retry to let it get scheduled. maybeRetryWithWait(t, d.Expected, d.Rewrite, buf.String) @@ -121,6 +125,20 @@ func TestSnapshotQueue(t *testing.T) { }) return strconv.FormatBool(q.empty()) + case "set-min-rate": + var v int + d.ScanArgs(t, "v", &v) + minRate = int64(v) + return "" + + case "advance-time": + var millis int + d.ScanArgs(t, "millis", &millis) + ts.Advance(time.Duration(millis) * time.Millisecond) + // Need deterministic output since advancing time may trigger timer callbacks. + maybeRetryWithWait(t, d.Expected, d.Rewrite, buf.String) + return buf.stringAndReset() + default: return fmt.Sprintf("unknown command: %s", d.Cmd) } @@ -138,7 +156,11 @@ func TestSnapshotPacer(t *testing.T) { require.NoError(t, pacer.Pace(ctx, 1, false)) q := &testingSnapshotQueue{} - pacer = NewSnapshotPacer(q) + ts := timeutil.NewManualTime(time.Unix(0, 0)) + timer := ts.NewTimer() + + var minRate int64 = 10 << 20 // 10 MB/s + pacer = NewSnapshotPacer(q, minRate, timer) // Should not ask for admission since write bytes = burst size. writeBytes := int64(SnapshotBurstSize) @@ -150,35 +172,42 @@ func TestSnapshotPacer(t *testing.T) { // Do another write, should go over threshold and seek admission. require.NoError(t, pacer.Pace(ctx, 1, false)) require.True(t, q.admitted) + require.Equal(t, minRate, q.minRate) require.Equal(t, int64(0), pacer.intWriteBytes) require.Equal(t, writeBytes+1, q.admitCount) // Not enough bytes since last admission. Should not ask for admission. q.admitted = false q.admitCount = 0 + q.minRate = 0 require.NoError(t, pacer.Pace(ctx, 5, false)) require.False(t, q.admitted) + require.Equal(t, int64(0), q.minRate) require.Equal(t, int64(5), pacer.intWriteBytes) require.Equal(t, int64(0), q.admitCount) // We now go above the threshold again. Should ask for admission. require.NoError(t, pacer.Pace(ctx, writeBytes, false)) require.True(t, q.admitted) + require.Equal(t, minRate, q.minRate) require.Equal(t, writeBytes+5, q.admitCount) require.Equal(t, int64(0), pacer.intWriteBytes) // Do few more writes. q.admitted = false q.admitCount = 0 + q.minRate = 0 require.NoError(t, pacer.Pace(ctx, 10, false)) require.False(t, q.admitted) require.Equal(t, int64(10), pacer.intWriteBytes) + require.Equal(t, int64(0), q.minRate) require.Equal(t, int64(0), q.admitCount) // If final call to pacer, we should admit regardless of size. It should flush // all intWriteBytes. require.NoError(t, pacer.Pace(ctx, -1, true)) require.True(t, q.admitted) + require.Equal(t, minRate, q.minRate) require.Equal(t, int64(9), q.admitCount) } @@ -186,12 +215,16 @@ func TestSnapshotPacer(t *testing.T) { type testingSnapshotQueue struct { admitted bool admitCount int64 + minRate int64 } var _ snapshotRequester = &testingSnapshotQueue{} -func (ts *testingSnapshotQueue) Admit(ctx context.Context, count int64) error { +func (ts *testingSnapshotQueue) Admit( + ctx context.Context, count int64, minRate int64, timerForMinRate timeutil.TimerI, +) error { ts.admitted = true ts.admitCount = count + ts.minRate = minRate return nil } diff --git a/pkg/util/admission/store_grant_coordinator.go b/pkg/util/admission/store_grant_coordinator.go index 8966e0275aed..2e3468f8c846 100644 --- a/pkg/util/admission/store_grant_coordinator.go +++ b/pkg/util/admission/store_grant_coordinator.go @@ -298,6 +298,18 @@ func (sgc *StoreGrantCoordinators) TryGetSnapshotQueueForStore(storeID roachpb.S return nil } +// TryGetProvisionedBandwidthForStore returns the provisioned bandwidth for the +// given store in bytes/second, or 0 if the store is not found or bandwidth is +// not configured. +func (sgc *StoreGrantCoordinators) TryGetProvisionedBandwidthForStore( + storeID roachpb.StoreID, +) int64 { + if gc, ok := sgc.gcMap.Load(storeID); ok { + return gc.ioLoadListener.diskBandwidthLimiter.getProvisionedBandwidth() + } + return 0 +} + func (sgc *StoreGrantCoordinators) close() { // closeCh can be nil in tests that never called SetPebbleMetricsProvider. if sgc.closeCh != nil { diff --git a/pkg/util/admission/testdata/snapshot_queue b/pkg/util/admission/testdata/snapshot_queue index a5e7ef9bb3e3..ab484c6835ad 100644 --- a/pkg/util/admission/testdata/snapshot_queue +++ b/pkg/util/admission/testdata/snapshot_queue @@ -69,3 +69,67 @@ granted: returned 15 empty ---- true + +# Test minRate functionality: timer fires before grant. +# When tryGet returns false and minRate is set, after the timer fires +# (count/minRate seconds), the request proceeds via tookWithoutPermission. +init +---- + +set-try-get-return-value v=false +---- + +# Set minRate to 10 bytes/s. With count=5, wait time = 5/10 = 500ms +set-min-rate v=10 +---- + +admit id=10 count=5 create-time-millis=100 +---- +tryGet: returning false + +# Advance time by 501ms to trigger the minRate timer. +advance-time millis=501 +---- +tookWithoutPermission 5 +id 10: admit succeeded + +# Note: empty returns false because cancelled items remain in the queue +# (see TODO in snapshot_queue.go). The granted() method skips them. +empty +---- +false + +# Calling granted cleans up the cancelled item by skipping it. +granted +---- +granted: returned 0 + +empty +---- +true + +# Test minRate functionality: grant before timer fires. +# When tokens are granted before the timer fires, tookWithoutPermission is NOT called. +init +---- + +set-try-get-return-value v=false +---- + +set-min-rate v=10 +---- + +admit id=11 count=5 create-time-millis=200 +---- +tryGet: returning false + +# Grant before the timer fires (don't advance time). +# tookWithoutPermission should NOT be called. +granted +---- +id 11: admit succeeded +granted: returned 5 + +empty +---- +true