traefik/marathon.go

231 lines
6.3 KiB
Go
Raw Normal View History

2015-09-09 20:39:08 +00:00
package main
2015-09-12 13:10:03 +00:00
2015-09-09 20:39:08 +00:00
import (
"bytes"
"strconv"
"strings"
"text/template"
"errors"
2015-09-09 20:39:08 +00:00
"github.com/BurntSushi/toml"
2015-09-10 20:54:37 +00:00
"github.com/BurntSushi/ty/fun"
log "github.com/Sirupsen/logrus"
2015-09-12 13:10:03 +00:00
"github.com/gambol99/go-marathon"
2015-09-09 20:39:08 +00:00
)
type MarathonProvider struct {
2015-09-10 13:13:35 +00:00
Watch bool
Endpoint string
marathonClient marathon.Marathon
Domain string
Filename string
NetworkInterface string
2015-09-09 20:39:08 +00:00
}
func (provider *MarathonProvider) Provide(configurationChan chan<- configMessage) error {
2015-09-09 20:39:08 +00:00
config := marathon.NewDefaultConfig()
config.URL = provider.Endpoint
2015-09-10 13:13:35 +00:00
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 <- configMessage{"marathon", configuration}
2015-09-09 21:09:16 +00:00
}
}
}()
2015-09-09 21:09:16 +00:00
}
2015-09-09 20:39:08 +00:00
}
configuration := provider.loadMarathonConfig()
configurationChan <- configMessage{"marathon", configuration}
return nil
2015-09-09 20:39:08 +00:00
}
func (provider *MarathonProvider) loadMarathonConfig() *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 {
2015-10-26 23:26:35 +00:00
application, errApp := getApplication(task, applications)
if errApp != nil {
log.Errorf("Unable to get marathon application from task %s", task.AppID)
return "0"
}
2015-10-26 23:26:35 +00:00
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)
},
2015-10-23 16:59:08 +00:00
"getProtocol": func(task marathon.Task, applications []marathon.Application) string {
2015-10-26 23:26:35 +00:00
application, errApp := getApplication(task, applications)
if errApp != nil {
log.Errorf("Unable to get marathon application from task %s", task.AppID)
return "http"
}
2015-10-26 23:26:35 +00:00
if label, err := provider.getLabel(application, "traefik.protocol"); err == nil {
return label
}
return "http"
},
"getFrontendValue": provider.GetFrontendValue,
"getFrontendRule": provider.GetFrontendRule,
}
2015-09-09 20:39:08 +00:00
configuration := new(Configuration)
applications, err := provider.marathonClient.Applications(nil)
2015-09-12 13:10:03 +00:00
if err != nil {
log.Errorf("Failed to create a client for marathon, error: %s", err)
2015-09-09 20:39:08 +00:00
return nil
}
2015-09-10 20:54:37 +00:00
2015-09-09 20:39:08 +00:00
tasks, err := provider.marathonClient.AllTasks()
2015-09-12 13:10:03 +00:00
if err != nil {
log.Errorf("Failed to create a client for marathon, error: %s", err)
2015-09-09 20:39:08 +00:00
return nil
}
2015-09-10 20:54:37 +00:00
//filter tasks
filteredTasks := fun.Filter(func(task marathon.Task) bool {
2015-09-12 13:10:03 +00:00
if len(task.Ports) == 0 {
2015-09-11 14:37:13 +00:00
log.Debug("Filtering marathon task without port", task.AppID)
2015-09-10 20:54:37 +00:00
return false
}
2015-10-26 23:26:35 +00:00
application, errApp := getApplication(task, applications.Apps)
if errApp != nil {
log.Errorf("Unable to get marathon application from task %s", task.AppID)
return false
}
2015-09-10 20:54:37 +00:00
_, err := strconv.Atoi(application.Labels["traefik.port"])
2015-09-12 13:10:03 +00:00
if len(application.Ports) > 1 && err != nil {
2015-09-11 14:37:13 +00:00
log.Debug("Filtering marathon task with more than 1 port and no traefik.port label", task.AppID)
2015-09-10 20:54:37 +00:00
return false
}
2015-09-12 13:10:03 +00:00
if application.Labels["traefik.enable"] == "false" {
2015-09-11 14:37:13 +00:00
log.Debug("Filtering disabled marathon task", task.AppID)
2015-09-10 20:54:37 +00:00
return false
}
return true
}, tasks.Tasks).([]marathon.Task)
//filter apps
filteredApps := fun.Filter(func(app marathon.Application) bool {
//get ports from app tasks
2015-09-12 13:10:03 +00:00
if !fun.Exists(func(task marathon.Task) bool {
if task.AppID == app.ID {
2015-09-10 20:54:37 +00:00
return true
}
return false
2015-09-12 13:10:03 +00:00
}, filteredTasks) {
2015-09-10 20:54:37 +00:00
return false
}
return true
}, applications.Apps).([]marathon.Application)
2015-09-09 20:39:08 +00:00
templateObjects := struct {
Applications []marathon.Application
Tasks []marathon.Task
Domain string
}{
2015-09-10 20:54:37 +00:00
filteredApps,
filteredTasks,
2015-09-09 20:39:08 +00:00
provider.Domain,
}
2015-09-21 16:05:56 +00:00
tmpl := template.New(provider.Filename).Funcs(MarathonFuncMap)
2015-09-12 13:10:03 +00:00
if len(provider.Filename) > 0 {
2015-09-11 17:25:49 +00:00
_, err := tmpl.ParseFiles(provider.Filename)
if err != nil {
log.Error("Error reading file", err)
return nil
}
2015-09-12 13:10:03 +00:00
} else {
buf, err := Asset("templates/marathon.tmpl")
2015-09-11 17:25:49 +00:00
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
}
2015-09-09 20:39:08 +00:00
}
var buffer bytes.Buffer
err = tmpl.Execute(&buffer, templateObjects)
if err != nil {
2015-09-19 11:02:59 +00:00
log.Error("Error with marathon template:", err)
2015-09-09 20:39:08 +00:00
return nil
}
if _, err := toml.Decode(buffer.String(), configuration); err != nil {
2015-09-11 14:37:13 +00:00
log.Error("Error creating marathon configuration:", err)
2015-09-09 20:39:08 +00:00
return nil
}
return configuration
2015-09-10 20:54:37 +00:00
}
2015-10-26 23:26:35 +00:00
func getApplication(task marathon.Task, apps []marathon.Application) (marathon.Application, error) {
2015-09-10 20:54:37 +00:00
for _, application := range apps {
2015-09-12 13:10:03 +00:00
if application.ID == task.AppID {
2015-10-26 23:26:35 +00:00
return application, nil
2015-09-10 20:54:37 +00:00
}
}
2015-10-26 23:26:35 +00:00
return marathon.Application{}, errors.New("Application not found: " + task.AppID)
2015-09-12 13:10:03 +00:00
}
func (provider *MarathonProvider) 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 *MarathonProvider) getEscapedName(name string) string {
return strings.Replace(name, "/", "", -1)
}
func (provider *MarathonProvider) 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
}
func (provider *MarathonProvider) GetFrontendRule(application marathon.Application) string {
if label, err := provider.getLabel(application, "traefik.frontend.rule"); err == nil {
return label
}
return "Host"
}