traefik/pkg/provider/marathon/config.go

package marathon

import (
	"context"
	"errors"
	"fmt"
	"math"
	"net"
	"strconv"
	"strings"

	"github.com/gambol99/go-marathon"
	"github.com/rs/zerolog/log"
	"github.com/traefik/traefik/v2/pkg/config/dynamic"
	"github.com/traefik/traefik/v2/pkg/config/label"
	"github.com/traefik/traefik/v2/pkg/provider"
	"github.com/traefik/traefik/v2/pkg/provider/constraints"
)

func (p *Provider) buildConfiguration(ctx context.Context, applications *marathon.Applications) *dynamic.Configuration {
	configurations := make(map[string]*dynamic.Configuration)

	for _, app := range applications.Apps {
		logger := log.Ctx(ctx).With().Str("applicationID", app.ID).Logger()
		ctxApp := logger.WithContext(ctx)

		extraConf, err := p.getConfiguration(app)
		if err != nil {
			logger.Error().Err(err).Msg("Skip application")
			continue
		}

		labels := stringValueMap(app.Labels)

		if app.Constraints != nil {
			for i, constraintParts := range *app.Constraints {
				key := constraints.MarathonConstraintPrefix + "-" + strconv.Itoa(i)
				labels[key] = strings.Join(constraintParts, ":")
			}
		}

		if !p.keepApplication(ctxApp, extraConf, labels) {
			continue
		}

		confFromLabel, err := label.DecodeConfiguration(labels)
		if err != nil {
			logger.Error().Err(err).Send()
			continue
		}

		var tcpOrUDP bool
		if len(confFromLabel.TCP.Routers) > 0 || len(confFromLabel.TCP.Services) > 0 {
			tcpOrUDP = true

			err := p.buildTCPServiceConfiguration(ctxApp, app, extraConf, confFromLabel.TCP)
			if err != nil {
				logger.Error().Err(err).Send()
				continue
			}
			provider.BuildTCPRouterConfiguration(ctxApp, confFromLabel.TCP)
		}

		if len(confFromLabel.UDP.Routers) > 0 || len(confFromLabel.UDP.Services) > 0 {
			tcpOrUDP = true

			err := p.buildUDPServiceConfiguration(ctxApp, app, extraConf, confFromLabel.UDP)
			if err != nil {
				logger.Error().Err(err).Send()
			} else {
				provider.BuildUDPRouterConfiguration(ctxApp, confFromLabel.UDP)
			}
		}

		if tcpOrUDP && len(confFromLabel.HTTP.Routers) == 0 &&
			len(confFromLabel.HTTP.Middlewares) == 0 &&
			len(confFromLabel.HTTP.Services) == 0 {
			configurations[app.ID] = confFromLabel
			continue
		}

		err = p.buildServiceConfiguration(ctxApp, app, extraConf, confFromLabel.HTTP)
		if err != nil {
			logger.Error().Err(err).Send()
			continue
		}

		model := struct {
			Name   string
			Labels map[string]string
		}{
			Name:   app.ID,
			Labels: labels,
		}

		serviceName := getServiceName(app)

		provider.BuildRouterConfiguration(ctxApp, confFromLabel.HTTP, serviceName, p.defaultRuleTpl, model)

		configurations[app.ID] = confFromLabel
	}

	return provider.Merge(ctx, configurations)
}

func getServiceName(app marathon.Application) string {
	return strings.ReplaceAll(strings.TrimPrefix(app.ID, "/"), "/", "_")
}

