cluster.go

package swarm

import (
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/docker/swarm/common"

	"github.com/docker/docker/api/types"
	containertypes "github.com/docker/docker/api/types/container"
	networktypes "github.com/docker/docker/api/types/network"
	"github.com/docker/docker/api/types/volume"
	"github.com/docker/docker/client"
	engineapi "github.com/docker/docker/client"
	"github.com/docker/docker/pkg/discovery"
	"github.com/docker/docker/pkg/stringid"
	"github.com/docker/go-units"
	"github.com/docker/swarm/cluster"
	"github.com/docker/swarm/scheduler"
	"github.com/docker/swarm/scheduler/node"
	"github.com/samalba/dockerclient"
)

type pendingContainer struct {
	Config *cluster.ContainerConfig
	Name   string
	Engine *cluster.Engine
}

func (p *pendingContainer) ToContainer() *cluster.Container {
	container := &cluster.Container{
		Container: types.Container{
			Labels: p.Config.Labels,
		},
		Config: p.Config,
		Info: types.ContainerJSON{
			ContainerJSONBase: &types.ContainerJSONBase{
				HostConfig: &containertypes.HostConfig{},
			},
		},
		Engine: p.Engine,
	}

	if p.Name != "" {
		container.Container.Names = []string{"/" + p.Name}
	}

	return container
}

// Cluster is exported
type Cluster struct {
	sync.RWMutex

	eventHandlers     *cluster.EventHandlers
	engines           map[string]*cluster.Engine
	pendingEngines    map[string]*cluster.Engine
	scheduler         *scheduler.Scheduler
	discovery         discovery.Backend
	pendingContainers map[string]*pendingContainer

	overcommitRatio float64
	engineOpts      *cluster.EngineOpts
	createRetry     int64
	TLSConfig       *tls.Config
}

// NewCluster is exported
func NewCluster(scheduler *scheduler.Scheduler, TLSConfig *tls.Config, discovery discovery.Backend, options cluster.DriverOpts, engineOptions *cluster.EngineOpts) (cluster.Cluster, error) {
	log.WithFields(log.Fields{"name": "swarm"}).Debug("Initializing cluster")

	cluster := &Cluster{
		eventHandlers:     cluster.NewEventHandlers(),
		engines:           make(map[string]*cluster.Engine),
		pendingEngines:    make(map[string]*cluster.Engine),
		scheduler:         scheduler,
		TLSConfig:         TLSConfig,
		discovery:         discovery,
		pendingContainers: make(map[string]*pendingContainer),
		overcommitRatio:   0.05,
		engineOpts:        engineOptions,
		createRetry:       0,
	}

	if val, ok := options.Float("swarm.overcommit", ""); ok {
		if val <= float64(-1) {
			log.Fatalf("swarm.overcommit should be larger than -1, %f is invalid", val)
		} else if val < float64(0) {
			log.Warn("-1 < swarm.overcommit < 0 will make swarm take less resource than docker engine offers")
			cluster.overcommitRatio = val
		} else {
			cluster.overcommitRatio = val
		}
	}

	if val, ok := options.Int("swarm.createretry", ""); ok {
		if val < 0 {
			log.Fatalf("swarm.createretry can not be negative, %d is invalid", val)
		}
		cluster.createRetry = val
	}

	discoveryCh, errCh := cluster.discovery.Watch(nil)
	go cluster.monitorDiscovery(discoveryCh, errCh)
	go cluster.monitorPendingEngines()

	return cluster, nil
}

// Handle callbacks for the events
func (c *Cluster) Handle(e *cluster.Event) error {
	c.eventHandlers.Handle(e)
	return nil
}

// RegisterEventHandler registers an event handler.
func (c *Cluster) RegisterEventHandler(h cluster.EventHandler) error {
	return c.eventHandlers.RegisterEventHandler(h)
}

// UnregisterEventHandler unregisters a previously registered event handler.
func (c *Cluster) UnregisterEventHandler(h cluster.EventHandler) {
	c.eventHandlers.UnregisterEventHandler(h)
}

// Generate a globally (across the cluster) unique ID.
func (c *Cluster) generateUniqueID() string {
	for {
		id := stringid.GenerateRandomID()
		if c.Container(id) == nil {
			return id
		}
	}
}

// StartContainer starts a container
func (c *Cluster) StartContainer(container *cluster.Container, hostConfig *dockerclient.HostConfig) error {
	return container.Engine.StartContainer(container.ID, hostConfig)
}

