marathon.go

package provider

import (
	"errors"
	"net/url"
	"strconv"
	"strings"
	"text/template"

	"math"
	"net/http"
	"time"

	"github.com/BurntSushi/ty/fun"
	"github.com/cenk/backoff"
	"github.com/containous/traefik/job"
	"github.com/containous/traefik/log"
	"github.com/containous/traefik/safe"
	"github.com/containous/traefik/types"
	"github.com/gambol99/go-marathon"
)

var _ Provider = (*Marathon)(nil)

// Marathon holds configuration of the Marathon provider.
type Marathon struct {
	BaseProvider
	Endpoint                string     `description:"Marathon server endpoint. You can also specify multiple endpoint for Marathon"`
	Domain                  string     `description:"Default domain used"`
	ExposedByDefault        bool       `description:"Expose Marathon apps by default"`
	GroupsAsSubDomains      bool       `description:"Convert Marathon groups to subdomains"`
	DCOSToken               string     `description:"DCOSToken for DCOS environment, This will override the Authorization header"`
	MarathonLBCompatibility bool       `description:"Add compatibility with marathon-lb labels"`
	TLS                     *ClientTLS `description:"Enable Docker TLS support"`
	Basic                   *MarathonBasic
	marathonClient          marathon.Marathon
}

// MarathonBasic holds basic authentication specific configurations
type MarathonBasic struct {
	HTTPBasicAuthUser string
	HTTPBasicPassword string
}

type lightMarathonClient interface {
	AllTasks(v url.Values) (*marathon.Tasks, error)
	Applications(url.Values) (*marathon.Applications, error)
}

// Provide allows the provider to provide configurations to traefik
// using the given configuration channel.
func (provider *Marathon) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool, constraints types.Constraints) error {
	provider.Constraints = append(provider.Constraints, constraints...)
	operation := func() error {
		config := marathon.NewDefaultConfig()
		config.URL = provider.Endpoint
		config.EventsTransport = marathon.EventsTransportSSE
		if provider.Basic != nil {
			config.HTTPBasicAuthUser = provider.Basic.HTTPBasicAuthUser
			config.HTTPBasicPassword = provider.Basic.HTTPBasicPassword
		}
		if len(provider.DCOSToken) > 0 {
			config.DCOSToken = provider.DCOSToken
		}
		TLSConfig, err := provider.TLS.CreateTLSConfig()
		if err != nil {
			return err
		}
		config.HTTPClient = &http.Client{
			Transport: &http.Transport{
				TLSClientConfig: TLSConfig,
			},
		}
		client, err := marathon.NewClient(config)
		if err != nil {
			log.Errorf("Failed to create a client for marathon, error: %s", err)
			return err
		}
		provider.marathonClient = client
		update := make(marathon.EventsChannel, 5)
		if provider.Watch {
			if err := client.AddEventsListener(update, marathon.EventIDApplications); err != nil {
				log.Errorf("Failed to register for events, %s", err)
				return err
			}
			pool.Go(func(stop chan bool) {
				defer close(update)
				for {
					select {
					case <-stop:
						return
					case event := <-update:
						log.Debug("Marathon event receveived", event)
						configuration := provider.loadMarathonConfig()
						if configuration != nil {
							configurationChan <- types.ConfigMessage{
								ProviderName:  "marathon",
								Configuration: configuration,
							}
						}
					}
				}
			})
		}
		configuration := provider.loadMarathonConfig()
		configurationChan <- types.ConfigMessage{
			ProviderName:  "marathon",
			Configuration: configuration,
		}
		return nil
	}

	notify := func(err error, time time.Duration) {
		log.Errorf("Marathon connection error %+v, retrying in %s", err, time)
	}
	err := backoff.RetryNotify(safe.OperationWithRecover(operation), job.NewBackOff(backoff.NewExponentialBackOff()), notify)
	if err != nil {
		log.Errorf("Cannot connect to Marathon server %+v", err)
	}
	return nil
}

func (provider *Marathon) loadMarathonConfig() *types.Configuration {
	var MarathonFuncMap = template.FuncMap{
		"getBackend":                  provider.getBackend,
		"getPort":                     provider.getPort,
		"getWeight":                   provider.getWeight,
		"getDomain":                   provider.getDomain,
		"getProtocol":                 provider.getProtocol,
		"getPassHostHeader":           provider.getPassHostHeader,
		"getPriority":                 provider.getPriority,
		"getEntryPoints":              provider.getEntryPoints,
		"getFrontendRule":             provider.getFrontendRule,
		"getFrontendBackend":          provider.getFrontendBackend,
		"replace":                     replace,
		"hasCircuitBreakerLabels":     provider.hasCircuitBreakerLabels,
		"hasLoadBalancerLabels":       provider.hasLoadBalancerLabels,
		"hasMaxConnLabels":            provider.hasMaxConnLabels,
		"getMaxConnExtractorFunc":     provider.getMaxConnExtractorFunc,
		"getMaxConnAmount":            provider.getMaxConnAmount,
		"getLoadBalancerMethod":       provider.getLoadBalancerMethod,
		"getCircuitBreakerExpression": provider.getCircuitBreakerExpression,
		"getSticky":                   provider.getSticky,
	}

	applications, err := provider.marathonClient.Applications(nil)
	if err != nil {
		log.Errorf("Failed to create a client for marathon, error: %s", err)
		return nil
	}

	tasks, err := provider.marathonClient.AllTasks(&marathon.AllTasksOpts{Status: "running"})
	if err != nil {
		log.Errorf("Failed to create a client for marathon, error: %s", err)
		return nil
	}

	//filter tasks
	filteredTasks := fun.Filter(func(task marathon.Task) bool {
		return provider.taskFilter(task, applications, provider.ExposedByDefault)
	}, tasks.Tasks).([]marathon.Task)

	//filter apps
	filteredApps := fun.Filter(func(app marathon.Application) bool {
		return provider.applicationFilter(app, filteredTasks)
	}, applications.Apps).([]marathon.Application)

	templateObjects := struct {
		Applications []marathon.Application
		Tasks        []marathon.Task
		Domain       string
	}{
		filteredApps,
		filteredTasks,
		provider.Domain,
	}

	configuration, err := provider.getConfiguration("templates/marathon.tmpl", MarathonFuncMap, templateObjects)
	if err != nil {
		log.Error(err)
	}
	return configuration
}

func (provider *Marathon) taskFilter(task marathon.Task, applications *marathon.Applications, exposedByDefaultFlag bool) bool {
	if len(task.Ports) == 0 {
		log.Debug("Filtering marathon task without port %s", task.AppID)
		return false
	}
	application, err := getApplication(task, applications.Apps)
	if err != nil {
		log.Errorf("Unable to get marathon application from task %s", task.AppID)
		return false
	}
	label, _ := provider.getLabel(application, "traefik.tags")
	constraintTags := strings.Split(label, ",")
	if provider.MarathonLBCompatibility {
		if label, err := provider.getLabel(application, "HAPROXY_GROUP"); err == nil {
			constraintTags = append(constraintTags, label)
		}
	}
	if ok, failingConstraint := provider.MatchConstraints(constraintTags); !ok {
		if failingConstraint != nil {
			log.Debugf("Application %v pruned by '%v' constraint", application.ID, failingConstraint.String())
		}
		return false
	}

	if !isApplicationEnabled(application, exposedByDefaultFlag) {
		log.Debugf("Filtering disabled marathon task %s", task.AppID)
		return false
	}

	//filter indeterminable task port
	portIndexLabel := (*application.Labels)["traefik.portIndex"]
	portValueLabel := (*application.Labels)["traefik.port"]
	if portIndexLabel != "" && portValueLabel != "" {
		log.Debugf("Filtering marathon task %s specifying both traefik.portIndex and traefik.port labels", task.AppID)
		return false
	}

	if portIndexLabel != "" {
		index, err := strconv.Atoi((*application.Labels)["traefik.portIndex"])
		if err != nil || index < 0 || index > len(application.Ports)-1 {
			log.Debugf("Filtering marathon task %s with unexpected value for traefik.portIndex label", task.AppID)
			return false
		}
	}
	if portValueLabel != "" {
		port, err := strconv.Atoi((*application.Labels)["traefik.port"])
		if err != nil {
			log.Debugf("Filtering marathon task %s with unexpected value for traefik.port label", task.AppID)
			return false
		}

		var foundPort bool
		for _, exposedPort := range task.Ports {
			if port == exposedPort {
				foundPort = true
				break
			}
		}

		if !foundPort {
			log.Debugf("Filtering marathon task %s without a matching port for traefik.port label", task.AppID)
			return false
		}
	}

	//filter healthchecks
	if application.HasHealthChecks() {
		if task.HasHealthCheckResults() {
			for _, healthcheck := range task.HealthCheckResults {
				// found one bad healthcheck, return false
				if !healthcheck.Alive {
					log.Debugf("Filtering marathon task %s with bad healthcheck", task.AppID)
					return false
				}
			}
		}
	}
	return true
}

func (provider *Marathon) applicationFilter(app marathon.Application, filteredTasks []marathon.Task) bool {
	label, _ := provider.getLabel(app, "traefik.tags")
	constraintTags := strings.Split(label, ",")
	if provider.MarathonLBCompatibility {
		if label, err := provider.getLabel(app, "HAPROXY_GROUP"); err == nil {
			constraintTags = append(constraintTags, label)
		}
	}
	if ok, failingConstraint := provider.MatchConstraints(constraintTags); !ok {
		if failingConstraint != nil {
			log.Debugf("Application %v pruned by '%v' constraint", app.ID, failingConstraint.String())
		}
		return false
	}

	return fun.Exists(func(task marathon.Task) bool {
		return task.AppID == app.ID
	}, filteredTasks)
}

func getApplication(task marathon.Task, apps []marathon.Application) (marathon.Application, error) {
	for _, application := range apps {
		if application.ID == task.AppID {
			return application, nil
		}
	}
	return marathon.Application{}, errors.New("Application not found: " + task.AppID)
}

func isApplicationEnabled(application marathon.Application, exposedByDefault bool) bool {
	return exposedByDefault && (*application.Labels)["traefik.enable"] != "false" || (*application.Labels)["traefik.enable"] == "true"
}

func (provider *Marathon) getLabel(application marathon.Application, label string) (string, error) {
	for key, value := range *application.Labels {
		if key == label {
			return value, nil
		}
	}
	return "", errors.New("Label not found:" + label)
}

func (provider *Marathon) getPort(task marathon.Task, applications []marathon.Application) string {
	application, err := getApplication(task, applications)
	if err != nil {
		log.Errorf("Unable to get marathon application from task %s", task.AppID)
		return ""
	}

	if portIndexLabel, err := provider.getLabel(application, "traefik.portIndex"); err == nil {
		if index, err := strconv.Atoi(portIndexLabel); err == nil {
			return strconv.Itoa(task.Ports[index])
		}
	}
	if portValueLabel, err := provider.getLabel(application, "traefik.port"); err == nil {
		return portValueLabel
	}

	for _, port := range task.Ports {
		return strconv.Itoa(port)
	}
	return ""
}

func (provider *Marathon) getWeight(task marathon.Task, applications []marathon.Application) string {
	application, errApp := getApplication(task, applications)
	if errApp != nil {
		log.Errorf("Unable to get marathon application from task %s", task.AppID)
		return "0"
	}
	if label, err := provider.getLabel(application, "traefik.weight"); err == nil {
		return label
	}
	return "0"
}

func (provider *Marathon) getDomain(application marathon.Application) string {
	if label, err := provider.getLabel(application, "traefik.domain"); err == nil {
		return label
	}
	return provider.Domain
}

func (provider *Marathon) getProtocol(task marathon.Task, applications []marathon.Application) string {
	application, errApp := getApplication(task, applications)
	if errApp != nil {
		log.Errorf("Unable to get marathon application from task %s", task.AppID)
		return "http"
	}
	if label, err := provider.getLabel(application, "traefik.protocol"); err == nil {
		return label
	}
	return "http"
}

func (provider *Marathon) getSticky(application marathon.Application) string {
	if sticky, err := provider.getLabel(application, "traefik.backend.loadbalancer.sticky"); err == nil {
		return sticky
	}
	return "false"
}

