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 Core
• type Manager
  • func New(infra Core, 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
  • func (d *Manager) CancelDrain(name string) error
  • func (d *Manager) ClearState()
  • func (d *Manager) CompletedBackends() map[string]bool
  • func (d *Manager) DrainOneObject(ctx context.Context, srcBackend backend.ObjectBackend, srcName string, obj *core.ObjectLocation) bool
  • func (d *Manager) GetDrainProgress(ctx context.Context, name string) (*Progress, error)
  • func (d *Manager) IsDraining(name string) bool
  • func (d *Manager) PurgeBackendObjects(ctx context.Context, be backend.ObjectBackend, name string)
  • func (d *Manager) RemoveBackend(ctx context.Context, name string, purge bool) error
  • func (d *Manager) SeedActiveForTest(name string)
  • func (d *Manager) SeedCompletedForTest(name string)
  • func (d *Manager) StartDrain(ctx context.Context, name string) error
• type Progress

type Core

Core is the slice of proxy infrastructure the Manager needs. Defined here at the consumer so the proxy package can satisfy it structurally without exporting a god interface.

type Core 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
    DeleteOrEnqueue(ctx context.Context, be backend.ObjectBackend, backendName, key, reason string, sizeBytes int64)
    MoveObject(ctx context.Context, req *writepath.MoveRequest) (int64, 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 Core, 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)

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) 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.

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 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