func runLogin(dockerCli *client.DockerCli, opts loginOptions) error {
	ctx := context.Background()
	clnt := dockerCli.Client()

	var serverAddress string
	var isDefaultRegistry bool
	if opts.serverAddress != "" {
		serverAddress = opts.serverAddress
	} else {
		serverAddress = dockerCli.ElectAuthServer(ctx)
		isDefaultRegistry = true
	}
	authConfig, err := dockerCli.ConfigureAuth(opts.user, opts.password, serverAddress, isDefaultRegistry)
	if err != nil {
		return err
	}
	response, err := clnt.RegistryLogin(ctx, authConfig)
	if err != nil {
		return err
	}
	if response.IdentityToken != "" {
		authConfig.Password = ""
		authConfig.IdentityToken = response.IdentityToken
	}
	if err := client.StoreCredentials(dockerCli.ConfigFile(), authConfig); err != nil {
		return fmt.Errorf("Error saving credentials: %v", err)
	}

	if response.Status != "" {
		fmt.Fprintln(dockerCli.Out(), response.Status)
	}
	return nil
}

func newCreateCommand(dockerCli *client.DockerCli) *cobra.Command {
	opts := createOptions{
		driverOpts: *opts.NewMapOpts(nil, nil),
	}

	cmd := &cobra.Command{
		Use:   "create [OPTIONS] [VOLUME]",
		Short: "Create a volume",
		Long:  createDescription,
		Args:  cli.RequiresMaxArgs(1),
		RunE: func(cmd *cobra.Command, args []string) error {
			if len(args) == 1 {
				if opts.name != "" {
					fmt.Fprint(dockerCli.Err(), "Conflicting options: either specify --name or provide positional arg, not both\n")
					return cli.StatusError{StatusCode: 1}
				}
				opts.name = args[0]
			}
			return runCreate(dockerCli, opts)
		},
	}
	flags := cmd.Flags()
	flags.StringVarP(&opts.driver, "driver", "d", "local", "Specify volume driver name")
	flags.StringVar(&opts.name, "name", "", "Specify volume name")
	flags.Lookup("name").Hidden = true
	flags.VarP(&opts.driverOpts, "opt", "o", "Set driver specific options")
	flags.StringSliceVar(&opts.labels, "label", []string{}, "Set metadata for a volume")

	return cmd
}

func runPS(dockerCli *client.DockerCli, opts psOptions) error {
	client := dockerCli.Client()
	ctx := context.Background()

	service, _, err := client.ServiceInspectWithRaw(ctx, opts.serviceID)
	if err != nil {
		return err
	}

	filter := opts.filter.Value()
	filter.Add("service", service.ID)
	if filter.Include("node") {
		nodeFilters := filter.Get("node")
		for _, nodeFilter := range nodeFilters {
			nodeReference, err := node.Reference(client, ctx, nodeFilter)
			if err != nil {
				return err
			}
			filter.Del("node", nodeFilter)
			filter.Add("node", nodeReference)
		}
	}

	tasks, err := client.TaskList(ctx, types.TaskListOptions{Filter: filter})
	if err != nil {
		return err
	}

	return task.Print(dockerCli, ctx, tasks, idresolver.New(client, opts.noResolve))
}

func inspectAll(ctx context.Context, dockerCli *client.DockerCli, getSize bool) inspect.GetRefFunc {
	client := dockerCli.Client()

	return func(ref string) (interface{}, []byte, error) {
		c, rawContainer, err := client.ContainerInspectWithRaw(ctx, ref, getSize)
		if err == nil || !apiclient.IsErrNotFound(err) {
			return c, rawContainer, err
		}
		// Search for image with that id if a container doesn't exist.
		i, rawImage, err := client.ImageInspectWithRaw(ctx, ref)
		if err == nil || !apiclient.IsErrNotFound(err) {
			return i, rawImage, err
		}

		// Search for task with that id if an image doesn't exist.
		t, rawTask, err := client.TaskInspectWithRaw(ctx, ref)
		if err == nil || !(apiclient.IsErrNotFound(err) || isErrorNoSwarmMode(err)) {
			if getSize {
				fmt.Fprintln(dockerCli.Err(), "WARNING: --size ignored for tasks")
			}
			return t, rawTask, err
		}
		return nil, nil, fmt.Errorf("Error: No such container, image or task: %s", ref)
	}
}

