Adds Marathon support.

Co-authored-by: Julien Salleyron <julien@containo.us>

provider/aggregator/aggregator.go

@@ -27,6 +27,10 @@ func NewProviderAggregator(conf static.Providers) ProviderAggregator {
 		p.quietAddProvider(conf.Docker)
 	}
 
+	if conf.Marathon != nil {
+		p.quietAddProvider(conf.Marathon)
+	}
+
 	if conf.Rest != nil {
 		p.quietAddProvider(conf.Rest)
 	}

provider/marathon/builder_test.go

@@ -0,0 +1,204 @@
+package marathon
+
+import (
+	"strings"
+	"time"
+
+	"github.com/gambol99/go-marathon"
+)
+
+const testTaskName = "taskID"
+
+// Functions related to building applications.
+
+func withApplications(apps ...marathon.Application) *marathon.Applications {
+	return &marathon.Applications{Apps: apps}
+}
+
+func application(ops ...func(*marathon.Application)) marathon.Application {
+	app := marathon.Application{}
+	app.EmptyLabels()
+	app.Deployments = []map[string]string{}
+	app.ReadinessChecks = &[]marathon.ReadinessCheck{}
+	app.ReadinessCheckResults = &[]marathon.ReadinessCheckResult{}
+
+	for _, op := range ops {
+		op(&app)
+	}
+
+	return app
+}
+
+func appID(name string) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.Name(name)
+	}
+}
+
+func appPorts(ports ...int) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.Ports = append(app.Ports, ports...)
+	}
+}
+
+func withLabel(key, value string) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.AddLabel(key, value)
+	}
+}
+
+func constraint(value string) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.AddConstraint(strings.Split(value, ":")...)
+	}
+}
+
+func portDefinition(port int) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.AddPortDefinition(marathon.PortDefinition{
+			Port: &port,
+		})
+	}
+}
+
+func bridgeNetwork() func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.SetNetwork("bridge", marathon.BridgeNetworkMode)
+	}
+}
+
+func containerNetwork() func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.SetNetwork("cni", marathon.ContainerNetworkMode)
+	}
+}
+
+func ipAddrPerTask(port int) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		p := marathon.Port{
+			Number: port,
+			Name:   "port",
+		}
+		disc := marathon.Discovery{}
+		disc.AddPort(p)
+		ipAddr := marathon.IPAddressPerTask{}
+		ipAddr.SetDiscovery(disc)
+		app.SetIPAddressPerTask(ipAddr)
+	}
+}
+
+func deployments(ids ...string) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		for _, id := range ids {
+			app.Deployments = append(app.Deployments, map[string]string{
+				"ID": id,
+			})
+		}
+	}
+}
+
+func readinessCheck(timeout time.Duration) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		app.ReadinessChecks = &[]marathon.ReadinessCheck{
+			{
+				Path:           "/ready",
+				TimeoutSeconds: int(timeout.Seconds()),
+			},
+		}
+	}
+}
+
+func readinessCheckResult(taskID string, ready bool) func(*marathon.Application) {
+	return func(app *marathon.Application) {
+		*app.ReadinessCheckResults = append(*app.ReadinessCheckResults, marathon.ReadinessCheckResult{
+			TaskID: taskID,
+			Ready:  ready,
+		})
+	}
+}
+
+func withTasks(tasks ...marathon.Task) func(*marathon.Application) {
+	return func(application *marathon.Application) {
+		for _, task := range tasks {
+			tu := task
+			application.Tasks = append(application.Tasks, &tu)
+		}
+	}
+}
+
+// Functions related to building tasks.
+
+func task(ops ...func(*marathon.Task)) marathon.Task {
+	t := &marathon.Task{
+		ID: testTaskName,
+		// The vast majority of tests expect the task state to be TASK_RUNNING.
+		State: string(taskStateRunning),
+	}
+
+	for _, op := range ops {
+		op(t)
+	}
+
+	return *t
+}
+
+func withTaskID(id string) func(*marathon.Task) {
+	return func(task *marathon.Task) {
+		task.ID = id
+	}
+}
+
+func localhostTask(ops ...func(*marathon.Task)) marathon.Task {
+	t := task(
+		host("localhost"),
+		ipAddresses("127.0.0.1"),
+		taskState(taskStateRunning),
+	)
+
+	for _, op := range ops {
+		op(&t)
+	}
+
+	return t
+}
+
+func taskPorts(ports ...int) func(*marathon.Task) {
+	return func(t *marathon.Task) {
+		t.Ports = append(t.Ports, ports...)
+	}
+}
+
+func taskState(state TaskState) func(*marathon.Task) {
+	return func(t *marathon.Task) {
+		t.State = string(state)
+	}
+}
+
+func host(h string) func(*marathon.Task) {
+	return func(t *marathon.Task) {
+		t.Host = h
+	}
+}
+
+func ipAddresses(addresses ...string) func(*marathon.Task) {
+	return func(t *marathon.Task) {
+		for _, addr := range addresses {
+			t.IPAddresses = append(t.IPAddresses, &marathon.IPAddress{
+				IPAddress: addr,
+				Protocol:  "tcp",
+			})
+		}
+	}
+}
+
+func startedAt(timestamp string) func(*marathon.Task) {
+	return func(t *marathon.Task) {
+		t.StartedAt = timestamp
+	}
+}
+
+func startedAtFromNow(offset time.Duration) func(*marathon.Task) {
+	return func(t *marathon.Task) {
+		t.StartedAt = time.Now().Add(-offset).Format(time.RFC3339)
+	}
+}

