epoch

Documentation

Overview

Package epoch manages repository epochs. It implements protocol described https://github.com/kopia/kopia/issues/1090 and is intentionally separate from 'content' package to be able to test in isolation.

Index

• Constants
• Variables
• func UncompactedEpochBlobPrefix(epoch int) blob.ID
• type CompactionFunc
• type CurrentSnapshot
• type Manager
  • func NewManager(st blob.Storage, paramProvider ParametersProvider, compactor CompactionFunc, ...) *Manager
  • func (e *Manager) AdvanceDeletionWatermark(ctx context.Context, ts time.Time) error
  • func (e *Manager) CleanupSupersededIndexes(ctx context.Context) error
  • func (e *Manager) Current(ctx context.Context) (CurrentSnapshot, error)
  • func (e *Manager) Flush()
  • func (e *Manager) ForceAdvanceEpoch(ctx context.Context) error
  • func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob.Metadata, time.Time, error)
  • func (e *Manager) Invalidate()
  • func (e *Manager) Refresh(ctx context.Context) error
  • func (e *Manager) WriteIndex(ctx context.Context, dataShards map[blob.ID]blob.Bytes) ([]blob.Metadata, error)
• type Parameters
  • func DefaultParameters() Parameters
  • func (p *Parameters) GetEpochAdvanceOnCountThreshold() int
  • func (p *Parameters) GetEpochAdvanceOnTotalSizeBytesThreshold() int64
  • func (p *Parameters) GetEpochCleanupSafetyMargin() time.Duration
  • func (p *Parameters) GetEpochDeleteParallelism() int
  • func (p *Parameters) GetEpochFullCheckpointFrequency() int
  • func (p *Parameters) GetEpochManagerEnabled() bool
  • func (p *Parameters) GetEpochRefreshFrequency() time.Duration
  • func (p *Parameters) GetMinEpochDuration() time.Duration
  • func (p *Parameters) Validate() error
• type ParametersProvider
• type RangeMetadata

Constants

const (
	EpochManagerIndexUberPrefix = "x"

	EpochMarkerIndexBlobPrefix      blob.ID = EpochManagerIndexUberPrefix + "e"
	UncompactedIndexBlobPrefix      blob.ID = EpochManagerIndexUberPrefix + "n"
	SingleEpochCompactionBlobPrefix blob.ID = EpochManagerIndexUberPrefix + "s"
	RangeCheckpointIndexBlobPrefix  blob.ID = EpochManagerIndexUberPrefix + "r"
	DeletionWatermarkBlobPrefix     blob.ID = EpochManagerIndexUberPrefix + "w"
)

Index blob prefixes.

const FirstEpoch = 0

FirstEpoch is the number of the first epoch in a repository.

const LatestEpoch = -1

LatestEpoch represents the current epoch number in GetCompleteIndexSet.

Variables

var ErrVerySlowIndexWrite = errors.Errorf("extremely slow index write - index write took more than two epochs")

ErrVerySlowIndexWrite is returned by WriteIndex if a write takes more than 2 epochs (usually >48h). This is theoretically possible with laptops going to sleep, etc.

Functions

func UncompactedEpochBlobPrefix

func UncompactedEpochBlobPrefix(epoch int) blob.ID

UncompactedEpochBlobPrefix returns the prefix for uncompacted blobs of a given epoch.

Types

type CompactionFunc

type CompactionFunc func(ctx context.Context, blobIDs []blob.ID, outputPrefix blob.ID) error

CompactionFunc merges the given set of index blobs into a new index blob set with a given prefix and writes them out as a set following naming convention established in 'complete_set.go'.

type CurrentSnapshot

type CurrentSnapshot struct {
	WriteEpoch                 int                     `json:"writeEpoch"`
	UncompactedEpochSets       map[int][]blob.Metadata `json:"unsettled"`
	LongestRangeCheckpointSets []*RangeMetadata        `json:"longestRangeCheckpointSets"`
	SingleEpochCompactionSets  map[int][]blob.Metadata `json:"singleEpochCompactionSets"`
	EpochStartTime             map[int]time.Time       `json:"epochStartTimes"`
	DeletionWatermark          time.Time               `json:"deletionWatermark"`
	ValidUntil                 time.Time               `json:"validUntil"`             // time after which the contents of this struct are no longer valid
	EpochMarkerBlobs           []blob.Metadata         `json:"epochMarkers"`           // list of epoch markers
	DeletionWatermarkBlobs     []blob.Metadata         `json:"deletionWatermarkBlobs"` // list of deletion watermark blobs
}

CurrentSnapshot captures a point-in time snapshot of a repository indexes, including current epoch information and compaction set.

type Manager

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

Manager manages repository epochs.

func NewManager

func NewManager(st blob.Storage, paramProvider ParametersProvider, compactor CompactionFunc, log logging.Logger, timeNow func() time.Time) *Manager

NewManager creates new epoch manager.

