s3-orchestrator

drain

import "github.com/afreidah/s3-orchestrator/internal/proxy/drain"

Package drain owns the backend drain/remove lifecycle. It tracks the draining state map, runs the migration goroutine, and exposes IsDraining for the proxy core’s eligibility filters.

Index

type DrainRuntime

DrainRuntime is the backend-runtime slice the Manager needs: fleet enumeration, per-backend copy/delete primitives, and usage accounting. Defined here at the consumer so *infra.BackendRuntime satisfies it structurally without the drain package importing infra.

type DrainRuntime interface {
    Backends() map[string]backend.ObjectBackend
    GetBackend(name string) (backend.ObjectBackend, error)
    BackendOrder() []string
    StreamCopy(ctx context.Context, src, dst backend.ObjectBackend, key string) error
    DeleteWithTimeout(ctx context.Context, be backend.ObjectBackend, key string) error
    Acct() *accounting.Recorder
}

type Manager

Manager handles draining and removing backends.

type Manager struct {
    // contains filtered or unexported fields
}

func New

func New(infra DrainRuntime, mover Mover, objects core.ObjectStore, quota core.QuotaStore, backendLifecycle core.BackendLifecycleStore, abortMultipartUploads func(ctx context.Context, backendName string), processCleanupQueue func(ctx context.Context) (processed, failed int)) *Manager

New creates a Manager.

func (*Manager) CancelDrain

func (d *Manager) CancelDrain(name string) error

CancelDrain stops an active drain operation. If the drain has already completed, it clears the “drained” state so the backend becomes eligible for writes again. Objects already moved are not rolled back.

func (*Manager) ClearState

func (d *Manager) ClearState()

ClearState removes all entries from the draining map. Used by tests to reset state between runs.

func (*Manager) CompletedBackends

func (d *Manager) CompletedBackends() map[string]bool

CompletedBackends returns the names of backends whose drain has finished (the goroutine closed state.done) and whose state is still in the map. FlushUsage uses this to skip backends whose backend_usage rows have already been deleted by the drain finalizer.

func (*Manager) DrainOneObject

func (d *Manager) DrainOneObject(ctx context.Context, srcBackend backend.ObjectBackend, srcName string, obj *core.ObjectLocation) bool

DrainOneObject moves a single object from the draining backend to another. If the object already has a replica on another backend, the source copy is simply removed (no data transfer needed). Returns true on success.

func (*Manager) GetDrainProgress

func (d *Manager) GetDrainProgress(ctx context.Context, name string) (*Progress, error)

GetDrainProgress returns the current state of a drain operation.

func (*Manager) IsDraining

func (d *Manager) IsDraining(name string) bool

IsDraining reports whether the named backend is currently being drained.

func (*Manager) PurgeBackendObjects

func (d *Manager) PurgeBackendObjects(ctx context.Context, be backend.ObjectBackend, name string, observer progress.Observer)

PurgeBackendObjects deletes all objects from a backend’s S3 storage and their metadata rows. Best-effort: per-key failures are logged and the loop continues. Bails on a page whose every DeleteObjectLocation fails so a persistent DB error (constraint, partition, conflict) cannot pin the loop on the same 100 rows forever; the same rows would otherwise list-and-fail until the process was restarted.

func (*Manager) RemoveBackend

func (d *Manager) RemoveBackend(ctx context.Context, name string, purge bool, observer progress.Observer) error

RemoveBackend deletes all database records for a backend. If purge is true and the backend is reachable, also deletes objects from the backend’s S3 storage. This is destructive and cannot be undone. observer, when non-nil, receives a start and end step per object purged.

func (*Manager) SeedActiveForTest

func (d *Manager) SeedActiveForTest(name string)

SeedActiveForTest stores an active (not-yet-completed) drain entry for the named backend. Lets tests put the manager in the “draining” state without launching the full StartDrain goroutine.

func (*Manager) SeedCompletedForTest

func (d *Manager) SeedCompletedForTest(name string)

SeedCompletedForTest stores a drain-completed entry (done channel closed) for the named backend. FlushUsage and the dashboard treat the backend as fully drained.

func (*Manager) StartDrain

func (d *Manager) StartDrain(ctx context.Context, name string) error

StartDrain begins draining a backend by migrating all objects to other backends. The drain runs in a background goroutine. New writes are excluded from the draining backend immediately.

type Mover

Mover funnels deletes and cross-backend moves through the write coordinator’s shared primitives (orphan cleanup, MoveObjectLocation CAS, source-delete accounting). *writepath.Coordinator satisfies it.

type Mover interface {
    DeleteOrEnqueue(ctx context.Context, be backend.ObjectBackend, backendName, key, reason string, sizeBytes int64)
    MoveObject(ctx context.Context, req *writepath.MoveRequest) (int64, error)
}

type Progress

Progress holds the current state of a drain operation.

type Progress struct {
    Active           bool   `json:"active"`
    ObjectsRemaining int64  `json:"objects_remaining"`
    BytesRemaining   int64  `json:"bytes_remaining"`
    ObjectsMoved     int64  `json:"objects_moved"`
    Error            string `json:"error,omitempty"`
}

Generated by gomarkdoc