func addContainersWithHost(args *changeUnitsPipelineArgs) ([]container, error) {
	a := args.app
	w := args.writer
	var units int
	processMsg := make([]string, 0, len(args.toAdd))
	for processName, v := range args.toAdd {
		units += v.Quantity
		processMsg = append(processMsg, fmt.Sprintf("[%s: %d]", processName, v.Quantity))
	}
	imageId := args.imageId
	var destinationHost []string
	if args.toHost != "" {
		destinationHost = []string{args.toHost}
	}
	if w == nil {
		w = ioutil.Discard
	}
	var plural string
	if units > 1 {
		plural = "s"
	}
	fmt.Fprintf(w, "\n---- Starting new unit%s %s ----\n", plural, strings.Join(processMsg, " "))
	oldContainers := make([]container, 0, units)
	for processName, cont := range args.toAdd {
		for i := 0; i < cont.Quantity; i++ {
			oldContainers = append(oldContainers, container{ProcessName: processName, Status: cont.Status.String()})
		}
	}
	rollbackCallback := func(c *container) {
		log.Errorf("Removing container %q due failed add units.", c.ID)
		errRem := args.provisioner.removeContainer(c, a)
		if errRem != nil {
			log.Errorf("Unable to destroy container %q: %s", c.ID, errRem)
		}
	}
	var (
		createdContainers []*container
		m                 sync.Mutex
	)
	err := runInContainers(oldContainers, func(c *container, _ chan *container) error {
		c, err := args.provisioner.start(c, a, imageId, w, destinationHost...)
		if err != nil {
			return err
		}
		m.Lock()
		createdContainers = append(createdContainers, c)
		m.Unlock()
		return nil
	}, rollbackCallback, true)
	if err != nil {
		return nil, err
	}
	result := make([]container, len(createdContainers))
	i := 0
	for _, c := range createdContainers {
		result[i] = *c
		i++
	}
	return result, nil
}

func deploy(app provision.App, version string, w io.Writer) (string, error) {
	commands, err := deployCmds(app, version)
	if err != nil {
		return "", err
	}
	imageId := getImage(app)
	actions := []*action.Action{&insertEmptyContainerInDB, &createContainer, &startContainer, &updateContainerInDB}
	pipeline := action.NewPipeline(actions...)
	err = pipeline.Execute(app, imageId, commands)
	if err != nil {
		log.Errorf("error on execute deploy pipeline for app %s - %s", app.GetName(), err)
		return "", err
	}
	c := pipeline.Result().(container)
	err = c.logs(w)
	if err != nil {
		log.Errorf("error on get logs for container %s - %s", c.ID, err)
		return "", err
	}
	_, err = dockerCluster().WaitContainer(c.ID)
	if err != nil {
		log.Errorf("Process failed for container %q: %s", c.ID, err)
		return "", err
	}
	imageId, err = c.commit()
	if err != nil {
		log.Errorf("error on commit container %s - %s", c.ID, err)
		return "", err
	}
	c.remove()
	return imageId, nil
}

