retain

Documentation

Index

• Variables
• func DeleteFile(path, filename string) error
• func SaveRequest(path string, req Request) error
• type Config
• type Queue
• type Request
  • func (req *Request) Filename() string
• type RequestStore
  • func NewRequestStore(path string) (RequestStore, error)
  • func (store *RequestStore) Add(req Request) (bool, error)
  • func (store *RequestStore) Data() map[storj.NodeID]Request
  • func (store *RequestStore) DeleteCache(req Request) error
  • func (store *RequestStore) Len() int
  • func (store *RequestStore) Next() (Request, bool)
  • func (store *RequestStore) Remove(req Request) bool
• type Service
  • func NewService(log *zap.Logger, store *pieces.Store, config Config) *Service
  • func (s *Service) Close() error
  • func (s *Service) Queue(req Request) bool
  • func (s *Service) Run(ctx context.Context) (err error)
  • func (s *Service) Status() Status
  • func (s *Service) TestWaitUntilEmpty()
  • func (s *Service) TestingHowManyQueued() int
• type Status
  • func (v *Status) Set(s string) error
  • func (v *Status) String() string
  • func (*Status) Type() string

Variables

var (

	// Error is the default error class for retain errors.
	Error = errs.Class("retain")
)

Functions

func DeleteFile

func DeleteFile(path, filename string) error

DeleteFile removes a file from the filesystem.

func SaveRequest

func SaveRequest(path string, req Request) error

SaveRequest stores the request to the filesystem.

Types

type Config

type Config struct {
	MaxTimeSkew time.Duration `help:"allows for small differences in the satellite and storagenode clocks" default:"72h0m0s"`
	Status      Status        `` /* 143-byte string literal not displayed */
	Concurrency int           `help:"how many concurrent retain requests can be processed at the same time." default:"5"`
	CachePath   string        `help:"path to the cache directory for retain requests." default:"$CONFDIR/retain"`
}

Config defines parameters for the retain service.

type Queue

type Queue interface {
	// Add adds a request to the queue.
	Add(request Request) (bool, error)
	// Remove removes a request from the queue.
	// Returns true if there was a request to remove.
	Remove(request Request) bool
	// Next returns the next request from the queue.
	Next() (Request, bool)
	// Len returns the number of requests in the queue.
	Len() int
	// DeleteCache removes the request from the queue and deletes the cache file.
	DeleteCache(request Request) error
}

Queue manages the retain requests queue.

type Request

type Request struct {
	SatelliteID   storj.NodeID
	CreatedBefore time.Time
	Filter        *bloomfilter.Filter
}

Request contains all the info necessary to process a retain request.

func (*Request) Filename

func (req *Request) Filename() string

Filename returns the filename used to store the request in the cache directory.

type RequestStore

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

RequestStore is a cache of requests to retain pieces.

func NewRequestStore

func NewRequestStore(path string) (RequestStore, error)

NewRequestStore loads the request caches from disk.

func (*RequestStore) Add

func (store *RequestStore) Add(req Request) (bool, error)

Add adds a request to the store. It returns true if the request was added, and an error if the file could not be saved.

func (*RequestStore) Data

func (store *RequestStore) Data() map[storj.NodeID]Request

Data returns the data in the store.

func (*RequestStore) DeleteCache

func (store *RequestStore) DeleteCache(req Request) error

DeleteCache removes the request from the store and deletes the cache file.

func (*RequestStore) Len

func (store *RequestStore) Len() int

Len returns the number of requests in the store.

func (*RequestStore) Next

func (store *RequestStore) Next() (Request, bool)

Next returns the next request from the store.

func (*RequestStore) Remove

func (store *RequestStore) Remove(req Request) bool

Remove removes a request from the queue. It returns true if the request was found in the queue. It does not remove the cache file from the filesystem.

type Service

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

Service queues and processes retain requests from satellites.

architecture: Worker

func NewService

func NewService(log *zap.Logger, store *pieces.Store, config Config) *Service

NewService creates a new retain service.

func (*Service) Close

func (s *Service) Close() error

Close causes any pending Run to exit and waits for any retain requests to clean up.

func (*Service) Queue

func (s *Service) Queue(req Request) bool

Queue adds a retain request to the queue. true is returned if the request is added to the queue, false if queue is closed.

func (*Service) Run

func (s *Service) Run(ctx context.Context) (err error)

Run listens for queued retain requests and processes them as they come in.

func (*Service) Status

func (s *Service) Status() Status

Status returns the retain status.

func (*Service) TestWaitUntilEmpty

func (s *Service) TestWaitUntilEmpty()

TestWaitUntilEmpty blocks until the queue and working is empty. When Run exits, it empties the queue.

func (*Service) TestingHowManyQueued

func (s *Service) TestingHowManyQueued() int

TestingHowManyQueued peeks at the number of bloom filters queued.

type Status

type Status uint32

Status is a type defining the enabled/disabled status of retain requests.

const (
	// Disabled means we do not do anything with retain requests.
	Disabled Status = iota + 1
	// Enabled means we fully enable retain requests and delete data not defined by bloom filter.
	Enabled
	// Debug means we partially enable retain requests, and print out pieces we should delete, without actually deleting them.
	Debug
)

func (*Status) Set

func (v *Status) Set(s string) error

Set implements pflag.Value.

func (*Status) String

func (v *Status) String() string

String implements pflag.Value.

func (*Status) Type

func (*Status) Type() string

Type implements pflag.Value.