// ForwardAllSignals forwards signals to the container
func ForwardAllSignals(ctx context.Context, cli *client.DockerCli, cid string) chan os.Signal {
	sigc := make(chan os.Signal, 128)
	signal.CatchAll(sigc)
	go func() {
		for s := range sigc {
			if s == signal.SIGCHLD || s == signal.SIGPIPE {
				continue
			}
			var sig string
			for sigStr, sigN := range signal.SignalMap {
				if sigN == s {
					sig = sigStr
					break
				}
			}
			if sig == "" {
				fmt.Fprintf(cli.Err(), "Unsupported signal: %v. Discarding.\n", s)
				continue
			}

			if err := cli.Client().ContainerKill(ctx, cid, sig); err != nil {
				logrus.Debugf("Error sending signal: %s", err)
			}
		}
	}()
	return sigc
}

func updateNodes(dockerCli *client.DockerCli, nodes []string, mergeNode func(node *swarm.Node) error, success func(nodeID string)) error {
	client := dockerCli.Client()
	ctx := context.Background()

	for _, nodeID := range nodes {
		node, _, err := client.NodeInspectWithRaw(ctx, nodeID)
		if err != nil {
			return err
		}

		err = mergeNode(&node)
		if err != nil {
			if err == errNoRoleChange {
				continue
			}
			return err
		}
		err = client.NodeUpdate(ctx, node.ID, node.Version, node.Spec)
		if err != nil {
			return err
		}
		success(nodeID)
	}
	return nil
}

func runUpdate(dockerCli *client.DockerCli, flags *pflag.FlagSet, opts swarmOptions) error {
	client := dockerCli.Client()
	ctx := context.Background()

	var updateFlags swarm.UpdateFlags

	swarm, err := client.SwarmInspect(ctx)
	if err != nil {
		return err
	}

	err = mergeSwarm(&swarm, flags)
	if err != nil {
		return err
	}

	err = client.SwarmUpdate(ctx, swarm.Version, swarm.Spec, updateFlags)
	if err != nil {
		return err
	}

	fmt.Fprintln(dockerCli.Out(), "Swarm updated.")

	return nil
}

func runLogs(dockerCli *client.DockerCli, opts *logsOptions) error {
	ctx := context.Background()

	c, err := dockerCli.Client().ContainerInspect(ctx, opts.container)
	if err != nil {
		return err
	}

	if !validDrivers[c.HostConfig.LogConfig.Type] {
		return fmt.Errorf("\"logs\" command is supported only for \"json-file\" and \"journald\" logging drivers (got: %s)", c.HostConfig.LogConfig.Type)
	}

	options := types.ContainerLogsOptions{
		ShowStdout: true,
		ShowStderr: true,
		Since:      opts.since,
		Timestamps: opts.timestamps,
		Follow:     opts.follow,
		Tail:       opts.tail,
		Details:    opts.details,
	}
	responseBody, err := dockerCli.Client().ContainerLogs(ctx, opts.container, options)
	if err != nil {
		return err
	}
	defer responseBody.Close()

	if c.Config.Tty {
		_, err = io.Copy(dockerCli.Out(), responseBody)
	} else {
		_, err = stdcopy.StdCopy(dockerCli.Out(), dockerCli.Err(), responseBody)
	}
	return err
}

func runPush(dockerCli *client.DockerCli, name string) error {
	named, err := reference.ParseNamed(name) // FIXME: validate
	if err != nil {
		return err
	}
	if reference.IsNameOnly(named) {
		named = reference.WithDefaultTag(named)
	}
	ref, ok := named.(reference.NamedTagged)
	if !ok {
		return fmt.Errorf("invalid name: %s", named.String())
	}

	ctx := context.Background()

	repoInfo, err := registry.ParseRepositoryInfo(named)
	if err != nil {
		return err
	}
	authConfig := dockerCli.ResolveAuthConfig(ctx, repoInfo.Index)

	encodedAuth, err := client.EncodeAuthToBase64(authConfig)
	if err != nil {
		return err
	}
	return dockerCli.Client().PluginPush(ctx, ref.String(), encodedAuth)
}

func runList(dockerCli *client.DockerCli, opts listOptions) error {
	ctx := context.Background()
	client := dockerCli.Client()

	services, err := client.ServiceList(ctx, types.ServiceListOptions{Filter: opts.filter.Value()})
	if err != nil {
		return err
	}

	out := dockerCli.Out()
	if opts.quiet {
		printQuiet(out, services)
	} else {
		taskFilter := filters.NewArgs()
		for _, service := range services {
			taskFilter.Add("service", service.ID)
		}

		tasks, err := client.TaskList(ctx, types.TaskListOptions{Filter: taskFilter})
		if err != nil {
			return err
		}

		running := map[string]int{}
		for _, task := range tasks {
			if task.Status.State == "running" {
				running[task.ServiceID]++
			}
		}

		printTable(out, services, running)
	}
	return nil
}

