sender

Documentation

Index

• Variables
• func IterateZipContent(ctx context.Context, project uplink.Project, bucket string, objectKey string, ...) (err error)
• func IterateZipObjectKeys(ctx context.Context, project uplink.Project, bucket string, prefix string, ...) (err error)
• func UnpackZipEntry(file *zip.File) (retainInfo *internalpb.RetainInfo, err error)
• type Config
• type Service
  • func NewService(log *zap.Logger, config Config, dialer rpc.Dialer, overlay overlay.DB) *Service
  • func (service *Service) Run(ctx context.Context) (err error)
  • func (service *Service) RunOnce(ctx context.Context) (err error)

Variables

var (
	// Error defines the gc service errors class.
	Error = errs.Class("gc")
)

Functions

func IterateZipContent

func IterateZipContent(
	ctx context.Context,
	project uplink.Project,
	bucket string,
	objectKey string,
	fn func(file *zip.File) error,
) (err error)

IterateZipContent opens a zip file at an object key and yields the files inside.

func IterateZipObjectKeys

func IterateZipObjectKeys(
	ctx context.Context,
	project uplink.Project,
	bucket string,
	prefix string,
	fn func(objectKey string) error,
) (err error)

IterateZipObjectKeys checks inside the top-level of a bucket and yields the keys which look like zip files.

func UnpackZipEntry

func UnpackZipEntry(
	file *zip.File,
) (retainInfo *internalpb.RetainInfo, err error)

UnpackZipEntry deserialized a retainInfo struct from a single file inside a zip file.

Types

type Config

type Config struct {
	Interval time.Duration `` /* 173-byte string literal not displayed */
	Enabled  bool          `help:"set if loop to send garbage collection retain filters is enabled" default:"false" devDefault:"false"`

	// We suspect this currently not to be the critical path w.r.t. garbage collection, so no paralellism is implemented.
	ConcurrentSends   int           `help:"the number of nodes to concurrently send garbage collection retain filters to" releaseDefault:"1" devDefault:"1"`
	RetainSendTimeout time.Duration `help:"the amount of time to allow a node to handle a retain request" default:"1m"`

	AccessGrant string        `help:"Access to download the bloom filters. Needs read and write permission."`
	Bucket      string        `help:"bucket where retain info is stored" default:"" testDefault:"gc-queue"`
	ExpireIn    time.Duration `help:"Expiration of newly created objects. These objects store error messages." default:"336h"`
}

Config contains configurable values for garbage collection.

type Service

type Service struct {
	Config Config
	Loop   *sync2.Cycle
	// contains filtered or unexported fields
}

Service reads bloom filters of piece IDs to retain from a Storj bucket and sends them out to the storage nodes.

The split between creating retain info and sending it out to storagenodes is made so that the bloom filter can be created from a backup database.

architecture: Service

func NewService

func NewService(log *zap.Logger, config Config, dialer rpc.Dialer, overlay overlay.DB) *Service

NewService creates a new instance of the gc sender service.

func (*Service) Run

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

Run continuously polls for new retain filters and sends them out.

func (*Service) RunOnce

func (service *Service) RunOnce(ctx context.Context) (err error)

RunOnce opens the bucket and sends out all the retain filters located in it to the storage nodes.