// CreateContainer aka schedule a brand new container into the cluster.
func (c *Cluster) CreateContainer(config *cluster.ContainerConfig, name string, authConfig *types.AuthConfig) (*cluster.Container, error) {
	container, err := c.createContainer(config, name, false, authConfig)

	if err != nil {
		var retries int64
		//  fails with image not found, then try to reschedule with image affinity
		// ENGINEAPIFIXME: The first error can be removed once dockerclient is removed
		bImageNotFoundError, _ := regexp.MatchString(`image \S* not found`, err.Error())

		// Since docker engine 1.13 the error message has been changed. We have to check both for backwards compatibility.
		bImageNotFoundError113, _ := regexp.MatchString(`repository \S* not found`, err.Error())
		if (bImageNotFoundError || bImageNotFoundError113 || client.IsErrImageNotFound(err)) && !config.HaveNodeConstraint() {
			// Check if the image exists in the cluster
			// If exists, retry with an image affinity
			if c.Image(config.Image) != nil {
				container, err = c.createContainer(config, name, true, authConfig)
				retries++
			}
		}

		for ; retries < c.createRetry && err != nil; retries++ {
			log.WithFields(log.Fields{"Name": "Swarm"}).Warnf("Failed to create container: %s, retrying", err)
			container, err = c.createContainer(config, name, false, authConfig)
		}
	}
	return container, err
}

func (c *Cluster) createContainer(config *cluster.ContainerConfig, name string, withImageAffinity bool, authConfig *types.AuthConfig) (*cluster.Container, error) {
	c.scheduler.Lock()

	// Ensure the name is available
	if !c.checkNameUniqueness(name) {
		c.scheduler.Unlock()
		return nil, fmt.Errorf("Conflict: The name %s is already assigned. You have to delete (or rename) that container to be able to assign %s to a container again.", name, name)
	}

	swarmID := config.SwarmID()
	if swarmID == "" {
		// Associate a Swarm ID to the container we are creating.
		swarmID = c.generateUniqueID()
		config.SetSwarmID(swarmID)
	}

	if network := c.Networks().Get(string(config.HostConfig.NetworkMode)); network != nil && network.Scope == "local" {
		if !config.HaveNodeConstraint() {
			config.AddConstraint("node==~" + network.Engine.Name)
		}
		config.HostConfig.NetworkMode = containertypes.NetworkMode(network.Name)
	}

	if withImageAffinity {
		config.AddAffinity("image==" + config.Image)
	}

	nodes, err := c.scheduler.SelectNodesForContainer(c.listNodes(), config)

	if withImageAffinity {
		config.RemoveAffinity("image==" + config.Image)
	}

	if err != nil {
		c.scheduler.Unlock()
		return nil, err
	}
	n := nodes[0]
	engine, ok := c.engines[n.ID]
	if !ok {
		c.scheduler.Unlock()
		return nil, fmt.Errorf("error creating container")
	}

	c.pendingContainers[swarmID] = &pendingContainer{
		Name:   name,
		Config: config,
		Engine: engine,
	}

	c.scheduler.Unlock()

	container, err := engine.CreateContainer(config, name, true, authConfig)

	if err != nil {
		log.WithFields(log.Fields{"NodeName": n.Name, "NodeID": n.ID}).WithError(err).Error("Failed to create container")
	} else {
		containerFlag := name
		if containerFlag == "" {
			containerFlag = stringid.TruncateID(container.ID)
		}
		log.WithFields(log.Fields{"NodeName": n.Name, "NodeID": n.ID}).Debugf("Scheduling container %s to ", containerFlag)
	}

	c.scheduler.Lock()
	delete(c.pendingContainers, swarmID)
	c.scheduler.Unlock()

	return container, err
}

// RemoveContainer aka Remove a container from the cluster.
func (c *Cluster) RemoveContainer(container *cluster.Container, force, volumes bool) error {
	return container.Engine.RemoveContainer(container, force, volumes)
}

