local

Documentation

Index

• Constants
• Variables
• func ChangeSourceString(changeSource ChangeSource) string
• func ChangeTypeString(changeType ChangeType) string
• func FindIndices(root string) ([]string, error)
• func IndexExists(baseAbs string) (bool, error)
• func ListRemote(ctx context.Context, client apigen.ClientWithResponsesInterface, loc *uri.URI, ...) error
• type Change
  • func (c *Change) String() string
• type ChangeSource
• type ChangeType
  • func ChangeTypeFromString(changeType string) ChangeType
• type Changes
  • func DiffLocalWithHead(left <-chan apigen.ObjectStats, rightPath string) (Changes, error)
  • func Undo(c Changes) Changes
  • func (c Changes) MergeWith(other Changes, strategy MergeStrategy) Changes
  • func (c Changes) String() string
• type Index
  • func ReadIndex(path string) (*Index, error)
  • func WriteIndex(path string, remote *uri.URI, atHead string, operation string) (*Index, error)
  • func (l *Index) GetCurrentURI() (*uri.URI, error)
  • func (l *Index) LocalPath() string
• type MergeStrategy
• type ProgressPool
  • func NewProgressPool() *ProgressPool
  • func (p *ProgressPool) AddReader(name string, sizeBytes int64) *ProgressUpdater
  • func (p *ProgressPool) AddSpinner(name string) *ProgressSpinner
  • func (p *ProgressPool) Start()
  • func (p *ProgressPool) Stop()
• type ProgressSpinner
  • func (p *ProgressSpinner) Done()
  • func (p *ProgressSpinner) Error()
• type ProgressUpdater
  • func (p *ProgressUpdater) Done()
  • func (p *ProgressUpdater) Error()
  • func (p *ProgressUpdater) Reader(reader io.Reader) io.Reader
• type ProgressUpdaterReader
  • func (pu *ProgressUpdaterReader) Read(p []byte) (n int, err error)
• type SyncFlags
• type SyncManager
  • func NewSyncManager(ctx context.Context, client *apigen.ClientWithResponses, ...) *SyncManager
  • func (s *SyncManager) Summary() Tasks
  • func (s *SyncManager) Sync(rootPath string, remote *uri.URI, changeSet <-chan *Change) error
• type Tasks

Constants

const (
	IndexFileName = ".lakefs_ref.yaml"
	IgnoreMarker  = "ignored by lakectl local"
	IndexFileMode = 0o644
)

const (
	DefaultDirectoryMask   = 0o755
	ClientMtimeMetadataKey = apiutil.LakeFSMetadataPrefix + "client-mtime"
)

Variables

var (
	ErrConflict        = errors.New("conflict")
	ErrDownloadingFile = errors.New("error downloading file")
	ErrRemoteFailure   = errors.New("remote failure")
)

Functions

func ChangeSourceString

func ChangeSourceString(changeSource ChangeSource) string

func ChangeTypeString

func ChangeTypeString(changeType ChangeType) string

func FindIndices

func FindIndices(root string) ([]string, error)

FindIndices searches the specified root directory for index files, returning their relative directory paths while skipping hidden folders.

func IndexExists

func IndexExists(baseAbs string) (bool, error)

func ListRemote

func ListRemote(ctx context.Context, client apigen.ClientWithResponsesInterface, loc *uri.URI, objects chan<- apigen.ObjectStats) error

ListRemote - Lists objects from a remote uri and inserts them into the objects channel

Types

type Change

type Change struct {
	Source ChangeSource
	Path   string
	Type   ChangeType
}

func (*Change) String

func (c *Change) String() string

type ChangeSource

type ChangeSource int

const (
	ChangeSourceRemote ChangeSource = iota
	ChangeSourceLocal
)

type ChangeType

type ChangeType int

const (
	ChangeTypeAdded ChangeType = iota
	ChangeTypeModified
	ChangeTypeRemoved
	ChangeTypeConflict
)

func ChangeTypeFromString

func ChangeTypeFromString(changeType string) ChangeType

type Changes

type Changes []*Change

func DiffLocalWithHead

func DiffLocalWithHead(left <-chan apigen.ObjectStats, rightPath string) (Changes, error)

DiffLocalWithHead Checks changes between a local directory and the head it is pointing to. The diff check assumes the remote is an immutable set so any changes found resulted from changes in the local directory left is an object channel which contains results from a remote source. rightPath is the local directory to diff with

func Undo

func Undo(c Changes) Changes

Undo Creates a new list of changes that reverses the given changes list.

func (Changes) MergeWith

func (c Changes) MergeWith(other Changes, strategy MergeStrategy) Changes

MergeWith combines changes from two diffs, sorting by lexicographic order. If the same path appears in both diffs, it's marked as a conflict.

func (Changes) String

func (c Changes) String() string

type Index

type Index struct {
	PathURI         string `yaml:"src"`
	AtHead          string `yaml:"at_head"`
	ActiveOperation string `yaml:"active_operation"`
	// contains filtered or unexported fields
}

Index defines the structure of the lakefs local reference file consisting of the information linking local directory with lakefs path

func ReadIndex

func ReadIndex(path string) (*Index, error)

func WriteIndex

func WriteIndex(path string, remote *uri.URI, atHead string, operation string) (*Index, error)

func (*Index) GetCurrentURI

func (l *Index) GetCurrentURI() (*uri.URI, error)

func (*Index) LocalPath

func (l *Index) LocalPath() string

type MergeStrategy

type MergeStrategy int

const (
	MergeStrategyNone MergeStrategy = iota
	MergeStrategyThis
	MergeStrategyOther
)

type ProgressPool

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

func NewProgressPool

func NewProgressPool() *ProgressPool

func (*ProgressPool) AddReader

func (p *ProgressPool) AddReader(name string, sizeBytes int64) *ProgressUpdater

func (*ProgressPool) AddSpinner

func (p *ProgressPool) AddSpinner(name string) *ProgressSpinner

func (*ProgressPool) Start

func (p *ProgressPool) Start()

func (*ProgressPool) Stop

func (p *ProgressPool) Stop()

type ProgressSpinner

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

func (*ProgressSpinner) Done

func (p *ProgressSpinner) Done()

func (*ProgressSpinner) Error

func (p *ProgressSpinner) Error()

type ProgressUpdater

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

func (*ProgressUpdater) Done

func (p *ProgressUpdater) Done()

func (*ProgressUpdater) Error

func (p *ProgressUpdater) Error()

func (*ProgressUpdater) Reader

func (p *ProgressUpdater) Reader(reader io.Reader) io.Reader

type ProgressUpdaterReader

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

func (*ProgressUpdaterReader) Read

func (pu *ProgressUpdaterReader) Read(p []byte) (n int, err error)

type SyncFlags

type SyncFlags struct {
	Parallelism      int
	Presign          bool
	PresignMultipart bool
}

type SyncManager

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

func NewSyncManager

func NewSyncManager(ctx context.Context, client *apigen.ClientWithResponses, httpClient *http.Client, flags SyncFlags) *SyncManager

func (*SyncManager) Summary

func (s *SyncManager) Summary() Tasks

func (*SyncManager) Sync

func (s *SyncManager) Sync(rootPath string, remote *uri.URI, changeSet <-chan *Change) error

Sync - sync changes between remote and local directory given the Changes channel. For each change, will apply download, upload or delete according to the change type and change source

type Tasks

type Tasks struct {
	Downloaded uint64
	Uploaded   uint64
	Removed    uint64
}