provider/marathon/config.go

@@ -0,0 +1,276 @@
+package marathon
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"net"
+	"strconv"
+	"strings"
+
+	"github.com/containous/traefik/config"
+	"github.com/containous/traefik/log"
+	"github.com/containous/traefik/provider"
+	"github.com/containous/traefik/provider/label"
+	"github.com/gambol99/go-marathon"
+)
+
+func (p *Provider) buildConfiguration(ctx context.Context, applications *marathon.Applications) *config.Configuration {
+	configurations := make(map[string]*config.Configuration)
+
+	for _, app := range applications.Apps {
+		ctxApp := log.With(ctx, log.Str("applicationID", app.ID))
+		logger := log.FromContext(ctxApp)
+
+		extraConf, err := p.getConfiguration(app)
+		if err != nil {
+			logger.Errorf("Skip application: %v", err)
+			continue
+		}
+
+		if !p.keepApplication(ctxApp, extraConf) {
+			continue
+		}
+
+		confFromLabel, err := label.DecodeConfiguration(stringValueMap(app.Labels))
+		if err != nil {
+			logger.Error(err)
+			continue
+		}
+
+		err = p.buildServiceConfiguration(ctxApp, app, extraConf, confFromLabel)
+		if err != nil {
+			logger.Error(err)
+			continue
+		}
+
+		model := struct {
+			Name   string
+			Labels map[string]string
+		}{
+			Name:   app.ID,
+			Labels: stringValueMap(app.Labels),
+		}
+
+		serviceName := getServiceName(app)
+
+		provider.BuildRouterConfiguration(ctxApp, confFromLabel, serviceName, p.defaultRuleTpl, model)
+
+		configurations[app.ID] = confFromLabel
+	}
+
+	return provider.Merge(ctx, configurations)
+}
+
+func getServiceName(app marathon.Application) string {
+	return strings.Replace(strings.TrimPrefix(app.ID, "/"), "/", "_", -1)
+}
+
+func (p *Provider) buildServiceConfiguration(ctx context.Context, app marathon.Application, extraConf configuration, conf *config.Configuration) error {
+	appName := getServiceName(app)
+	appCtx := log.With(ctx, log.Str("ApplicationID", appName))
+
+	if len(conf.Services) == 0 {
+		conf.Services = make(map[string]*config.Service)
+		lb := &config.LoadBalancerService{}
+		lb.SetDefaults()
+		conf.Services[appName] = &config.Service{
+			LoadBalancer: lb,
+		}
+	}
+
+	for serviceName, service := range conf.Services {
+		var servers []config.Server
+
+		defaultServer := config.Server{}
+		defaultServer.SetDefaults()
+
+		if len(service.LoadBalancer.Servers) > 0 {
+			defaultServer = service.LoadBalancer.Servers[0]
+		}
+
+		for _, task := range app.Tasks {
+			if p.taskFilter(ctx, *task, app) {
+				server, err := p.getServer(app, *task, extraConf, defaultServer)
+				if err != nil {
+					log.FromContext(appCtx).Errorf("Skip task: %v", err)
+					continue
+				}
+				servers = append(servers, server)
+			}
+		}
+		if len(servers) == 0 {
+			return fmt.Errorf("no server for the service %s", serviceName)
+		}
+		service.LoadBalancer.Servers = servers
+	}
+
+	return nil
+}
+
+func (p *Provider) keepApplication(ctx context.Context, extraConf configuration) bool {
+	logger := log.FromContext(ctx)
+
+	// Filter disabled application.
+	if !extraConf.Enable {
+		logger.Debug("Filtering disabled Marathon application")
+		return false
+	}
+
+	// Filter by constraints.
+	if ok, failingConstraint := p.MatchConstraints(extraConf.Tags); !ok {
+		if failingConstraint != nil {
+			logger.Debugf("Filtering Marathon application, pruned by %q constraint", failingConstraint.String())
+		}
+		return false
+	}
+
+	return true
+}
+
+func (p *Provider) taskFilter(ctx context.Context, task marathon.Task, application marathon.Application) bool {
+	if task.State != string(taskStateRunning) {
+		return false
+	}
+
+	if ready := p.readyChecker.Do(task, application); !ready {
+		log.FromContext(ctx).Infof("Filtering unready task %s from application %s", task.ID, application.ID)
+		return false
+	}
+
+	return true
+}
+
+func (p *Provider) getServer(app marathon.Application, task marathon.Task, extraConf configuration, defaultServer config.Server) (config.Server, error) {
+	host, err := p.getServerHost(task, app, extraConf)
+	if len(host) == 0 {
+		return config.Server{}, err
+	}
+
+	port, err := getPort(task, app, defaultServer.Port)
+	if err != nil {
+		return config.Server{}, err
+	}
+
+	server := config.Server{
+		URL:    fmt.Sprintf("%s://%s", defaultServer.Scheme, net.JoinHostPort(host, port)),
+		Weight: 1,
+	}
+
+	return server, nil
+}
+
+func (p *Provider) getServerHost(task marathon.Task, app marathon.Application, extraConf configuration) (string, error) {
+	networks := app.Networks
+	var hostFlag bool
+
+	if networks == nil {
+		hostFlag = app.IPAddressPerTask == nil
+	} else {
+		hostFlag = (*networks)[0].Mode != marathon.ContainerNetworkMode
+	}
+
+	if hostFlag || p.ForceTaskHostname {
+		if len(task.Host) == 0 {
+			return "", fmt.Errorf("host is undefined for task %q app %q", task.ID, app.ID)
+		}
+		return task.Host, nil
+	}
+
+	numTaskIPAddresses := len(task.IPAddresses)
+	switch numTaskIPAddresses {
+	case 0:
+		return "", fmt.Errorf("missing IP address for Marathon application %s on task %s", app.ID, task.ID)
+	case 1:
+		return task.IPAddresses[0].IPAddress, nil
+	default:
+		if extraConf.Marathon.IPAddressIdx == math.MinInt32 {
+			return "", fmt.Errorf("found %d task IP addresses but missing IP address index for Marathon application %s on task %s",
+				numTaskIPAddresses, app.ID, task.ID)
+		}
+		if extraConf.Marathon.IPAddressIdx < 0 || extraConf.Marathon.IPAddressIdx > numTaskIPAddresses {
+			return "", fmt.Errorf("cannot use IP address index to select from %d task IP addresses for Marathon application %s on task %s",
+				numTaskIPAddresses, app.ID, task.ID)
+		}
+
+		return task.IPAddresses[extraConf.Marathon.IPAddressIdx].IPAddress, nil
+	}
+}
+
+func getPort(task marathon.Task, app marathon.Application, serverPort string) (string, error) {
+	port, err := processPorts(app, task, serverPort)
+	if err != nil {
+		return "", fmt.Errorf("unable to process ports for %s %s: %v", app.ID, task.ID, err)
+	}
+
+	return strconv.Itoa(port), nil
+}
+
+// processPorts returns the configured port.
+// An explicitly specified port is preferred. If none is specified, it selects
+// one of the available port. The first such found port is returned unless an
+// optional index is provided.
+func processPorts(app marathon.Application, task marathon.Task, serverPort string) (int, error) {
+	if len(serverPort) > 0 && !strings.HasPrefix(serverPort, "index:") {
+		port, err := strconv.Atoi(serverPort)
+		if err != nil {
+			return 0, err
+		}
+
+		if port <= 0 {
+			return 0, fmt.Errorf("explicitly specified port %d must be greater than zero", port)
+		} else if port > 0 {
+			return port, nil
+		}
+	}
+
+	ports := retrieveAvailablePorts(app, task)
+	if len(ports) == 0 {
+		return 0, errors.New("no port found")
+	}
+
+	portIndex := 0
+	if strings.HasPrefix(serverPort, "index:") {
+		split := strings.SplitN(serverPort, ":", 2)
+		index, err := strconv.Atoi(split[1])
+		if err != nil {
+			return 0, err
+		}
+
+		if index < 0 || index > len(ports)-1 {
+			return 0, fmt.Errorf("index %d must be within range (0, %d)", index, len(ports)-1)
+		}
+		portIndex = index
+	}
+	return ports[portIndex], nil
+}
+
+func retrieveAvailablePorts(app marathon.Application, task marathon.Task) []int {
+	// Using default port configuration
+	if len(task.Ports) > 0 {
+		return task.Ports
+	}
+
+	// Using port definition if available
+	if app.PortDefinitions != nil && len(*app.PortDefinitions) > 0 {
+		var ports []int
+		for _, def := range *app.PortDefinitions {
+			if def.Port != nil {
+				ports = append(ports, *def.Port)
+			}
+		}
+		return ports
+	}
+
+	// If using IP-per-task using this port definition
+	if app.IPAddressPerTask != nil && app.IPAddressPerTask.Discovery != nil && len(*(app.IPAddressPerTask.Discovery.Ports)) > 0 {
+		var ports []int
+		for _, def := range *(app.IPAddressPerTask.Discovery.Ports) {
+			ports = append(ports, def.Number)
+		}
+		return ports
+	}
+
+	return []int{}
+}

