画布 2
containerd阅读
  ctr pull index.docker.io/library/busybox:latest
    ref = clicontext.Args().First() // index.docker.io/library/busybox:latest
    ctx, cancel := appContext(clicontext)
    img, err := fetch(ctx, ref, clicontext)
      client, err := newClient(clicontext)
        return containerd.New(context.GlobalString("address"))
          client.go:
return NewWithConn(conn, opts...)
            client.go:
func NewWithConn(conn *grpc.ClientConn, opts ...ClientOpt) (*Client, error) {
return &Client{
conn: conn,
runtime: fmt.Sprintf("%s.%s", plugin.RuntimePlugin, runtime.GOOS),
}, nil
}
      resolver, err := getResolver(ctx, clicontext)
        options := docker.ResolverOptions{
PlainHTTP: clicontext.Bool("plain-http"),
Tracker: pushTracker,
}
          pushTracker = docker.NewInMemoryTracker()
            remotes/docker/status.go:
type Status struct {
content.Status
// UploadUUID is used by the Docker registry to reference blob uploads
UploadUUID string
}
func NewInMemoryTracker() StatusTracker {
return &memoryStatusTracker{
statuses: map[string]Status{},
}
}
        tr := &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
MaxIdleConns: 10,
IdleConnTimeout: 30 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
TLSClientConfig: &tls.Config{
InsecureSkipVerify: clicontext.Bool("insecure"),
},
ExpectContinueTimeout: 5 * time.Second,
}
options.Client = &http.Client{
Transport: tr,
}
return docker.NewResolver(options), nil
          remotes/docker/resolver.go:
func NewResolver(options ResolverOptions) remotes.Resolver {
tracker := options.Tracker
if tracker == nil {
tracker = NewInMemoryTracker()
}
return &dockerResolver{
credentials: options.Credentials,
plainHTTP: options.PlainHTTP,
client: options.Client,
tracker: tracker,
}
}
      ongoing := newJobs(ref)
        type jobs struct {
name string
added map[digest.Digest]struct{}
descs []ocispec.Descriptor
mu sync.Mutex
resolved bool
}
func newJobs(name string) *jobs {
return &jobs{
name: name, // index.docker.io/library/busybox:latest
added: map[digest.Digest]struct{}{},
}
}
      pctx, stopProgress := context.WithCancel(ctx)
      go func() {
if !clicontext.GlobalBool("debug") {
// no progress bar, because it hides some debug logs
showProgress(pctx, ongoing, client.ContentStore(), os.Stdout)
}
close(progress)
}()
      h := images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
if desc.MediaType != images.MediaTypeDockerSchema1Manifest {
ongoing.add(desc)
}
return nil, nil
})
        cmd/ctr/fetch.go:
func (j *jobs) add(desc ocispec.Descriptor) {
j.mu.Lock()
defer j.mu.Unlock()
j.resolved = true
if _, ok := j.added[desc.Digest]; ok {
return
}
j.descs = append(j.descs, desc)
j.added[desc.Digest] = struct{}{}
}
      img, err := client.Pull(pctx, ref, containerd.WithResolver(resolver), containerd.WithImageHandler(h), containerd.WithSchema1Conversion)
        client.go:
pullCtx := defaultRemoteContext()
202 for _, o := range opts {
203 if err := o(c, pullCtx); err != nil {
204 return nil, err
205 }
206 }
          func defaultRemoteContext() *RemoteContext {
return &RemoteContext{
Resolver: docker.NewResolver(docker.ResolverOptions{
Client: http.DefaultClient,
}),
Snapshotter: DefaultSnapshotter, // "overlayfs"
}
}
          pullCtx.ImageHandler = []images.Handler {h}
          pullCtx.Resolver = dockerResolver
          pullCtx.ConvertSchema1 = true
        store := c.ContentStore()
          client.go:
return contentservice.NewStoreFromClient(contentapi.NewContentClient(c.conn))
            api/services/content/v1/content.pb.go:
466 func NewContentClient(cc *grpc.ClientConn) ContentClient {
467 return &contentClient{cc}
468 }
// 这是一个content service client
// 对应的server端实现在services/content/service.go
            services/content/store.go:
14 type remoteStore struct {
15 client contentapi.ContentClient
16 }
17
18 func NewStoreFromClient(client contentapi.ContentClient) content.Store {
19 return &remoteStore{
20 client: client,
21 }
22 }
// 返回一个content.Store interface, 定义在content/content.go
        name, desc, err := pullCtx.Resolver.Resolve(ctx, ref)
          refspec, err := reference.Parse(ref)
            reference/reference.go:
// 以docker.io/library/busybox:latest为例
return Spec{
Locator: path.Join(u.Host, u.Path), // docker.io/library/busybox
Object: object, // latest
}, nil
          base, err := r.base(refspec)
            return &dockerBase{
base: base, // url.URL (docker.io/v2/library/busybox)
client: r.client, // http client, 定制了Transport
username: username, // 无
secret: secret, // 密码无
}, nil
          fetcher := dockerFetcher{
dockerBase: base,
}
          dgst = refspec.Digest() // 如果是busybox:latest@abcdef, 那么dgst就是abcdef
          if dgst != "" {
urls = append(urls, fetcher.url("manifests", dgst.String())) // 加一个url： docker.io/v2/library/busybox/manifests/abcdef
// fallback to blobs on not found.
urls = append(urls, fetcher.url("blobs", dgst.String()))
} else {
urls = append(urls, fetcher.url("manifests", refspec.Object)) // 加一个url: docker.io/v2/library/busybox/manifests/latest
// note: docker.io 实质上会被替换成registry-1.docker.io
}
          开始发送http请求，获取latest这个tag对应的manifest
119 for _, u := range urls {
120 req, err := http.NewRequest(http.MethodHead, u, nil)
125 // set headers for all the types we support for resolution.
126 req.Header.Set("Accept", strings.Join([]string{
127 images.MediaTypeDockerSchema2Manifest,
128 images.MediaTypeDockerSchema2ManifestList,
129 ocispec.MediaTypeImageManifest,
130 ocispec.MediaTypeImageIndex, "*"}, ", "))
131
132 log.G(ctx).Debug("resolving")
133 resp, err := fetcher.doRequestWithRetries(ctx, req, nil)
176 desc := ocispec.Descriptor{
177 Digest: dgst, // sha256:b82b5740006c1ab823596d2c07f081084ecdb32fd258072707b99f52a3cb8692
178 MediaType: resp.Header.Get("Content-Type"), // application/vnd.docker.distribution.manifest.v2+json
179 Size: size, // 527
180 }
181
182 log.G(ctx).WithField("desc.digest", desc.Digest).Debug("resolved")
183 return ref, desc, nil
// return "docker.io/library/busybox:latest", desc, nil
184 }
}
            日志：
12 DEBU[0008] fetch digest=sha256:b82b5740006c1ab823596d2c07f081084ecdb32fd258072707b99f52a3cb8692 med iatype="application/vnd.docker.distribution.manifest.v2+json" size=527
        fetcher, err := pullCtx.Resolver.Fetcher(ctx, name)
          refspec, err := reference.Parse(ref)
          base, err := r.base(refspec)
          200 return dockerFetcher{
201 dockerBase: base,
202 }, nil
        218 var (
219 schema1Converter *schema1.Converter
220 handler images.Handler
221 )
222 if desc.MediaType == images.MediaTypeDockerSchema1Manifest && pullCtx.ConvertSchema1 {
223 schema1Converter = schema1.NewConverter(store, fetcher)
224 handler = images.Handlers(append(pullCtx.BaseHandlers, schema1Converter)...)
225 } else {
// 在这里我们的例子里，是Schema2,所以不需要转换，走到这一步。handler里面有三个handler了
226 handler = images.Handlers(append(pullCtx.BaseHandlers,
227 remotes.FetchHandler(store, fetcher),
228 images.ChildrenHandler(store))...,
229 )
230 }
          remotes/handler.go:
44 func FetchHandler(ingester content.Ingester, fetcher Fetcher) images.HandlerFunc {
45 return func(ctx context.Context, desc ocispec.Descriptor) (subdescs []ocispec.Descriptor, err error) {
46 ctx = log.WithLogger(ctx, log.G(ctx).WithFields(logrus.Fields{
47 "digest": desc.Digest,
48 "mediatype": desc.MediaType,
49 "size": desc.Size,
50 }))
51
52 switch desc.MediaType {
53 case images.MediaTypeDockerSchema1Manifest:
54 return nil, fmt.Errorf("%v not supported", desc.MediaType)
55 default:
56 err := fetch(ctx, ingester, fetcher, desc)
57 return nil, err
58 }
59 }
60 }
            func fetch(ctx context.Context, ingester content.Ingester, fetcher Fetcher, desc ocispec.Descriptor) error {}
              remotes/handler.go:
ref = MakeRefKey(ctx, desc) // 制作了一个key，比如： manifest-<digest>
              cw, err = ingester.Writer(ctx, ref, desc.Size, desc.Digest)
// ref=manifest-ab234acd44
// desc.Size = 527
// desc.Digest = sha256:b82b5740006c1ab823596d2c07f081084ecdb32fd258072707b99f52a3cb8692
                services/content/store.go:
142 func (rs *remoteStore) Writer(ctx context.Context, ref string, size int64, expected digest.Digest) (content.Writer, error) {
143 wrclient, offset, err := rs.negotiate(ctx, ref, size, expected)
148 return &remoteWriter{
149 ref: ref, // manifest-<digest>
150 client: wrclient,
151 offset: offset,
152 }, nil
153 }
                  wrclient, err := rs.client.Write(ctx)
                    发送grpc给请求给server端。
server端的代码移步services/content/service.go
                  172 if err := wrclient.Send(&contentapi.WriteContentRequest{
173 Action: contentapi.WriteActionStat,
174 Ref: ref, // manifest-<digest>
175 Total: size, // 527
176 Expected: expected, // digest
177 });
181 resp, err := wrclient.Recv()
186 return wrclient, resp.Offset, nil
              rc, err := fetcher.Fetch(ctx, desc)
                remotes/docker/fetcher.go:
func (r dockerFetcher) Fetch(ctx context.Context, desc ocispec.Descriptor) (io.ReadCloser, error) {}
// 实际上返回的是 responseBody
// 例如，把manifest获取成功
                  paths, err := getV2URLPaths(desc)
                    switch desc.MediaType {
case images.MediaTypeDockerSchema2Manifest, images.MediaTypeDockerSchema2ManifestList,
images.MediaTypeDockerSchema1Manifest,
ocispec.MediaTypeImageManifest, ocispec.MediaTypeImageIndex:
urls = append(urls, path.Join("manifests", desc.Digest.String()))
}
// always fallback to attempting to get the object out of the blobs store.
urls = append(urls, path.Join("blobs", desc.Digest.String()))
return urls, nil
                  for _, path := range paths {
u := r.url(path)
req, err := http.NewRequest(http.MethodGet, u, nil)
resp, err := r.doRequestWithRetries(ctx, req, nil)
if resp.StatusCode == http.StatusNotFound {
continue // try one of the other urls.
}
}
// 把上一步的paths从第一个开始试。如果找不到就下一个。找到了就直接返回
              return content.Copy(cw, rc, desc.Size, desc.Digest)
                content/helpers.go:
func Copy(cw Writer, r io.Reader, size int64, expected digest.Digest) error {}
// 未完待续
          130 func ChildrenHandler(provider content.Provider) HandlerFunc {
131 return func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
132 var descs []ocispec.Descriptor
133 switch desc.MediaType {
134 case MediaTypeDockerSchema2Manifest, ocispec.MediaTypeImageManifest:
135 p, err := content.ReadBlob(ctx, provider, desc.Digest)
136 if err != nil {
137 return nil, err
138 }
142 var manifest ocispec.Manifest
143 if err := json.Unmarshal(p, &manifest); err != nil {
144 return nil, err
145 }
147 descs = append(descs, manifest.Config)
148 descs = append(descs, manifest.Layers...)
149 case MediaTypeDockerSchema2ManifestList, ocispec.MediaTypeImageIndex:
150 p, err := content.ReadBlob(ctx, provider, desc.Digest)
155 var index ocispec.Index
156 if err := json.Unmarshal(p, &index); err != nil {
157 return nil, err
158 }
160 descs = append(descs, index.Manifests...)
161 case MediaTypeDockerSchema2Layer, MediaTypeDockerSchema2LayerGzip,
162 MediaTypeDockerSchema2Config, ocispec.MediaTypeImageConfig,
163 ocispec.MediaTypeImageLayer, ocispec.MediaTypeImageLayerGzip:
164 // childless data types.
165 return nil, nil
166 default:
167 log.G(ctx).Warnf("encountered unknown type %v; children may not be fetched", desc.MediaType)
168 }
170 return descs, nil
171 }
172 }
        images.Dispatch(ctx, handler, desc);
// 定义在images/image.go
// 就是调用上面的handler循环解析ocispec，如果有children就继续深度优先遍历children
      stopProgress()
if err != nil {
return nil, err
}
<-progress
return img, nil
    err = img.Unpack(ctx, clicontext.String("snapshotter"))