// RemoveNetwork removes a network from the cluster
func (c *Cluster) RemoveNetwork(network *cluster.Network) error {
	err := network.Engine.RemoveNetwork(network)
	if err == nil {
		if network.Scope == "global" {
			for _, engine := range c.engines {
				engine.DeleteNetwork(network)
			}
		}
	} else if engineapi.IsErrConnectionFailed(err) && network.Scope == "global" {
		log.Debug("The original engine is unreachable - Attempting to remove global network from the reachable engines...")
		for _, engine := range c.engines {
			e1 := engine.RemoveNetwork(network)
			if e1 == nil {
				for _, engine := range c.engines {
					engine.DeleteNetwork(network)
				}
				err = nil
				break
			}
		}
	}
	if err != nil && network.Scope == "global" {
		log.Debugf("Failed to remove global scope network %s from any engine...", network.ID)
	}
	return err
}

func (c *Cluster) getEngineByAddr(addr string) *cluster.Engine {
	c.RLock()
	defer c.RUnlock()

	if engine, ok := c.pendingEngines[addr]; ok {
		return engine
	}
	for _, engine := range c.engines {
		if engine.Addr == addr {
			return engine
		}
	}
	return nil
}

func (c *Cluster) hasEngineByAddr(addr string) bool {
	return c.getEngineByAddr(addr) != nil
}

func (c *Cluster) addEngine(addr string) bool {
	// Check the engine is already registered by address.
	if c.hasEngineByAddr(addr) {
		return false
	}

	engine := cluster.NewEngine(addr, c.overcommitRatio, c.engineOpts)
	if err := engine.RegisterEventHandler(c); err != nil {
		log.Error(err)
	}
	// Add it to pending engine map, indexed by address. This will prevent
	// duplicates from entering
	c.Lock()
	c.pendingEngines[addr] = engine
	c.Unlock()

	// validatePendingEngine will start a thread to validate the engine.
	// If the engine is reachable and valid, it'll be monitored and updated in a loop.
	// If engine is not reachable, pending engines will be examined once in a while
	go c.validatePendingEngine(engine)

	return true
}

// validatePendingEngine connects to the engine,
func (c *Cluster) validatePendingEngine(engine *cluster.Engine) bool {
	// Attempt a connection to the engine. Since this is slow, don't get a hold
	// of the lock yet.
	if err := engine.Connect(c.TLSConfig); err != nil {
		log.WithFields(log.Fields{"Addr": engine.Addr}).Debugf("Failed to validate pending node: %s", err)
		return false
	}

	// The following is critical and fast. Grab a lock.
	c.Lock()
	defer c.Unlock()

	// Only validate engines from pendingEngines list
	if _, exists := c.pendingEngines[engine.Addr]; !exists {
		return false
	}

	// Make sure the engine ID is unique.
	if old, exists := c.engines[engine.ID]; exists {
		if old.Addr != engine.Addr {
			log.Errorf("ID duplicated. %s shared by %s and %s", engine.ID, old.Addr, engine.Addr)
			// Keep this engine in pendingEngines table and show its error.
			// If it's ID duplication from VM clone, user see this message and can fix it.
			// If the engine is rebooted and get new IP from DHCP, previous address will be removed
			// from discovery after a while.
			// In both cases, retry may fix the problem.
			engine.HandleIDConflict(old.Addr)
		} else {
			log.Debugf("node %q (name: %q) with address %q is already registered", engine.ID, engine.Name, engine.Addr)
			engine.Disconnect()
			// Remove it from pendingEngines table
			delete(c.pendingEngines, engine.Addr)
		}
		return false
	}

	// Engine validated, move from pendingEngines table to engines table
	delete(c.pendingEngines, engine.Addr)
	// set engine state to healthy, and start refresh loop
	engine.ValidationComplete()
	c.engines[engine.ID] = engine

	log.Infof("Registered Engine %s at %s", engine.Name, engine.Addr)
	return true
}

func (c *Cluster) removeEngine(addr string) bool {
	engine := c.getEngineByAddr(addr)
	if engine == nil {
		return false
	}
	c.Lock()
	defer c.Unlock()

	engine.Disconnect()
	// it could be in pendingEngines or engines
	if _, ok := c.pendingEngines[addr]; ok {
		delete(c.pendingEngines, addr)
	} else {
		delete(c.engines, engine.ID)
	}
	log.Infof("Removed Engine %s", engine.Name)
	return true
}

// Entries are Docker Engines
func (c *Cluster) monitorDiscovery(ch <-chan discovery.Entries, errCh <-chan error) {
	// Watch changes on the discovery channel.
	currentEntries := discovery.Entries{}
	for {
		select {
		case entries := <-ch:
			added, removed := currentEntries.Diff(entries)
			currentEntries = entries

			// Remove engines first. `addEngine` will refuse to add an engine
			// if there's already an engine with the same ID.  If an engine
			// changes address, we have to first remove it then add it back.
			for _, entry := range removed {
				c.removeEngine(entry.String())
			}

			for _, entry := range added {
				c.addEngine(entry.String())
			}
		case err := <-errCh:
			log.Errorf("Discovery error: %v", err)
		}
	}
}

