scheduler

Documentation

Index

• Variables
• type Action
• type ActionStore
• type CidInfoStore
• type JobStore
• type Scheduler
  • func New(js JobStore, as ActionStore, cis CidInfoStore, l ffs.CidLogger, ...) *Scheduler
  • func (s *Scheduler) Close() error
  • func (s *Scheduler) GetCidFromHot(ctx context.Context, c cid.Cid) (io.Reader, error)
  • func (s *Scheduler) GetCidInfo(c cid.Cid) (ffs.CidInfo, error)
  • func (s *Scheduler) GetJob(jid ffs.JobID) (ffs.Job, error)
  • func (s *Scheduler) GetLogs(ctx context.Context, c cid.Cid) ([]ffs.LogEntry, error)
  • func (s *Scheduler) PushConfig(iid ffs.APIID, cfg ffs.CidConfig) (ffs.JobID, error)
  • func (s *Scheduler) PushReplace(iid ffs.APIID, cfg ffs.CidConfig, oldCid cid.Cid) (ffs.JobID, error)
  • func (s *Scheduler) Untrack(c cid.Cid) error
  • func (s *Scheduler) WatchJobs(ctx context.Context, c chan<- ffs.Job, iid ffs.APIID) error
  • func (s *Scheduler) WatchLogs(ctx context.Context, c chan<- ffs.LogEntry) error

Variables

var (
	// ErrNotFound is returned when an item isn't found on a Store.
	ErrNotFound = errors.New("item not found")
)

Types

type Action

type Action struct {
	APIID       ffs.APIID
	Cfg         ffs.CidConfig
	ReplacedCid cid.Cid
}

Action represents an action to be executed by the Scheduler.

type ActionStore

type ActionStore interface {
	// Put saves a new state for a Job.
	Put(ffs.JobID, Action) error
	// Get returns the current state of a Job.
	Get(ffs.JobID) (Action, error)
	// Remove removes the action associated with a Cid.
	Remove(cid.Cid) error
	// GetRenewable returns the known pushed configs that have enabled
	// renew Filecoin flag for their deals.
	GetRenewable() ([]Action, error)
	// GetRepairable returns the known pushed configs that have enabled
	// auto-repair.
	GetRepairable() ([]Action, error)
}

ActionStore persist actions for Cids.

type CidInfoStore

type CidInfoStore interface {
	Put(ffs.CidInfo) error
	Get(cid.Cid) (ffs.CidInfo, error)
}

CidInfoStore persists CidInfo which represent the current storage state of a Cid.

type JobStore

type JobStore interface {
	Enqueue(ffs.Job) error
	// Dequeue returns the next queued Job. It automatically changes
	// its state to Executing.
	Dequeue() (*ffs.Job, error)
	// Finalize sets the Job to a final state, i.e. success or failed.
	Finalize(ffs.JobID, ffs.JobStatus, error, []ffs.DealError) error

	// Get retrieves job data from the store.
	Get(ffs.JobID) (ffs.Job, error)
	// Watch subscribes to all job state changes within an instance.
	Watch(context.Context, chan<- ffs.Job, ffs.APIID) error
}

JobStore persist Job information and allows to watch for Job state updates.

type Scheduler

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

Scheduler receives actions to store a Cid in Hot and Cold layers. These actions are created as Jobs which have a lifecycle that can be watched by external actors. This Jobs are executed by delegating the work to the Hot and Cold layers configured for the scheduler.

func New

func New(js JobStore, as ActionStore, cis CidInfoStore, l ffs.CidLogger, hs ffs.HotStorage, cs ffs.ColdStorage) *Scheduler

New returns a new instance of Scheduler which uses JobStore as its backing repository for state, HotStorage for the hot layer, and ColdStorage for the cold layer.

func (*Scheduler) Close

func (s *Scheduler) Close() error

Close terminates the scheduler.

func (*Scheduler) GetCidFromHot

func (s *Scheduler) GetCidFromHot(ctx context.Context, c cid.Cid) (io.Reader, error)

GetCidFromHot returns an io.Reader of the data from the hot layer.

func (*Scheduler) GetCidInfo

func (s *Scheduler) GetCidInfo(c cid.Cid) (ffs.CidInfo, error)

GetCidInfo returns the current storage state of a Cid. Returns ErrNotFound if there isn't information for a Cid.

func (*Scheduler) GetJob

func (s *Scheduler) GetJob(jid ffs.JobID) (ffs.Job, error)

GetJob the current state of a Job.

func (*Scheduler) GetLogs

func (s *Scheduler) GetLogs(ctx context.Context, c cid.Cid) ([]ffs.LogEntry, error)

GetLogs returns history logs of a Cid.

func (*Scheduler) PushConfig

func (s *Scheduler) PushConfig(iid ffs.APIID, cfg ffs.CidConfig) (ffs.JobID, error)

PushConfig queues the specified CidConfig to be executed as a new Job. It returns the created JobID for further tracking of its state.

func (*Scheduler) PushReplace

func (s *Scheduler) PushReplace(iid ffs.APIID, cfg ffs.CidConfig, oldCid cid.Cid) (ffs.JobID, error)

PushReplace queues a new CidConfig to be executed as a new Job, replacing an oldCid that will be untrack in the Scheduler (i.e: deal renewals, repairing).

func (*Scheduler) Untrack

func (s *Scheduler) Untrack(c cid.Cid) error

Untrack untracks a Cid for renewal and repair background crons.

func (*Scheduler) WatchJobs

func (s *Scheduler) WatchJobs(ctx context.Context, c chan<- ffs.Job, iid ffs.APIID) error

WatchJobs returns a channel to listen to Job status changes from a specified API instance. It immediately pushes the current Job state to the channel.

func (*Scheduler) WatchLogs

func (s *Scheduler) WatchLogs(ctx context.Context, c chan<- ffs.LogEntry) error

WatchLogs writes to a channel all new logs for Cids. The context should be canceled when wanting to stop receiving updates to the channel.