func (p *Provider) buildServiceConfiguration(ctx context.Context, app marathon.Application, extraConf configuration, conf *dynamic.HTTPConfiguration) error {
	appName := getServiceName(app)

	logger := log.Ctx(ctx).With().Str("applicationName", appName).Logger()

	if len(conf.Services) == 0 {
		conf.Services = make(map[string]*dynamic.Service)
		lb := &dynamic.ServersLoadBalancer{}
		lb.SetDefaults()
		conf.Services[appName] = &dynamic.Service{
			LoadBalancer: lb,
		}
	}

	for serviceName, service := range conf.Services {
		var servers []dynamic.Server

		defaultServer := dynamic.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) {
				continue
			}
			server, err := p.getServer(app, *task, extraConf, defaultServer)
			if err != nil {
				logger.Error().Err(err).Msg("Skip task")
				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) buildTCPServiceConfiguration(ctx context.Context, app marathon.Application, extraConf configuration, conf *dynamic.TCPConfiguration) error {
	appName := getServiceName(app)

	logger := log.Ctx(ctx).With().Str("applicationName", appName).Logger()

	if len(conf.Services) == 0 {
		conf.Services = make(map[string]*dynamic.TCPService)
		lb := &dynamic.TCPServersLoadBalancer{}
		lb.SetDefaults()
		conf.Services[appName] = &dynamic.TCPService{
			LoadBalancer: lb,
		}
	}

	for serviceName, service := range conf.Services {
		var servers []dynamic.TCPServer

		defaultServer := dynamic.TCPServer{}

		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.getTCPServer(app, *task, extraConf, defaultServer)
				if err != nil {
					logger.Error().Err(err).Msg("Skip task")
					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) buildUDPServiceConfiguration(ctx context.Context, app marathon.Application, extraConf configuration, conf *dynamic.UDPConfiguration) error {
	appName := getServiceName(app)
	logger := log.Ctx(ctx).With().Str("applicationName", appName).Logger()

	if len(conf.Services) == 0 {
		conf.Services = make(map[string]*dynamic.UDPService)
		lb := &dynamic.UDPServersLoadBalancer{}

		conf.Services[appName] = &dynamic.UDPService{
			LoadBalancer: lb,
		}
	}

	for serviceName, service := range conf.Services {
		var servers []dynamic.UDPServer

		defaultServer := dynamic.UDPServer{}

		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.getUDPServer(app, *task, extraConf, defaultServer)
				if err != nil {
					logger.Error().Err(err).Msg("Skip task")
					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, labels map[string]string) bool {
	logger := log.Ctx(ctx)

	// Filter disabled application.
	if !extraConf.Enable {
		logger.Debug().Msg("Filtering disabled Marathon application")
		return false
	}

	// Filter by constraints.
	matches, err := constraints.MatchLabels(labels, p.Constraints)
	if err != nil {
		logger.Error().Err(err).Msg("Error matching constraints expression")
		return false
	}
	if !matches {
		logger.Debug().Msgf("Marathon application filtered by constraint expression: %q", p.Constraints)
		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.Ctx(ctx).Info().Msgf("Filtering unready task %s from application %s", task.ID, application.ID)
		return false
	}

	return true
}

func (p *Provider) getTCPServer(app marathon.Application, task marathon.Task, extraConf configuration, defaultServer dynamic.TCPServer) (dynamic.TCPServer, error) {
	host, err := p.getServerHost(task, app, extraConf)
	if len(host) == 0 {
		return dynamic.TCPServer{}, err
	}

	port, err := getPort(task, app, defaultServer.Port)
	if err != nil {
		return dynamic.TCPServer{}, err
	}

	server := dynamic.TCPServer{
		Address: net.JoinHostPort(host, port),
	}

	return server, nil
}

func (p *Provider) getUDPServer(app marathon.Application, task marathon.Task, extraConf configuration, defaultServer dynamic.UDPServer) (dynamic.UDPServer, error) {
	host, err := p.getServerHost(task, app, extraConf)
	if len(host) == 0 {
		return dynamic.UDPServer{}, err
	}

	port, err := getPort(task, app, defaultServer.Port)
	if err != nil {
		return dynamic.UDPServer{}, err
	}

	server := dynamic.UDPServer{
		Address: net.JoinHostPort(host, port),
	}

	return server, nil
}

func (p *Provider) getServer(app marathon.Application, task marathon.Task, extraConf configuration, defaultServer dynamic.Server) (dynamic.Server, error) {
	host, err := p.getServerHost(task, app, extraConf)
	if len(host) == 0 {
		return dynamic.Server{}, err
	}

	port, err := getPort(task, app, defaultServer.Port)
	if err != nil {
		return dynamic.Server{}, err
	}

	server := dynamic.Server{
		URL: fmt.Sprintf("%s://%s", defaultServer.Scheme, net.JoinHostPort(host, port)),
	}

	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: %w", 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:") || strings.HasPrefix(serverPort, "name:")) {
		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
		}
	}

	if strings.HasPrefix(serverPort, "name:") {
		name := strings.TrimPrefix(serverPort, "name:")
		port := retrieveNamedPort(app, name)

		if port == 0 {
			return 0, fmt.Errorf("no port with name %s", name)
		}

		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:") {
		indexString := strings.TrimPrefix(serverPort, "index:")
		index, err := strconv.Atoi(indexString)
		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 retrieveNamedPort(app marathon.Application, name string) int {
	// Using port definition if available
	if app.PortDefinitions != nil && len(*app.PortDefinitions) > 0 {
		for _, def := range *app.PortDefinitions {
			if def.Port != nil && *def.Port > 0 && def.Name == name {
				return *def.Port
			}
		}
	}

	// If using IP-per-task using this port definition
	if app.IPAddressPerTask != nil && app.IPAddressPerTask.Discovery != nil && len(*(app.IPAddressPerTask.Discovery.Ports)) > 0 {
		for _, def := range *(app.IPAddressPerTask.Discovery.Ports) {
			if def.Number > 0 && def.Name == name {
				return def.Number
			}
		}
	}

	return 0
}

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{}
}