Repository: containerd/containerd
File: internal/cri/server/images/image_pull.go (Go)

Code before:
		return nil, fmt.Errorf("failed to create OCI registry: %w", err)
	}

	doTransfer := func() (containerd.Image, error) {
		rctx, rcancel := context.WithCancel(ctx)
		defer rcancel()
		transferProgressReporter := newTransferProgressReporter(ref, rcancel, imagePullProgressTimeout)
		transferProgressReporter.start(rctx)
		log.G(ctx).Debugf("Calling cri transfer service")
		if err := c.transferrer.Transfer(rctx, reg, is, transfer.WithProgress(transferProgressReporter.createProgressFunc(rctx))); err != nil {
			return nil, err
		}
		return c.client.GetImage(ctx, ref)
	}

	image, err := c.pullWithStaleSnapshotRetry(ctx, ref, snapshotter, doTransfer)
	if err != nil {
		return nil, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
	}
	return image, nil
}

// getImageIDIfUnpacked checks if an image is already properly unpacked for the given
// snapshotter and returns its ID if so.
// Returns the image ID on success, or an error if a pull is needed.
func (c *CRIImageService) getImageIDIfUnpacked(ctx context.Context, ref, snapshotter string) (string, error) {
	unpacked, err := c.IsImageUnpackedForSnapshotter(ctx, ref, snapshotter)
	if err != nil {
		return "", err
	}

	if !unpacked {
		return "", fmt.Errorf("image exists but not unpacked for snapshotter %q: %w", snapshotter, errdefs.ErrFailedPrecondition)
	}

	// Image is ready - get its ID
	existingImage, err := c.client.GetImage(ctx, ref)
	if err != nil {
		return "", err
	}

	configDesc, err := existingImage.Config(ctx)
	if err != nil {
		return "", err
	}

	imageID := configDesc.Digest.String()

	// Ensure image has proper labels and update image store
	if err := c.ensureImageLabels(ctx, existingImage, ref); err != nil {
		log.G(ctx).WithError(err).Warnf("Failed to ensure image labels for %s", ref)

Review comment (suggestion):
```suggestion
func (c *CRIImageService) imageForSnapshotter(ctx context.Context, ref, snapshotter string) (string, error) {
```

Code after:
		u, err := url.Parse(auth.ServerAddress)
		if err != nil {
			return "", "", fmt.Errorf("parse server address: %w", err)
		}
		if host != u.Host {
			return "", "", nil
		}
	}
	if auth.Username != "" {
		return auth.Username, auth.Password, nil
	}
	if auth.IdentityToken != "" {
		return "", auth.IdentityToken, nil
	}
	if auth.Auth != "" {
		decLen := base64.StdEncoding.DecodedLen(len(auth.Auth))
		decoded := make([]byte, decLen)
		_, err := base64.StdEncoding.Decode(decoded, []byte(auth.Auth))
		if err != nil {
			return "", "", err
		}
		user, passwd, ok := strings.Cut(string(decoded), ":")
		if !ok {
			return "", "", fmt.Errorf("invalid decoded auth: %q", decoded)
		}
		return user, strings.Trim(passwd, "\x00"), nil
	}
	// TODO(random-liu): Support RegistryToken.
	// An empty auth config is valid for anonymous registry
	return "", "", nil
}

// createOrUpdateImageReference creates or updates image reference inside containerd image store.
// Note that because create and update are not finished in one transaction, there could be race. E.g.
// the image reference is deleted by someone else after create returns already exists, but before update
// happens.
func (c *CRIImageService) createOrUpdateImageReference(ctx context.Context, name string, desc imagespec.Descriptor, labels map[string]string) error {
	img := containerdimages.Image{
		Name:   name,
		Target: desc,
		// Add a label to indicate that the image is managed by the cri plugin.
		Labels: labels,
	}
	// TODO(random-liu): Figure out which is the more performant sequence create then update or
	// update then create.
	// TODO: Call CRIImageService directly
	_, err := c.images.Create(ctx, img)
	if err == nil {
		return nil
	} else if !errdefs.IsAlreadyExists(err) {
		return err