func (p *JujuProvisioner) heal(units []provision.Unit) {
	var inst instance
	coll := p.unitsCollection()
	for _, unit := range units {
		err := coll.FindId(unit.Name).One(&inst)
		if err != nil {
			coll.Insert(instance{UnitName: unit.Name, InstanceId: unit.InstanceId})
		} else if unit.InstanceId == inst.InstanceId {
			continue
		} else {
			format := "[juju] instance-id of unit %q changed from %q to %q. Healing."
			log.Printf(format, unit.Name, inst.InstanceId, unit.InstanceId)
			if p.elbSupport() {
				a := qApp{unit.AppName}
				manager := p.LoadBalancer()
				manager.Deregister(&a, provision.Unit{InstanceId: inst.InstanceId})
				err := manager.Register(&a, provision.Unit{InstanceId: unit.InstanceId})
				if err != nil {
					format := "[juju] Could not register instance %q in the load balancer: %s."
					log.Printf(format, unit.InstanceId, err)
					continue
				}
			}
			if inst.InstanceId != "pending" {
				msg := queue.Message{
					Action: app.RegenerateApprcAndStart,
					Args:   []string{unit.AppName, unit.Name},
				}
				app.Enqueue(msg)
			}
			inst.InstanceId = unit.InstanceId
			coll.UpdateId(unit.Name, inst)
		}
	}
}

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.Printf("error on getting hipache domin 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.Printf("error on add route for %s - %s", backendName, address)
		return &routeError{"add", err}
	}
	cname, err := r.getCName(backendName)
	if err != nil {
		log.Printf("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 handle(msg *queue.Message) {
	switch msg.Action {
	case app.RegenerateApprc:
		if len(msg.Args) < 1 {
			log.Printf("Error handling %q: this action requires at least 1 argument.", msg.Action)
			return
		}
		app, err := ensureAppIsStarted(msg)
		if err != nil {
			log.Print(err)
			return
		}
		app.SerializeEnvVars()
	case app.StartApp:
		if len(msg.Args) < 1 {
			log.Printf("Error handling %q: this action requires at least 1 argument.", msg.Action)
		}
		app, err := ensureAppIsStarted(msg)
		if err != nil {
			log.Print(err)
			return
		}
		err = app.Restart(ioutil.Discard)
		if err != nil {
			log.Printf("Error handling %q. App failed to start:\n%s.", msg.Action, err)
		}
	default:
		log.Printf("Error handling %q: invalid action.", msg.Action)
	}
}

// newContainer creates a new container in Docker and stores it in the database.
func newContainer(app provision.App, imageId string, cmds []string) (container, error) {
	cont := container{
		AppName: app.GetName(),
		Type:    app.GetPlatform(),
	}
	port, err := getPort()
	if err != nil {
		log.Printf("error on getting port for container %s - %s", cont.AppName, port)
		return container{}, err
	}
	config := docker.Config{
		Image:        imageId,
		Cmd:          cmds,
		PortSpecs:    []string{port},
		AttachStdin:  false,
		AttachStdout: false,
		AttachStderr: false,
	}
	hostID, c, err := dockerCluster().CreateContainer(&config)
	if err != nil {
		log.Printf("error on creating container in docker %s - %s", cont.AppName, err.Error())
		return container{}, err
	}
	cont.ID = c.ID
	cont.Port = port
	cont.HostAddr = getHostAddr(hostID)
	return cont, nil
}

func handle(msg *queue.Message) {
	if msg.Action == addUnitToLoadBalancer {
		if len(msg.Args) < 1 {
			log.Printf("Failed to handle %q: it requires at least one argument.", msg.Action)
			msg.Delete()
			return
		}
		a := qApp{name: msg.Args[0]}
		unitNames := msg.Args[1:]
		sort.Strings(unitNames)
		status, err := (&JujuProvisioner{}).collectStatus()
		if err != nil {
			log.Printf("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.Printf("Failed to handle %q: units not found.", msg.Action)
			msg.Delete()
			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).Release(msg, 0)
		} else {
			manager := ELBManager{}
			manager.Register(&a, ok...)
			msg.Delete()
			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)
			}
		}
	} else {
		msg.Delete()
	}
}

