traefik/provider/marathon.go

package provider

import (
	"bytes"
	"errors"
	"strconv"
	"strings"
	"text/template"

	"github.com/BurntSushi/toml"
	"github.com/BurntSushi/ty/fun"
	log "github.com/Sirupsen/logrus"
	"github.com/emilevauge/traefik/autogen"
	"github.com/emilevauge/traefik/types"
	"github.com/gambol99/go-marathon"
)

// Marathon holds configuration of the Marathon provider.
type Marathon struct {
	Watch            bool
	Endpoint         string
	Domain           string
	Filename         string
	NetworkInterface string
	marathonClient   marathon.Marathon
}

// Provide allows the provider to provide configurations to traefik
// using the given configuration channel.
func (provider *Marathon) Provide(configurationChan chan<- types.ConfigMessage) error {
	config := marathon.NewDefaultConfig()
	config.URL = provider.Endpoint
	config.EventsInterface = provider.NetworkInterface
	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.EVENTS_APPLICATIONS); err != nil {
			log.Errorf("Failed to register for subscriptions, %s", err)
		} else {
			go func() {
				for {
					event := <-update
					log.Debug("Marathon event receveived", event)
					configuration := provider.loadMarathonConfig()
					if configuration != nil {
						configurationChan <- types.ConfigMessage{"marathon", configuration}
					}
				}
			}()
		}
	}

	configuration := provider.loadMarathonConfig()
	configurationChan <- types.ConfigMessage{"marathon", configuration}
	return nil
}

func (provider *Marathon) loadMarathonConfig() *types.Configuration {
	var MarathonFuncMap = template.FuncMap{
		"getPort": func(task marathon.Task) string {
			for _, port := range task.Ports {
				return strconv.Itoa(port)
			}
			return ""
		},
		"getWeight": func(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"
		},
		"getDomain": func(application marathon.Application) string {
			if label, err := provider.getLabel(application, "traefik.domain"); err == nil {
				return label
			}
			return provider.Domain
		},
		"replace": func(s1 string, s2 string, s3 string) string {
			return strings.Replace(s3, s1, s2, -1)
		},
		"getProtocol": func(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"
		},
		"getPassHostHeader": func(application marathon.Application) string {
			if passHostHeader, err := provider.getLabel(application, "traefik.frontend.passHostHeader"); err == nil {
				return passHostHeader
			}
			return "false"
		},
		"getFrontendValue": provider.GetFrontendValue,
		"getFrontendRule":  provider.GetFrontendRule,
	}
	configuration := new(types.Configuration)

	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()
	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 {
		if len(task.Ports) == 0 {
			log.Debug("Filtering marathon task without port %s", task.AppID)
			return false
		}
		application, errApp := getApplication(task, applications.Apps)
		if errApp != nil {
			log.Errorf("Unable to get marathon application from task %s", task.AppID)
			return false
		}
		_, err := strconv.Atoi(application.Labels["traefik.port"])
		if len(application.Ports) > 1 && err != nil {
			log.Debugf("Filtering marathon task %s with more than 1 port and no traefik.port label", task.AppID)
			return false
		}
		if application.Labels["traefik.enable"] == "false" {
			log.Debugf("Filtering disabled marathon task %s", task.AppID)
			return false
		}
		//filter healthchecks
		if application.HasHealthChecks() {
			if task.HasHealthCheckResults() {
				for _, healthcheck := range task.HealthCheckResult {
					// found one bad healthcheck, return false
					if !healthcheck.Alive {
						log.Debugf("Filtering marathon task %s with bad healthcheck", task.AppID)
						return false
					}
				}
			} else {
				log.Debugf("Filtering marathon task %s with bad healthcheck", task.AppID)
				return false
			}
		}
		return true
	}, tasks.Tasks).([]marathon.Task)

	//filter apps
	filteredApps := fun.Filter(func(app marathon.Application) bool {
		//get ports from app tasks
		if !fun.Exists(func(task marathon.Task) bool {
			if task.AppID == app.ID {
				return true
			}
			return false
		}, filteredTasks) {
			return false
		}
		return true
	}, applications.Apps).([]marathon.Application)

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

	tmpl := template.New(provider.Filename).Funcs(MarathonFuncMap)
	if len(provider.Filename) > 0 {
		_, err := tmpl.ParseFiles(provider.Filename)
		if err != nil {
			log.Error("Error reading file", err)
			return nil
		}
	} else {
		buf, err := autogen.Asset("templates/marathon.tmpl")
		if err != nil {
			log.Error("Error reading file", err)
		}
		_, err = tmpl.Parse(string(buf))
		if err != nil {
			log.Error("Error reading file", err)
			return nil
		}
	}

	var buffer bytes.Buffer

	err = tmpl.Execute(&buffer, templateObjects)
	if err != nil {
		log.Error("Error with marathon template:", err)
		return nil
	}

	if _, err := toml.Decode(buffer.String(), configuration); err != nil {
		log.Error("Error creating marathon configuration:", err)
		return nil
	}

	return configuration
}

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 (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) getEscapedName(name string) string {
	return strings.Replace(name, "/", "", -1)
}

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

// 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
	}
	return "Host"
}