// LoadImage checks the client for an image matching from. If not found,
// attempts to pull the image and then tries to inspect again.
func (e *ClientExecutor) LoadImage(from string) (*docker.Image, error) {
	image, err := e.Client.InspectImage(from)
	if err == nil {
		return image, nil
	}
	if err != docker.ErrNoSuchImage {
		return nil, err
	}

	if !e.AllowPull {
		glog.V(4).Infof("image %s did not exist", from)
		return nil, docker.ErrNoSuchImage
	}

	repository, tag := docker.ParseRepositoryTag(from)
	if len(tag) == 0 {
		tag = "latest"
	}

	glog.V(4).Infof("attempting to pull %s with auth from repository %s:%s", from, repository, tag)

	// TODO: we may want to abstract looping over multiple credentials
	auth, _ := e.AuthFn(repository)
	if len(auth) == 0 {
		auth = append(auth, credentialprovider.LazyAuthConfiguration{})
	}

	if e.LogFn != nil {
		e.LogFn("Image %s was not found, pulling ...", from)
	}

	var lastErr error
	outputProgress := func(s string) {
		e.LogFn("%s", s)
	}
	for _, config := range auth {
		// TODO: handle IDs?
		pullImageOptions := docker.PullImageOptions{
			Repository:    repository,
			Tag:           tag,
			OutputStream:  imageprogress.NewPullWriter(outputProgress),
			RawJSONStream: true,
		}
		if glog.V(5) {
			pullImageOptions.OutputStream = os.Stderr
			pullImageOptions.RawJSONStream = false
		}
		authConfig := docker.AuthConfiguration{Username: config.Username, ServerAddress: config.ServerAddress, Password: config.Password}
		if err = e.Client.PullImage(pullImageOptions, authConfig); err == nil {
			break
		}
		lastErr = err
		continue
	}
	if lastErr != nil {
		return nil, fmt.Errorf("unable to pull image (from: %s, tag: %s): %v", repository, tag, lastErr)
	}

	return e.Client.InspectImage(from)
}

// CommitContainer commits a container to an image with a specific tag.
// The new image ID is returned
func (d *stiDocker) CommitContainer(opts CommitContainerOptions) (string, error) {
	repository, tag := docker.ParseRepositoryTag(opts.Repository)
	dockerOpts := docker.CommitContainerOptions{
		Container:  opts.ContainerID,
		Repository: repository,
		Tag:        tag,
	}
	if opts.Command != nil || opts.Entrypoint != nil {
		config := docker.Config{
			Cmd:        opts.Command,
			Entrypoint: opts.Entrypoint,
			Env:        opts.Env,
			Labels:     opts.Labels,
			User:       opts.User,
		}
		dockerOpts.Run = &config
		glog.V(2).Infof("Committing container with dockerOpts: %+v, config: %+v", dockerOpts, config)
	}

	image, err := d.client.CommitContainer(dockerOpts)
	if err == nil && image != nil {
		return image.ID, nil
	}
	return "", err
}

// pushImage pushes a docker image to the registry specified in its tag.
// The method will retry to push the image when following scenarios occur:
// - Docker registry is down temporarily or permanently
// - other image is being pushed to the registry
// If any other scenario the push will fail, without retries.
func pushImage(client DockerClient, name string, authConfig docker.AuthConfiguration) error {
	repository, tag := docker.ParseRepositoryTag(name)
	opts := docker.PushImageOptions{
		Name: repository,
		Tag:  tag,
	}
	if glog.V(5) {
		opts.OutputStream = os.Stderr
	}
	var err error

	for retries := 0; retries <= DefaultPushRetryCount; retries++ {
		err = client.PushImage(opts, authConfig)
		if err == nil {
			return nil
		}

		errMsg := fmt.Sprintf("%s", err)
		if !strings.Contains(errMsg, "ping attempt failed with error") && !strings.Contains(errMsg, "is already in progress") {
			return err
		}

		util.HandleError(fmt.Errorf("push for image %s failed, will retry in %s seconds ...", name, DefaultPushRetryDelay))
		glog.Flush()
		time.Sleep(DefaultPushRetryDelay)
	}
	return err
}