// replicateImage replicates the given image through all nodes in the cluster.
func replicateImage(name string) error {
	var buf bytes.Buffer
	if registry, err := config.GetString("docker:registry"); err == nil {
		if !strings.HasPrefix(name, registry) {
			name = registry + "/" + name
		}
		pushOpts := dclient.PushImageOptions{Name: name}
		for i := 0; i < maxTry; i++ {
			err = dockerCluster().PushImage(pushOpts, dclient.AuthConfiguration{}, &buf)
			if err == nil {
				buf.Reset()
				break
			}
			log.Printf("[docker] Failed to push image %q (%s): %s", name, err, buf.String())
			buf.Reset()
		}
		if err != nil {
			return err
		}
		pullOpts := dclient.PullImageOptions{Repository: name}
		for i := 0; i < maxTry; i++ {
			err = dockerCluster().PullImage(pullOpts, &buf)
			if err == nil {
				break
			}
			buf.Reset()
		}
		if err != nil {
			log.Printf("[docker] Failed to replicate image %q through nodes (%s): %s", name, err, buf.String())
			return err
		}
	}
	return nil
}

func (h *MessageHandler) handle(msg queue.Message) {
	if msg.Visits >= MaxVisits {
		log.Printf("Error handling %q: this message has been visited more than %d times.", msg.Action, MaxVisits)
		return
	}
	switch msg.Action {
	case app.RegenerateApprc:
		if len(msg.Args) < 1 {
			log.Printf("Error handling %q: this action requires at least 1 argument.", msg.Action)
			return
		}
		app, err := h.ensureAppIsStarted(msg)
		if err != nil {
			log.Print(err)
			return
		}
		app.SerializeEnvVars()
	case app.StartApp:
		if len(msg.Args) < 1 {
			log.Printf("Error handling %q: this action requires at least 1 argument.", msg.Action)
		}
		app, err := h.ensureAppIsStarted(msg)
		if err != nil {
			log.Print(err)
			return
		}
		err = app.Restart(ioutil.Discard)
		if err != nil {
			log.Printf("Error handling %q. App failed to start:\n%s.", msg.Action, err)
		}
	default:
		log.Printf("Error handling %q: invalid action.", msg.Action)
	}
}

// Execute executes the pipeline.
//
// The execution starts in the forward phase, calling the Forward function of
// all actions. If none of the Forward calls return error, the pipeline
// execution ends in the forward phase and is "committed".
//
// If any of the Forward call fail, the executor switches to the backward phase
// (roll back) and call the Backward function for each action completed. It
// does not call the Backward function of the action that has failed.
//
// After rolling back all completed actions, it returns the original error
// returned by the action that failed.
func (p *Pipeline) Execute(params ...interface{}) error {
	var (
		r   Result
		err error
	)
	if len(p.actions) == 0 {
		return errors.New("No actions to execute.")
	}
	fwCtx := FWContext{Params: params}
	for i, a := range p.actions {
		log.Printf("[pipeline] running the Forward for the %s action", a.Name)
		if a.Forward == nil {
			err = errors.New("All actions must define the forward function.")
		} else if len(fwCtx.Params) < a.MinParams {
			err = errors.New("Not enough parameters to call Action.Forward.")
		} else {
			r, err = a.Forward(fwCtx)
			a.rMutex.Lock()
			a.result = r
			a.rMutex.Unlock()
			fwCtx.Previous = r
		}
		if err != nil {
			log.Printf("[pipeline] error running the Forward for the %s action - %s", a.Name, err.Error())
			p.rollback(i-1, params)
			return err
		}
	}
	return nil
}

