sync

Documentation

Index

• Constants
• Variables
• func RunSyncGroup(ctx context.Context, name string, syncGroup *SyncGroup) ([]*storage.Object, error)
• type SyncGroup
  • func PrepareSyncGroup(ctx context.Context, source *kubernetes.BucketDetails, sourceEndpoint string, ...) (*SyncGroup, error)

Constants

const KEYS_PER_REQ = 1000

Variables

var AdvancedObjectFilter pipeline.StepFn = func(group *pipeline.Group, stepNum int, input <-chan *storage.Object, output chan<- *storage.Object, errChan chan<- error) {
	for obj := range input {

		destObj := &storage.Object{
			Key:       obj.Key,
			VersionId: obj.VersionId,
		}
		err := group.Target.GetObjectMeta(destObj)

		if err != nil || obj.ETag == nil {

			output <- obj
			continue
		}

		if destObj.ETag != nil && *obj.ETag == *destObj.ETag {

			continue
		}

		if origETag, ok := destObj.Metadata["X-Original-Etag"]; ok && origETag != nil && *obj.ETag == *origETag {

			continue
		}

		output <- obj
	}
}

FilterObjectsModified accepts an input object and checks if it matches the filter This filter read object meta from target storage and compare object ETags. If Etags are equal object will be skipped For FS storage xattr support are required for proper work.

var AnnotateETag pipeline.StepFn = func(group *pipeline.Group, stepNum int, input <-chan *storage.Object, output chan<- *storage.Object, errChan chan<- error) {
	for obj := range input {

		if obj.ETag != nil {
			if obj.Metadata == nil {
				obj.Metadata = make(map[string]*string)
			}
			obj.Metadata["X-Original-Etag"] = obj.ETag
		}
		output <- obj
	}
}

var ListRetryFiles = func(retryFiles []*storage.Object) pipeline.StepFn {
	return func(group *pipeline.Group, stepNum int, input <-chan *storage.Object, output chan<- *storage.Object, errChan chan<- error) {
		for obj := range lo.SliceToChannel(32, retryFiles) {
			output <- obj
		}
	}
}

Functions

func RunSyncGroup

func RunSyncGroup(ctx context.Context, name string, syncGroup *SyncGroup) ([]*storage.Object, error)

Types

type SyncGroup

type SyncGroup struct {
	pipeline.Group
	// contains filtered or unexported fields
}

func PrepareSyncGroup

func PrepareSyncGroup(
	ctx context.Context,
	source *kubernetes.BucketDetails,
	sourceEndpoint string,
	target *kubernetes.BucketDetails,
	targetEndpoint string,
	numWorkers int,
	retryFiles []*storage.Object,
) (*SyncGroup, error)