// monitorPendingEngines checks if some previous unreachable/invalid engines have been fixed
func (c *Cluster) monitorPendingEngines() {
	const minimumValidationInterval time.Duration = 10 * time.Second
	for {
		// Don't need to do it frequently
		time.Sleep(minimumValidationInterval)
		// Get the list of pendingEngines
		c.RLock()
		pEngines := make([]*cluster.Engine, 0, len(c.pendingEngines))
		for _, e := range c.pendingEngines {
			pEngines = append(pEngines, e)
		}
		c.RUnlock()
		for _, e := range pEngines {
			if e.TimeToValidate() {
				go c.validatePendingEngine(e)
			}
		}
	}
}

// Images returns all the images in the cluster.
func (c *Cluster) Images() cluster.Images {
	c.RLock()
	defer c.RUnlock()

	out := []*cluster.Image{}
	for _, e := range c.engines {
		out = append(out, e.Images()...)
	}
	return out
}

// Image returns an image with IDOrName in the cluster
func (c *Cluster) Image(IDOrName string) *cluster.Image {
	// Abort immediately if the name is empty.
	if len(IDOrName) == 0 {
		return nil
	}

	c.RLock()
	defer c.RUnlock()
	for _, e := range c.engines {
		if image := e.Image(IDOrName); image != nil {
			return image
		}
	}

	return nil
}

// RemoveImages removes all the images that match `name` from the cluster
func (c *Cluster) RemoveImages(name string, force bool) ([]types.ImageDelete, error) {
	c.Lock()
	defer c.Unlock()

	out := []types.ImageDelete{}
	errs := []string{}
	var err error
	for _, e := range c.engines {
		for _, image := range e.Images() {
			if image.Match(name, true) {
				content, err := image.Engine.RemoveImage(name, force)
				if err != nil {
					errs = append(errs, fmt.Sprintf("%s: %s", image.Engine.Name, err.Error()))
					continue
				}
				out = append(out, content...)
			}
		}
	}

	if len(errs) > 0 {
		err = errors.New(strings.Join(errs, "\n"))
	}

	return out, err
}

// CreateNetwork creates a network in the cluster
func (c *Cluster) CreateNetwork(name string, request *types.NetworkCreate) (response *types.NetworkCreateResponse, err error) {
	var (
		parts  = strings.SplitN(name, "/", 2)
		config = &cluster.ContainerConfig{}
	)

	if len(parts) == 2 {
		// a node was specified, create the container only on this node
		name = parts[1]
		config = cluster.BuildContainerConfig(containertypes.Config{Env: []string{"constraint:node==" + parts[0]}}, containertypes.HostConfig{}, networktypes.NetworkingConfig{})
	}

	nodes, err := c.scheduler.SelectNodesForContainer(c.listNodes(), config)
	if err != nil {
		return nil, err
	}
	if nodes != nil {
		resp, err := c.engines[nodes[0].ID].CreateNetwork(name, request)
		if err == nil {
			if network := c.engines[nodes[0].ID].Networks().Get(resp.ID); network != nil && network.Scope == "global" {
				for id, engine := range c.engines {
					if id != nodes[0].ID {
						engine.AddNetwork(network)
					}
				}
			}
		}
		return resp, err
	}
	return nil, nil
}

// CreateVolume creates a volume in the cluster
func (c *Cluster) CreateVolume(request *volume.VolumesCreateBody) (*types.Volume, error) {
	var (
		wg     sync.WaitGroup
		volume *types.Volume
		err    error
		parts  = strings.SplitN(request.Name, "/", 2)
		node   = ""
	)

	if request.Name == "" {
		request.Name = stringid.GenerateRandomID()
	} else if len(parts) == 2 {
		node = parts[0]
		request.Name = parts[1]
	}
	if node == "" {
		c.RLock()
		for _, e := range c.engines {
			wg.Add(1)

			go func(engine *cluster.Engine) {
				defer wg.Done()

				v, er := engine.CreateVolume(request)
				if v != nil {
					volume = v
					err = nil
				}
				if er != nil && volume == nil {
					err = er
				}
			}(e)
		}
		c.RUnlock()

		wg.Wait()
	} else {
		config := cluster.BuildContainerConfig(containertypes.Config{Env: []string{"constraint:node==" + parts[0]}}, containertypes.HostConfig{}, networktypes.NetworkingConfig{})
		nodes, err := c.scheduler.SelectNodesForContainer(c.listNodes(), config)
		if err != nil {
			return nil, err
		}
		if nodes != nil {
			v, er := c.engines[nodes[0].ID].CreateVolume(request)
			if v != nil {
				volume = v
				err = nil
			}
			if er != nil && volume == nil {
				err = er
			}
		}
	}

	return volume, err
}

