scheduler

Documentation

Index

• Variables
• type GCConfig
• type Scheduler
  • func New(ds datastore.TxnDatastore, l ffs.JobLogger, hs ffs.HotStorage, ...) (*Scheduler, error)
  • func (s *Scheduler) Cancel(jid ffs.JobID) error
  • func (s *Scheduler) Close() error
  • func (s *Scheduler) ExecutingStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Scheduler) GCStaged(ctx context.Context) ([]cid.Cid, error)
  • func (s *Scheduler) GetCidFromHot(ctx context.Context, c cid.Cid) (io.Reader, error)
  • func (s *Scheduler) GetLogsByCid(ctx context.Context, iid ffs.APIID, c cid.Cid) ([]ffs.LogEntry, error)
  • func (s *Scheduler) GetRetrievalInfo(rid ffs.RetrievalID) (ffs.RetrievalInfo, error)
  • func (s *Scheduler) GetStorageInfo(iid ffs.APIID, c cid.Cid) (ffs.StorageInfo, error)
  • func (s *Scheduler) ImportStorageInfo(iid ffs.APIID, ci ffs.StorageInfo) error
  • func (s *Scheduler) LatestFinalStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Scheduler) LatestSuccessfulStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Scheduler) PinnedCids(ctx context.Context) ([]ffs.PinnedCid, error)
  • func (s *Scheduler) PushConfig(iid ffs.APIID, c cid.Cid, cfg ffs.StorageConfig) (ffs.JobID, error)
  • func (s *Scheduler) PushReplace(iid ffs.APIID, c cid.Cid, cfg ffs.StorageConfig, oldCid cid.Cid) (ffs.JobID, error)
  • func (s *Scheduler) QueuedStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Scheduler) StartRetrieval(iid ffs.APIID, rid ffs.RetrievalID, pyCid, piCid cid.Cid, sel string, ...) (ffs.JobID, error)
  • func (s *Scheduler) StorageConfig(jid ffs.JobID) (ffs.StorageConfig, error)
  • func (s *Scheduler) StorageJob(jid ffs.JobID) (ffs.StorageJob, error)
  • func (s *Scheduler) Untrack(iid ffs.APIID, c cid.Cid) error
  • func (s *Scheduler) WatchJobs(ctx context.Context, c chan<- ffs.StorageJob, 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")

	// RenewalEvalFrequency is the frequency in which renewable StorageConfigs
	// will be evaluated.
	RenewalEvalFrequency = time.Hour * 24

	// RepairEvalFrequency is the frequency in which repairable StorageConfigs
	// will be evaluated.
	RepairEvalFrequency = time.Hour * 24
)

var (
	// HardcodedHotTimeout is a temporary override of storage configs
	// value for AddTimeout.
	HardcodedHotTimeout = time.Second * 300
)

Types

type GCConfig

type GCConfig struct {
	StageGracePeriod time.Duration
	AutoGCInterval   time.Duration
}

GCConfig provides configuration for FFS GC.

type Scheduler

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

Scheduler receives actions to store a Cid in hot and cold storage. 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 storage configured for the scheduler.

func New

func New(ds datastore.TxnDatastore, l ffs.JobLogger, hs ffs.HotStorage, cs ffs.ColdStorage, maxParallel int, dealFinalityTimeout time.Duration, sr2rf func() (int, error), gcConfig GCConfig) (*Scheduler, error)

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

func (s *Scheduler) Cancel(jid ffs.JobID) error

Cancel cancels an executing Job.

func (*Scheduler) Close

func (s *Scheduler) Close() error

Close terminates the scheduler.

func (*Scheduler) ExecutingStorageJobs

func (s *Scheduler) ExecutingStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

ExecutingStorageJobs returns executing jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

func (*Scheduler) GCStaged

func (s *Scheduler) GCStaged(ctx context.Context) ([]cid.Cid, error)

GCStaged runs a unpinned garbage collection of stage-pins.

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 hot storage.

func (*Scheduler) GetLogsByCid

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

GetLogsByCid returns history logs of a Cid.

func (*Scheduler) GetRetrievalInfo

func (s *Scheduler) GetRetrievalInfo(rid ffs.RetrievalID) (ffs.RetrievalInfo, error)

GetRetrievalInfo returns the information about an executed retrieval.

func (*Scheduler) GetStorageInfo

func (s *Scheduler) GetStorageInfo(iid ffs.APIID, c cid.Cid) (ffs.StorageInfo, error)

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

func (*Scheduler) ImportStorageInfo

func (s *Scheduler) ImportStorageInfo(iid ffs.APIID, ci ffs.StorageInfo) error

ImportStorageInfo imports Cid information manually. That's to say, will be StorageInfo which wasn't generated by executing a Job, but provided externally.

func (*Scheduler) LatestFinalStorageJobs

func (s *Scheduler) LatestFinalStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

LatestFinalStorageJobs returns the most recent finished jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

func (*Scheduler) LatestSuccessfulStorageJobs

func (s *Scheduler) LatestSuccessfulStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

LatestSuccessfulStorageJobs returns the most recent successful jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

func (*Scheduler) PinnedCids

func (s *Scheduler) PinnedCids(ctx context.Context) ([]ffs.PinnedCid, error)

PinnedCids returns the pinned cids from Hot-Storage.

func (*Scheduler) PushConfig

func (s *Scheduler) PushConfig(iid ffs.APIID, c cid.Cid, cfg ffs.StorageConfig) (ffs.JobID, error)

PushConfig queues the specified StorageConfig 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, c cid.Cid, cfg ffs.StorageConfig, oldCid cid.Cid) (ffs.JobID, error)

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

func (*Scheduler) QueuedStorageJobs

func (s *Scheduler) QueuedStorageJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

QueuedStorageJobs returns queued jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

func (*Scheduler) StartRetrieval

func (s *Scheduler) StartRetrieval(iid ffs.APIID, rid ffs.RetrievalID, pyCid, piCid cid.Cid, sel string, miners []string, walletAddr string, maxPrice uint64) (ffs.JobID, error)

StartRetrieval schedules a new RetrievalJob to execute a Filecoin retrieval.

func (*Scheduler) StorageConfig

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

StorageConfig returns the storage config for a job.

func (*Scheduler) StorageJob

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

StorageJob the current storage state of a Job.

func (*Scheduler) Untrack

func (s *Scheduler) Untrack(iid ffs.APIID, c cid.Cid) error

Untrack untracks a Cid from iid for renewal and repair background crons.

func (*Scheduler) WatchJobs

func (s *Scheduler) WatchJobs(ctx context.Context, c chan<- ffs.StorageJob, 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.