// pushImage pushes a docker image to the registry specified in its tag.
// The method will retry to push the image when following scenarios occur:
// - Docker registry is down temporarily or permanently
// - other image is being pushed to the registry
// If any other scenario the push will fail, without retries.
func pushImage(client DockerClient, name string, authConfig docker.AuthConfiguration) error {
	repository, tag := docker.ParseRepositoryTag(name)
	opts := docker.PushImageOptions{
		Name: repository,
		Tag:  tag,
	}
	if glog.V(5) {
		opts.OutputStream = os.Stderr
	}
	var err error
	var retriableError = false

	for retries := 0; retries <= DefaultPushRetryCount; retries++ {
		err = client.PushImage(opts, authConfig)
		if err == nil {
			return nil
		}

		errMsg := fmt.Sprintf("%s", err)
		for _, errorString := range RetriableErrors {
			if strings.Contains(errMsg, errorString) {
				retriableError = true
				break
			}
		}
		if !retriableError {
			return err
		}

		utilruntime.HandleError(fmt.Errorf("push for image %s failed, will retry in %s ...", name, DefaultPushRetryDelay))
		glog.Flush()
		time.Sleep(DefaultPushRetryDelay)
	}
	return err
}

// createImage creates a docker image either by pulling it from a registry or by
// loading it from the file system
func (d *DockerDriver) createImage(driverConfig *DockerDriverConfig, client *docker.Client, taskDir string) error {
	image := driverConfig.ImageName
	repo, tag := docker.ParseRepositoryTag(image)
	if tag == "" {
		tag = "latest"
	}

	var dockerImage *docker.Image
	var err error
	// We're going to check whether the image is already downloaded. If the tag
	// is "latest" we have to check for a new version every time so we don't
	// bother to check and cache the id here. We'll download first, then cache.
	if tag != "latest" {
		dockerImage, err = client.InspectImage(image)
	}

	// Download the image
	if dockerImage == nil {
		if len(driverConfig.LoadImages) > 0 {
			return d.loadImage(driverConfig, client, taskDir)
		}

		return d.pullImage(driverConfig, client, repo, tag)
	}
	return err
}

// pullImage pulls the latest image for the given repotag from a public
// registry.
func (c *client) pullImage(repoTag string) error {
	// configuration options that get passed to client.PullImage
	repository, tag := docker.ParseRepositoryTag(repoTag)
	pullImageOptions := docker.PullImageOptions{Repository: repository, Tag: tag}
	// pull image from registry
	return c.client.PullImage(pullImageOptions, docker.AuthConfiguration{})
}

// getImageName checks the image name and adds DefaultTag if none is specified
func getImageName(name string) string {
	_, tag := docker.ParseRepositoryTag(name)
	if len(tag) == 0 {
		return strings.Join([]string{name, DefaultTag}, ":")
	}

	return name
}

// ValidateImage validates the image field does not include a tag
func (c *ImageConfig) ValidateImage() error {
	_, tag := docker.ParseRepositoryTag(c.Image)
	if tag != "" {
		return fmt.Errorf(
			"tag %q must be specified in the `tags` field, not in `image`", tag)
	}
	return nil
}

// ValidateTags to ensure the first tag is a basic tag without an image name.
func (c *ImageConfig) ValidateTags() error {
	if len(c.Tags) == 0 {
		return nil
	}
	_, tag := docker.ParseRepositoryTag(c.Tags[0])
	if tag != "" {
		return fmt.Errorf("the first tag %q may not include an image name", tag)
	}
	return nil

}

// tagImage uses the dockerClient to tag a Docker image with name. It is a
// helper to facilitate the usage of dockerClient.TagImage, because the former
// requires the name to be split into more explicit parts.
func tagImage(dockerClient DockerClient, image, name string) error {
	repo, tag := docker.ParseRepositoryTag(name)
	return dockerClient.TagImage(image, docker.TagImageOptions{
		Repo: repo,
		Tag:  tag,
		// We need to set Force to true to update the tag even if it
		// already exists. This is the same behavior as `docker build -t
		// tag .`.
		Force: true,
	})
}

