401 lines
11 KiB
Go
401 lines
11 KiB
Go
// Copyright (c) HashiCorp, Inc.
|
||
// SPDX-License-Identifier: BUSL-1.1
|
||
|
||
/*
|
||
Package batch implements a batch processor for jobs that update or delete
|
||
multiple rows in the database using a single SQL UPDATE or DELETE
|
||
statement.
|
||
|
||
It defines a type, [Batch], which is used by a job to execute a SQL
|
||
statement in batches. SQL commands are executed in batches by providing an
|
||
[Exec] function that executes a SQL statement, which must contain a
|
||
parameterized LIMIT clause, and returns the number of rows affected by the
|
||
query or an error if the query failed. Batch adjusts the batch size in an
|
||
effort to get the query execution time close to the [Config.Target]
|
||
duration. When the batch size is adjusted, [Store] is called. Jobs using
|
||
the batch processor should persist this value and use it as the starting
|
||
batch size in subsequent calls to Batch.
|
||
|
||
See session.deleteTerminatedJob for an example of how to use the batch
|
||
processor.
|
||
|
||
# SQL
|
||
|
||
The SQL LIMIT clause can only be used in query statements. It cannot be
|
||
used directly in a DELETE or UPDATE statement. Instead, a subquery or CTE
|
||
must be used to limit the number of rows affected by the query.
|
||
|
||
Here is an example SQL DELETE statement that uses a subquery with a LIMIT
|
||
clause:
|
||
|
||
delete
|
||
from alias_target_deleted
|
||
where public_id in (
|
||
select public_id
|
||
from alias_target_deleted
|
||
where delete_time < @delete_time
|
||
limit @batch_size
|
||
);
|
||
*/
|
||
package batch
|
||
|
||
import (
|
||
"context"
|
||
"math/rand/v2"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/hashicorp/boundary/internal/errors"
|
||
"github.com/hashicorp/boundary/internal/scheduler"
|
||
)
|
||
|
||
const (
|
||
DefaultStatusThreshold = 5 * time.Minute
|
||
DefaultTarget = 1 * time.Second
|
||
)
|
||
|
||
const (
|
||
DefaultSize = 5000
|
||
DefaultMin = 10
|
||
DefaultMax = 10000
|
||
)
|
||
|
||
// Exec is the function type used for executing the batch query. An Exec
|
||
// function must return the number of rows affected by the query or an
|
||
// error if the query failed.
|
||
type Exec func(ctx context.Context, batchSize int) (rowCount int, err error)
|
||
|
||
// Store is the function type used for storing the batch size in the
|
||
// database. A Store function must return an error if the store failed.
|
||
type Store func(ctx context.Context, batchSize int) error
|
||
|
||
// Config is a structure used to configure a [Batch].
|
||
type Config struct {
|
||
// Size is the initial batch size.
|
||
//
|
||
// If Size is less than 1, the initial batch size will be set to
|
||
// DefaultSize.
|
||
//
|
||
// If Size is less than Min, the initial batch size will be set to Min.
|
||
//
|
||
// If Size is greater than Max, the initial batch size will be set to
|
||
// Max.
|
||
Size int
|
||
|
||
// Min and Max are the minimum and maximum batch sizes allowed. The
|
||
// batch size will be clamped to the range [Min, Max].
|
||
//
|
||
// If Min is less than 1, it will be set to DefaultMin.
|
||
//
|
||
// If Max is less than or equal to Min, it will be set to DefaultMax.
|
||
Min int
|
||
Max int
|
||
|
||
// TotalToComplete is the total number of rows to be processed by the
|
||
// job. This is reported in the JobStatus returned by the Status
|
||
// method.
|
||
TotalToComplete int
|
||
|
||
// StatusThreshold is the amount of time the job has to return a
|
||
// JobStatus with values different from the previous JobStatus. If a
|
||
// call to Exec approaches this threshold, Batch will interrupt the
|
||
// call to Exec, reduce the batch size, and call Exec again. It will
|
||
// also increment the Retries value reported in the Status method.
|
||
//
|
||
// If StatusThreshold is less than or equal to zero, it will be set to
|
||
// DefaultStatusThreshold.
|
||
StatusThreshold time.Duration
|
||
|
||
// Target is the target duration for the query to run in. The batch
|
||
// size will be adjusted to keep the query duration within the target
|
||
// range.
|
||
//
|
||
// If Target is less than or equal to zero, it will be set to
|
||
// DefaultTarget.
|
||
//
|
||
// If Target is greater than or equal to StatusThreshold, it will be
|
||
// set to StatusThreshold - 5ms.
|
||
Target time.Duration
|
||
|
||
// Exec is called to execute the query. Exec is called by the Run
|
||
// method in a loop until the row count returned by Exec is less than
|
||
// the current batch size or Exec returns an error other than
|
||
// context.DeadlineExceeded. The configuration must supply this
|
||
// callback for batch to succeed.
|
||
//
|
||
// If Exec returns a context.DeadlineExceeded error, the batch size
|
||
// will be reduced, the Retries value reported in the Status method
|
||
// will be incremented, and then Exec will be called again.
|
||
//
|
||
// If Exec returns an error other than context.DeadlineExceeded, the
|
||
// batch size will not be changed and the Run method will exit
|
||
// returning the error.
|
||
Exec Exec
|
||
|
||
// Store, if not nil, is called when the batch size has changed and the
|
||
// new batch size should be stored in the database. If Store returns an
|
||
// error, the Run method will exit and return with the error.
|
||
Store Store
|
||
}
|
||
|
||
func (c *Config) size() int {
|
||
switch {
|
||
case c.Size < 1:
|
||
return DefaultSize
|
||
case c.Size < c.min():
|
||
return c.min()
|
||
case c.Size > c.max():
|
||
return c.max()
|
||
}
|
||
return c.Size
|
||
}
|
||
|
||
func (c *Config) min() int {
|
||
if c.Min < 1 {
|
||
return DefaultMin
|
||
}
|
||
return c.Min
|
||
}
|
||
|
||
func (c *Config) max() int {
|
||
if c.Max <= c.Min {
|
||
return DefaultMax
|
||
}
|
||
return c.Max
|
||
}
|
||
|
||
// Arbitrary constants
|
||
const (
|
||
statusThresholdBuffer = 250 * time.Millisecond
|
||
|
||
// both ranges are a percentage of the target duration
|
||
lowerRange = 10
|
||
upperRange = 10
|
||
)
|
||
|
||
func (c *Config) statusThreshold() time.Duration {
|
||
if c.StatusThreshold <= 0 {
|
||
return DefaultStatusThreshold - statusThresholdBuffer
|
||
}
|
||
return c.StatusThreshold - statusThresholdBuffer
|
||
}
|
||
|
||
func (c *Config) target() time.Duration {
|
||
switch {
|
||
case c.Target <= 0:
|
||
return DefaultTarget
|
||
case c.Target >= c.statusThreshold():
|
||
return c.statusThreshold()
|
||
}
|
||
return c.Target
|
||
}
|
||
|
||
func (c *Config) targetRange() (lower time.Duration, upper time.Duration) {
|
||
target := c.target()
|
||
return target - (target / lowerRange), target + (target / upperRange)
|
||
}
|
||
|
||
func (c *Config) store() Store {
|
||
if c.Store == nil {
|
||
return func(_ context.Context, _ int) error { return nil }
|
||
}
|
||
return c.Store
|
||
}
|
||
|
||
func (c *Config) clone() *Config {
|
||
if c == nil {
|
||
return nil
|
||
}
|
||
return &Config{
|
||
Size: c.Size,
|
||
Min: c.Min,
|
||
Max: c.Max,
|
||
TotalToComplete: c.TotalToComplete,
|
||
StatusThreshold: c.StatusThreshold,
|
||
Target: c.Target,
|
||
Exec: c.Exec,
|
||
Store: c.Store,
|
||
}
|
||
}
|
||
|
||
// Batch is a batch job processor for SQL jobs that update or delete
|
||
// multiple rows in the database using a single SQL UPDATE or DELETE
|
||
// statement.
|
||
type Batch struct {
|
||
c *Config
|
||
|
||
slowExecutions int
|
||
fastExecutions int
|
||
|
||
mu sync.Mutex
|
||
retries int
|
||
totalCompleted int
|
||
}
|
||
|
||
// New creates a [Batch] that uses the given configuration to execute a SQL
|
||
// job in batches. An error is returned if c contains a nil Exec.
|
||
func New(ctx context.Context, c *Config) (*Batch, error) {
|
||
const op = "batch.New"
|
||
switch {
|
||
case c == nil:
|
||
return nil, errors.New(ctx, errors.InvalidParameter, op, "nil Config")
|
||
case c.Exec == nil:
|
||
return nil, errors.New(ctx, errors.InvalidParameter, op, "nil Exec")
|
||
}
|
||
return &Batch{
|
||
c: c.clone(),
|
||
}, nil
|
||
}
|
||
|
||
// Status reports the job’s current status.
|
||
func (b *Batch) Status() scheduler.JobStatus {
|
||
b.mu.Lock()
|
||
defer b.mu.Unlock()
|
||
return scheduler.JobStatus{
|
||
Completed: b.totalCompleted,
|
||
Total: b.c.TotalToComplete,
|
||
Retries: b.retries,
|
||
}
|
||
}
|
||
|
||
// Run runs the batch processor. It calls the [Exec] function in a loop
|
||
// until the row count returned by Exec is less than the current batch size
|
||
// or Exec returns an error other than context.DeadlineExceeded.
|
||
//
|
||
// Each call to Run resets the values returned in [Batch.Status].
|
||
func (b *Batch) Run(ctx context.Context) error {
|
||
const op = "batch.Run"
|
||
b.reset()
|
||
|
||
for {
|
||
count, runDuration, err := b.run(ctx)
|
||
if err != nil {
|
||
if errors.Is(err, context.DeadlineExceeded) {
|
||
if err := b.timedOut(ctx); err != nil {
|
||
return errors.Wrap(ctx, err, op)
|
||
}
|
||
continue
|
||
}
|
||
return errors.Wrap(ctx, err, op)
|
||
}
|
||
|
||
b.successful(count)
|
||
|
||
// batch is not complete
|
||
if count == b.c.size() {
|
||
if err := b.adjustSize(ctx, runDuration); err != nil {
|
||
return errors.Wrap(ctx, err, op)
|
||
}
|
||
continue
|
||
}
|
||
|
||
// batch is complete
|
||
return nil
|
||
}
|
||
}
|
||
|
||
func (b *Batch) run(ctx context.Context) (int, time.Duration, error) {
|
||
queryCtx, cancel := context.WithTimeout(ctx, b.c.statusThreshold())
|
||
defer cancel()
|
||
start := time.Now()
|
||
n, err := b.c.Exec(queryCtx, b.c.size())
|
||
return n, time.Since(start), err
|
||
}
|
||
|
||
func (b *Batch) reset() {
|
||
b.mu.Lock()
|
||
b.retries = 0
|
||
b.totalCompleted = 0
|
||
b.mu.Unlock()
|
||
b.fastExecutions = 0
|
||
b.slowExecutions = 0
|
||
}
|
||
|
||
func (b *Batch) timedOut(ctx context.Context) error {
|
||
b.mu.Lock()
|
||
b.retries++
|
||
b.mu.Unlock()
|
||
return b.c.exponentialDecrease(ctx, b.retries)
|
||
}
|
||
|
||
func (b *Batch) successful(rowCount int) {
|
||
b.mu.Lock()
|
||
b.totalCompleted += rowCount
|
||
b.mu.Unlock()
|
||
}
|
||
|
||
func (b *Batch) adjustSize(ctx context.Context, runDuration time.Duration) error {
|
||
lower, upper := b.c.targetRange()
|
||
|
||
switch {
|
||
case runDuration < lower: // too fast
|
||
// increase the batch size to go slower
|
||
b.fastExecutions++
|
||
b.slowExecutions = 0
|
||
return b.c.linearIncrease(ctx, b.fastExecutions)
|
||
case runDuration > upper: // too slow
|
||
// decrease the batch size to go faster
|
||
b.slowExecutions++
|
||
b.fastExecutions = 0
|
||
return b.c.linearDecrease(ctx, b.slowExecutions)
|
||
}
|
||
|
||
// within target range so reset the counters
|
||
b.fastExecutions = 0
|
||
b.slowExecutions = 0
|
||
return nil
|
||
}
|
||
|
||
func (c *Config) exponentialDecrease(ctx context.Context, attempt int) error {
|
||
if attempt < 1 {
|
||
attempt = 1
|
||
}
|
||
newSize := (c.size() / (1 << uint(attempt))) - c.jitter()
|
||
return c.setSize(ctx, newSize)
|
||
}
|
||
|
||
func (c *Config) linearIncrease(ctx context.Context, attempt int) error {
|
||
if attempt < 1 {
|
||
attempt = 1
|
||
}
|
||
newSize := c.size() + (c.size() / 10 * attempt) + c.jitter()
|
||
return c.setSize(ctx, newSize)
|
||
}
|
||
|
||
func (c *Config) linearDecrease(ctx context.Context, attempt int) error {
|
||
if attempt < 1 {
|
||
attempt = 1
|
||
}
|
||
newSize := c.size() - (c.size() / 10 * attempt) - c.jitter()
|
||
return c.setSize(ctx, newSize)
|
||
}
|
||
|
||
// jitter returns a random number between 0 and 10% of the current batch
|
||
// size.
|
||
func (c *Config) jitter() int {
|
||
return rand.N(c.size() / 10)
|
||
}
|
||
|
||
// setSize sets the batch size to newSize and calls Store if newSize is
|
||
// different from the current size. If newSize is less than Min, the batch
|
||
// size will be set to Min. If newSize is greater than Max, the batch size
|
||
// will be set to Max. If Store returns an error, it will be returned by
|
||
// setSize.
|
||
func (c *Config) setSize(ctx context.Context, newSize int) error {
|
||
currentSize := c.Size
|
||
if newSize == currentSize {
|
||
return nil
|
||
}
|
||
switch {
|
||
case newSize < c.min():
|
||
newSize = c.min()
|
||
case newSize > c.max():
|
||
newSize = c.max()
|
||
}
|
||
if newSize != currentSize {
|
||
c.Size = newSize
|
||
return c.store()(ctx, newSize)
|
||
}
|
||
return nil
|
||
}
|