// RemoveVolumes removes all the volumes that match `name` from the cluster
func (c *Cluster) RemoveVolumes(name string) (bool, error) {
	c.Lock()
	defer c.Unlock()

	found := false
	errs := []string{}
	var err error
	for _, e := range c.engines {
		if volume := e.Volumes().Get(name); volume != nil {
			if err := volume.Engine.RemoveVolume(volume.Name); err != nil {
				errs = append(errs, fmt.Sprintf("%s: %s", volume.Engine.Name, err.Error()))
				continue
			}
			found = true
		}
	}
	if len(errs) > 0 {
		err = errors.New(strings.Join(errs, "\n"))
	}
	return found, err
}

// Pull is exported
func (c *Cluster) Pull(name string, authConfig *types.AuthConfig, callback func(where, status string, err error)) {
	var wg sync.WaitGroup

	c.RLock()
	for _, e := range c.engines {
		wg.Add(1)

		go func(engine *cluster.Engine) {
			defer wg.Done()

			if callback != nil {
				callback(engine.Name, "", nil)
			}
			err := engine.Pull(name, authConfig)
			if callback != nil {
				if err != nil {
					callback(engine.Name, "", err)
				} else {
					callback(engine.Name, "downloaded", nil)
				}
			}
		}(e)
	}
	c.RUnlock()

	wg.Wait()
}

// Load image
func (c *Cluster) Load(imageReader io.Reader, callback func(where, status string, err error)) {
	var wg sync.WaitGroup

	c.RLock()
	pipeWriters := []*io.PipeWriter{}
	for _, e := range c.engines {
		wg.Add(1)

		pipeReader, pipeWriter := io.Pipe()
		pipeWriters = append(pipeWriters, pipeWriter)

		go func(reader *io.PipeReader, engine *cluster.Engine) {
			defer wg.Done()
			defer reader.Close()

			// call engine load image
			err := engine.Load(reader)
			if callback != nil {
				if err != nil {
					callback(engine.Name, "", err)
				}
			}
		}(pipeReader, e)
	}
	c.RUnlock()

	// create multi-writer
	listWriter := []io.Writer{}
	for _, pipeW := range pipeWriters {
		listWriter = append(listWriter, pipeW)
	}
	multiWriter := io.MultiWriter(listWriter...)

	// copy image-reader to multi-writer
	_, err := io.Copy(multiWriter, imageReader)
	if err != nil {
		log.Error(err)
	}

	// close pipe writers
	for _, pipeW := range pipeWriters {
		pipeW.Close()
	}

	wg.Wait()
}

// Import image
func (c *Cluster) Import(source string, ref string, tag string, imageReader io.Reader, callback func(what, status string, err error)) {
	var wg sync.WaitGroup
	c.RLock()
	pipeWriters := []*io.PipeWriter{}

	for _, e := range c.engines {
		wg.Add(1)

		pipeReader, pipeWriter := io.Pipe()
		pipeWriters = append(pipeWriters, pipeWriter)

		go func(reader *io.PipeReader, engine *cluster.Engine) {
			defer wg.Done()
			defer reader.Close()

			// call engine import
			err := engine.Import(source, ref, tag, reader)
			if callback != nil {
				if err != nil {
					callback(engine.Name, "", err)
				} else {
					callback(engine.Name, "Import success", nil)
				}
			}

		}(pipeReader, e)
	}
	c.RUnlock()

	// create multi-writer
	listWriter := []io.Writer{}
	for _, pipeW := range pipeWriters {
		listWriter = append(listWriter, pipeW)
	}
	multiWriter := io.MultiWriter(listWriter...)

	// copy image-reader to muti-writer
	_, err := io.Copy(multiWriter, imageReader)
	if err != nil {
		log.Error(err)
	}

	// close pipe writers
	for _, pipeW := range pipeWriters {
		pipeW.Close()
	}

	wg.Wait()
}