func tagImage(ctx *context.ExecuteContext, config *config.ImageConfig, imageTag string) error {
	canonicalImageTag := GetImageName(ctx, config)
	if imageTag == canonicalImageTag {
		return nil
	}

	repo, tag := docker.ParseRepositoryTag(imageTag)
	err := ctx.Client.TagImage(canonicalImageTag, docker.TagImageOptions{
		Repo:  repo,
		Tag:   tag,
		Force: true,
	})
	if err != nil {
		return fmt.Errorf("failed to add tag %q: %s", imageTag, err)
	}
	return nil
}

func pullImage(ctx *context.ExecuteContext, t *Task, imageTag string) error {
	registry, err := parseAuthRepo(t.config.Image)
	if err != nil {
		return err
	}

	repo, tag := docker.ParseRepositoryTag(imageTag)
	return Stream(os.Stdout, func(out io.Writer) error {
		return ctx.Client.PullImage(docker.PullImageOptions{
			Repository:    repo,
			Tag:           tag,
			OutputStream:  out,
			RawJSONStream: true,
			// TODO: timeout
		}, ctx.GetAuthConfig(registry))
	})
}

func importImage(client *dockerClient.Client, name, fileName string) error {
	file, err := os.Open(fileName)
	if err != nil {
		return err
	}

	defer file.Close()

	log.Debugf("Importing image for %s", fileName)
	repo, tag := dockerClient.ParseRepositoryTag(name)
	return client.ImportImage(dockerClient.ImportImageOptions{
		Source:      "-",
		Repository:  repo,
		Tag:         tag,
		InputStream: file,
	})
}

// ForEachTag runs a function for each tag
func (t *Task) ForEachTag(ctx *context.ExecuteContext, each func(string) error) error {
	if len(t.config.Tags) == 0 {
		return each(GetImageName(ctx, t.config))
	}

	for _, tag := range t.config.Tags {
		imageTag := t.config.Image + ":" + tag
		// If the tag is already a complete image name then use it directly
		if _, hasTag := docker.ParseRepositoryTag(tag); hasTag != "" {
			imageTag = tag
		}

		if err := each(imageTag); err != nil {
			return err
		}
	}
	return nil
}

// LoadImage checks the client for an image matching from. If not found,
// attempts to pull the image and then tries to inspect again.
func (e *ClientExecutor) LoadImage(from string) (*docker.Image, error) {
	image, err := e.Client.InspectImage(from)
	if err == nil {
		return image, nil
	}
	if err != docker.ErrNoSuchImage {
		return nil, err
	}

	if !e.AllowPull {
		glog.V(4).Infof("image %s did not exist", from)
		return nil, docker.ErrNoSuchImage
	}

	repository, _ := docker.ParseRepositoryTag(from)

	glog.V(4).Infof("attempting to pull %s with auth from repository %s", from, repository)

	// TODO: we may want to abstract looping over multiple credentials
	auth, _ := e.AuthFn(repository)
	if len(auth) == 0 {
		auth = append(auth, docker.AuthConfiguration{})
	}

	if e.LogFn != nil {
		e.LogFn("Image %s was not found, pulling ...", from)
	}

	var lastErr error
	for _, config := range auth {
		// TODO: handle IDs?
		// TODO: use RawJSONStream:true and handle the output nicely
		if err = e.Client.PullImage(docker.PullImageOptions{Repository: from, OutputStream: e.Out}, config); err == nil {
			break
		}
		lastErr = err
		continue
	}
	if lastErr != nil {
		return nil, lastErr
	}

	return e.Client.InspectImage(from)
}

// GetAuthconfig retrieves the correct auth configuration for the given repository
func (authProvider *dockerAuthProvider) GetAuthconfig(image string) (docker.AuthConfiguration, error) {
	// Ignore 'tag', not used in auth determination
	repository, _ := docker.ParseRepositoryTag(image)
	authDataMap := authProvider.authMap

	// Ignore repo/image name for some auth checks (see use of 'image' below for where it's not ignored.
	indexName, _ := splitReposName(repository)

	if isDockerhubHostname(indexName) {
		return authDataMap[dockerRegistryKey], nil
	}

	// Try to find the longest match that at least matches the hostname
	// This is to support e.g. 'registry.tld: auth1, registry.tld/username:
	// auth2' for multiple different auths on the same registry.
	longestKey := ""
	authConfigKey := indexName

	// Take a direct match of the index hostname as a sane default
	if _, found := authDataMap[authConfigKey]; found && len(authConfigKey) > len(longestKey) {
		longestKey = authConfigKey
	}

	for registry, _ := range authDataMap {
		nameParts := strings.SplitN(registry, "/", 2)
		hostname := nameParts[0]

		// Only ever take a new key if the hostname matches in normal cases
		if authConfigKey == hostname {
			if longestKey == "" {
				longestKey = registry
			} else if len(registry) > len(longestKey) && strings.HasPrefix(image, registry) {
				// If we have a longer match, that indicates a username / namespace appended
				longestKey = registry
			}
		}
	}
	if longestKey != "" {
		return authDataMap[longestKey], nil
	}
	return docker.AuthConfiguration{}, nil
}