func (*Manager) AdvanceDeletionWatermark

func (e *Manager) AdvanceDeletionWatermark(ctx context.Context, ts time.Time) error

AdvanceDeletionWatermark moves the deletion watermark time to a given timestamp this causes all deleted content entries before given time to be treated as non-existent.

func (*Manager) CleanupSupersededIndexes

func (e *Manager) CleanupSupersededIndexes(ctx context.Context) error

CleanupSupersededIndexes cleans up the indexes which have been superseded by compacted ones.

func (*Manager) Current

func (e *Manager) Current(ctx context.Context) (CurrentSnapshot, error)

Current retrieves current snapshot.

func (*Manager) Flush

func (e *Manager) Flush()

Flush waits for all in-process compaction work to complete.

func (*Manager) ForceAdvanceEpoch

func (e *Manager) ForceAdvanceEpoch(ctx context.Context) error

ForceAdvanceEpoch advances current epoch unconditionally.

func (*Manager) GetCompleteIndexSet

func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob.Metadata, time.Time, error)

GetCompleteIndexSet returns the set of blobs forming a complete index set up to the provided epoch number.

func (*Manager) Invalidate

func (e *Manager) Invalidate()

Invalidate ensures that all cached index information is discarded.

func (*Manager) Refresh

func (e *Manager) Refresh(ctx context.Context) error

Refresh refreshes information about current epoch.

func (*Manager) WriteIndex

func (e *Manager) WriteIndex(ctx context.Context, dataShards map[blob.ID]blob.Bytes) ([]blob.Metadata, error)

WriteIndex writes new index blob by picking the appropriate prefix based on current epoch.

type Parameters

type Parameters struct {
	// whether epoch manager is enabled, must be true.
	Enabled bool

	// how frequently each client will list blobs to determine the current epoch.
	EpochRefreshFrequency time.Duration

	// number of epochs between full checkpoints.
	FullCheckpointFrequency int

	// do not delete uncompacted blobs if the corresponding compacted blob age is less than this.
	CleanupSafetyMargin time.Duration

	// minimum duration of an epoch
	MinEpochDuration time.Duration

	// advance epoch if number of files exceeds this
	EpochAdvanceOnCountThreshold int

	// advance epoch if total size of files exceeds this.
	EpochAdvanceOnTotalSizeBytesThreshold int64

	// number of blobs to delete in parallel during cleanup
	DeleteParallelism int
}

Parameters encapsulates all parameters that influence the behavior of epoch manager.

func DefaultParameters

func DefaultParameters() Parameters

DefaultParameters contains default epoch manager parameters.

func (*Parameters) GetEpochAdvanceOnCountThreshold

func (p *Parameters) GetEpochAdvanceOnCountThreshold() int

GetEpochAdvanceOnCountThreshold returns the number of files above which epoch should be advanced.

func (*Parameters) GetEpochAdvanceOnTotalSizeBytesThreshold

func (p *Parameters) GetEpochAdvanceOnTotalSizeBytesThreshold() int64

GetEpochAdvanceOnTotalSizeBytesThreshold returns the total size of files above which the epoch should be advanced.

func (*Parameters) GetEpochCleanupSafetyMargin

func (p *Parameters) GetEpochCleanupSafetyMargin() time.Duration

GetEpochCleanupSafetyMargin returns safety margin to prevent uncompacted blobs from being deleted if the corresponding compacted blob age is less than this.

func (*Parameters) GetEpochDeleteParallelism

func (p *Parameters) GetEpochDeleteParallelism() int

GetEpochDeleteParallelism returns the number of blobs to delete in parallel during cleanup.

func (*Parameters) GetEpochFullCheckpointFrequency

func (p *Parameters) GetEpochFullCheckpointFrequency() int

GetEpochFullCheckpointFrequency returns the number of epochs between full checkpoints.

func (*Parameters) GetEpochManagerEnabled

func (p *Parameters) GetEpochManagerEnabled() bool

GetEpochManagerEnabled returns whether epoch manager is enabled, must be true.

func (*Parameters) GetEpochRefreshFrequency

func (p *Parameters) GetEpochRefreshFrequency() time.Duration

GetEpochRefreshFrequency determines how frequently each client will list blobs to determine the current epoch.

func (*Parameters) GetMinEpochDuration

func (p *Parameters) GetMinEpochDuration() time.Duration

GetMinEpochDuration returns the minimum duration of an epoch.

func (*Parameters) Validate

func (p *Parameters) Validate() error

Validate validates epoch parameters.

type ParametersProvider

type ParametersProvider interface {
	GetParameters() (*Parameters, error)
}

ParametersProvider provides epoch manager parameters.

type RangeMetadata

type RangeMetadata struct {
	MinEpoch int             `json:"min"`
	MaxEpoch int             `json:"max"`
	Blobs    []blob.Metadata `json:"blobs"`
}

RangeMetadata represents a range of indexes for [min,max] epoch range. Both min and max are inclusive.