backfill

package

v0.11.754 Latest Latest Go to latest Published: Sep 10, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/bruin-data/bruin

Links

Documentation ¶

Rendered for

Overview ¶

Package backfill plans and records resumable local partitioned runs.

Index ¶

Constants
func AcquireChildLease(dir, generation string) (io.Closer, error)
func Eligible(status Status, rerun string) bool
func ParseRange(start, end, zone string) (time.Time, time.Time, error)
type Attempt
type Interval
type LockedWriter
- func (w *LockedWriter) Write(p []byte) (int, error)
type Manifest
type Options
- func (o Options) Validate() error
type Plan
- func (p Plan) Count(ctx context.Context) (int, error)
- func (p Plan) Intervals(reverse bool) iter.Seq[Interval]
- func (p Plan) Validate() error
type Record
type Run
type Status
type Store
- func Open(root, id string) (*Store, error)
- func (s *Store) BeginExecution(generation string) error
- func (s *Store) Create(m Manifest) error
- func (s *Store) Lock() (*flock.Flock, error)
- func (s *Store) Manifest() (Manifest, error)
- func (s *Store) Read(i Interval) (Record, error)
- func (s *Store) Records(ctx context.Context) iter.Seq2[Record, error]
- func (s *Store) Save(r Record) error
type Summary
- func Execute(ctx context.Context, m Manifest, s *Store, o Options, run Run, ...) (Summary, error)
- func Summarize(ctx context.Context, p Plan, s *Store, rerun string) (Summary, error)

Constants ¶

View Source

const Version = 1

Variables ¶

This section is empty.

Functions ¶

func AcquireChildLease ¶

func AcquireChildLease(dir, generation string) (io.Closer, error)

AcquireChildLease holds a shared execution lock until Close is called and rejects children from a superseded executor generation.

func Eligible ¶

func Eligible(status Status, rerun string) bool

func ParseRange ¶

func ParseRange(start, end, zone string) (time.Time, time.Time, error)

ParseRange treats date-only ends as inclusive calendar dates and timestamp ends as exclusive instants. Unzoned inputs use the explicit IANA timezone.

Types ¶

type Attempt ¶

type Attempt struct {
	RunID      string     `json:"run_id"`
	Status     Status     `json:"status"`
	StartedAt  time.Time  `json:"started_at"`
	FinishedAt *time.Time `json:"finished_at,omitempty"`
	Error      string     `json:"error,omitempty"`
}

type Interval ¶

type Interval struct {
	ID    string    `json:"id"`
	Start time.Time `json:"start"`
	End   time.Time `json:"end"` // exclusive
}

type LockedWriter ¶

type LockedWriter struct {
	Mu        sync.Mutex
	WriteFunc func([]byte) (int, error)
}

LockedWriter prevents interleaved child output and races in custom writers.

func (*LockedWriter) Write ¶

func (w *LockedWriter) Write(p []byte) (int, error)

type Manifest ¶

type Manifest struct {
	Version   int       `json:"version"`
	ID        string    `json:"id"`
	CreatedAt time.Time `json:"created_at"`
	Plan      Plan      `json:"plan"`
}

type Options ¶

type Options struct {
	MaxParallel int    `json:"max_parallel"`
	Workers     int    `json:"workers"`
	Retries     int    `json:"retries"`
	Reverse     bool   `json:"reverse"`
	Rerun       string `json:"rerun"`
	OnFailure   string `json:"on_failure"`
}

func (Options) Validate ¶

func (o Options) Validate() error

type Plan ¶

type Plan struct {
	Target           string              `json:"target"`
	WorkingDirectory string              `json:"working_directory"`
	Environment      string              `json:"environment"`
	RunFlags         map[string][]string `json:"run_flags"`
	Start            time.Time           `json:"start"`
	End              time.Time           `json:"end"` // exclusive
	Timezone         string              `json:"timezone"`
	Partition        string              `json:"partition"`
}

Plan contains the immutable inputs needed to regenerate a backfill. Intervals are generated lazily; neither planning nor execution retains the whole range.

func (Plan) Count ¶

func (p Plan) Count(ctx context.Context) (int, error)

Count calculates elapsed-time partition counts directly and walks calendar boundaries without creating partition IDs or reading queued records.

func (Plan) Intervals ¶

func (p Plan) Intervals(reverse bool) iter.Seq[Interval]

Intervals yields the same partitions in either order. Calendar partitions align to local boundaries (Monday for weeks); duration partitions anchor at Start. Call Validate before iterating an externally supplied plan.

func (Plan) Validate ¶

func (p Plan) Validate() error

type Record ¶

type Record struct {
	Interval
	Status   Status    `json:"status"`
	Attempts []Attempt `json:"attempts"`
}

type Run ¶

type Run func(context.Context, Interval, string) error

Run executes a single attempt. The run ID has already been persisted before this is called; implementations must wait for the child to exit on cancellation.

type Status ¶

type Status string

const (
	Queued    Status = "queued"
	Running   Status = "running"
	Succeeded Status = "succeeded"
	Failed    Status = "failed"
	Cancelled Status = "cancelled"
)

type Store ¶

type Store struct{ Dir string }

Store writes independent atomic partition snapshots. The OS lock protects a backfill from concurrent executors and is automatically released after a crash.

func Open ¶

func Open(root, id string) (*Store, error)

func (*Store) BeginExecution ¶

func (s *Store) BeginExecution(generation string) error

BeginExecution changes the token accepted by child runs while holding an exclusive lock. Child processes hold shared locks for their whole lifetime, so an executor cannot resume while a child orphaned by an abrupt parent exit is still active. A child that had not acquired its lock before the resume sees the changed token and exits before doing any work.

func (*Store) Create ¶

func (s *Store) Create(m Manifest) error

func (*Store) Lock ¶

func (s *Store) Lock() (*flock.Flock, error)

func (*Store) Manifest ¶

func (s *Store) Manifest() (Manifest, error)

func (*Store) Read ¶

func (s *Store) Read(i Interval) (Record, error)

func (*Store) Records ¶

func (s *Store) Records(ctx context.Context) iter.Seq2[Record, error]

Records pages through materialized records only. Queued partitions are implicit in the plan, so even a huge mostly-unstarted backfill has a cheap summary.

func (*Store) Save ¶

func (s *Store) Save(r Record) error

type Summary ¶

type Summary struct {
	Total     int `json:"total"`
	Succeeded int `json:"succeeded"`
	Failed    int `json:"failed"`
	Cancelled int `json:"cancelled"`
	Queued    int `json:"queued"`
	Running   int `json:"running"`
	Skipped   int `json:"skipped"` // not selected in this invocation, including previous successes
}

func Execute ¶

func Execute(ctx context.Context, m Manifest, s *Store, o Options, run Run, observers ...func(Status, Record)) (Summary, error)

Execute keeps at most MaxParallel records in memory. The caller holds the store lock. A stopped or interrupted invocation leaves unstarted partitions queued.

func Summarize ¶

func Summarize(ctx context.Context, p Plan, s *Store, rerun string) (Summary, error)

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL