docs/cluster/mesos/task/task.go

package task

import (
	"errors"
	"fmt"
	"strconv"
	"strings"

	log "github.com/Sirupsen/logrus"
	"github.com/docker/docker/pkg/stringid"
	"github.com/docker/swarm/cluster"
	"github.com/gogo/protobuf/proto"
	"github.com/mesos/mesos-go/mesosproto"
	"github.com/mesos/mesos-go/mesosutil"
)

// Task struct inherits from TaskInfo and represents a mesos task
type Task struct {
	mesosproto.TaskInfo

	updates chan *mesosproto.TaskStatus

	config    *cluster.ContainerConfig
	Error     chan error
	container chan *cluster.Container
	done      bool
}

// GetContainer returns the container channel from the task
// where the Swarm API sends the created container
func (t *Task) GetContainer() chan *cluster.Container {
	return t.container
}

// SetContainer writes on the container channel from the task
func (t *Task) SetContainer(container *cluster.Container) {
	t.container <- container
}

// GetConfig returns the container configuration of the task
func (t *Task) GetConfig() *cluster.ContainerConfig {
	return t.config
}

// ID method returns the taskId
func (t *Task) ID() string {
	return t.TaskId.GetValue()
}

// Stopped method returns a boolean determining if the task
// is done
func (t *Task) Stopped() bool {
	return t.done
}

// Stop method sets the boolean determining if the task is done
func (t *Task) Stop() {
	t.done = true
}

// Build method builds the task
func (t *Task) Build(slaveID string, offers map[string]*mesosproto.Offer) {
	t.Command = &mesosproto.CommandInfo{Shell: proto.Bool(false)}

	t.Container = &mesosproto.ContainerInfo{
		Type: mesosproto.ContainerInfo_DOCKER.Enum(),
		Docker: &mesosproto.ContainerInfo_DockerInfo{
			Image: &t.config.Image,
		},
	}

	if t.config.Hostname != "" {
		t.Container.Hostname = proto.String(t.config.Hostname)
		if t.config.Domainname != "" {
			t.Container.Hostname = proto.String(t.config.Hostname + "." + t.config.Domainname)
		}
	}

	switch t.config.HostConfig.NetworkMode {
	case "none":
		t.Container.Docker.Network = mesosproto.ContainerInfo_DockerInfo_NONE.Enum()
	case "host":
		t.Container.Docker.Network = mesosproto.ContainerInfo_DockerInfo_HOST.Enum()
	case "default", "bridge", "":
		var ports []uint64

		for _, offer := range offers {
			ports = append(ports, getPorts(offer)...)
		}

		for containerProtoPort, bindings := range t.config.HostConfig.PortBindings {
			for _, binding := range bindings {
				containerInfo := strings.SplitN(containerProtoPort, "/", 2)
				containerPort, err := strconv.ParseUint(containerInfo[0], 10, 32)
				if err != nil {
					log.Warn(err)
					continue
				}

				var hostPort uint64

				if binding.HostPort != "" {
					hostPort, err = strconv.ParseUint(binding.HostPort, 10, 32)
					if err != nil {
						log.Warn(err)
						continue
					}
				} else if len(ports) > 0 {
					hostPort = ports[0]
					ports = ports[1:]
				}

				if hostPort == 0 {
					log.Warn("cannot find port to bind on the host")
					continue
				}

				protocol := "tcp"
				if len(containerInfo) == 2 {
					protocol = containerInfo[1]
				}
				t.Container.Docker.PortMappings = append(t.Container.Docker.PortMappings, &mesosproto.ContainerInfo_DockerInfo_PortMapping{
					HostPort:      proto.Uint32(uint32(hostPort)),
					ContainerPort: proto.Uint32(uint32(containerPort)),
					Protocol:      proto.String(protocol),
				})
				t.Resources = append(t.Resources, mesosutil.NewRangesResource("ports", []*mesosproto.Value_Range{mesosutil.NewValueRange(hostPort, hostPort)}))
			}
		}
		// TODO handle -P here
		t.Container.Docker.Network = mesosproto.ContainerInfo_DockerInfo_BRIDGE.Enum()
	default:
		log.Errorf("Unsupported network mode %q", t.config.HostConfig.NetworkMode)
		t.Container.Docker.Network = mesosproto.ContainerInfo_DockerInfo_BRIDGE.Enum()
	}

	if cpus := t.config.CpuShares; cpus > 0 {
		t.Resources = append(t.Resources, mesosutil.NewScalarResource("cpus", float64(cpus)))
	}

	if mem := t.config.Memory; mem > 0 {
		t.Resources = append(t.Resources, mesosutil.NewScalarResource("mem", float64(mem/1024/1024)))
	}

	if len(t.config.Cmd) > 0 && t.config.Cmd[0] != "" {
		t.Command.Value = &t.config.Cmd[0]
	}

	if len(t.config.Cmd) > 1 {
		t.Command.Arguments = t.config.Cmd[1:]
	}

	for key, value := range t.config.Labels {
		t.Container.Docker.Parameters = append(t.Container.Docker.Parameters, &mesosproto.Parameter{Key: proto.String("label"), Value: proto.String(fmt.Sprintf("%s=%s", key, value))})
	}

	for _, value := range t.config.Env {
		t.Container.Docker.Parameters = append(t.Container.Docker.Parameters, &mesosproto.Parameter{Key: proto.String("env"), Value: proto.String(value)})
	}

	t.SlaveId = &mesosproto.SlaveID{Value: &slaveID}
}

// NewTask fucntion creates a task
func NewTask(config *cluster.ContainerConfig, name string) (*Task, error) {
	id := stringid.TruncateID(stringid.GenerateRandomID())

	if name != "" {
		id = name + "." + id
	}
	// save the name in labels as the mesos containerizer will override it
	config.Labels[cluster.SwarmLabelNamespace+".mesos.name"] = name
	// FIXME: once Mesos changes merged no need to save the task id to know which container we launched
	config.Labels[cluster.SwarmLabelNamespace+".mesos.task"] = id

	task := &Task{
		config:    config,
		container: make(chan *cluster.Container),
		Error:     make(chan error),
		updates:   make(chan *mesosproto.TaskStatus),
	}

	task.Name = &name
	task.TaskId = &mesosproto.TaskID{Value: &id}
	task.Labels = &mesosproto.Labels{Labels: []*mesosproto.Label{{Key: proto.String("SWARM_CONTAINER_NAME"), Value: &name}}}
	return task, nil
}

// SendStatus method writes the task status in the updates channel
func (t *Task) SendStatus(status *mesosproto.TaskStatus) {
	t.updates <- status
}

// GetStatus method reads the task status on the updates channel
func (t *Task) GetStatus() *mesosproto.TaskStatus {
	return <-t.updates
}

// Monitor method monitors task statuses
func (t *Task) Monitor() (bool, []byte, error) {
	taskStatus := t.GetStatus()

	switch taskStatus.GetState() {
	case mesosproto.TaskState_TASK_STAGING:
	case mesosproto.TaskState_TASK_STARTING:
	case mesosproto.TaskState_TASK_RUNNING:
	case mesosproto.TaskState_TASK_FINISHED:
		return true, taskStatus.Data, nil
	case mesosproto.TaskState_TASK_FAILED:
		errorMessage := taskStatus.GetMessage()
		if errorMessage == "Abnormal executor termination" {
			errorMessage += " : please verify your SWARM_MESOS_USER is correctly set"
		}
		return true, nil, errors.New(errorMessage)
	case mesosproto.TaskState_TASK_KILLED:
		return true, taskStatus.Data, nil
	case mesosproto.TaskState_TASK_LOST:
		return true, nil, errors.New(taskStatus.GetMessage())
	case mesosproto.TaskState_TASK_ERROR:
		return true, nil, errors.New(taskStatus.GetMessage())
	}

	return false, taskStatus.Data, nil
}