func (provider *Marathon) getPassHostHeader(application marathon.Application) string {
	if passHostHeader, err := provider.getLabel(application, "traefik.frontend.passHostHeader"); err == nil {
		return passHostHeader
	}
	return "true"
}

func (provider *Marathon) getPriority(application marathon.Application) string {
	if priority, err := provider.getLabel(application, "traefik.frontend.priority"); err == nil {
		return priority
	}
	return "0"
}

func (provider *Marathon) getEntryPoints(application marathon.Application) []string {
	if entryPoints, err := provider.getLabel(application, "traefik.frontend.entryPoints"); err == nil {
		return strings.Split(entryPoints, ",")
	}
	return []string{}
}

// getFrontendRule returns the frontend rule for the specified application, using
// it's label. It returns a default one (Host) if the label is not present.
func (provider *Marathon) getFrontendRule(application marathon.Application) string {
	if label, err := provider.getLabel(application, "traefik.frontend.rule"); err == nil {
		return label
	}
	if provider.MarathonLBCompatibility {
		if label, err := provider.getLabel(application, "HAPROXY_0_VHOST"); err == nil {
			return "Host:" + label
		}
	}
	return "Host:" + provider.getSubDomain(application.ID) + "." + provider.Domain
}

func (provider *Marathon) getBackend(task marathon.Task, applications []marathon.Application) string {
	application, errApp := getApplication(task, applications)
	if errApp != nil {
		log.Errorf("Unable to get marathon application from task %s", task.AppID)
		return ""
	}
	return provider.getFrontendBackend(application)
}

func (provider *Marathon) getFrontendBackend(application marathon.Application) string {
	if label, err := provider.getLabel(application, "traefik.backend"); err == nil {
		return label
	}
	return replace("/", "-", application.ID)
}

func (provider *Marathon) getSubDomain(name string) string {
	if provider.GroupsAsSubDomains {
		splitedName := strings.Split(strings.TrimPrefix(name, "/"), "/")
		reverseStringSlice(&splitedName)
		reverseName := strings.Join(splitedName, ".")
		return reverseName
	}
	return strings.Replace(strings.TrimPrefix(name, "/"), "/", "-", -1)
}

func (provider *Marathon) hasCircuitBreakerLabels(application marathon.Application) bool {
	if _, err := provider.getLabel(application, "traefik.backend.circuitbreaker.expression"); err != nil {
		return false
	}
	return true
}

func (provider *Marathon) hasLoadBalancerLabels(application marathon.Application) bool {
	_, errMethod := provider.getLabel(application, "traefik.backend.loadbalancer.method")
	_, errSticky := provider.getLabel(application, "traefik.backend.loadbalancer.sticky")
	if errMethod != nil && errSticky != nil {
		return false
	}
	return true
}

func (provider *Marathon) hasMaxConnLabels(application marathon.Application) bool {
	if _, err := provider.getLabel(application, "traefik.backend.maxconn.amount"); err != nil {
		return false
	}
	if _, err := provider.getLabel(application, "traefik.backend.maxconn.extractorfunc"); err != nil {
		return false
	}
	return true
}

func (provider *Marathon) getMaxConnAmount(application marathon.Application) int64 {
	if label, err := provider.getLabel(application, "traefik.backend.maxconn.amount"); err == nil {
		i, errConv := strconv.ParseInt(label, 10, 64)
		if errConv != nil {
			log.Errorf("Unable to parse traefik.backend.maxconn.amount %s", label)
			return math.MaxInt64
		}
		return i
	}
	return math.MaxInt64
}

func (provider *Marathon) getMaxConnExtractorFunc(application marathon.Application) string {
	if label, err := provider.getLabel(application, "traefik.backend.maxconn.extractorfunc"); err == nil {
		return label
	}
	return "request.host"
}

func (provider *Marathon) getLoadBalancerMethod(application marathon.Application) string {
	if label, err := provider.getLabel(application, "traefik.backend.loadbalancer.method"); err == nil {
		return label
	}
	return "wrr"
}

func (provider *Marathon) getCircuitBreakerExpression(application marathon.Application) string {
	if label, err := provider.getLabel(application, "traefik.backend.circuitbreaker.expression"); err == nil {
		return label
	}
	return "NetworkErrorRatio() > 1"
}