Initial push
This commit is contained in:
@@ -0,0 +1,348 @@
|
||||
// Package job tracks long-running migrations and streams their progress to
|
||||
// the UI. A job is a tree: job -> per-container item -> per-operation step,
|
||||
// with byte counters on the steps that move data.
|
||||
package job
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// State is the lifecycle of a job, item or step.
|
||||
type State string
|
||||
|
||||
const (
|
||||
StatePending State = "pending"
|
||||
StateRunning State = "running"
|
||||
StateSucceeded State = "succeeded"
|
||||
StateFailed State = "failed"
|
||||
StateSkipped State = "skipped"
|
||||
StateCanceled State = "canceled"
|
||||
)
|
||||
|
||||
// Terminal reports whether no further transitions are expected.
|
||||
func (s State) Terminal() bool {
|
||||
switch s {
|
||||
case StateSucceeded, StateFailed, StateSkipped, StateCanceled:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Kind distinguishes the two migration modes.
|
||||
type Kind string
|
||||
|
||||
const (
|
||||
KindSSH Kind = "ssh"
|
||||
KindPackage Kind = "package"
|
||||
KindRestore Kind = "restore"
|
||||
)
|
||||
|
||||
// Level classifies a log line.
|
||||
type Level string
|
||||
|
||||
const (
|
||||
LevelInfo Level = "info"
|
||||
LevelWarn Level = "warn"
|
||||
LevelError Level = "error"
|
||||
LevelCmd Level = "cmd" // a command that was (or would be) run on a host
|
||||
)
|
||||
|
||||
// LogEntry is one line in the job log.
|
||||
type LogEntry struct {
|
||||
Seq int64 `json:"seq"`
|
||||
At time.Time `json:"at"`
|
||||
Level Level `json:"level"`
|
||||
Item string `json:"item,omitempty"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
// Step is one unit of work inside an item, e.g. "transfer volume pgdata".
|
||||
type Step struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
State State `json:"state"`
|
||||
BytesDone int64 `json:"bytesDone"`
|
||||
BytesTotal int64 `json:"bytesTotal"` // -1 when unknown
|
||||
Error string `json:"error,omitempty"`
|
||||
StartedAt *time.Time `json:"startedAt,omitempty"`
|
||||
EndedAt *time.Time `json:"endedAt,omitempty"`
|
||||
}
|
||||
|
||||
// Item is the migration of one container.
|
||||
type Item struct {
|
||||
ID string `json:"id"` // container id on the source
|
||||
Name string `json:"name"`
|
||||
State State `json:"state"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Steps []*Step `json:"steps"`
|
||||
Warnings []string `json:"warnings,omitempty"`
|
||||
}
|
||||
|
||||
// Snapshot is the serializable view of a job handed to the UI.
|
||||
type Snapshot struct {
|
||||
ID string `json:"id"`
|
||||
Kind Kind `json:"kind"`
|
||||
Title string `json:"title"`
|
||||
State State `json:"state"`
|
||||
DryRun bool `json:"dryRun"`
|
||||
Error string `json:"error,omitempty"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
StartedAt *time.Time `json:"startedAt,omitempty"`
|
||||
EndedAt *time.Time `json:"endedAt,omitempty"`
|
||||
Items []*Item `json:"items"`
|
||||
Log []LogEntry `json:"log"`
|
||||
BytesDone int64 `json:"bytesDone"`
|
||||
BytesTotal int64 `json:"bytesTotal"`
|
||||
// Artifact is the produced package path, for package jobs.
|
||||
Artifact string `json:"artifact,omitempty"`
|
||||
// ArtifactBytes is the package size on disk.
|
||||
ArtifactBytes int64 `json:"artifactBytes,omitempty"`
|
||||
Revision int64 `json:"revision"`
|
||||
}
|
||||
|
||||
// Job is a running or finished migration.
|
||||
type Job struct {
|
||||
mu sync.RWMutex
|
||||
snap Snapshot
|
||||
seq int64
|
||||
revision int64
|
||||
maxLog int
|
||||
|
||||
cancel context.CancelFunc
|
||||
done chan struct{}
|
||||
|
||||
subsMu sync.Mutex
|
||||
subs map[int]chan struct{}
|
||||
nextID int
|
||||
}
|
||||
|
||||
func newJob(id string, kind Kind, title string, dryRun bool) *Job {
|
||||
return &Job{
|
||||
snap: Snapshot{
|
||||
ID: id, Kind: kind, Title: title, State: StatePending,
|
||||
DryRun: dryRun, CreatedAt: time.Now(), Items: []*Item{}, Log: []LogEntry{},
|
||||
BytesTotal: 0,
|
||||
},
|
||||
maxLog: 5000,
|
||||
done: make(chan struct{}),
|
||||
subs: map[int]chan struct{}{},
|
||||
}
|
||||
}
|
||||
|
||||
// ID returns the job identifier.
|
||||
func (j *Job) ID() string { return j.snap.ID }
|
||||
|
||||
// Done is closed once the job reaches a terminal state.
|
||||
func (j *Job) Done() <-chan struct{} { return j.done }
|
||||
|
||||
// Snapshot returns a deep-enough copy for JSON serialization.
|
||||
func (j *Job) Snapshot() Snapshot {
|
||||
j.mu.RLock()
|
||||
defer j.mu.RUnlock()
|
||||
s := j.snap
|
||||
s.Items = make([]*Item, len(j.snap.Items))
|
||||
var done, total int64
|
||||
for i, it := range j.snap.Items {
|
||||
cp := *it
|
||||
cp.Steps = make([]*Step, len(it.Steps))
|
||||
for k, st := range it.Steps {
|
||||
sc := *st
|
||||
cp.Steps[k] = &sc
|
||||
done += sc.BytesDone
|
||||
if sc.BytesTotal > 0 {
|
||||
total += sc.BytesTotal
|
||||
}
|
||||
}
|
||||
s.Items[i] = &cp
|
||||
}
|
||||
s.Log = append([]LogEntry(nil), j.snap.Log...)
|
||||
s.BytesDone, s.BytesTotal = done, total
|
||||
s.Revision = atomic.LoadInt64(&j.revision)
|
||||
return s
|
||||
}
|
||||
|
||||
// Subscribe returns a channel that receives a signal whenever the job changes,
|
||||
// plus a function to unsubscribe.
|
||||
func (j *Job) Subscribe() (<-chan struct{}, func()) {
|
||||
j.subsMu.Lock()
|
||||
defer j.subsMu.Unlock()
|
||||
id := j.nextID
|
||||
j.nextID++
|
||||
ch := make(chan struct{}, 1)
|
||||
j.subs[id] = ch
|
||||
return ch, func() {
|
||||
j.subsMu.Lock()
|
||||
defer j.subsMu.Unlock()
|
||||
if c, ok := j.subs[id]; ok {
|
||||
delete(j.subs, id)
|
||||
close(c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (j *Job) touch() {
|
||||
atomic.AddInt64(&j.revision, 1)
|
||||
j.subsMu.Lock()
|
||||
for _, ch := range j.subs {
|
||||
select {
|
||||
case ch <- struct{}{}:
|
||||
default: // a signal is already pending; the reader will see the latest state
|
||||
}
|
||||
}
|
||||
j.subsMu.Unlock()
|
||||
}
|
||||
|
||||
// Logf appends a line to the job log.
|
||||
func (j *Job) Logf(level Level, item, format string, args ...any) {
|
||||
j.mu.Lock()
|
||||
j.seq++
|
||||
e := LogEntry{Seq: j.seq, At: time.Now(), Level: level, Item: item, Message: fmt.Sprintf(format, args...)}
|
||||
j.snap.Log = append(j.snap.Log, e)
|
||||
if len(j.snap.Log) > j.maxLog {
|
||||
j.snap.Log = j.snap.Log[len(j.snap.Log)-j.maxLog:]
|
||||
}
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
// AddItem registers a container in the job and returns its handle.
|
||||
func (j *Job) AddItem(id, name string) *Item {
|
||||
j.mu.Lock()
|
||||
it := &Item{ID: id, Name: name, State: StatePending, Steps: []*Step{}}
|
||||
j.snap.Items = append(j.snap.Items, it)
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
return it
|
||||
}
|
||||
|
||||
// AddStep registers a unit of work under an item. bytesTotal may be -1 when
|
||||
// the size is not known ahead of time.
|
||||
func (j *Job) AddStep(it *Item, id, label string, bytesTotal int64) *Step {
|
||||
j.mu.Lock()
|
||||
st := &Step{ID: id, Label: label, State: StatePending, BytesTotal: bytesTotal}
|
||||
it.Steps = append(it.Steps, st)
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
return st
|
||||
}
|
||||
|
||||
// StartStep marks a step as running.
|
||||
func (j *Job) StartStep(st *Step) {
|
||||
now := time.Now()
|
||||
j.mu.Lock()
|
||||
st.State = StateRunning
|
||||
st.StartedAt = &now
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
// FinishStep closes a step, recording an error when one occurred.
|
||||
func (j *Job) FinishStep(st *Step, err error) {
|
||||
now := time.Now()
|
||||
j.mu.Lock()
|
||||
st.EndedAt = &now
|
||||
if err != nil {
|
||||
st.State = StateFailed
|
||||
st.Error = err.Error()
|
||||
} else {
|
||||
st.State = StateSucceeded
|
||||
if st.BytesTotal < 0 {
|
||||
st.BytesTotal = st.BytesDone
|
||||
}
|
||||
}
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
// SkipStep marks a step as deliberately not performed.
|
||||
func (j *Job) SkipStep(st *Step, reason string) {
|
||||
now := time.Now()
|
||||
j.mu.Lock()
|
||||
st.State = StateSkipped
|
||||
st.EndedAt = &now
|
||||
st.Error = reason
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
// AddBytes advances a step's byte counter. It is safe to call at high rates
|
||||
// from the transfer goroutine.
|
||||
func (j *Job) AddBytes(st *Step, n int64) {
|
||||
j.mu.Lock()
|
||||
st.BytesDone += n
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
// SetItemState transitions an item.
|
||||
func (j *Job) SetItemState(it *Item, s State, err error) {
|
||||
j.mu.Lock()
|
||||
it.State = s
|
||||
if err != nil {
|
||||
it.Error = err.Error()
|
||||
}
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
// AddItemWarning attaches a non-fatal note to an item.
|
||||
func (j *Job) AddItemWarning(it *Item, format string, args ...any) {
|
||||
msg := fmt.Sprintf(format, args...)
|
||||
j.mu.Lock()
|
||||
it.Warnings = append(it.Warnings, msg)
|
||||
j.mu.Unlock()
|
||||
j.Logf(LevelWarn, it.ID, "%s", msg)
|
||||
}
|
||||
|
||||
// SetArtifact records the produced package.
|
||||
func (j *Job) SetArtifact(path string, bytes int64) {
|
||||
j.mu.Lock()
|
||||
j.snap.Artifact = path
|
||||
j.snap.ArtifactBytes = bytes
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
func (j *Job) start() {
|
||||
now := time.Now()
|
||||
j.mu.Lock()
|
||||
j.snap.State = StateRunning
|
||||
j.snap.StartedAt = &now
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
}
|
||||
|
||||
func (j *Job) finish(err error) {
|
||||
now := time.Now()
|
||||
j.mu.Lock()
|
||||
if j.snap.State.Terminal() {
|
||||
j.mu.Unlock()
|
||||
return
|
||||
}
|
||||
j.snap.EndedAt = &now
|
||||
switch {
|
||||
case err == nil:
|
||||
j.snap.State = StateSucceeded
|
||||
case err == context.Canceled:
|
||||
j.snap.State = StateCanceled
|
||||
j.snap.Error = "canceled by operator"
|
||||
default:
|
||||
j.snap.State = StateFailed
|
||||
j.snap.Error = err.Error()
|
||||
}
|
||||
j.mu.Unlock()
|
||||
j.touch()
|
||||
close(j.done)
|
||||
}
|
||||
|
||||
// Cancel asks the job to stop. Work already in flight unwinds through context
|
||||
// cancellation.
|
||||
func (j *Job) Cancel() {
|
||||
if j.cancel != nil {
|
||||
j.cancel()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
package job
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"io"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Manager owns every job in the process.
|
||||
type Manager struct {
|
||||
mu sync.RWMutex
|
||||
jobs map[string]*Job
|
||||
// keep bounds how many finished jobs are retained.
|
||||
keep int
|
||||
}
|
||||
|
||||
// NewManager creates an empty job manager.
|
||||
func NewManager() *Manager {
|
||||
return &Manager{jobs: map[string]*Job{}, keep: 50}
|
||||
}
|
||||
|
||||
// ErrNotFound is returned for an unknown job id.
|
||||
var ErrNotFound = errors.New("job not found")
|
||||
|
||||
// Run creates a job and executes fn in the background. fn receives a context
|
||||
// that is canceled when the job is canceled, and the job handle for progress
|
||||
// reporting.
|
||||
func (m *Manager) Run(parent context.Context, kind Kind, title string, dryRun bool, fn func(context.Context, *Job) error) *Job {
|
||||
j := newJob(newID(), kind, title, dryRun)
|
||||
ctx, cancel := context.WithCancel(parent)
|
||||
j.cancel = cancel
|
||||
|
||||
m.mu.Lock()
|
||||
m.jobs[j.snap.ID] = j
|
||||
m.mu.Unlock()
|
||||
m.prune()
|
||||
|
||||
go func() {
|
||||
defer cancel()
|
||||
j.start()
|
||||
err := fn(ctx, j)
|
||||
if err == nil && ctx.Err() != nil {
|
||||
err = context.Canceled
|
||||
}
|
||||
if errors.Is(err, context.Canceled) {
|
||||
err = context.Canceled
|
||||
}
|
||||
j.finish(err)
|
||||
}()
|
||||
return j
|
||||
}
|
||||
|
||||
// Get returns a job by id.
|
||||
func (m *Manager) Get(id string) (*Job, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
j, ok := m.jobs[id]
|
||||
if !ok {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
return j, nil
|
||||
}
|
||||
|
||||
// List returns every job, newest first.
|
||||
func (m *Manager) List() []Snapshot {
|
||||
m.mu.RLock()
|
||||
jobs := make([]*Job, 0, len(m.jobs))
|
||||
for _, j := range m.jobs {
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
|
||||
out := make([]Snapshot, 0, len(jobs))
|
||||
for _, j := range jobs {
|
||||
s := j.Snapshot()
|
||||
// The list view does not need the full log.
|
||||
if len(s.Log) > 5 {
|
||||
s.Log = s.Log[len(s.Log)-5:]
|
||||
}
|
||||
out = append(out, s)
|
||||
}
|
||||
sort.Slice(out, func(i, k int) bool { return out[i].CreatedAt.After(out[k].CreatedAt) })
|
||||
return out
|
||||
}
|
||||
|
||||
// Delete removes a finished job. A running job is canceled instead.
|
||||
func (m *Manager) Delete(id string) error {
|
||||
m.mu.Lock()
|
||||
j, ok := m.jobs[id]
|
||||
if !ok {
|
||||
m.mu.Unlock()
|
||||
return ErrNotFound
|
||||
}
|
||||
if !j.Snapshot().State.Terminal() {
|
||||
m.mu.Unlock()
|
||||
j.Cancel()
|
||||
return nil
|
||||
}
|
||||
delete(m.jobs, id)
|
||||
m.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// prune drops the oldest finished jobs once the retention limit is exceeded.
|
||||
func (m *Manager) prune() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if len(m.jobs) <= m.keep {
|
||||
return
|
||||
}
|
||||
type entry struct {
|
||||
id string
|
||||
at time.Time
|
||||
}
|
||||
var finished []entry
|
||||
for id, j := range m.jobs {
|
||||
s := j.Snapshot()
|
||||
if s.State.Terminal() {
|
||||
finished = append(finished, entry{id, s.CreatedAt})
|
||||
}
|
||||
}
|
||||
sort.Slice(finished, func(i, k int) bool { return finished[i].at.Before(finished[k].at) })
|
||||
for i := 0; i < len(finished) && len(m.jobs) > m.keep; i++ {
|
||||
delete(m.jobs, finished[i].id)
|
||||
}
|
||||
}
|
||||
|
||||
func newID() string {
|
||||
b := make([]byte, 8)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
return hex.EncodeToString([]byte(time.Now().Format("150405.000000")))
|
||||
}
|
||||
return hex.EncodeToString(b)
|
||||
}
|
||||
|
||||
// CountingReader wraps a reader and reports every read to a job step, which is
|
||||
// how transfer progress reaches the UI.
|
||||
type CountingReader struct {
|
||||
R io.Reader
|
||||
Job *Job
|
||||
St *Step
|
||||
|
||||
pending int64
|
||||
lastFlush time.Time
|
||||
}
|
||||
|
||||
// NewCountingReader builds a progress-reporting reader.
|
||||
func NewCountingReader(r io.Reader, j *Job, st *Step) *CountingReader {
|
||||
return &CountingReader{R: r, Job: j, St: st, lastFlush: time.Now()}
|
||||
}
|
||||
|
||||
// Read implements io.Reader, batching counter updates so a fast transfer does
|
||||
// not flood subscribers with notifications.
|
||||
func (c *CountingReader) Read(p []byte) (int, error) {
|
||||
n, err := c.R.Read(p)
|
||||
if n > 0 {
|
||||
c.pending += int64(n)
|
||||
if c.pending >= 4<<20 || time.Since(c.lastFlush) > 200*time.Millisecond {
|
||||
c.flush()
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
c.flush()
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
// Flush pushes any buffered byte count to the job.
|
||||
func (c *CountingReader) Flush() { c.flush() }
|
||||
|
||||
func (c *CountingReader) flush() {
|
||||
if c.pending == 0 {
|
||||
return
|
||||
}
|
||||
c.Job.AddBytes(c.St, c.pending)
|
||||
c.pending = 0
|
||||
c.lastFlush = time.Now()
|
||||
}
|
||||
Reference in New Issue
Block a user