// Containers returns all the containers in the cluster.
func (c *Cluster) Containers() cluster.Containers {
	c.RLock()
	defer c.RUnlock()

	out := cluster.Containers{}
	for _, e := range c.engines {
		out = append(out, e.Containers()...)
	}

	return out
}

func (c *Cluster) checkNameUniqueness(name string) bool {
	// Abort immediately if the name is empty.
	if len(name) == 0 {
		return true
	}

	c.RLock()
	defer c.RUnlock()
	for _, e := range c.engines {
		for _, c := range e.Containers() {
			for _, cname := range c.Names {
				if cname == name || cname == "/"+name {
					return false
				}
			}
		}
	}

	// check pending containers.
	for _, c := range c.pendingContainers {
		if c.Name == name {
			return false
		}
	}

	return true
}

// Container returns the container with IDOrName in the cluster
func (c *Cluster) Container(IDOrName string) *cluster.Container {
	// Abort immediately if the name is empty.
	if len(IDOrName) == 0 {
		return nil
	}
	return c.Containers().Get(IDOrName)
}

// Networks returns all the networks in the cluster.
func (c *Cluster) Networks() cluster.Networks {
	c.RLock()
	defer c.RUnlock()

	out := cluster.Networks{}
	for _, e := range c.engines {
		out = append(out, e.Networks()...)
	}

	return out
}

// Volumes returns all the volumes in the cluster.
func (c *Cluster) Volumes() cluster.Volumes {
	c.RLock()
	defer c.RUnlock()

	out := cluster.Volumes{}
	for _, e := range c.engines {
		out = append(out, e.Volumes()...)
	}

	return out
}

// listNodes returns all validated engines in the cluster, excluding pendingEngines.
func (c *Cluster) listNodes() []*node.Node {
	c.RLock()
	defer c.RUnlock()

	out := make([]*node.Node, 0, len(c.engines))
	for _, e := range c.engines {
		node := node.NewNode(e)
		for _, pc := range c.pendingContainers {
			if pc.Engine.ID == e.ID && node.Container(pc.Config.SwarmID()) == nil {
				node.AddContainer(pc.ToContainer())
			}
		}
		out = append(out, node)
	}

	return out
}

// listEngines returns all the engines in the cluster.
// This is for reporting, not scheduling, hence pendingEngines are included.
func (c *Cluster) listEngines() []*cluster.Engine {
	c.RLock()
	defer c.RUnlock()

	out := make([]*cluster.Engine, 0, len(c.engines)+len(c.pendingEngines))
	for _, n := range c.engines {
		out = append(out, n)
	}
	for _, n := range c.pendingEngines {
		out = append(out, n)
	}
	return out
}

// TotalMemory returns the total memory of the cluster
func (c *Cluster) TotalMemory() int64 {
	var totalMemory int64
	for _, engine := range c.engines {
		totalMemory += engine.TotalMemory()
	}
	return totalMemory
}

// TotalCpus returns the total CPUs of the cluster
func (c *Cluster) TotalCpus() int64 {
	var totalCpus int64
	for _, engine := range c.engines {
		totalCpus += engine.TotalCpus()
	}
	return totalCpus
}

