package provider import ( "errors" "fmt" "strconv" "strings" "text/template" "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/mesos/mesos-go/detector" // Register mesos zoo the detector _ "github.com/mesos/mesos-go/detector/zoo" "github.com/mesosphere/mesos-dns/detect" "github.com/mesosphere/mesos-dns/logging" "github.com/mesosphere/mesos-dns/records" "github.com/mesosphere/mesos-dns/records/state" "github.com/mesosphere/mesos-dns/util" ) var _ Provider = (*Mesos)(nil) //Mesos holds configuration of the mesos provider. type Mesos struct { BaseProvider Endpoint string `description:"Mesos server endpoint. You can also specify multiple endpoint for Mesos"` Domain string `description:"Default domain used"` ExposedByDefault bool `description:"Expose Mesos apps by default"` GroupsAsSubDomains bool `description:"Convert Mesos groups to subdomains"` ZkDetectionTimeout int `description:"Zookeeper timeout (in seconds)"` RefreshSeconds int `description:"Polling interval (in seconds)"` IPSources string `description:"IPSources (e.g. host, docker, mesos, rkt)"` // e.g. "host", "docker", "mesos", "rkt" StateTimeoutSecond int `description:"HTTP Timeout (in seconds)"` Masters []string } // Provide allows the provider to provide configurations to traefik // using the given configuration channel. func (provider *Mesos) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool, constraints types.Constraints) error { operation := func() error { // initialize logging logging.SetupLogs() log.Debugf("%s", provider.IPSources) var zk string var masters []string if strings.HasPrefix(provider.Endpoint, "zk://") { zk = provider.Endpoint } else { masters = strings.Split(provider.Endpoint, ",") } errch := make(chan error) changed := detectMasters(zk, masters) reload := time.NewTicker(time.Second * time.Duration(provider.RefreshSeconds)) zkTimeout := time.Second * time.Duration(provider.ZkDetectionTimeout) timeout := time.AfterFunc(zkTimeout, func() { if zkTimeout > 0 { errch <- fmt.Errorf("master detection timed out after %s", zkTimeout) } }) defer reload.Stop() defer util.HandleCrash() if !provider.Watch { reload.Stop() timeout.Stop() } for { select { case <-reload.C: configuration := provider.loadMesosConfig() if configuration != nil { configurationChan <- types.ConfigMessage{ ProviderName: "mesos", Configuration: configuration, } } case masters := <-changed: if len(masters) == 0 || masters[0] == "" { // no leader timeout.Reset(zkTimeout) } else { timeout.Stop() } log.Debugf("new masters detected: %v", masters) provider.Masters = masters configuration := provider.loadMesosConfig() if configuration != nil { configurationChan <- types.ConfigMessage{ ProviderName: "mesos", Configuration: configuration, } } case err := <-errch: log.Errorf("%s", err) } } } notify := func(err error, time time.Duration) { log.Errorf("mesos 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 mesos server %+v", err) } return nil } func (provider *Mesos) loadMesosConfig() *types.Configuration { var mesosFuncMap = template.FuncMap{ "getBackend": provider.getBackend, "getPort": provider.getPort, "getHost": provider.getHost, "getWeight": provider.getWeight, "getDomain": provider.getDomain, "getProtocol": provider.getProtocol, "getPassHostHeader": provider.getPassHostHeader, "getPriority": provider.getPriority, "getEntryPoints": provider.getEntryPoints, "getFrontendRule": provider.getFrontendRule, "getFrontendBackend": provider.getFrontendBackend, "getID": provider.getID, "getFrontEndName": provider.getFrontEndName, } t := records.NewRecordGenerator(time.Duration(provider.StateTimeoutSecond) * time.Second) sj, err := t.FindMaster(provider.Masters...) if err != nil { log.Errorf("Failed to create a client for mesos, error: %s", err) return nil } tasks := provider.taskRecords(sj) //filter tasks filteredTasks := fun.Filter(func(task state.Task) bool { return mesosTaskFilter(task, provider.ExposedByDefault) }, tasks).([]state.Task) filteredApps := []state.Task{} for _, value := range filteredTasks { if !taskInSlice(value, filteredApps) { filteredApps = append(filteredApps, value) } } templateObjects := struct { Applications []state.Task Tasks []state.Task Domain string }{ filteredApps, filteredTasks, provider.Domain, } configuration, err := provider.getConfiguration("templates/mesos.tmpl", mesosFuncMap, templateObjects) if err != nil { log.Error(err) } return configuration } func taskInSlice(a state.Task, list []state.Task) bool { for _, b := range list { if b.DiscoveryInfo.Name == a.DiscoveryInfo.Name { return true } } return false } // labels returns all given Status.[]Labels' values whose keys are equal // to the given key func labels(task state.Task, key string) string { for _, l := range task.Labels { if l.Key == key { return l.Value } } return "" } func mesosTaskFilter(task state.Task, exposedByDefaultFlag bool) bool { if len(task.DiscoveryInfo.Ports.DiscoveryPorts) == 0 { log.Debugf("Filtering mesos task without port %s", task.Name) return false } if !isMesosApplicationEnabled(task, exposedByDefaultFlag) { log.Debugf("Filtering disabled mesos task %s", task.DiscoveryInfo.Name) return false } //filter indeterminable task port portIndexLabel := labels(task, "traefik.portIndex") portValueLabel := labels(task, "traefik.port") if portIndexLabel != "" && portValueLabel != "" { log.Debugf("Filtering mesos task %s specifying both traefik.portIndex and traefik.port labels", task.Name) return false } if portIndexLabel != "" { index, err := strconv.Atoi(labels(task, "traefik.portIndex")) if err != nil || index < 0 || index > len(task.DiscoveryInfo.Ports.DiscoveryPorts)-1 { log.Debugf("Filtering mesos task %s with unexpected value for traefik.portIndex label", task.Name) return false } } if portValueLabel != "" { port, err := strconv.Atoi(labels(task, "traefik.port")) if err != nil { log.Debugf("Filtering mesos task %s with unexpected value for traefik.port label", task.Name) return false } var foundPort bool for _, exposedPort := range task.DiscoveryInfo.Ports.DiscoveryPorts { if port == exposedPort.Number { foundPort = true break } } if !foundPort { log.Debugf("Filtering mesos task %s without a matching port for traefik.port label", task.Name) return false } } //filter healthchecks if task.Statuses != nil && len(task.Statuses) > 0 && task.Statuses[0].Healthy != nil && !*task.Statuses[0].Healthy { log.Debugf("Filtering mesos task %s with bad healthcheck", task.DiscoveryInfo.Name) return false } return true } func getMesos(task state.Task, apps []state.Task) (state.Task, error) { for _, application := range apps { if application.DiscoveryInfo.Name == task.DiscoveryInfo.Name { return application, nil } } return state.Task{}, errors.New("Application not found: " + task.DiscoveryInfo.Name) } func isMesosApplicationEnabled(task state.Task, exposedByDefault bool) bool { return exposedByDefault && labels(task, "traefik.enable") != "false" || labels(task, "traefik.enable") == "true" } func (provider *Mesos) getLabel(task state.Task, label string) (string, error) { for _, tmpLabel := range task.Labels { if tmpLabel.Key == label { return tmpLabel.Value, nil } } return "", errors.New("Label not found:" + label) } func (provider *Mesos) getPort(task state.Task, applications []state.Task) string { application, err := getMesos(task, applications) if err != nil { log.Errorf("Unable to get mesos application from task %s", task.DiscoveryInfo.Name) return "" } if portIndexLabel, err := provider.getLabel(application, "traefik.portIndex"); err == nil { if index, err := strconv.Atoi(portIndexLabel); err == nil { return strconv.Itoa(task.DiscoveryInfo.Ports.DiscoveryPorts[index].Number) } } if portValueLabel, err := provider.getLabel(application, "traefik.port"); err == nil { return portValueLabel } for _, port := range task.DiscoveryInfo.Ports.DiscoveryPorts { return strconv.Itoa(port.Number) } return "" } func (provider *Mesos) getWeight(task state.Task, applications []state.Task) string { application, errApp := getMesos(task, applications) if errApp != nil { log.Errorf("Unable to get mesos application from task %s", task.DiscoveryInfo.Name) return "0" } if label, err := provider.getLabel(application, "traefik.weight"); err == nil { return label } return "0" } func (provider *Mesos) getDomain(task state.Task) string { if label, err := provider.getLabel(task, "traefik.domain"); err == nil { return label } return provider.Domain } func (provider *Mesos) getProtocol(task state.Task, applications []state.Task) string { application, errApp := getMesos(task, applications) if errApp != nil { log.Errorf("Unable to get mesos application from task %s", task.DiscoveryInfo.Name) return "http" } if label, err := provider.getLabel(application, "traefik.protocol"); err == nil { return label } return "http" } func (provider *Mesos) getPassHostHeader(task state.Task) string { if passHostHeader, err := provider.getLabel(task, "traefik.frontend.passHostHeader"); err == nil { return passHostHeader } return "false" } func (provider *Mesos) getPriority(task state.Task) string { if priority, err := provider.getLabel(task, "traefik.frontend.priority"); err == nil { return priority } return "0" } func (provider *Mesos) getEntryPoints(task state.Task) []string { if entryPoints, err := provider.getLabel(task, "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 *Mesos) getFrontendRule(task state.Task) string { if label, err := provider.getLabel(task, "traefik.frontend.rule"); err == nil { return label } return "Host:" + strings.ToLower(strings.Replace(provider.getSubDomain(task.DiscoveryInfo.Name), "_", "-", -1)) + "." + provider.Domain } func (provider *Mesos) getBackend(task state.Task, applications []state.Task) string { application, errApp := getMesos(task, applications) if errApp != nil { log.Errorf("Unable to get mesos application from task %s", task.DiscoveryInfo.Name) return "" } return provider.getFrontendBackend(application) } func (provider *Mesos) getFrontendBackend(task state.Task) string { if label, err := provider.getLabel(task, "traefik.backend"); err == nil { return label } return "-" + cleanupSpecialChars(task.DiscoveryInfo.Name) } func (provider *Mesos) getHost(task state.Task) string { return task.IP(strings.Split(provider.IPSources, ",")...) } func (provider *Mesos) getID(task state.Task) string { return cleanupSpecialChars(task.ID) } func (provider *Mesos) getFrontEndName(task state.Task) string { return strings.Replace(cleanupSpecialChars(task.ID), "/", "-", -1) } func cleanupSpecialChars(s string) string { return strings.Replace(strings.Replace(strings.Replace(s, ".", "-", -1), ":", "-", -1), "_", "-", -1) } func detectMasters(zk string, masters []string) <-chan []string { changed := make(chan []string, 1) if zk != "" { log.Debugf("Starting master detector for ZK ", zk) if md, err := detector.New(zk); err != nil { log.Errorf("failed to create master detector: %v", err) } else if err := md.Detect(detect.NewMasters(masters, changed)); err != nil { log.Errorf("failed to initialize master detector: %v", err) } } else { changed <- masters } return changed } func (provider *Mesos) taskRecords(sj state.State) []state.Task { var p []state.Task // == nil for _, f := range sj.Frameworks { for _, task := range f.Tasks { for _, slave := range sj.Slaves { if task.SlaveID == slave.ID { task.SlaveIP = slave.Hostname } } // only do running and discoverable tasks if task.State == "TASK_RUNNING" { p = append(p, task) } } } return p } // ErrorFunction A function definition that returns an error // to be passed to the Ignore or Panic error handler type ErrorFunction func() error // Ignore Calls an ErrorFunction, and ignores the result. // This allows us to be more explicit when there is no error // handling to be done, for example in defers func Ignore(f ErrorFunction) { _ = f() } func (provider *Mesos) 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) }