-
Notifications
You must be signed in to change notification settings - Fork 18.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #44756 from rumpl/containerd-image-pull
containerd integration: image pull
- Loading branch information
Showing
6 changed files
with
267 additions
and
25 deletions.
There are no files selected for viewing
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,150 @@ | ||
| package containerd | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/containerd/containerd/content" | ||
| cerrdefs "github.com/containerd/containerd/errdefs" | ||
| "github.com/containerd/containerd/remotes" | ||
| "github.com/docker/docker/pkg/progress" | ||
| "github.com/docker/docker/pkg/stringid" | ||
| "github.com/opencontainers/go-digest" | ||
| ocispec "github.com/opencontainers/image-spec/specs-go/v1" | ||
| "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| type progressUpdater interface { | ||
| UpdateProgress(context.Context, *jobs, progress.Output, time.Time) error | ||
| } | ||
|
|
||
| type jobs struct { | ||
| descs map[digest.Digest]ocispec.Descriptor | ||
| mu sync.Mutex | ||
| } | ||
|
|
||
| // newJobs creates a new instance of the job status tracker | ||
| func newJobs() *jobs { | ||
| return &jobs{ | ||
| descs: map[digest.Digest]ocispec.Descriptor{}, | ||
| } | ||
| } | ||
|
|
||
| func (j *jobs) showProgress(ctx context.Context, out progress.Output, updater progressUpdater) func() { | ||
| ctx, cancelProgress := context.WithCancel(ctx) | ||
|
|
||
| start := time.Now() | ||
|
|
||
| go func() { | ||
| ticker := time.NewTicker(100 * time.Millisecond) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| if err := updater.UpdateProgress(ctx, j, out, start); err != nil { | ||
| if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { | ||
| logrus.WithError(err).Error("Updating progress failed") | ||
| } | ||
| return | ||
| } | ||
| case <-ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| return cancelProgress | ||
| } | ||
|
|
||
| // Add adds a descriptor to be tracked | ||
| func (j *jobs) Add(desc ocispec.Descriptor) { | ||
| j.mu.Lock() | ||
| defer j.mu.Unlock() | ||
|
|
||
| if _, ok := j.descs[desc.Digest]; ok { | ||
| return | ||
| } | ||
| j.descs[desc.Digest] = desc | ||
| } | ||
|
|
||
| // Remove removes a descriptor | ||
| func (j *jobs) Remove(desc ocispec.Descriptor) { | ||
| j.mu.Lock() | ||
| defer j.mu.Unlock() | ||
|
|
||
| delete(j.descs, desc.Digest) | ||
| } | ||
|
|
||
| // Jobs returns a list of all tracked descriptors | ||
| func (j *jobs) Jobs() []ocispec.Descriptor { | ||
| j.mu.Lock() | ||
| defer j.mu.Unlock() | ||
|
|
||
| descs := make([]ocispec.Descriptor, 0, len(j.descs)) | ||
| for _, d := range j.descs { | ||
| descs = append(descs, d) | ||
| } | ||
| return descs | ||
| } | ||
|
|
||
| type pullProgress struct { | ||
| Store content.Store | ||
| ShowExists bool | ||
| } | ||
|
|
||
| func (p pullProgress) UpdateProgress(ctx context.Context, ongoing *jobs, out progress.Output, start time.Time) error { | ||
| actives, err := p.Store.ListStatuses(ctx, "") | ||
| if err != nil { | ||
| if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { | ||
| return err | ||
| } | ||
| logrus.WithError(err).Error("status check failed") | ||
| return nil | ||
| } | ||
| pulling := make(map[string]content.Status, len(actives)) | ||
|
|
||
| // update status of status entries! | ||
| for _, status := range actives { | ||
| pulling[status.Ref] = status | ||
| } | ||
|
|
||
| for _, j := range ongoing.Jobs() { | ||
| key := remotes.MakeRefKey(ctx, j) | ||
| if info, ok := pulling[key]; ok { | ||
| out.WriteProgress(progress.Progress{ | ||
| ID: stringid.TruncateID(j.Digest.Encoded()), | ||
| Action: "Downloading", | ||
| Current: info.Offset, | ||
| Total: info.Total, | ||
| }) | ||
| continue | ||
| } | ||
|
|
||
| info, err := p.Store.Info(ctx, j.Digest) | ||
| if err != nil { | ||
| if !cerrdefs.IsNotFound(err) { | ||
| return err | ||
| } | ||
| } else if info.CreatedAt.After(start) { | ||
| out.WriteProgress(progress.Progress{ | ||
| ID: stringid.TruncateID(j.Digest.Encoded()), | ||
| Action: "Download complete", | ||
| HideCounts: true, | ||
| LastUpdate: true, | ||
| }) | ||
| ongoing.Remove(j) | ||
| } else if p.ShowExists { | ||
| out.WriteProgress(progress.Progress{ | ||
| ID: stringid.TruncateID(j.Digest.Encoded()), | ||
| Action: "Exists", | ||
| HideCounts: true, | ||
| LastUpdate: true, | ||
| }) | ||
| ongoing.Remove(j) | ||
| } | ||
| } | ||
| return nil | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,35 +1,84 @@ | ||
| package containerd | ||
|
|
||
| import ( | ||
| "net/http" | ||
| "strings" | ||
|
|
||
| "github.com/containerd/containerd/remotes" | ||
| "github.com/containerd/containerd/remotes/docker" | ||
| registrytypes "github.com/docker/docker/api/types/registry" | ||
| "github.com/docker/docker/registry" | ||
| "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| func newResolverFromAuthConfig(authConfig *registrytypes.AuthConfig) remotes.Resolver { | ||
| opts := []docker.RegistryOpt{} | ||
| if authConfig != nil { | ||
| cfgHost := registry.ConvertToHostname(authConfig.ServerAddress) | ||
| if cfgHost == registry.IndexHostname { | ||
| cfgHost = registry.DefaultRegistryHost | ||
| func (i *ImageService) newResolverFromAuthConfig(authConfig *registrytypes.AuthConfig) (remotes.Resolver, docker.StatusTracker) { | ||
| tracker := docker.NewInMemoryTracker() | ||
| hostsFn := i.registryHosts.RegistryHosts() | ||
|
|
||
| hosts := hostsWrapper(hostsFn, authConfig, i.registryService) | ||
|
|
||
| return docker.NewResolver(docker.ResolverOptions{ | ||
| Hosts: hosts, | ||
| Tracker: tracker, | ||
| }), tracker | ||
| } | ||
|
|
||
| func hostsWrapper(hostsFn docker.RegistryHosts, authConfig *registrytypes.AuthConfig, regService registry.Service) docker.RegistryHosts { | ||
| return func(n string) ([]docker.RegistryHost, error) { | ||
| hosts, err := hostsFn(n) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| authorizer := docker.NewDockerAuthorizer(docker.WithAuthCreds(func(host string) (string, string, error) { | ||
| if cfgHost != host { | ||
| logrus.WithField("host", host).WithField("cfgHost", cfgHost).Warn("Host doesn't match") | ||
| return "", "", nil | ||
| } | ||
| if authConfig.IdentityToken != "" { | ||
| return "", authConfig.IdentityToken, nil | ||
|
|
||
| for i := range hosts { | ||
| if hosts[i].Authorizer == nil { | ||
| var opts []docker.AuthorizerOpt | ||
| if authConfig != nil { | ||
| opts = append(opts, authorizationCredsFromAuthConfig(*authConfig)) | ||
| } | ||
| hosts[i].Authorizer = docker.NewDockerAuthorizer(opts...) | ||
|
|
||
| isInsecure := regService.IsInsecureRegistry(hosts[i].Host) | ||
| if hosts[i].Client.Transport != nil && isInsecure { | ||
| hosts[i].Client.Transport = httpFallback{super: hosts[i].Client.Transport} | ||
| } | ||
| } | ||
| return authConfig.Username, authConfig.Password, nil | ||
| })) | ||
| } | ||
| return hosts, nil | ||
| } | ||
| } | ||
|
|
||
| opts = append(opts, docker.WithAuthorizer(authorizer)) | ||
| func authorizationCredsFromAuthConfig(authConfig registrytypes.AuthConfig) docker.AuthorizerOpt { | ||
| cfgHost := registry.ConvertToHostname(authConfig.ServerAddress) | ||
| if cfgHost == registry.IndexHostname { | ||
| cfgHost = registry.DefaultRegistryHost | ||
| } | ||
|
|
||
| return docker.NewResolver(docker.ResolverOptions{ | ||
| Hosts: docker.ConfigureDefaultRegistries(opts...), | ||
| return docker.WithAuthCreds(func(host string) (string, string, error) { | ||
| if cfgHost != host { | ||
| logrus.WithField("host", host).WithField("cfgHost", cfgHost).Warn("Host doesn't match") | ||
| return "", "", nil | ||
| } | ||
| if authConfig.IdentityToken != "" { | ||
| return "", authConfig.IdentityToken, nil | ||
| } | ||
| return authConfig.Username, authConfig.Password, nil | ||
| }) | ||
| } | ||
|
|
||
| type httpFallback struct { | ||
| super http.RoundTripper | ||
| } | ||
|
|
||
| func (f httpFallback) RoundTrip(r *http.Request) (*http.Response, error) { | ||
| resp, err := f.super.RoundTrip(r) | ||
| if err != nil { | ||
| if strings.Contains(err.Error(), "http: server gave HTTP response to HTTPS client") { | ||
| plain := r.Clone(r.Context()) | ||
| plain.URL.Scheme = "http" | ||
| return http.DefaultTransport.RoundTrip(plain) | ||
| } | ||
| } | ||
|
|
||
| return resp, err | ||
| } |
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
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
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