func (r *hipacheRouter) AddRoute(name string, address *url.URL) error {
	backendName, err := router.Retrieve(name)
	if err != nil {
		return err
	}
	domain, err := config.GetString(r.prefix + ":domain")
	if err != nil {
		log.Errorf("error on getting hipache domain in add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	frontend := "frontend:" + backendName + "." + domain
	if err := r.addRoute(frontend, address.String()); err != nil {
		log.Errorf("error on add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	cnames, err := r.getCNames(backendName)
	if err != nil {
		log.Errorf("error on get cname in add route for %s - %s", backendName, address)
		return err
	}
	if cnames == nil {
		return nil
	}
	for _, cname := range cnames {
		err = r.addRoute("frontend:"+cname, address.String())
		if err != nil {
			return err
		}
	}
	return nil
}

func (p *dockerProvisioner) Destroy(app provision.App) error {
	containers, err := listContainersByApp(app.GetName())
	if err != nil {
		log.Errorf("Failed to list app containers: %s", err.Error())
		return err
	}
	var containersGroup sync.WaitGroup
	containersGroup.Add(len(containers))
	for _, c := range containers {
		go func(c container) {
			defer containersGroup.Done()
			err := removeContainer(&c)
			if err != nil {
				log.Errorf("Unable to destroy container %s: %s", c.ID, err.Error())
			}
		}(c)
	}
	containersGroup.Wait()
	err = removeImage(assembleImageName(app.GetName()))
	if err != nil {
		log.Errorf("Failed to remove image: %s", err.Error())
	}
	r, err := getRouter()
	if err != nil {
		log.Errorf("Failed to get router: %s", err.Error())
		return err
	}
	err = r.RemoveBackend(app.GetName())
	if err != nil {
		log.Errorf("Failed to remove route backend: %s", err.Error())
		return err
	}
	return nil
}

func notify(appName string, messages []interface{}) {
	factory, err := queue.Factory()
	if err != nil {
		log.Errorf("Error on logs notify: %s", err.Error())
		return
	}
	q, err := factory.Get(logQueueName(appName))
	if err != nil {
		log.Errorf("Error on logs notify: %s", err.Error())
		return
	}
	pubSubQ, ok := q.(queue.PubSubQ)
	if !ok {
		log.Debugf("Queue does not support pubsub, logs only in database.")
		return
	}
	for _, msg := range messages {
		bytes, err := json.Marshal(msg)
		if err != nil {
			log.Errorf("Error on logs notify: %s", err.Error())
			continue
		}
		err = pubSubQ.Pub(bytes)
		if err != nil {
			log.Errorf("Error on logs notify: %s", err.Error())
		}
	}
}

func (l *lockUpdater) spin() {
	set := map[Target]struct{}{}
	for {
		select {
		case added := <-l.addCh:
			set[*added] = struct{}{}
		case removed := <-l.removeCh:
			delete(set, *removed)
		case <-l.stopCh:
			return
		case <-time.After(lockUpdateInterval):
		}
		conn, err := db.Conn()
		if err != nil {
			log.Errorf("[events] [lock update] error getting db conn: %s", err)
			continue
		}
		coll := conn.Events()
		slice := make([]interface{}, len(set))
		i := 0
		for id := range set {
			slice[i], _ = id.GetBSON()
			i++
		}
		err = coll.Update(bson.M{"_id": bson.M{"$in": slice}}, bson.M{"$set": bson.M{"lockupdatetime": time.Now().UTC()}})
		if err != nil && err != mgo.ErrNotFound {
			log.Errorf("[events] [lock update] error updating: %s", err)
		}
		conn.Close()
	}
}

func (r hipacheRouter) AddRoute(name, address string) error {
	backendName, err := router.Retrieve(name)
	if err != nil {
		return err
	}
	domain, err := config.GetString("hipache:domain")
	if err != nil {
		log.Errorf("error on getting hipache domain in add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	frontend := "frontend:" + backendName + "." + domain
	if err := r.addRoute(frontend, address); err != nil {
		log.Errorf("error on add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	cname, err := r.getCName(backendName)
	if err != nil {
		log.Errorf("error on get cname in add route for %s - %s", backendName, address)
		return err
	}
	if cname == "" {
		return nil
	}
	return r.addRoute("frontend:"+cname, address)
}

func runRoutesRebuildOnce(appName string, lock bool) bool {
	if appFinder == nil {
		return false
	}
	a, err := appFinder(appName)
	if err != nil {
		log.Errorf("[routes-rebuild-task] error getting app %q: %s", appName, err)
		return false
	}
	if a == nil {
		return true
	}
	if lock {
		var locked bool
		locked, err = a.InternalLock("rebuild-routes-task")
		if err != nil || !locked {
			return false
		}
		defer a.Unlock()
	}
	if err != nil {
		log.Errorf("[routes-rebuild-task] error getting app %q: %s", appName, err)
		return false
	}
	_, err = RebuildRoutes(a)
	if err != nil {
		log.Errorf("[routes-rebuild-task] error rebuilding app %q: %s", appName, err)
		return false
	}
	return true
}

// handle is the function called by the queue handler on each message.
func handle(msg *queue.Message) {
	switch msg.Action {
	case regenerateApprc:
		if len(msg.Args) < 1 {
			log.Errorf("Error handling %q: this action requires at least 1 argument.", msg.Action)
			return
		}
		app, err := ensureAppIsStarted(msg)
		if err != nil {
			log.Error(err.Error())
			return
		}
		err = app.SerializeEnvVars()
		if err != nil {
			log.Error(err.Error())
		}
	case BindService:
		err := bindUnit(msg)
		if err != nil {
			log.Error(err.Error())
			return
		}
	default:
		log.Errorf("Error handling %q: invalid action.", msg.Action)
	}
}

// Rename renames a repository.
func Rename(oldName, newName string) error {
	log.Debugf("Renaming repository %q to %q", oldName, newName)
	repo, err := Get(oldName)
	if err != nil {
		log.Errorf("repository.Rename: Repository %q not found: %s", oldName, err)
		return err
	}
	newRepo := repo
	newRepo.Name = newName
	conn, err := db.Conn()
	if err != nil {
		return err
	}
	defer conn.Close()
	err = conn.Repository().Insert(newRepo)
	if err != nil {
		log.Errorf("repository.Rename: Error adding new repository %q: %s", newName, err)
		return err
	}
	err = conn.Repository().RemoveId(oldName)
	if err != nil {
		log.Errorf("repository.Rename: Error removing old repository %q: %s", oldName, err)
		return err
	}
	return fs.Filesystem().Rename(barePath(oldName), barePath(newName))
}

// New creates a representation of a git repository. It creates a Git
// repository using the "bare-dir" setting and saves repository's meta data in
// the database.
func New(name string, users []string, isPublic bool) (*Repository, error) {
	log.Debugf("Creating repository %q", name)
	r := &Repository{Name: name, Users: users, IsPublic: isPublic}
	if v, err := r.isValid(); !v {
		log.Errorf("repository.New: Invalid repository %q: %s", name, err)
		return r, err
	}
	if err := newBare(name); err != nil {
		log.Errorf("repository.New: Error creating bare repository for %q: %s", name, err)
		return r, err
	}
	barePath := barePath(name)
	if barePath != "" && isPublic {
		ioutil.WriteFile(barePath+"/git-daemon-export-ok", []byte(""), 0644)
		if f, err := fs.Filesystem().Create(barePath + "/git-daemon-export-ok"); err == nil {
			f.Close()
		}
	}
	conn, err := db.Conn()
	if err != nil {
		return nil, err
	}
	defer conn.Close()
	err = conn.Repository().Insert(&r)
	if mgo.IsDup(err) {
		log.Errorf("repository.New: Duplicate repository %q", name)
		return r, fmt.Errorf("A repository with this name already exists.")
	}
	return r, err
}

// RemoveUnits removes n units from the app. It's a process composed of
// multiple steps:
//
//     1. Remove units from the provisioner
//     2. Remove units from the app list
//     3. Update quota
func (app *App) RemoveUnits(n uint) error {
	if n == 0 {
		ReleaseApplicationLock(app.Name)
		return stderr.New("Cannot remove zero units.")
	} else if l := uint(len(app.Units())); l == n {
		ReleaseApplicationLock(app.Name)
		return stderr.New("Cannot remove all units from an app.")
	} else if n > l {
		ReleaseApplicationLock(app.Name)
		return fmt.Errorf("Cannot remove %d units from this app, it has only %d units.", n, l)
	}
	go func() {
		defer ReleaseApplicationLock(app.Name)
		Provisioner.RemoveUnits(app, n)
		conn, err := db.Conn()
		if err != nil {
			log.Errorf("Error: %s", err)
		}
		defer conn.Close()
		dbErr := conn.Apps().Update(
			bson.M{"name": app.Name},
			bson.M{
				"$set": bson.M{
					"quota.inuse": len(app.Units()),
				},
			},
		)
		if dbErr != nil {
			log.Errorf("Error: %s", dbErr)
		}
	}()
	return nil
}

// Deploy runs a deployment of an application. It will first try to run an
// archive based deploy (if opts.ArchiveURL is not empty), and then fallback to
// the Git based deployment.
func Deploy(opts DeployOptions) error {
	var outBuffer bytes.Buffer
	start := time.Now()
	logWriter := LogWriter{App: opts.App}
	logWriter.Async()
	defer logWriter.Close()
	writer := io.MultiWriter(&tsuruIo.NoErrorWriter{Writer: opts.OutputStream}, &outBuffer, &logWriter)
	imageId, err := deployToProvisioner(&opts, writer)
	elapsed := time.Since(start)
	saveErr := saveDeployData(&opts, imageId, outBuffer.String(), elapsed, err)
	if saveErr != nil {
		log.Errorf("WARNING: couldn't save deploy data, deploy opts: %#v", opts)
	}
	if err != nil {
		return err
	}
	err = incrementDeploy(opts.App)
	if err != nil {
		log.Errorf("WARNING: couldn't increment deploy count, deploy opts: %#v", opts)
	}
	if opts.App.UpdatePlatform == true {
		opts.App.SetUpdatePlatform(false)
	}
	return nil
}

// New creates a representation of a git repository. It creates a Git
// repository using the "bare-dir" setting and saves repository's meta data in
// the database.
func New(name string, users, readOnlyUsers []string, isPublic bool) (*Repository, error) {
	log.Debugf("Creating repository %q", name)
	r := &Repository{Name: name, Users: users, ReadOnlyUsers: readOnlyUsers, IsPublic: isPublic}
	if v, err := r.isValid(); !v {
		log.Errorf("repository.New: Invalid repository %q: %s", name, err)
		return r, err
	}
	conn, err := db.Conn()
	if err != nil {
		return nil, err
	}
	defer conn.Close()
	err = conn.Repository().Insert(r)
	if err != nil {
		if mgo.IsDup(err) {
			return nil, ErrRepositoryAlreadyExists
		}
		return nil, err
	}
	if err = newBare(name); err != nil {
		log.Errorf("repository.New: Error creating bare repository for %q: %s", name, err)
		conn.Repository().Remove(bson.M{"_id": r.Name})
		return r, err
	}
	barePath := barePath(name)
	if barePath != "" && isPublic {
		if f, createErr := fs.Filesystem().Create(barePath + "/git-daemon-export-ok"); createErr == nil {
			f.Close()
		}
	}
	return r, err
}

func syncRepositoryApps(user *auth.User, beforeApps []string, roleCache map[string]*permission.Role) error {
	err := user.Reload()
	if err != nil {
		return err
	}
	afterApps, err := deployableApps(user, roleCache)
	if err != nil {
		return err
	}
	afterMap := map[string]struct{}{}
	for _, a := range afterApps {
		afterMap[a] = struct{}{}
	}
	manager := repository.Manager()
	for _, a := range beforeApps {
		var err error
		if _, ok := afterMap[a]; !ok {
			err = manager.RevokeAccess(a, user.Email)
		}
		if err != nil {
			log.Errorf("error revoking gandalf access for app %s, user %s: %s", a, user.Email, err)
		}
	}
	for _, a := range afterApps {
		err := manager.GrantAccess(a, user.Email)
		if err != nil {
			log.Errorf("error granting gandalf access for app %s, user %s: %s", a, user.Email, err)
		}
	}
	return nil

}

func (app *App) BindUnit(unit *provision.Unit) error {
	instances, err := app.serviceInstances()
	if err != nil {
		return err
	}
	var i int
	var instance service.ServiceInstance
	for i, instance = range instances {
		err = instance.BindUnit(app, unit)
		if err != nil {
			log.Errorf("Error binding the unit %s with the service instance %s: %s", unit.ID, instance.Name, err)
			break
		}
	}
	if err != nil {
		for j := i - 1; j >= 0; j-- {
			instance = instances[j]
			rollbackErr := instance.UnbindUnit(app, unit)
			if rollbackErr != nil {
				log.Errorf("Error unbinding unit %s with the service instance %s during rollback: %s", unit.ID, instance.Name, rollbackErr)
			}
		}
	}
	return err
}

func (h *NodeHealer) tryHealingNode(node provision.Node, reason string, lastCheck *NodeChecks) error {
	_, hasIaas := node.Metadata()["iaas"]
	if !hasIaas {
		log.Debugf("node %q doesn't have IaaS information, healing (%s) won't run on it.", node.Address(), reason)
		return nil
	}
	poolName := node.Metadata()[poolMetadataName]
	evt, err := event.NewInternal(&event.Opts{
		Target:       event.Target{Type: event.TargetTypeNode, Value: node.Address()},
		InternalKind: "healer",
		CustomData: NodeHealerCustomData{
			Node:      provision.NodeToSpec(node),
			Reason:    reason,
			LastCheck: lastCheck,
		},
		Allowed: event.Allowed(permission.PermPoolReadEvents, permission.Context(permission.CtxPool, poolName)),
	})
	if err != nil {
		if _, ok := err.(event.ErrEventLocked); ok {
			// Healing in progress.
			return nil
		}
		return errors.Wrap(err, "Error trying to insert node healing event, healing aborted")
	}
	var createdNode *provision.NodeSpec
	var evtErr error
	defer func() {
		var updateErr error
		if evtErr == nil && createdNode == nil {
			updateErr = evt.Abort()
		} else {
			updateErr = evt.DoneCustomData(evtErr, createdNode)
		}
		if updateErr != nil {
			log.Errorf("error trying to update healing event: %s", updateErr)
		}
	}()
	_, err = node.Provisioner().GetNode(node.Address())
	if err != nil {
		if err == provision.ErrNodeNotFound {
			return nil
		}
		evtErr = errors.Wrap(err, "unable to check if node still exists")
		return evtErr
	}
	shouldHeal, err := h.shouldHealNode(node)
	if err != nil {
		evtErr = errors.Wrap(err, "unable to check if node should be healed")
		return evtErr
	}
	if !shouldHeal {
		return nil
	}
	log.Errorf("initiating healing process for node %q due to: %s", node.Address(), reason)
	createdNode, evtErr = h.healNode(node)
	return evtErr
}

func handle(msg *queue.Message) {
	if msg.Action == addUnitToLoadBalancer {
		if len(msg.Args) < 1 {
			log.Errorf("Failed to handle %q: it requires at least one argument.", msg.Action)
			return
		}
		a := qApp{name: msg.Args[0]}
		unitNames := msg.Args[1:]
		sort.Strings(unitNames)
		status, err := (&JujuProvisioner{}).collectStatus()
		if err != nil {
			log.Errorf("Failed to handle %q: juju status failed.\n%s.", msg.Action, err)
			return
		}
		var units []provision.Unit
		for _, u := range status {
			if u.AppName != a.name {
				continue
			}
			n := sort.SearchStrings(unitNames, u.Name)
			if len(unitNames) == 0 ||
				n < len(unitNames) && unitNames[n] == u.Name {
				units = append(units, u)
			}
		}
		if len(units) == 0 {
			log.Errorf("Failed to handle %q: units not found.", msg.Action)
			return
		}
		var noID []string
		var ok []provision.Unit
		for _, u := range units {
			if u.InstanceId == "pending" || u.InstanceId == "" {
				noID = append(noID, u.Name)
			} else {
				ok = append(ok, u)
			}
		}
		if len(noID) == len(units) {
			getQueue(queueName).Put(msg, 0)
		} else {
			router, _ := Router()
			for _, u := range units {
				router.AddRoute(a.GetName(), u.InstanceId)
			}
			if len(noID) > 0 {
				args := []string{a.name}
				args = append(args, noID...)
				msg := queue.Message{
					Action: msg.Action,
					Args:   args,
				}
				getQueue(queueName).Put(&msg, 1e9)
			}
		}
	}
}

func (s *segregatedScheduler) filterByMemoryUsage(a *app.App, nodes []cluster.Node, maxMemoryRatio float32, TotalMemoryMetadata string) ([]cluster.Node, error) {
	if maxMemoryRatio == 0 || TotalMemoryMetadata == "" {
		return nodes, nil
	}
	hosts := make([]string, len(nodes))
	for i := range nodes {
		hosts[i] = urlToHost(nodes[i].Address)
	}
	containers, err := s.provisioner.ListContainers(bson.M{"hostaddr": bson.M{"$in": hosts}, "id": bson.M{"$nin": s.ignoredContainers}})
	if err != nil {
		return nil, err
	}
	hostReserved := make(map[string]int64)
	for _, cont := range containers {
		a, err := app.GetByName(cont.AppName)
		if err != nil {
			return nil, err
		}
		hostReserved[cont.HostAddr] += a.Plan.Memory
	}
	megabyte := float64(1024 * 1024)
	nodeList := make([]cluster.Node, 0, len(nodes))
	for _, node := range nodes {
		totalMemory, _ := strconv.ParseFloat(node.Metadata[TotalMemoryMetadata], 64)
		shouldAdd := true
		if totalMemory != 0 {
			maxMemory := totalMemory * float64(maxMemoryRatio)
			host := urlToHost(node.Address)
			nodeReserved := hostReserved[host] + a.Plan.Memory
			if nodeReserved > int64(maxMemory) {
				shouldAdd = false
				tryingToReserveMB := float64(a.Plan.Memory) / megabyte
				reservedMB := float64(hostReserved[host]) / megabyte
				limitMB := maxMemory / megabyte
				log.Errorf("Node %q has reached its memory limit. "+
					"Limit %0.4fMB. Reserved: %0.4fMB. Needed additional %0.4fMB",
					host, limitMB, reservedMB, tryingToReserveMB)
			}
		}
		if shouldAdd {
			nodeList = append(nodeList, node)
		}
	}
	if len(nodeList) == 0 {
		autoScaleEnabled, _ := config.GetBool("docker:auto-scale:enabled")
		errMsg := fmt.Sprintf("no nodes found with enough memory for container of %q: %0.4fMB",
			a.Name, float64(a.Plan.Memory)/megabyte)
		if autoScaleEnabled {
			// Allow going over quota temporarily because auto-scale will be
			// able to detect this and automatically add a new nodes.
			log.Errorf("WARNING: %s. Will ignore memory restrictions.", errMsg)
			return nodes, nil
		}
		return nil, errors.New(errMsg)
	}
	return nodeList, nil
}

// ReleaseApplicationLock releases a lock hold on an app, currently it's called
// by a middleware, however, ideally, it should be called individually by each
// handler since they might be doing operations in background.
func ReleaseApplicationLock(appName string) {
	conn, err := db.Conn()
	if err != nil {
		log.Errorf("Error getting DB, couldn't unlock %s: %s", appName, err.Error())
	}
	defer conn.Close()
	err = conn.Apps().Update(bson.M{"name": appName, "lock.locked": true}, bson.M{"$set": bson.M{"lock": AppLock{}}})
	if err != nil {
		log.Errorf("Error updating entry, couldn't unlock %s: %s", appName, err.Error())
	}
}