// Info returns some info about the cluster, like nb or containers / images
func (c *Cluster) Info() [][2]string {
	info := [][2]string{
		{"Strategy", c.scheduler.Strategy()},
		{"Filters", c.scheduler.Filters()},
		{"Nodes", fmt.Sprintf("%d", len(c.engines)+len(c.pendingEngines))},
	}

	engines := c.listEngines()
	sort.Sort(cluster.EngineSorter(engines))

	for _, engine := range engines {
		engineName := "(unknown)"
		if engine.Name != "" {
			engineName = engine.Name
		}
		info = append(info, [2]string{" " + engineName, engine.Addr})
		info = append(info, [2]string{"  └ ID", engine.ID})
		info = append(info, [2]string{"  └ Status", engine.Status()})

		// if engine's status is healthy, show container details of the node
		if engine.IsHealthy() {
			var paused, running, stopped int = 0, 0, 0
			for _, c := range engine.Containers() {
				if c.Info.State.Paused {
					paused++
				} else if c.Info.State.Running {
					running++
				} else {
					stopped++
				}
			}
			info = append(info, [2]string{"  └ Containers", fmt.Sprintf("%d (%d Running, %d Paused, %d Stopped)", len(engine.Containers()), running, paused, stopped)})
		} else {
			info = append(info, [2]string{"  └ Containers", fmt.Sprintf("%d", len(engine.Containers()))})
		}

		info = append(info, [2]string{"  └ Reserved CPUs", fmt.Sprintf("%d / %d", engine.UsedCpus(), engine.TotalCpus())})
		info = append(info, [2]string{"  └ Reserved Memory", fmt.Sprintf("%s / %s", units.BytesSize(float64(engine.UsedMemory())), units.BytesSize(float64(engine.TotalMemory())))})
		labels := make([]string, 0, len(engine.Labels))
		for k, v := range engine.Labels {
			labels = append(labels, k+"="+v)
		}
		sort.Strings(labels)
		info = append(info, [2]string{"  └ Labels", fmt.Sprintf("%s", strings.Join(labels, ", "))})
		errMsg := engine.ErrMsg()
		if len(errMsg) != 0 {
			info = append(info, [2]string{"  └ Error", errMsg})
		}
		info = append(info, [2]string{"  └ UpdatedAt", engine.UpdatedAt().UTC().Format(time.RFC3339)})
		info = append(info, [2]string{"  └ ServerVersion", engine.Version})
	}

	return info
}

// RANDOMENGINE returns a random engine.
func (c *Cluster) RANDOMENGINE() (*cluster.Engine, error) {
	nodes, err := c.scheduler.SelectNodesForContainer(c.listNodes(), &cluster.ContainerConfig{})
	if err != nil {
		return nil, err
	}
	return c.engines[nodes[0].ID], nil
}

// RenameContainer renames a container
func (c *Cluster) RenameContainer(container *cluster.Container, newName string) error {
	// check new name whether available
	if !c.checkNameUniqueness(newName) {
		return fmt.Errorf("Conflict: The name %s is already assigned. You have to delete (or rename) that container to be able to assign %s to a container again.", newName, newName)
	}

	// call engine rename
	err := container.Engine.RenameContainer(container, newName)
	return err
}

// BuildImage builds an image
func (c *Cluster) BuildImage(buildContext io.Reader, buildImage *types.ImageBuildOptions, out io.Writer) error {
	c.scheduler.Lock()

	// get an engine
	config := cluster.BuildContainerConfig(containertypes.Config{Env: convertMapToKVStrings(buildImage.BuildArgs)},
		containertypes.HostConfig{Resources: containertypes.Resources{CPUShares: buildImage.CPUShares, Memory: buildImage.Memory}},
		networktypes.NetworkingConfig{})
	buildImage.BuildArgs = convertKVStringsToMap(config.Env)
	nodes, err := c.scheduler.SelectNodesForContainer(c.listNodes(), config)
	c.scheduler.Unlock()
	if err != nil {
		return err
	}
	n := nodes[0]

	reader, err := c.engines[n.ID].BuildImage(buildContext, buildImage)
	if err != nil {
		return err
	}

	if _, err := io.Copy(out, reader); err != nil {
		return err
	}

	c.engines[n.ID].RefreshImages()
	return nil
}

// RefreshEngines refreshes all containers in the cluster
func (c *Cluster) RefreshEngines() error {
	for _, e := range c.engines {
		err := e.RefreshContainers(true)
		if err != nil {
			return err
		}
	}
	return nil
}

// RefreshEngine refreshes all containers in a specific engine
func (c *Cluster) RefreshEngine(hostname string) error {
	for _, e := range c.engines {
		if e.Name == hostname {
			err := e.RefreshContainers(true)
			if err != nil {
				return err
			}
			return nil
		}
	}
	return fmt.Errorf("no engine found with hostname %s", hostname)
}

// TagImage tags an image
func (c *Cluster) TagImage(IDOrName string, ref string, force bool) error {
	c.RLock()
	defer c.RUnlock()

	errs := []string{}
	var err error
	found := false
	for _, e := range c.engines {
		for _, image := range e.Images() {
			if image.Match(IDOrName, true) {
				found = true
				err := image.Engine.TagImage(IDOrName, ref, force)
				if err != nil {
					errs = append(errs, fmt.Sprintf("%s: %s", image.Engine.Name, err.Error()))
					continue
				}
			}
		}
	}
	if !found {
		return fmt.Errorf("No such image: %s", IDOrName)
	}
	if len(errs) > 0 {
		err = errors.New(strings.Join(errs, "\n"))
	}

	return err
}