// pushImage pushes a docker image to the registry specified in its tag
func pushImage(client DockerClient, name string, authConfig docker.AuthConfiguration) error {
	repository, tag := docker.ParseRepositoryTag(name)
	opts := docker.PushImageOptions{
		Name: repository,
		Tag:  tag,
	}
	if glog.V(5) {
		opts.OutputStream = os.Stderr
	}
	var err error
	for retries := 0; retries <= DefaultPushRetryCount; retries++ {
		err = client.PushImage(opts, authConfig)
		if err == nil {
			return nil
		}
		if retries == DefaultPushRetryCount {
			return err
		}
		util.HandleError(fmt.Errorf("push for image %s failed, will retry in %s ...", name, DefaultPushRetryDelay))
		glog.Flush()
		time.Sleep(DefaultPushRetryDelay)
	}
	return err
}

// Looks up *docker.APIImage which matches *DockerImg passed as paramter and
// calls an anonymous func on it
func withDockerAPIImage(i *DockerImg, fn func(*docker.APIImages) error) error {
	return withDockerClient(func(client *docker.Client) error {
		imgs, err := client.ListImages(docker.ListImagesOptions{All: false})
		if err != nil {
			return err
		}
		for _, img := range imgs {
			for _, repoTag := range img.RepoTags {
				repo, tag := docker.ParseRepositoryTag(repoTag)
				if repo == i.Repo {
					if i.Tag == "" {
						if tag == "latest" {
							return fn(&img)
						}
					}
					if tag == i.Tag {
						return fn(&img)
					}
				}
			}
		}
		return fmt.Errorf("Image %s not found", i)
	})
}

func (c *Client) pull(image string) ImageMetadata {
	reader, writer := io.Pipe()
	defer writer.Close()

	repository, tag := api.ParseRepositoryTag(image)
	tag = misc.NVL(tag, "latest")
	opts := api.PullImageOptions{
		Repository:   repository + ":" + tag,
		OutputStream: writer,
	}

	// check output goroutine
	began := make(chan bool, 1)
	once := sync.Once{}

	go func() {
		reader := bufio.NewReader(reader)
		var line string
		var err error
		for err == nil {
			line, err = reader.ReadString('\n')
			if err != nil {
				break
			}
			once.Do(func() {
				began <- true
			})
			if strings.Contains(line, "already being pulled by another client. Waiting.") {
				logs.Error.Printf("Image 'pull' status marked as already being pulled. image: %v, status: %v", opts.Repository, line)
			}
		}
		if err != nil && err != io.EOF {
			logs.Warn.Printf("Error reading pull image status. image: %v, err: %v", opts.Repository, err)
		}
	}()

	// pull the image
	timeout := time.After(cfg.DockerPullBeginTimeout)
	finished := make(chan error, 1)
	go func() {
		finished <- c.PullImage(opts, api.AuthConfiguration{})
		logs.Debug.Printf("Pull completed for image: %v", opts.Repository)
	}()

	// wait for the pulling to begin
	select {
	case <-began:
		break
	case err := <-finished:
		if err != nil {
			return ImageMetadata{
				Image: &api.Image{ID: ""},
				Error: CannotXContainerError{"Pull", err.Error()},
			}
		}
		return c.InspectImage(opts.Repository)
	case <-timeout:
		return ImageMetadata{
			Image: &api.Image{ID: ""},
			Error: &DockerTimeoutError{cfg.DockerPullBeginTimeout, "pullBegin"},
		}
	}

	// wait for the completion
	err := <-finished
	if err != nil {
		return ImageMetadata{
			Image: &api.Image{ID: ""},
			Error: CannotXContainerError{"Pull", err.Error()},
		}
	}
	return c.InspectImage(opts.Repository)
}

