func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig, opts dockertypes.ImagePullOptions) error {
	// RegistryAuth is the base64 encoded credentials for the registry
	base64Auth, err := base64EncodeAuth(auth)
	if err != nil {
		return err
	}
	opts.ImageID = image
	opts.RegistryAuth = base64Auth
	resp, err := d.client.ImagePull(getDefaultContext(), opts, nil)
	if err != nil {
		return err
	}
	defer resp.Close()
	// TODO(random-liu): Use the image pulling progress information.
	decoder := json.NewDecoder(resp)
	for {
		var msg dockermessage.JSONMessage
		err := decoder.Decode(&msg)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		if msg.Error != nil {
			return msg.Error
		}
	}
	return nil
}

// ImagePull requests the docker host to pull an image from a remote registry.
// It executes the privileged function if the operation is unauthorized
// and it tries one more time.
// It's up to the caller to handle the io.ReadCloser and close it properly.
//
// FIXME(vdemeester): there is currently used in a few way in docker/docker
// - if not in trusted content, ref is used to pass the whole reference, and tag is empty
// - if in trusted content, ref is used to pass the reference name, and tag for the digest
func (cli *Client) ImagePull(ctx context.Context, ref string, options types.ImagePullOptions) (io.ReadCloser, error) {
	repository, tag, err := reference.Parse(ref)
	if err != nil {
		return nil, err
	}

	query := url.Values{}
	query.Set("fromImage", repository)
	if tag != "" {
		query.Set("tag", tag)
	}

	resp, err := cli.tryImageCreate(ctx, query, options.RegistryAuth)
	if resp.statusCode == http.StatusUnauthorized {
		newAuthHeader, privilegeErr := options.PrivilegeFunc()
		if privilegeErr != nil {
			return nil, privilegeErr
		}
		resp, err = cli.tryImageCreate(ctx, query, newAuthHeader)
	}
	if err != nil {
		return nil, err
	}
	return resp.body, nil
}

func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig, opts dockertypes.ImagePullOptions) error {
	// RegistryAuth is the base64 encoded credentials for the registry
	base64Auth, err := base64EncodeAuth(auth)
	if err != nil {
		return err
	}
	opts.RegistryAuth = base64Auth
	ctx, cancel := d.getCancelableContext()
	defer cancel()
	resp, err := d.client.ImagePull(ctx, image, opts)
	if err != nil {
		return err
	}
	defer resp.Close()
	reporter := newProgressReporter(image, cancel)
	reporter.start()
	defer reporter.stop()
	decoder := json.NewDecoder(resp)
	for {
		var msg dockermessage.JSONMessage
		err := decoder.Decode(&msg)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		if msg.Error != nil {
			return msg.Error
		}
		reporter.set(&msg)
	}
	return nil
}

func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig, opts dockertypes.ImagePullOptions) error {
	// RegistryAuth is the base64 encoded credentials for the registry
	base64Auth, err := base64EncodeAuth(auth)
	if err != nil {
		return err
	}
	opts.RegistryAuth = base64Auth
	// Don't set timeout for the context because image pulling can be
	// take an arbitrarily long time.
	resp, err := d.client.ImagePull(context.Background(), image, opts)
	if err != nil {
		return err
	}
	defer resp.Close()
	reporter := newProgressReporter(image, defaultImagePullingProgressReportInterval)
	reporter.start()
	defer reporter.stop()
	decoder := json.NewDecoder(resp)
	for {
		var msg dockermessage.JSONMessage
		err := decoder.Decode(&msg)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		if msg.Error != nil {
			return msg.Error
		}
		reporter.set(&msg)
	}
	return nil
}

