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, params Parameters, compactor CompactionFunc, ...) *Manager
  • func (e *Manager) AdvanceDeletionWatermark(ctx context.Context, ts time.Time) error
  • func (e *Manager) Cleanup(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 (p *Parameters) Validate() error
• 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 DefaultParameters = Parameters{
	Enabled:                               true,
	EpochRefreshFrequency:                 20 * time.Minute,
	FullCheckpointFrequency:               7,
	CleanupSafetyMargin:                   4 * time.Hour,
	MinEpochDuration:                      24 * time.Hour,
	EpochAdvanceOnCountThreshold:          20,
	EpochAdvanceOnTotalSizeBytesThreshold: 10 << 20,
	DeleteParallelism:                     4,
}

DefaultParameters contains default epoch manager parameters. nolint:gomnd

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
}

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

type Manager

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

Manager manages repository epochs.

func NewManager

func NewManager(st blob.Storage, params Parameters, 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) Cleanup

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

Cleanup cleans up the old indexes for which there's a compacted replacement.

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 (*Parameters) Validate

func (p *Parameters) Validate() error

Validate validates epoch parameters. nolint:gomnd

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.