func update(units []provision.Unit) {
	log.Print("updating status from provisioner")
	var l AppList
	for _, unit := range units {
		a, index := l.Search(unit.AppName)
		if index > -1 {
			err := a.Get()
			if err != nil {
				log.Printf("collector: app %q not found. Skipping.\n", unit.AppName)
				continue
			}
		}
		u := app.Unit{}
		u.Name = unit.Name
		u.Type = unit.Type
		u.Machine = unit.Machine
		u.InstanceId = unit.InstanceId
		u.Ip = unit.Ip
		u.State = string(unit.Status)
		a.AddUnit(&u)
		if index > -1 {
			l.Add(a, index)
		}
	}
	conn, err := db.Conn()
	if err != nil {
		log.Printf("collector failed to connect to the database: %s", err)
		return
	}
	defer conn.Close()
	for _, a := range l {
		a.Ip, _ = app.Provisioner.Addr(a)
		conn.Apps().Update(bson.M{"name": a.Name}, a)
	}
}

// ip returns the ip for the container.
func (c *container) ip() (string, error) {
	docker, err := config.GetString("docker:binary")
	if err != nil {
		return "", err
	}
	log.Printf("Getting ipaddress to instance %s", c.id)
	instanceJson, err := runCmd(docker, "inspect", c.id)
	if err != nil {
		msg := "error(%s) trying to inspect docker instance(%s) to get ipaddress"
		log.Printf(msg, err)
		return "", errors.New(msg)
	}
	var result map[string]interface{}
	if err := json.Unmarshal([]byte(instanceJson), &result); err != nil {
		msg := "error(%s) parsing json from docker when trying to get ipaddress"
		log.Printf(msg, err)
		return "", errors.New(msg)
	}
	if ns, ok := result["NetworkSettings"]; !ok || ns == nil {
		msg := "Error when getting container information. NetworkSettings is missing."
		log.Printf(msg)
		return "", errors.New(msg)
	}
	networkSettings := result["NetworkSettings"].(map[string]interface{})
	instanceIp := networkSettings["IpAddress"].(string)
	if instanceIp == "" {
		msg := "error: Can't get ipaddress..."
		log.Print(msg)
		return "", errors.New(msg)
	}
	log.Printf("Instance IpAddress: %s", instanceIp)
	return instanceIp, nil
}

func collectUnit(container container, units chan<- provision.Unit, wg *sync.WaitGroup) {
	defer wg.Done()
	unit := provision.Unit{
		Name:    container.ID,
		AppName: container.AppName,
		Type:    container.Type,
	}
	switch container.Status {
	case "error":
		unit.Status = provision.StatusError
		units <- unit
		return
	case "created":
		return
	}
	unit.Ip = container.HostAddr
	if ip, hostPort, err := container.networkInfo(); err == nil &&
		(hostPort != container.HostPort || ip != container.IP) {
		err = fixContainer(&container, ip, hostPort)
		if err != nil {
			log.Printf("error on fix container hostport for [container %s]", container.ID)
			return
		}
	}
	addr := strings.Replace(container.getAddress(), "http://", "", 1)
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		unit.Status = provision.StatusInstalling
	} else {
		conn.Close()
		unit.Status = provision.StatusStarted
	}
	log.Printf("collected data for [container %s] - [app %s]", container.ID, container.AppName)
	units <- unit
}

func flatten(imageID string) error {
	config := docker.Config{
		Image:        imageID,
		Cmd:          []string{"/bin/bash"},
		AttachStdin:  false,
		AttachStdout: false,
		AttachStderr: false,
	}
	_, c, err := dockerCluster().CreateContainer(&config)
	if err != nil {
		return err
	}
	buf := &bytes.Buffer{}
	if err := dockerCluster().ExportContainer(c.ID, buf); err != nil {
		log.Printf("Flatten: Caugh error while exporting container %s: %s", c.ID, err.Error())
		return err
	}
	opts := dcli.ImportImageOptions{Repository: imageID, Source: "-"}
	if err := dockerCluster().ImportImage(opts, buf); err != nil {
		log.Printf("Flatten: Caugh error while importing image from container %s: %s", c.ID, err.Error())
		return err
	}
	if err := dockerCluster().RemoveContainer(c.ID); err != nil {
		log.Printf("Flatten: Caugh error while removing container %s: %s", c.ID, err.Error())
	}
	removeFromRegistry(imageID)
	return nil
}