provider/marathon/fake_client_test.go

@@ -0,0 +1,24 @@
+package marathon
+
+import (
+	"errors"
+
+	"github.com/containous/traefik/provider/marathon/mocks"
+	"github.com/gambol99/go-marathon"
+	"github.com/stretchr/testify/mock"
+)
+
+type fakeClient struct {
+	mocks.Marathon
+}
+
+func newFakeClient(applicationsError bool, applications marathon.Applications) *fakeClient {
+	// create an instance of our test object
+	fakeClient := new(fakeClient)
+	if applicationsError {
+		fakeClient.On("Applications", mock.Anything).Return(nil, errors.New("fake Marathon server error"))
+	} else {
+		fakeClient.On("Applications", mock.Anything).Return(&applications, nil)
+	}
+	return fakeClient
+}

provider/marathon/label.go

@@ -0,0 +1,51 @@
+package marathon
+
+import (
+	"math"
+	"strings"
+
+	"github.com/containous/traefik/provider/label"
+	"github.com/gambol99/go-marathon"
+)
+
+type configuration struct {
+	Enable   bool
+	Tags     []string
+	Marathon specificConfiguration
+}
+
+type specificConfiguration struct {
+	IPAddressIdx int
+}
+
+func (p *Provider) getConfiguration(app marathon.Application) (configuration, error) {
+	labels := stringValueMap(app.Labels)
+
+	conf := configuration{
+		Enable: p.ExposedByDefault,
+		Tags:   nil,
+		Marathon: specificConfiguration{
+			IPAddressIdx: math.MinInt32,
+		},
+	}
+
+	err := label.Decode(labels, &conf, "traefik.marathon.", "traefik.enable", "traefik.tags")
+	if err != nil {
+		return configuration{}, err
+	}
+
+	if p.FilterMarathonConstraints && app.Constraints != nil {
+		for _, constraintParts := range *app.Constraints {
+			conf.Tags = append(conf.Tags, strings.Join(constraintParts, ":"))
+		}
+	}
+
+	return conf, nil
+}
+
+func stringValueMap(mp *map[string]string) map[string]string {
+	if mp != nil {
+		return *mp
+	}
+	return make(map[string]string)
+}