// scale up or scale down may using different filter
func (c *Cluster) filterContainer(f []string, n int) (containers cluster.Containers) {
	log.Debugf("filter container: %v  n:%d", f, n)
	/*
	   case name==xxx filter container by container name
	   case service==xxx  may return multiple containers
	   other may filter container by container label
	*/
	isScaleService := hasPrifix(f, common.LabelKeyService)
	serviceApps := make(map[string]cluster.Containers)
	serviceAppSet := make(map[string]*cluster.Container)
	minNum := 1
	isContainersLeftBigger := false

	filters, err := parseFilterString(f)
	if err != nil {
		log.Errorf("parse Filter failed! %s", err)
		return nil
	}
	log.Debugf("got filters: %v", filters)

	if n > 0 {
		// scale up container
		for _, c := range c.Containers() {
			log.Debugln("container info: ", c.Names, c.Info.Config.Labels)
			if filterContainer(filters, c) {
				if !isScaleService {
					containers = append(containers, c)
					break
				} else {
					app, ok := c.Labels[common.LabelKeyApp]
					if ok {
						serviceAppSet[app] = c
					} else {
						log.Errorf("container has service label must has app label too!")
						return nil
					}
				}
			}
		}
	} else if n < 0 {
		n = -n
		//default app min number
		//scale down container
		if isScaleService {
			for _, c := range c.Containers() {
				if filterContainer(filters, c) {
					app, ok := c.Labels[common.LabelKeyApp]
					if !ok {
						log.Error("service must set app label")
						return nil
					}
					cs, ok := serviceApps[app]
					if !ok {
						serviceApps[app] = append(serviceApps[app], c)
					} else if len(cs) < n+getMinNum(cs[0].Config.Env) {
						serviceApps[app] = append(serviceApps[app], c)
					}
				}
			}
		} else {
			for _, c := range c.Containers() {
				if filterContainer(filters, c) {
					containers = append(containers, c)
					minNum = getMinNum(c.Config.Env)

					if len(containers) >= n+minNum {
						containers = containers[:n]
						log.Debugf("container num >= n + minNumber: %d", len(containers))
						isContainersLeftBigger = true
						break
					}
				}
			}
			if len(containers) < n+minNum && !isContainersLeftBigger {
				containers = containers[minNum:]
				log.Debugf("container num < n + minNumber: %d", len(containers))
			}
		}
	}

	//scale multiple containers in a same serivce, same apps scale once, scale up containers
	if len(serviceAppSet) != 0 {
		for _, v := range serviceAppSet {
			containers = append(containers, v)
		}
	}

	//scale down containers with service label
	if len(serviceApps) != 0 {
		for _, v := range serviceApps {
			minNum = getMinNum(v[0].Config.Env)
			if len(v) >= n+minNum {
				for _, vTemp := range v[:n] {
					containers = append(containers, vTemp)
				}
			} else if len(v) < n+minNum {
				for _, vTemp := range v[minNum:] {
					containers = append(containers, vTemp)
				}
			}
		}
	}

	log.Debugf("got scale containers:%v", containers)

	return containers
}

// generate task by scale config
/*
func (c *Cluster) product(scaleConfig common.ScaleAPI) (*scaleTask.Tasks, error) {
	tasks := &scaleTask.Tasks{*new([]scaleTask.Task), c}

	for _, item := range scaleConfig.Items {
		containers := c.filterContainer(item.Filters, item.Number)
	}
	return nil, nil
}
*/

func showContainers(cs cluster.Containers) {
	log.Infoln("\n\nFilter out containers:")
	for _, c := range cs {
		log.Infof("container name: %s\n", c.Names)
	}
	log.Infoln("\n\n")
}

// Scale containers
func (c *Cluster) Scale(scaleConfig common.ScaleAPI) []string {
	log.Debugf("swarm cluster scale: %v", scaleConfig)
	tasks := NewTasks(&LocalProcessor{c})

	for _, item := range scaleConfig.Items {
		log.Debugf("scale Item: %v", item)

		filter := NewFilter(c, &item)
		if filter == nil {
			continue
		}
		containers := filter.Filter()
		showContainers(containers)
		filter.AddTasks(tasks)
	}

	res, err := tasks.DoTasks()
	if err != nil {
		return res
	}

	return nil
}