mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-19 14:34:17 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

bump up kopia v0.13.0

Browse Source 
Signed-off-by: Lyndon-Li <lyonghui@vmware.com>
This commit is contained in:
Lyndon-Li
2023-05-15 07:23:38 +08:00
parent 4db1a781fc
commit 307b82a2ec
37 changed files with 939 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/uploader/kopia/progress.go
+2
View File
@@ -151,3 +151,5 @@ func (p *Progress) ProgressBytes(processedBytes int64, totalBytes int64) {
atomic.StoreInt64(&p.estimatedTotalBytes, totalBytes)
p.UpdateProgress()
}
func (p *Progress) FinishedFile(fname string, err error) {}
pkg/uploader/kopia/shim.go
+38 -8
View File
@@ -55,7 +55,7 @@ func NewShimRepo(repo udmrepo.BackupRepo) repo.RepositoryWriter {
// OpenObject open specific object
func (sr *shimRepository) OpenObject(ctx context.Context, id object.ID) (object.Reader, error) {
reader, err := sr.udmRepo.OpenObject(ctx, udmrepo.ID(id))
reader, err := sr.udmRepo.OpenObject(ctx, udmrepo.ID(id.String()))
if err != nil {
return nil, errors.Wrapf(err, "failed to open object with id %v", id)
}
@@ -70,7 +70,7 @@ func (sr *shimRepository) OpenObject(ctx context.Context, id object.ID) (object.
// VerifyObject not supported
func (sr *shimRepository) VerifyObject(ctx context.Context, id object.ID) ([]content.ID, error) {
return nil, errors.New("not supported")
return nil, errors.New("VerifyObject is not supported")
}
// Get one or more manifest data that match the specific manifest id
@@ -135,12 +135,12 @@ func (sr *shimRepository) ClientOptions() repo.ClientOptions {
// Refresh not supported
func (sr *shimRepository) Refresh(ctx context.Context) error {
return errors.New("not supported")
return errors.New("Refresh is not supported")
}
// ContentInfo not supported
func (sr *shimRepository) ContentInfo(ctx context.Context, contentID content.ID) (content.Info, error) {
return nil, errors.New("not supported")
return nil, errors.New("ContentInfo is not supported")
}
// PrefetchContents is not supported by unified repo
@@ -150,7 +150,7 @@ func (sr *shimRepository) PrefetchContents(ctx context.Context, contentIDs []con
// PrefetchObjects is not supported by unified repo
func (sr *shimRepository) PrefetchObjects(ctx context.Context, objectIDs []object.ID, hint string) ([]content.ID, error) {
return nil, errors.New("not supported")
return nil, errors.New("PrefetchObjects is not supported")
}
// UpdateDescription is not supported by unified repo
@@ -159,7 +159,7 @@ func (sr *shimRepository) UpdateDescription(d string) {
// NewWriter is not supported by unified repo
func (sr *shimRepository) NewWriter(ctx context.Context, option repo.WriteSessionOptions) (context.Context, repo.RepositoryWriter, error) {
return nil, nil, errors.New("not supported")
return nil, nil, errors.New("NewWriter is not supported")
}
// Close will close unified repo
@@ -174,6 +174,7 @@ func (sr *shimRepository) NewObjectWriter(ctx context.Context, option object.Wri
opt.Prefix = udmrepo.ID(option.Prefix)
opt.FullPath = ""
opt.AccessMode = udmrepo.ObjectDataAccessModeFile
opt.AsyncWrites = option.AsyncWrites
if strings.HasPrefix(option.Description, "DIR:") {
opt.DataType = udmrepo.ObjectDataTypeMetadata
@@ -208,11 +209,22 @@ func (sr *shimRepository) DeleteManifest(ctx context.Context, id manifest.ID) er
return sr.udmRepo.DeleteManifest(ctx, udmrepo.ID(id))
}
func (sr *shimRepository) ReplaceManifests(ctx context.Context, labels map[string]string, payload interface{}) (manifest.ID, error) {
return manifest.ID(""), errors.New("ReplaceManifests is not supported")
}
// Flush all the unifited repository data
func (sr *shimRepository) Flush(ctx context.Context) error {
return sr.udmRepo.Flush(ctx)
}
func (sr *shimRepository) ConcatenateObjects(ctx context.Context, objectIDs []object.ID) (object.ID, error) {
return object.ID{}, errors.New("ConcatenateObjects is not supported")
}
func (sr *shimRepository) OnSuccessfulFlush(callback repo.RepositoryWriterCallback) {
}
// Flush all the unifited repository data
func (sr *shimObjectReader) Read(p []byte) (n int, err error) {
return sr.repoReader.Read(p)
@@ -240,13 +252,31 @@ func (sr *shimObjectWriter) Write(p []byte) (n int, err error) {
// Periodically called to preserve the state of data written to the repo so far.
func (sr *shimObjectWriter) Checkpoint() (object.ID, error) {
id, err := sr.repoWriter.Checkpoint()
return object.ID(id), err
if err != nil {
return object.ID{}, err
}
objID, err := object.ParseID(string(id))
if err != nil {
return object.ID{}, errors.Wrapf(err, "error to parse object ID from %v", id)
}
return objID, err
}
// Result returns the object's unified identifier after the write completes.
func (sr *shimObjectWriter) Result() (object.ID, error) {
id, err := sr.repoWriter.Result()
return object.ID(id), err
if err != nil {
return object.ID{}, err
}
objID, err := object.ParseID(string(id))
if err != nil {
return object.ID{}, errors.Wrapf(err, "error to parse object ID from %v", id)
}
return objID, err
}
// Close closes the repository and releases all resources.
pkg/uploader/kopia/snapshot.go
+19 -30
View File
@@ -44,9 +44,7 @@ import (
)
// All function mainly used to make testing more convenient
var treeForSourceFunc = policy.TreeForSource
var applyRetentionPolicyFunc = policy.ApplyRetentionPolicy
var setPolicyFunc = policy.SetPolicy
var saveSnapshotFunc = snapshot.SaveSnapshot
var loadSnapshotFunc = snapshot.LoadSnapshot
@@ -72,24 +70,17 @@ func newOptionalBool(b bool) *policy.OptionalBool {
}
// setupDefaultPolicy set default policy for kopia
func setupDefaultPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snapshot.SourceInfo) error {
return setPolicyFunc(ctx, rep, sourceInfo, &policy.Policy{
RetentionPolicy: policy.RetentionPolicy{
KeepLatest: newOptionalInt(math.MaxInt32),
},
CompressionPolicy: policy.CompressionPolicy{
CompressorName: "none",
},
UploadPolicy: policy.UploadPolicy{
MaxParallelFileReads: newOptionalInt(runtime.NumCPU()),
},
SchedulingPolicy: policy.SchedulingPolicy{
Manual: true,
},
ErrorHandlingPolicy: policy.ErrorHandlingPolicy{
IgnoreUnknownTypes: newOptionalBool(true),
},
})
func setupDefaultPolicy() *policy.Tree {
defaultPolicy := *policy.DefaultPolicy
defaultPolicy.RetentionPolicy.KeepLatest = newOptionalInt(math.MaxInt32)
defaultPolicy.CompressionPolicy.CompressorName = "none"
defaultPolicy.UploadPolicy.MaxParallelFileReads = newOptionalInt(runtime.NumCPU())
defaultPolicy.UploadPolicy.ParallelUploadAboveSize = nil
defaultPolicy.SchedulingPolicy.Manual = true
defaultPolicy.ErrorHandlingPolicy.IgnoreUnknownTypes = newOptionalBool(true)
return policy.BuildTree(nil, &defaultPolicy)
}
// Backup backup specific sourcePath and update progress
@@ -193,17 +184,10 @@ func SnapshotSource(
previous = pre
}
var manifest *snapshot.Manifest
if err := setupDefaultPolicy(ctx, rep, sourceInfo); err != nil {
return "", 0, errors.Wrapf(err, "unable to set policy for si %v", sourceInfo)
}
policyTree, err := treeForSourceFunc(ctx, rep, sourceInfo)
if err != nil {
return "", 0, errors.Wrapf(err, "unable to create policy getter for si %v", sourceInfo)
}
policyTree := setupDefaultPolicy()
manifest, err = u.Upload(ctx, rootDir, policyTree, sourceInfo, previous...)
manifest, err := u.Upload(ctx, rootDir, policyTree, sourceInfo, previous...)
if err != nil {
return "", 0, errors.Wrapf(err, "Failed to upload the kopia snapshot for si %v", sourceInfo)
}
@@ -247,7 +231,7 @@ func reportSnapshotStatus(manifest *snapshot.Manifest, policyTree *policy.Tree)
// findPreviousSnapshotManifest returns the list of previous snapshots for a given source, including
// last complete snapshot following it.
func findPreviousSnapshotManifest(ctx context.Context, rep repo.Repository, sourceInfo snapshot.SourceInfo, noLaterThan *time.Time) ([]*snapshot.Manifest, error) {
func findPreviousSnapshotManifest(ctx context.Context, rep repo.Repository, sourceInfo snapshot.SourceInfo, noLaterThan *fs.UTCTimestamp) ([]*snapshot.Manifest, error) {
man, err := snapshot.ListSnapshots(ctx, rep, sourceInfo)
if err != nil {
return nil, err
@@ -297,6 +281,11 @@ func Restore(ctx context.Context, rep repo.RepositoryWriter, progress *Progress,
IgnorePermissionErrors: true,
}
err = output.Init(ctx)
if err != nil {
return 0, 0, errors.Wrap(err, "error to init output")
}
stat, err := restore.Entry(kopiaCtx, rep, output, rootEntry, restore.Options{
Parallel: runtime.NumCPU(),
RestoreDirEntryAtDepth: math.MaxInt32,
pkg/uploader/kopia/snapshot_test.go
-15
View File
@@ -50,8 +50,6 @@ func injectSnapshotFuncs() *snapshotMockes {
repoWriterMock: &repomocks.RepositoryWriter{},
}
setPolicyFunc = s.policyMock.SetPolicy
treeForSourceFunc = s.policyMock.TreeForSource
applyRetentionPolicyFunc = s.policyMock.ApplyRetentionPolicy
loadSnapshotFunc = s.snapshotMock.LoadSnapshot
saveSnapshotFunc = s.snapshotMock.SaveSnapshot
@@ -141,19 +139,6 @@ func TestSnapshotSource(t *testing.T) {
},
notError: false,
},
{
name: "failed to set policy",
args: []mockArgs{
{methodName: "LoadSnapshot", returns: []interface{}{manifest, nil}},
{methodName: "SaveSnapshot", returns: []interface{}{manifest.ID, nil}},
{methodName: "TreeForSource", returns: []interface{}{nil, nil}},
{methodName: "ApplyRetentionPolicy", returns: []interface{}{nil, nil}},
{methodName: "SetPolicy", returns: []interface{}{errors.New("failed to set policy")}},
{methodName: "Upload", returns: []interface{}{manifest, nil}},
{methodName: "Flush", returns: []interface{}{nil}},
},
notError: false,
},
{
name: "failed to upload snapshot",
args: []mockArgs{
pkg/uploader/provider/kopia.go
+4 -4
View File
@@ -177,9 +177,9 @@ func (kp *kopiaProvider) RunRestore(
"volumePath": volumePath,
})
repoWriter := kopia.NewShimRepo(kp.bkRepo)
prorgess := new(kopia.Progress)
prorgess.InitThrottle(restoreProgressCheckInterval)
prorgess.Updater = updater
progress := new(kopia.Progress)
progress.InitThrottle(restoreProgressCheckInterval)
progress.Updater = updater
restoreCancel := make(chan struct{})
quit := make(chan struct{})
@@ -193,7 +193,7 @@ func (kp *kopiaProvider) RunRestore(
close(quit)
}()
size, fileCount, err := RestoreFunc(ctx, repoWriter, prorgess, snapshotID, volumePath, log, restoreCancel)
size, fileCount, err := RestoreFunc(ctx, repoWriter, progress, snapshotID, volumePath, log, restoreCancel)
if err != nil {
return errors.Wrapf(err, "Failed to run kopia restore")