func (d *DockerDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
	// Get the image from config
	image, ok := task.Config["image"]
	if !ok || image == "" {
		return nil, fmt.Errorf("Image not specified")
	}
	if task.Resources == nil {
		return nil, fmt.Errorf("Resources are not specified")
	}
	if task.Resources.MemoryMB == 0 {
		return nil, fmt.Errorf("Memory limit cannot be zero")
	}
	if task.Resources.CPU == 0 {
		return nil, fmt.Errorf("CPU limit cannot be zero")
	}

	cleanupContainer, err := strconv.ParseBool(d.config.ReadDefault("docker.cleanup.container", "true"))
	if err != nil {
		return nil, fmt.Errorf("Unable to parse docker.cleanup.container: %s", err)
	}
	cleanupImage, err := strconv.ParseBool(d.config.ReadDefault("docker.cleanup.image", "true"))
	if err != nil {
		return nil, fmt.Errorf("Unable to parse docker.cleanup.image: %s", err)
	}

	// Initialize docker API client
	dockerEndpoint := d.config.ReadDefault("docker.endpoint", "unix:///var/run/docker.sock")
	client, err := docker.NewClient(dockerEndpoint)
	if err != nil {
		return nil, fmt.Errorf("Failed to connect to docker.endpoint (%s): %s", dockerEndpoint, err)
	}

	repo, tag := docker.ParseRepositoryTag(image)
	// Make sure tag is always explicitly set. We'll default to "latest" if it
	// isn't, which is the expected behavior.
	if tag == "" {
		tag = "latest"
	}

	var dockerImage *docker.Image
	// We're going to check whether the image is already downloaded. If the tag
	// is "latest" we have to check for a new version every time so we don't
	// bother to check and cache the id here. We'll download first, then cache.
	if tag != "latest" {
		dockerImage, err = client.InspectImage(image)
	}

	// Download the image
	if dockerImage == nil {
		pullOptions := docker.PullImageOptions{
			Repository: repo,
			Tag:        tag,
		}
		// TODO add auth configuration for private repos
		authOptions := docker.AuthConfiguration{}
		err = client.PullImage(pullOptions, authOptions)
		if err != nil {
			d.logger.Printf("[ERR] driver.docker: pulling container %s", err)
			return nil, fmt.Errorf("Failed to pull `%s`: %s", image, err)
		}
		d.logger.Printf("[DEBUG] driver.docker: docker pull %s:%s succeeded", repo, tag)

		// Now that we have the image we can get the image id
		dockerImage, err = client.InspectImage(image)
		if err != nil {
			d.logger.Printf("[ERR] driver.docker: getting image id for %s", image)
			return nil, fmt.Errorf("Failed to determine image id for `%s`: %s", image, err)
		}
	}
	d.logger.Printf("[DEBUG] driver.docker: using image %s", dockerImage.ID)
	d.logger.Printf("[INFO] driver.docker: identified image %s as %s", image, dockerImage.ID)

	// Create a container
	container, err := client.CreateContainer(createContainer(ctx, task, d.logger))
	if err != nil {
		d.logger.Printf("[ERR] driver.docker: %s", err)
		return nil, fmt.Errorf("Failed to create container from image %s", image)
	}
	d.logger.Printf("[INFO] driver.docker: created container %s", container.ID)

	// Start the container
	err = client.StartContainer(container.ID, container.HostConfig)
	if err != nil {
		d.logger.Printf("[ERR] driver.docker: starting container %s", container.ID)
		return nil, fmt.Errorf("Failed to start container %s", container.ID)
	}
	d.logger.Printf("[INFO] driver.docker: started container %s", container.ID)

	// Return a driver handle
	h := &dockerHandle{
		client:           client,
		cleanupContainer: cleanupContainer,
		cleanupImage:     cleanupImage,
		logger:           d.logger,
		imageID:          dockerImage.ID,
		containerID:      container.ID,
		doneCh:           make(chan struct{}),
		waitCh:           make(chan error, 1),
	}
	go h.run()
//.........这里部分代码省略.........