Add support for resumable downloads; clean up ingest directory; add --concurrency flag
#408
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
amisevsk
force-pushed
the
resumable-download
branch
from
2282937
to
9a588e1
Compare
gorkem
requested changes
Jul 15, 2024
pkg/cmd/options/network.go
Outdated
| @@ -43,6 +44,7 @@ func (o *NetworkOptions) AddNetworkFlags(cmd *cobra.Command) { | |||
| fmt.Sprintf("Path to client certificate used for authentication (can also be set via environment variable %s)", constants.ClientCertEnvVar)) | |||
| cmd.Flags().StringVar(&o.ClientCertKeyPath, "key", "", | |||
| fmt.Sprintf("Path to client certificate key used for authentication (can also be set via environment variable %s)", constants.ClientCertKeyEnvVar)) | |||
| cmd.Flags().IntVar(&o.Concurrency, "concurrency", 5, "Maximum number of simultaeous uploads/downloads") | |||
There was a problem hiding this comment.
typo
Suggested change
| cmd.Flags().IntVar(&o.Concurrency, "concurrency", 5, "Maximum number of simultaeous uploads/downloads") | |
| cmd.Flags().IntVar(&o.Concurrency, "concurrency", 5, "Maximum number of simultaneous uploads/downloads") |
| @@ -58,6 +60,9 @@ func (o *NetworkOptions) Complete(ctx context.Context, args []string) error { | |||
| if certKeyPath := os.Getenv(constants.ClientCertKeyEnvVar); certKeyPath != "" { | |||
| o.ClientCertKeyPath = certKeyPath | |||
| } | |||
| if o.Concurrency < 1 { | |||
There was a problem hiding this comment.
Is there a max concurrency?
There was a problem hiding this comment.
Not really. We don't have a lot of the usual number-of-threads concurrency worries since there's at most one goroutine per layer and if it crashes you can just rerun it with a sane number.
We're not allocating a threadpool or anything, so saying 999999 here would have no real effect.
pkg/cmd/pull/cmd.go
Outdated
| modelRef *registry.Reference | ||
| configHome string | ||
| modelRef *registry.Reference | ||
| Concurrency int64 |
There was a problem hiding this comment.
Ah this one is actually unused/redeclares the one in the NetworkOptions block.
pkg/cmd/pull/pull.go
Outdated
| } | ||
| logger.Wait() | ||
|
|
||
| return desc, err |
There was a problem hiding this comment.
Suggested change
| return desc, err | |
| return desc, nil |
| @@ -123,7 +125,7 @@ func (r *localRepo) GetRepoName() string { | |||
| } | |||
|
|
|||
| func (r *localRepo) BlobPath(desc ocispec.Descriptor) string { | |||
| return filepath.Join(r.storagePath, "blobs", "sha256", desc.Digest.Encoded()) | |||
| return filepath.Join(r.storagePath, ocispec.ImageBlobsDir, desc.Digest.Algorithm().String(), desc.Digest.Encoded()) | |||
There was a problem hiding this comment.
This is something I want to check, if we can easily support different digest levels. Filed #414
pkg/lib/repo/local/pull.go
Outdated
|
|
||
| func (l *localRepo) ensurePullDirs() error { | ||
| blobsPath := filepath.Join(l.storagePath, ocispec.ImageBlobsDir, "sha256") | ||
| if err := os.MkdirAll(blobsPath, 0777); err != nil { |
There was a problem hiding this comment.
Would it work if we had 0755 as permissions?
There was a problem hiding this comment.
I think it would, this is just copying what oras-go does.
pkg/lib/repo/local/pull.go
Outdated
| func (l *localRepo) resumeAndDownloadFile(desc ocispec.Descriptor, blob io.ReadSeekCloser, p *output.PullProgress) error { | ||
| ingestDir := constants.IngestPath(l.storagePath) | ||
| ingestFilename := filepath.Join(ingestDir, desc.Digest.Encoded()) | ||
| ingestFile, err := os.OpenFile(ingestFilename, os.O_CREATE|os.O_RDWR, 0666) |
There was a problem hiding this comment.
Would 0644 work as permission?
amisevsk
force-pushed
the
resumable-download
branch
from
9a588e1
to
db4ab86
Compare
Add basic implementation for PullModel, which pulls a modelkit from a remote reference. This is basically equivalent to the Copy() utility function in the oras library, which uses Fetch() and Push(); however, this base allows us control over how files are saved and enables implementing better cleanup and resumable downloads.
If a remote response with Accept-Ranges: "bytes", treat downloads as resumable. In this case: * Temporary ingest file is saved with the digest name in ingest directory * If a file already exists, seek to the end of it and seek to that point in the download * If a download is cancelled or encounters an error, the ingest file is not removed To avoid excessive slowdown, instead of seeking to the end of the file, we read it into a digester to resume calculating the digest of the downloaded data at the same time.
If a pull is cancelled via signal (instead of an error), the ingest file will not be cleaned up. To avoid accrual of partially-downloaded ingest files (generated by the non-resumable download case), clear the ingest directory on every successful pull. This may remove a partially-completed resumable download, but this should only occur if the user downloads some _other_ modelkit before attempting to resume the download.
Since we're no longer using a basic copy operation, we need to redo progress bars for pull from scratch.
Add flag --concurrency to allow configuring maximum number of simultaneous uploads/downloads. Default value is 5.
amisevsk
force-pushed
the
resumable-download
branch
from
db4ab86
to
cc597dc
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
Add support for resuming downloads that were cancelled/hit an error in an earlier invocation. Doing this required implementing our own Pull routine, separate from oras.Copy, which is used elsewhere.
If a server responds with header
Accept-Ranges: "bytes", treat the download as resumable; otherwise, pull behaves as before. For resumable downloads:$KITOPS_HOME/storage/ingest/<blob-digest>There's some risk of corruption if e.g. two separate kit pulls are pulling the same resumable download simultaneously, though I think the risk is fairly low and I'm not sure we care to support running
kitin parallel at this moment.When a pull is completed, all data in the ingest directory is removed. This is to ensure we don't leave files lying around with no easy path to their removal, which can happen with non-resumable downloads that are e.g. cancelled via ctrl-C. This means that completing a pull effectively cancels all other downloads.
Finally, since I had to add reimplement concurrency for the pull operation, I've added a
--concurrencyflag to allow configuring it. It's available on all commands that take the standard network options, but is currently only used for pull and push.Linked issues
Closes #311
Closes #387