func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig, opts dockertypes.ImagePullOptions) error {
	// RegistryAuth is the base64 encoded credentials for the registry
	base64Auth, err := base64EncodeAuth(auth)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	opts.RegistryAuth = base64Auth
	resp, err := d.client.ImagePull(ctx, image, opts)
	if ctxErr := contextError(ctx); ctxErr != nil {
		return ctxErr
	}
	if err != nil {
		return err
	}
	defer resp.Close()
	// TODO(random-liu): Use the image pulling progress information.
	decoder := json.NewDecoder(resp)
	for {
		var msg dockermessage.JSONMessage
		err := decoder.Decode(&msg)
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		if msg.Error != nil {
			return msg.Error
		}
	}
	return nil
}

// Spool spools an image with a tag latest
func (b *Box) Spool(ctx context.Context, name string, opts isolate.Profile) (err error) {
	profile, err := convertProfile(opts)
	if err != nil {
		apexctx.GetLogger(ctx).WithError(err).WithFields(log.Fields{"name": name}).Info("unbale to convert raw profile to Docker specific profile")
		return err
	}

	if profile.Registry == "" {
		apexctx.GetLogger(ctx).WithFields(log.Fields{"name": name}).Info("local image will be used")
		return nil
	}

	defer apexctx.GetLogger(ctx).WithField("name", name).Trace("spooling an image").Stop(&err)

	pullOpts := types.ImagePullOptions{
		All: false,
	}

	if registryAuth, ok := b.config.RegistryAuth[profile.Registry]; ok {
		pullOpts.RegistryAuth = registryAuth
	}

	ref := fmt.Sprintf("%s:%s", filepath.Join(profile.Registry, profile.Repository, name), "latest")

	body, err := b.client.ImagePull(ctx, ref, pullOpts)
	if err != nil {
		apexctx.GetLogger(ctx).WithError(err).WithFields(
			log.Fields{"name": name, "ref": ref}).Error("unable to pull an image")
		return err
	}
	defer body.Close()

	if err := decodeImagePull(ctx, body); err != nil {
		return err
	}

	return nil
}

func pullImage(ctx context.Context, client client.APIClient, service *Service, image string) error {
	fmt.Fprintf(os.Stderr, "Pulling %s (%s)...\n", service.name, image)
	distributionRef, err := reference.ParseNamed(image)
	if err != nil {
		return err
	}

	repoInfo, err := registry.ParseRepositoryInfo(distributionRef)
	if err != nil {
		return err
	}

	authConfig := service.context.AuthLookup.Lookup(repoInfo)

	encodedAuth, err := encodeAuthToBase64(authConfig)
	if err != nil {
		return err
	}

	options := types.ImagePullOptions{
		ImageID:      distributionRef.String(),
		Tag:          "latest",
		RegistryAuth: encodedAuth,
	}
	if named, ok := distributionRef.(reference.Named); ok {
		options.ImageID = named.FullName()
	}
	if tagged, ok := distributionRef.(reference.NamedTagged); ok {
		options.Tag = tagged.Tag()
	}

	timeoutsRemaining := 3
	for i := 0; i < 100; i++ {
		responseBody, err := client.ImagePull(ctx, options, nil)
		if err != nil {
			logrus.Errorf("Failed to pull image %s: %v", image, err)
			return err
		}

		var writeBuff io.Writer = os.Stderr

		outFd, isTerminalOut := term.GetFdInfo(os.Stderr)

		err = jsonmessage.DisplayJSONMessagesStream(responseBody, writeBuff, outFd, isTerminalOut, nil)
		responseBody.Close()
		if err == nil {
			return nil
		} else if strings.Contains(err.Error(), "timed out") {
			timeoutsRemaining -= 1
			if timeoutsRemaining == 0 {
				return err
			}
			continue
		} else if strings.Contains(err.Error(), "connection") || strings.Contains(err.Error(), "unreachable") {
			time.Sleep(300 * time.Millisecond)
			continue
		} else {
			if jerr, ok := err.(*jsonmessage.JSONError); ok {
				// If no error code is set, default to 1
				if jerr.Code == 0 {
					jerr.Code = 1
				}
				fmt.Fprintf(os.Stderr, "%s", writeBuff)
				return fmt.Errorf("Status: %s, Code: %d", jerr.Message, jerr.Code)
			}
		}
	}

	return err
}