// Heal executes the action for heal the bootstrap machine agent.
func (h bootstrapMachineHealer) Heal() error {
	if h.needsHeal() {
		bootstrapMachine := getBootstrapMachine()
		log.Printf("Healing bootstrap juju-machine-agent")
		upStartCmd("stop", "juju-machine-agent", bootstrapMachine.IPAddress)
		return upStartCmd("start", "juju-machine-agent", bootstrapMachine.IPAddress)
	}
	log.Printf("Bootstrap juju-machine-agent needs no cure, skipping...")
	return nil
}

// getPrivateDns returns the private dns for an instance.
func (h *instanceAgentsConfigHealer) getPrivateDns(instanceId string) (string, error) {
	log.Printf("getting dns for %s", instanceId)
	resp, err := h.ec2().Instances([]string{instanceId}, nil)
	if err != nil {
		log.Printf("error in gettings dns for %s", instanceId)
		log.Print(err)
		return "", err
	}
	dns := resp.Reservations[0].Instances[0].PrivateDNSName
	return dns, nil
}

// stop stops a docker container.
func (c *container) stop() error {
	docker, err := config.GetString("docker:binary")
	if err != nil {
		return err
	}
	//TODO: better error handling
	log.Printf("trying to stop instance %s", c.id)
	output, err := runCmd(docker, "stop", c.id)
	log.Printf("docker stop=%s", output)
	return err
}

func removeContainer(c *container) error {
	err := c.stop()
	if err != nil {
		log.Printf("error on stop unit %s - %s", c.ID, err)
	}
	err = c.remove()
	if err != nil {
		log.Printf("error on remove container %s - %s", c.ID, err)
	}
	return err
}

func (p *LocalProvisioner) install(ip string) error {
	log.Printf("executing the install hook for %s", ip)
	cmd := exec.Command("ssh", "-q", "-o", "StrictHostKeyChecking no", "-l", "ubuntu", ip, "sudo /var/lib/tsuru/hooks/install")
	err := cmd.Run()
	if err != nil {
		log.Printf("error on install for %s", ip)
		log.Print(err)
		return err
	}
	return nil
}

func (u *User) sendResetPassword(t *passwordToken) {
	var body bytes.Buffer
	err := resetEmailData.Execute(&body, t)
	if err != nil {
		log.Printf("Failed to send password token to user %q: %s", u.Email, err)
		return
	}
	err = sendEmail(u.Email, body.Bytes())
	if err != nil {
		log.Printf("Failed to send password token for user %q: %s", u.Email, err)
	}
}

func (p *LXCProvisioner) install(ip string) error {
	log.Printf("executing the install hook for %s", ip)
	cmd := "ssh"
	args := []string{"-q", "-o", "StrictHostKeyChecking no", "-l", "ubuntu", ip, "sudo /var/lib/tsuru/hooks/install"}
	err := executor().Execute(cmd, args, nil, nil, nil)
	if err != nil {
		log.Printf("error on install for %s", ip)
		log.Print(err)
		return err
	}
	return nil
}

func (p *LXCProvisioner) Destroy(app provision.App) error {
	c := container{name: app.GetName()}
	go func(c container) {
		log.Printf("stoping container %s", c.name)
		c.stop()
		log.Printf("destroying container %s", c.name)
		c.destroy()
		log.Printf("removing container %s from the database", c.name)
		p.collection().Remove(bson.M{"name": c.name})
	}(c)
	return nil
}