func runServiceScale(dockerCli *client.DockerCli, serviceID string, scale string) error {
	client := dockerCli.Client()
	ctx := context.Background()

	service, err := client.ServiceInspect(ctx, serviceID)
	if err != nil {
		return err
	}

	serviceMode := &service.Spec.Mode
	if serviceMode.Replicated == nil {
		return fmt.Errorf("scale can only be used with replicated mode")
	}
	uintScale, err := strconv.ParseUint(scale, 10, 64)
	if err != nil {
		return fmt.Errorf("invalid replicas value %s: %s", scale, err.Error())
	}
	serviceMode.Replicated.Replicas = &uintScale

	err = client.ServiceUpdate(ctx, service.ID, service.Version, service.Spec)
	if err != nil {
		return err
	}

	fmt.Fprintf(dockerCli.Out(), "%s scaled to %s\n", serviceID, scale)
	return nil
}

func runCreate(dockerCli *client.DockerCli, opts createOptions) error {
	client := dockerCli.Client()

	ipamCfg, err := consolidateIpam(opts.ipamSubnet, opts.ipamIPRange, opts.ipamGateway, opts.ipamAux.GetAll())
	if err != nil {
		return err
	}

	// Construct network create request body
	nc := types.NetworkCreate{
		Driver:  opts.driver,
		Options: opts.driverOpts.GetAll(),
		IPAM: network.IPAM{
			Driver:  opts.ipamDriver,
			Config:  ipamCfg,
			Options: opts.ipamOpt.GetAll(),
		},
		CheckDuplicate: true,
		Internal:       opts.internal,
		EnableIPv6:     opts.ipv6,
		Labels:         runconfigopts.ConvertKVStringsToMap(opts.labels),
	}

	resp, err := client.NetworkCreate(context.Background(), opts.name, nc)
	if err != nil {
		return err
	}
	fmt.Fprintf(dockerCli.Out(), "%s\n", resp.ID)
	return nil
}

func runEvents(dockerCli *client.DockerCli, opts *eventsOptions) error {
	eventFilterArgs := filters.NewArgs()

	// Consolidate all filter flags, and sanity check them early.
	// They'll get process in the daemon/server.
	for _, f := range opts.filter {
		var err error
		eventFilterArgs, err = filters.ParseFlag(f, eventFilterArgs)
		if err != nil {
			return err
		}
	}

	options := types.EventsOptions{
		Since:   opts.since,
		Until:   opts.until,
		Filters: eventFilterArgs,
	}

	responseBody, err := dockerCli.Client().Events(context.Background(), options)
	if err != nil {
		return err
	}
	defer responseBody.Close()

	return streamEvents(responseBody, dockerCli.Out())
}

func runList(dockerCli *client.DockerCli, opts listOptions) error {
	ctx := context.Background()
	client := dockerCli.Client()

	services, err := client.ServiceList(ctx, types.ServiceListOptions{Filter: opts.filter.Value()})
	if err != nil {
		return err
	}

	out := dockerCli.Out()
	if opts.quiet {
		PrintQuiet(out, services)
	} else {
		taskFilter := filters.NewArgs()
		for _, service := range services {
			taskFilter.Add("service", service.ID)
		}

		tasks, err := client.TaskList(ctx, types.TaskListOptions{Filter: taskFilter})
		if err != nil {
			return err
		}

		nodes, err := client.NodeList(ctx, types.NodeListOptions{})
		if err != nil {
			return err
		}

		PrintNotQuiet(out, services, nodes, tasks)
	}
	return nil
}

func runTasks(dockerCli *client.DockerCli, opts tasksOptions) error {
	client := dockerCli.Client()
	ctx := context.Background()

	nodeRef, err := nodeReference(client, ctx, opts.nodeID)
	if err != nil {
		return nil
	}
	node, err := client.NodeInspect(ctx, nodeRef)
	if err != nil {
		return err
	}

	filter := opts.filter.Value()
	filter.Add("node", node.ID)
	if !opts.all {
		filter.Add("desired_state", string(swarm.TaskStateRunning))
		filter.Add("desired_state", string(swarm.TaskStateAccepted))

	}

	tasks, err := client.TaskList(
		ctx,
		types.TaskListOptions{Filter: filter})
	if err != nil {
		return err
	}

	return task.Print(dockerCli, ctx, tasks, idresolver.New(client, opts.noResolve))
}