provider/marathon/label_test.go

@@ -0,0 +1,178 @@
+package marathon
+
+import (
+	"math"
+	"testing"
+
+	"github.com/containous/traefik/provider"
+	"github.com/gambol99/go-marathon"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetConfiguration(t *testing.T) {
+	testCases := []struct {
+		desc     string
+		app      marathon.Application
+		p        Provider
+		expected configuration
+	}{
+		{
+			desc: "Empty labels",
+			app: marathon.Application{
+				Constraints: &[][]string{},
+				Labels:      &map[string]string{},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          false,
+				FilterMarathonConstraints: false,
+			},
+			expected: configuration{
+				Enable: false,
+				Tags:   nil,
+				Marathon: specificConfiguration{
+					IPAddressIdx: math.MinInt32,
+				},
+			},
+		},
+		{
+			desc: "label enable",
+			app: marathon.Application{
+				Constraints: &[][]string{},
+				Labels: &map[string]string{
+					"traefik.enable": "true",
+				},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          false,
+				FilterMarathonConstraints: false,
+			},
+			expected: configuration{
+				Enable: true,
+				Tags:   nil,
+				Marathon: specificConfiguration{
+					IPAddressIdx: math.MinInt32,
+				},
+			},
+		},
+		{
+			desc: "Use ip address index",
+			app: marathon.Application{
+				Constraints: &[][]string{},
+				Labels: &map[string]string{
+					"traefik.marathon.IPAddressIdx": "4",
+				},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          false,
+				FilterMarathonConstraints: false,
+			},
+			expected: configuration{
+				Enable: false,
+				Tags:   nil,
+				Marathon: specificConfiguration{
+					IPAddressIdx: 4,
+				},
+			},
+		},
+		{
+			desc: "Use marathon constraints",
+			app: marathon.Application{
+				Constraints: &[][]string{
+					{"key", "value"},
+				},
+				Labels: &map[string]string{},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          false,
+				FilterMarathonConstraints: true,
+			},
+			expected: configuration{
+				Enable: false,
+				Tags: []string{
+					"key:value",
+				},
+				Marathon: specificConfiguration{
+					IPAddressIdx: math.MinInt32,
+				},
+			},
+		},
+		{
+			desc: "ExposedByDefault and no enable label",
+			app: marathon.Application{
+				Constraints: &[][]string{},
+				Labels:      &map[string]string{},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          true,
+				FilterMarathonConstraints: false,
+			},
+			expected: configuration{
+				Enable: true,
+				Tags:   nil,
+				Marathon: specificConfiguration{
+					IPAddressIdx: math.MinInt32,
+				},
+			},
+		},
+		{
+			desc: "ExposedByDefault and enable label false",
+			app: marathon.Application{
+				Constraints: &[][]string{},
+				Labels: &map[string]string{
+					"traefik.enable": "false",
+				},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          true,
+				FilterMarathonConstraints: false,
+			},
+			expected: configuration{
+				Enable: false,
+				Tags:   nil,
+				Marathon: specificConfiguration{
+					IPAddressIdx: math.MinInt32,
+				},
+			},
+		},
+		{
+			desc: "Tags in label",
+			app: marathon.Application{
+				Constraints: &[][]string{},
+				Labels: &map[string]string{
+					"traefik.tags": "mytags",
+				},
+			},
+			p: Provider{
+				BaseProvider:              provider.BaseProvider{},
+				ExposedByDefault:          true,
+				FilterMarathonConstraints: false,
+			},
+			expected: configuration{
+				Enable: true,
+				Tags:   []string{"mytags"},
+				Marathon: specificConfiguration{
+					IPAddressIdx: math.MinInt32,
+				},
+			},
+		},
+	}
+
+	for _, test := range testCases {
+		test := test
+		t.Run(test.desc, func(t *testing.T) {
+			t.Parallel()
+
+			extraConf, err := test.p.getConfiguration(test.app)
+			require.NoError(t, err)
+
+			assert.Equal(t, test.expected, extraConf)
+		})
+	}
+}

provider/marathon/marathon.go

@@ -0,0 +1,207 @@
+package marathon
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"text/template"
+	"time"
+
+	"github.com/cenk/backoff"
+	"github.com/containous/flaeg/parse"
+	"github.com/containous/traefik/config"
+	"github.com/containous/traefik/job"
+	"github.com/containous/traefik/log"
+	"github.com/containous/traefik/old/types"
+	"github.com/containous/traefik/provider"
+	"github.com/containous/traefik/safe"
+	"github.com/gambol99/go-marathon"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	// DefaultTemplateRule The default template for the default rule.
+	DefaultTemplateRule   = "Host:{{ normalize .Name }}"
+	traceMaxScanTokenSize = 1024 * 1024
+	marathonEventIDs      = marathon.EventIDApplications |
+		marathon.EventIDAddHealthCheck |
+		marathon.EventIDDeploymentSuccess |
+		marathon.EventIDDeploymentFailed |
+		marathon.EventIDDeploymentInfo |
+		marathon.EventIDDeploymentStepSuccess |
+		marathon.EventIDDeploymentStepFailed
+)
+
+// TaskState denotes the Mesos state a task can have.
+type TaskState string
+
+const (
+	taskStateRunning TaskState = "TASK_RUNNING"
+	taskStateStaging TaskState = "TASK_STAGING"
+)
+
+var _ provider.Provider = (*Provider)(nil)
+
+// Provider holds configuration of the provider.
+type Provider struct {
+	provider.BaseProvider
+	Endpoint                  string           `description:"Marathon server endpoint. You can also specify multiple endpoint for Marathon" export:"true"`
+	DefaultRule               string           `description:"Default rule"`
+	ExposedByDefault          bool             `description:"Expose Marathon apps by default" export:"true"`
+	DCOSToken                 string           `description:"DCOSToken for DCOS environment, This will override the Authorization header" export:"true"`
+	FilterMarathonConstraints bool             `description:"Enable use of Marathon constraints in constraint filtering" export:"true"`
+	TLS                       *types.ClientTLS `description:"Enable TLS support" export:"true"`
+	DialerTimeout             parse.Duration   `description:"Set a dialer timeout for Marathon" export:"true"`
+	ResponseHeaderTimeout     parse.Duration   `description:"Set a response header timeout for Marathon" export:"true"`
+	TLSHandshakeTimeout       parse.Duration   `description:"Set a TLS handhsake timeout for Marathon" export:"true"`
+	KeepAlive                 parse.Duration   `description:"Set a TCP Keep Alive time in seconds" export:"true"`
+	ForceTaskHostname         bool             `description:"Force to use the task's hostname." export:"true"`
+	Basic                     *Basic           `description:"Enable basic authentication" export:"true"`
+	RespectReadinessChecks    bool             `description:"Filter out tasks with non-successful readiness checks during deployments" export:"true"`
+	readyChecker              *readinessChecker
+	marathonClient            marathon.Marathon
+	defaultRuleTpl            *template.Template
+}
+
+// Basic holds basic authentication specific configurations
+type Basic struct {
+	HTTPBasicAuthUser string `description:"Basic authentication User"`
+	HTTPBasicPassword string `description:"Basic authentication Password"`
+}
+
+// Init the provider
+func (p *Provider) Init() error {
+	fm := template.FuncMap{
+		"strsToItfs": func(values []string) []interface{} {
+			var r []interface{}
+			for _, v := range values {
+				r = append(r, v)
+			}
+			return r
+		},
+	}
+
+	defaultRuleTpl, err := provider.MakeDefaultRuleTemplate(p.DefaultRule, fm)
+	if err != nil {
+		return fmt.Errorf("error while parsing default rule: %v", err)
+	}
+
+	p.defaultRuleTpl = defaultRuleTpl
+	return p.BaseProvider.Init()
+}
+
+// Provide allows the marathon provider to provide configurations to traefik
+// using the given configuration channel.
+func (p *Provider) Provide(configurationChan chan<- config.Message, pool *safe.Pool) error {
+	ctx := log.With(context.Background(), log.Str(log.ProviderName, "marathon"))
+	logger := log.FromContext(ctx)
+
+	operation := func() error {
+
+		confg := marathon.NewDefaultConfig()
+		confg.URL = p.Endpoint
+		confg.EventsTransport = marathon.EventsTransportSSE
+		if p.Trace {
+			confg.LogOutput = log.CustomWriterLevel(logrus.DebugLevel, traceMaxScanTokenSize)
+		}
+		if p.Basic != nil {
+			confg.HTTPBasicAuthUser = p.Basic.HTTPBasicAuthUser
+			confg.HTTPBasicPassword = p.Basic.HTTPBasicPassword
+		}
+		var rc *readinessChecker
+		if p.RespectReadinessChecks {
+			logger.Debug("Enabling Marathon readiness checker")
+			rc = defaultReadinessChecker(p.Trace)
+		}
+		p.readyChecker = rc
+
+		if len(p.DCOSToken) > 0 {
+			confg.DCOSToken = p.DCOSToken
+		}
+		TLSConfig, err := p.TLS.CreateTLSConfig()
+		if err != nil {
+			return err
+		}
+		confg.HTTPClient = &http.Client{
+			Transport: &http.Transport{
+				DialContext: (&net.Dialer{
+					KeepAlive: time.Duration(p.KeepAlive),
+					Timeout:   time.Duration(p.DialerTimeout),
+				}).DialContext,
+				ResponseHeaderTimeout: time.Duration(p.ResponseHeaderTimeout),
+				TLSHandshakeTimeout:   time.Duration(p.TLSHandshakeTimeout),
+				TLSClientConfig:       TLSConfig,
+			},
+		}
+		client, err := marathon.NewClient(confg)
+		if err != nil {
+			logger.Errorf("Failed to create a client for marathon, error: %s", err)
+			return err
+		}
+		p.marathonClient = client
+
+		if p.Watch {
+			update, err := client.AddEventsListener(marathonEventIDs)
+			if err != nil {
+				logger.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:
+						logger.Debugf("Received provider event %s", event)
+
+						conf := p.getConfigurations(ctx)
+						if conf != nil {
+							configurationChan <- config.Message{
+								ProviderName:  "marathon",
+								Configuration: conf,
+							}
+						}
+					}
+				}
+			})
+		}
+
+		configuration := p.getConfigurations(ctx)
+		configurationChan <- config.Message{
+			ProviderName:  "marathon",
+			Configuration: configuration,
+		}
+		return nil
+	}
+
+	notify := func(err error, time time.Duration) {
+		logger.Errorf("Provider connection error %+v, retrying in %s", err, time)
+	}
+	err := backoff.RetryNotify(safe.OperationWithRecover(operation), job.NewBackOff(backoff.NewExponentialBackOff()), notify)
+	if err != nil {
+		logger.Errorf("Cannot connect to Provider server: %+v", err)
+	}
+	return nil
+}
+
+func (p *Provider) getConfigurations(ctx context.Context) *config.Configuration {
+	applications, err := p.getApplications()
+	if err != nil {
+		log.FromContext(ctx).Errorf("Failed to retrieve Marathon applications: %v", err)
+		return nil
+	}
+
+	return p.buildConfiguration(ctx, applications)
+}
+
+func (p *Provider) getApplications() (*marathon.Applications, error) {
+	v := url.Values{}
+	v.Add("embed", "apps.tasks")
+	v.Add("embed", "apps.deployments")
+	v.Add("embed", "apps.readiness")
+
+	return p.marathonClient.Applications(v)
+}