func runCommit(dockerCli *client.DockerCli, opts *commitOptions) error {
	ctx := context.Background()

	name := opts.container
	reference := opts.reference

	var config *containertypes.Config
	if opts.config != "" {
		config = &containertypes.Config{}
		if err := json.Unmarshal([]byte(opts.config), config); err != nil {
			return err
		}
	}

	options := types.ContainerCommitOptions{
		Reference: reference,
		Comment:   opts.comment,
		Author:    opts.author,
		Changes:   opts.changes.GetAll(),
		Pause:     opts.pause,
		Config:    config,
	}

	response, err := dockerCli.Client().ContainerCommit(ctx, name, options)
	if err != nil {
		return err
	}

	fmt.Fprintln(dockerCli.Out(), response.ID)
	return nil
}

func runLoad(dockerCli *client.DockerCli, opts loadOptions) error {

	var input io.Reader = dockerCli.In()
	if opts.input != "" {
		file, err := os.Open(opts.input)
		if err != nil {
			return err
		}
		defer file.Close()
		input = file
	}
	if !dockerCli.Out().IsTerminal() {
		opts.quiet = true
	}
	response, err := dockerCli.Client().ImageLoad(context.Background(), input, opts.quiet)
	if err != nil {
		return err
	}
	defer response.Body.Close()

	if response.Body != nil && response.JSON {
		return jsonmessage.DisplayJSONMessagesToStream(response.Body, dockerCli.Out(), nil)
	}

	_, err = io.Copy(dockerCli.Out(), response.Body)
	return err
}

func runConfig(dockerCli *client.DockerCli, opts configOptions) error {
	bundle, err := loadBundlefile(dockerCli.Err(), opts.namespace, opts.bundlefile)
	if err != nil {
		return err
	}
	return bundlefile.Print(dockerCli.Out(), bundle)
}

func copyFromContainer(ctx context.Context, dockerCli *client.DockerCli, srcContainer, srcPath, dstPath string, cpParam *cpConfig) (err error) {
	if dstPath != "-" {
		// Get an absolute destination path.
		dstPath, err = resolveLocalPath(dstPath)
		if err != nil {
			return err
		}
	}

	// if client requests to follow symbol link, then must decide target file to be copied
	var rebaseName string
	if cpParam.followLink {
		srcStat, err := statContainerPath(ctx, dockerCli, srcContainer, srcPath)

		// If the destination is a symbolic link, we should follow it.
		if err == nil && srcStat.Mode&os.ModeSymlink != 0 {
			linkTarget := srcStat.LinkTarget
			if !system.IsAbs(linkTarget) {
				// Join with the parent directory.
				srcParent, _ := archive.SplitPathDirEntry(srcPath)
				linkTarget = filepath.Join(srcParent, linkTarget)
			}

			linkTarget, rebaseName = archive.GetRebaseName(srcPath, linkTarget)
			srcPath = linkTarget
		}

	}

	content, stat, err := dockerCli.Client().CopyFromContainer(ctx, srcContainer, srcPath)
	if err != nil {
		return err
	}
	defer content.Close()

	if dstPath == "-" {
		// Send the response to STDOUT.
		_, err = io.Copy(os.Stdout, content)

		return err
	}

	// Prepare source copy info.
	srcInfo := archive.CopyInfo{
		Path:       srcPath,
		Exists:     true,
		IsDir:      stat.Mode.IsDir(),
		RebaseName: rebaseName,
	}

	preArchive := content
	if len(srcInfo.RebaseName) != 0 {
		_, srcBase := archive.SplitPathDirEntry(srcInfo.Path)
		preArchive = archive.RebaseArchiveEntries(content, srcBase, srcInfo.RebaseName)
	}
	// See comments in the implementation of `archive.CopyTo` for exactly what
	// goes into deciding how and whether the source archive needs to be
	// altered for the correct copy behavior.
	return archive.CopyTo(preArchive, srcInfo, dstPath)
}

// RunMockCeph runs a set of containers used to mock a Ceph storage cluster
func RunMockCeph(t *testing.T, name string, cli *client.DockerCli, etcdPort string) {

	etcdutils.SetSingle(t, "/deis/store/hosts/"+utils.HostAddress(), utils.HostAddress(), etcdPort)
	var err error
	cli, stdout, stdoutPipe := dockercli.NewClient()
	cephImage := "deis/mock-store:latest"
	ipaddr := utils.HostAddress()
	fmt.Printf("--- Running deis/mock-store at %s\n", ipaddr)
	done2 := make(chan bool, 1)
	go func() {
		done2 <- true
		_ = cli.CmdRm("-f", name)
		err = dockercli.RunContainer(cli,
			"--name", name,
			"--rm",
			"-e", "HOST="+ipaddr,
			"-e", "ETCD_PORT="+etcdPort,
			"--net=host",
			cephImage)
	}()
	dockercli.PrintToStdout(t, stdout, stdoutPipe, "deis-store-gateway running...")
	if err != nil {
		t.Fatal(err)
	}
}