provider/marathon/readiness.go

@@ -0,0 +1,122 @@
+package marathon
+
+import (
+	"time"
+
+	"github.com/containous/traefik/old/log"
+	"github.com/gambol99/go-marathon"
+)
+
+const (
+	// readinessCheckDefaultTimeout is the default timeout for a readiness
+	// check if no check timeout is specified on the application spec. This
+	// should really never be the case, but better be safe than sorry.
+	readinessCheckDefaultTimeout = 10 * time.Second
+	// readinessCheckSafetyMargin is some buffer duration to account for
+	// small offsets in readiness check execution.
+	readinessCheckSafetyMargin = 5 * time.Second
+	readinessLogHeader         = "Marathon readiness check: "
+)
+
+type readinessChecker struct {
+	checkDefaultTimeout time.Duration
+	checkSafetyMargin   time.Duration
+	traceLogging        bool
+}
+
+func defaultReadinessChecker(isTraceLogging bool) *readinessChecker {
+	return &readinessChecker{
+		checkDefaultTimeout: readinessCheckDefaultTimeout,
+		checkSafetyMargin:   readinessCheckSafetyMargin,
+		traceLogging:        isTraceLogging,
+	}
+}
+
+func (rc *readinessChecker) Do(task marathon.Task, app marathon.Application) bool {
+	if rc == nil {
+		// Readiness checker disabled.
+		return true
+	}
+
+	switch {
+	case len(app.Deployments) == 0:
+		// We only care about readiness during deployments; post-deployment readiness
+		// can be covered by a periodic post-deployment probe (i.e., Traefik health checks).
+		rc.tracef("task %s app %s: ready = true [no deployment ongoing]", task.ID, app.ID)
+		return true
+
+	case app.ReadinessChecks == nil || len(*app.ReadinessChecks) == 0:
+		// Applications without configured readiness checks are always considered
+		// ready.
+		rc.tracef("task %s app %s: ready = true [no readiness checks on app]", task.ID, app.ID)
+		return true
+	}
+
+	// Loop through all readiness check results and return the results for
+	// matching task IDs.
+	if app.ReadinessCheckResults != nil {
+		for _, readinessCheckResult := range *app.ReadinessCheckResults {
+			if readinessCheckResult.TaskID == task.ID {
+				rc.tracef("task %s app %s: ready = %t [evaluating readiness check ready state]", task.ID, app.ID, readinessCheckResult.Ready)
+				return readinessCheckResult.Ready
+			}
+		}
+	}
+
+	// There's a corner case sometimes hit where the first new task of a
+	// deployment goes from TASK_STAGING to TASK_RUNNING without a corresponding
+	// readiness check result being included in the API response. This only happens
+	// in a very short (yet unlucky) time frame and does not repeat for subsequent
+	// tasks of the same deployment.
+	// Complicating matters, the situation may occur for both initially deploying
+	// applications as well as rolling-upgraded ones where one or more tasks from
+	// a previous deployment exist already and are joined by new tasks from a
+	// subsequent deployment. We must always make sure that pre-existing tasks
+	// maintain their ready state while newly launched tasks must be considered
+	// unready until a check result appears.
+	// We distinguish the two cases by comparing the current time with the start
+	// time of the task: It should take Marathon at most one readiness check timeout
+	// interval (plus some safety margin to account for the delayed nature of
+	// distributed systems) for readiness check results to be returned along the API
+	// response. Once the task turns old enough, we assume it to be part of a
+	// pre-existing deployment and mark it as ready. Note that it is okay to err
+	// on the side of caution and consider a task unready until the safety time
+	// window has elapsed because a newly created task should be readiness-checked
+	// and be given a result fairly shortly after its creation (i.e., on the scale
+	// of seconds).
+	readinessCheckTimeoutSecs := (*app.ReadinessChecks)[0].TimeoutSeconds
+	readinessCheckTimeout := time.Duration(readinessCheckTimeoutSecs) * time.Second
+	if readinessCheckTimeout == 0 {
+		rc.tracef("task %s app %s: readiness check timeout not set, using default value %s", task.ID, app.ID, rc.checkDefaultTimeout)
+		readinessCheckTimeout = rc.checkDefaultTimeout
+	} else {
+		readinessCheckTimeout += rc.checkSafetyMargin
+	}
+
+	startTime, err := time.Parse(time.RFC3339, task.StartedAt)
+	if err != nil {
+		// An unparseable start time should never occur; if it does, we assume the
+		// problem should be surfaced as quickly as possible, which is easiest if
+		// we shun the task from rotation.
+		log.Warnf("Failed to parse start-time %s of task %s from application %s: %s (assuming unready)", task.StartedAt, task.ID, app.ID, err)
+		return false
+	}
+
+	since := time.Since(startTime)
+	if since < readinessCheckTimeout {
+		rc.tracef("task %s app %s: ready = false [task with start-time %s not within assumed check timeout window of %s (elapsed time since task start: %s)]", task.ID, app.ID, startTime.Format(time.RFC3339), readinessCheckTimeout, since)
+		return false
+	}
+
+	// Finally, we can be certain this task is not part of the deployment (i.e.,
+	// it's an old task that's going to transition into the TASK_KILLING and/or
+	// TASK_KILLED state as new tasks' readiness checks gradually turn green.)
+	rc.tracef("task %s app %s: ready = true [task with start-time %s not involved in deployment (elapsed time since task start: %s)]", task.ID, app.ID, startTime.Format(time.RFC3339), since)
+	return true
+}
+
+func (rc *readinessChecker) tracef(format string, args ...interface{}) {
+	if rc.traceLogging {
+		log.Debugf(readinessLogHeader+format, args...)
+	}
+}

provider/marathon/readiness_test.go

@@ -0,0 +1,134 @@
+package marathon
+
+import (
+	"testing"
+	"time"
+
+	"github.com/gambol99/go-marathon"
+)
+
+func testReadinessChecker() *readinessChecker {
+	return defaultReadinessChecker(false)
+}
+
+func TestDisabledReadinessChecker(t *testing.T) {
+	var rc *readinessChecker
+	tsk := task()
+	app := application(
+		deployments("deploymentId"),
+		readinessCheck(0),
+		readinessCheckResult(testTaskName, false),
+	)
+
+	if ready := rc.Do(tsk, app); !ready {
+		t.Error("expected ready = true")
+	}
+}
+
+func TestEnabledReadinessChecker(t *testing.T) {
+	tests := []struct {
+		desc          string
+		task          marathon.Task
+		app           marathon.Application
+		rc            readinessChecker
+		expectedReady bool
+	}{
+		{
+			desc:          "no deployment running",
+			task:          task(),
+			app:           application(),
+			expectedReady: true,
+		},
+		{
+			desc:          "no readiness checks defined",
+			task:          task(),
+			app:           application(deployments("deploymentId")),
+			expectedReady: true,
+		},
+		{
+			desc: "readiness check result negative",
+			task: task(),
+			app: application(
+				deployments("deploymentId"),
+				readinessCheck(0),
+				readinessCheckResult("otherTaskID", true),
+				readinessCheckResult(testTaskName, false),
+			),
+			expectedReady: false,
+		},
+		{
+			desc: "readiness check result positive",
+			task: task(),
+			app: application(
+				deployments("deploymentId"),
+				readinessCheck(0),
+				readinessCheckResult("otherTaskID", false),
+				readinessCheckResult(testTaskName, true),
+			),
+			expectedReady: true,
+		},
+		{
+			desc: "no readiness check result with default timeout",
+			task: task(startedAtFromNow(3 * time.Minute)),
+			app: application(
+				deployments("deploymentId"),
+				readinessCheck(0),
+			),
+			rc: readinessChecker{
+				checkDefaultTimeout: 5 * time.Minute,
+			},
+			expectedReady: false,
+		},
+		{
+			desc: "no readiness check result with readiness check timeout",
+			task: task(startedAtFromNow(4 * time.Minute)),
+			app: application(
+				deployments("deploymentId"),
+				readinessCheck(3*time.Minute),
+			),
+			rc: readinessChecker{
+				checkSafetyMargin: 3 * time.Minute,
+			},
+			expectedReady: false,
+		},
+		{
+			desc: "invalid task start time",
+			task: task(startedAt("invalid")),
+			app: application(
+				deployments("deploymentId"),
+				readinessCheck(0),
+			),
+			expectedReady: false,
+		},
+		{
+			desc: "task not involved in deployment",
+			task: task(startedAtFromNow(1 * time.Hour)),
+			app: application(
+				deployments("deploymentId"),
+				readinessCheck(0),
+			),
+			rc: readinessChecker{
+				checkDefaultTimeout: 10 * time.Second,
+			},
+			expectedReady: true,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+		t.Run(test.desc, func(t *testing.T) {
+			t.Parallel()
+			rc := testReadinessChecker()
+			if test.rc.checkDefaultTimeout > 0 {
+				rc.checkDefaultTimeout = test.rc.checkDefaultTimeout
+			}
+			if test.rc.checkSafetyMargin > 0 {
+				rc.checkSafetyMargin = test.rc.checkSafetyMargin
+			}
+			actualReady := test.rc.Do(test.task, test.app)
+			if actualReady != test.expectedReady {
+				t.Errorf("actual ready = %t, expected ready = %t", actualReady, test.expectedReady)
+			}
+		})
+	}
+}