Skip to content

Commit

Permalink
Merge pull request traefik#1474 from containous/marathon-check-port-l…
Browse files Browse the repository at this point in the history 
…abel-overwrite-earlier

Check for explicitly defined Marathon port first.
  • Loading branch information
@timoreimann
timoreimann authored Apr 26, 2017
2 parents 49a9aeb + 099d605 commit 750fa22
Show file tree
Hide file tree
Showing 3 changed files with 306 additions and 252 deletions.
169 changes: 90 additions & 79 deletions provider/marathon/marathon.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package marathon

import (
"errors"
"fmt"
"math"
"net"
"net/http"
Expand All @@ -22,6 +23,11 @@ import (
"github.com/gambol99/go-marathon"
)

const (
labelPort = "traefik.port"
labelPortIndex = "traefik.portIndex"
)

var _ provider.Provider = (*Provider)(nil)

// Provider holds configuration of the provider.
Expand Down Expand Up @@ -194,75 +200,63 @@ func (p *Provider) loadMarathonConfig() *types.Configuration {
func (p *Provider) taskFilter(task marathon.Task, applications *marathon.Applications, exposedByDefaultFlag bool) bool {
application, err := getApplication(task, applications.Apps)
if err != nil {
log.Errorf("Unable to get marathon application from task %s", task.AppID)
log.Errorf("Unable to get Marathon application %s for task %s", task.AppID, task.ID)
return false
}
ports := processPorts(application, task)
if len(ports) == 0 {
log.Debug("Filtering marathon task without port %s", task.AppID)
if _, err = processPorts(application, task); err != nil {
log.Errorf("Filtering Marathon task %s from application %s without port: %s", task.ID, application.ID, err)
return false
}

// Filter illegal port label specification.
_, hasPortIndexLabel := p.getLabel(application, labelPortIndex)
_, hasPortLabel := p.getLabel(application, labelPort)
if hasPortIndexLabel && hasPortLabel {
log.Debugf("Filtering Marathon task %s from application %s specifying both traefik.portIndex and traefik.port labels", task.ID, application.ID)
return false
}

// Filter by constraints.
label, _ := p.getLabel(application, "traefik.tags")
constraintTags := strings.Split(label, ",")
if p.MarathonLBCompatibility {
if label, err := p.getLabel(application, "HAPROXY_GROUP"); err == nil {
if label, ok := p.getLabel(application, "HAPROXY_GROUP"); ok {
constraintTags = append(constraintTags, label)
}
}
if ok, failingConstraint := p.MatchConstraints(constraintTags); !ok {
if failingConstraint != nil {
log.Debugf("Application %v pruned by '%v' constraint", application.ID, failingConstraint.String())
log.Debugf("Filtering Marathon task %s from application %s pruned by '%v' constraint", task.ID, application.ID, failingConstraint.String())
}
return false
}

// Filter disabled application.
if !isApplicationEnabled(application, exposedByDefaultFlag) {
log.Debugf("Filtering disabled marathon task %s", task.AppID)
return false
}

//filter indeterminable task port
portIndexLabel := (*application.Labels)["traefik.portIndex"]
portValueLabel := (*application.Labels)["traefik.port"]
if portIndexLabel != "" && portValueLabel != "" {
log.Debugf("Filtering marathon task %s specifying both traefik.portIndex and traefik.port labels", task.AppID)
log.Debugf("Filtering disabled Marathon task %s from application %s", task.ID, application.ID)
return false
}
if portIndexLabel != "" {
index, err := strconv.Atoi((*application.Labels)["traefik.portIndex"])
if err != nil || index < 0 || index > len(ports)-1 {
log.Debugf("Filtering marathon task %s with unexpected value for traefik.portIndex label", task.AppID)
return false
}
}
if portValueLabel != "" {
_, err := strconv.Atoi((*application.Labels)["traefik.port"])
if err != nil {
log.Debugf("Filtering marathon task %s with unexpected value for traefik.port label", task.AppID)
return false
}
}

//filter healthchecks
// Filter task with existing, bad health check results.
if application.HasHealthChecks() {
if task.HasHealthCheckResults() {
for _, healthcheck := range task.HealthCheckResults {
// found one bad healthcheck, return false
if !healthcheck.Alive {
log.Debugf("Filtering marathon task %s with bad healthcheck", task.AppID)
log.Debugf("Filtering Marathon task %s from application %s with bad health check", task.ID, application.ID)
return false
}
}
}
}

return true
}

func (p *Provider) applicationFilter(app marathon.Application, filteredTasks []marathon.Task) bool {
label, _ := p.getLabel(app, "traefik.tags")
constraintTags := strings.Split(label, ",")
if p.MarathonLBCompatibility {
if label, err := p.getLabel(app, "HAPROXY_GROUP"); err == nil {
if label, ok := p.getLabel(app, "HAPROXY_GROUP"); ok {
constraintTags = append(constraintTags, label)
}
}
Expand Down Expand Up @@ -291,35 +285,28 @@ func isApplicationEnabled(application marathon.Application, exposedByDefault boo
return exposedByDefault && (*application.Labels)["traefik.enable"] != "false" || (*application.Labels)["traefik.enable"] == "true"
}

func (p *Provider) getLabel(application marathon.Application, label string) (string, error) {
func (p *Provider) getLabel(application marathon.Application, label string) (string, bool) {
for key, value := range *application.Labels {
if key == label {
return value, nil
return value, true
}
}
return "", errors.New("Label not found:" + label)
return "", false
}

func (p *Provider) getPort(task marathon.Task, applications []marathon.Application) string {
application, err := getApplication(task, applications)
if err != nil {
log.Errorf("Unable to get marathon application from task %s", task.AppID)
log.Errorf("Unable to get Marathon application %s for task %s", application.ID, task.ID)
return ""
}
ports := processPorts(application, task)
if portIndexLabel, err := p.getLabel(application, "traefik.portIndex"); err == nil {
if index, err := strconv.Atoi(portIndexLabel); err == nil {
return strconv.Itoa(ports[index])
}
}
if portValueLabel, err := p.getLabel(application, "traefik.port"); err == nil {
return portValueLabel
port, err := processPorts(application, task)
if err != nil {
log.Errorf("Unable to process ports for Marathon application %s and task %s: %s", application.ID, task.ID, err)
return ""
}

for _, port := range ports {
return strconv.Itoa(port)
}
return ""
return strconv.Itoa(port)
}

func (p *Provider) getWeight(task marathon.Task, applications []marathon.Application) string {
Expand All @@ -328,14 +315,14 @@ func (p *Provider) getWeight(task marathon.Task, applications []marathon.Applica
log.Errorf("Unable to get marathon application from task %s", task.AppID)
return "0"
}
if label, err := p.getLabel(application, "traefik.weight"); err == nil {
if label, ok := p.getLabel(application, "traefik.weight"); ok {
return label
}
return "0"
}

func (p *Provider) getDomain(application marathon.Application) string {
if label, err := p.getLabel(application, "traefik.domain"); err == nil {
if label, ok := p.getLabel(application, "traefik.domain"); ok {
return label
}
return p.Domain
Expand All @@ -347,35 +334,35 @@ func (p *Provider) getProtocol(task marathon.Task, applications []marathon.Appli
log.Errorf("Unable to get marathon application from task %s", task.AppID)
return "http"
}
if label, err := p.getLabel(application, "traefik.protocol"); err == nil {
if label, ok := p.getLabel(application, "traefik.protocol"); ok {
return label
}
return "http"
}

func (p *Provider) getSticky(application marathon.Application) string {
if sticky, err := p.getLabel(application, "traefik.backend.loadbalancer.sticky"); err == nil {
if sticky, ok := p.getLabel(application, "traefik.backend.loadbalancer.sticky"); ok {
return sticky
}
return "false"
}

func (p *Provider) getPassHostHeader(application marathon.Application) string {
if passHostHeader, err := p.getLabel(application, "traefik.frontend.passHostHeader"); err == nil {
if passHostHeader, ok := p.getLabel(application, "traefik.frontend.passHostHeader"); ok {
return passHostHeader
}
return "true"
}

func (p *Provider) getPriority(application marathon.Application) string {
if priority, err := p.getLabel(application, "traefik.frontend.priority"); err == nil {
if priority, ok := p.getLabel(application, "traefik.frontend.priority"); ok {
return priority
}
return "0"
}

func (p *Provider) getEntryPoints(application marathon.Application) []string {
if entryPoints, err := p.getLabel(application, "traefik.frontend.entryPoints"); err == nil {
if entryPoints, ok := p.getLabel(application, "traefik.frontend.entryPoints"); ok {
return strings.Split(entryPoints, ",")
}
return []string{}
Expand All @@ -384,11 +371,11 @@ func (p *Provider) getEntryPoints(application marathon.Application) []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 (p *Provider) getFrontendRule(application marathon.Application) string {
if label, err := p.getLabel(application, "traefik.frontend.rule"); err == nil {
if label, ok := p.getLabel(application, "traefik.frontend.rule"); ok {
return label
}
if p.MarathonLBCompatibility {
if label, err := p.getLabel(application, "HAPROXY_0_VHOST"); err == nil {
if label, ok := p.getLabel(application, "HAPROXY_0_VHOST"); ok {
return "Host:" + label
}
}
Expand All @@ -405,7 +392,7 @@ func (p *Provider) getBackend(task marathon.Task, applications []marathon.Applic
}

func (p *Provider) getFrontendBackend(application marathon.Application) string {
if label, err := p.getLabel(application, "traefik.backend"); err == nil {
if label, ok := p.getLabel(application, "traefik.backend"); ok {
return label
}
return provider.Replace("/", "-", application.ID)
Expand All @@ -422,33 +409,26 @@ func (p *Provider) getSubDomain(name string) string {
}

func (p *Provider) hasCircuitBreakerLabels(application marathon.Application) bool {
if _, err := p.getLabel(application, "traefik.backend.circuitbreaker.expression"); err != nil {
return false
}
return true
_, ok := p.getLabel(application, "traefik.backend.circuitbreaker.expression")
return ok
}

func (p *Provider) hasLoadBalancerLabels(application marathon.Application) bool {
_, errMethod := p.getLabel(application, "traefik.backend.loadbalancer.method")
_, errSticky := p.getLabel(application, "traefik.backend.loadbalancer.sticky")
if errMethod != nil && errSticky != nil {
return false
}
return true
return errMethod || errSticky
}

func (p *Provider) hasMaxConnLabels(application marathon.Application) bool {
if _, err := p.getLabel(application, "traefik.backend.maxconn.amount"); err != nil {
return false
}
if _, err := p.getLabel(application, "traefik.backend.maxconn.extractorfunc"); err != nil {
if _, ok := p.getLabel(application, "traefik.backend.maxconn.amount"); !ok {
return false
}
return true
_, ok := p.getLabel(application, "traefik.backend.maxconn.extractorfunc")
return ok
}

func (p *Provider) getMaxConnAmount(application marathon.Application) int64 {
if label, err := p.getLabel(application, "traefik.backend.maxconn.amount"); err == nil {
if label, ok := p.getLabel(application, "traefik.backend.maxconn.amount"); ok {
i, errConv := strconv.ParseInt(label, 10, 64)
if errConv != nil {
log.Errorf("Unable to parse traefik.backend.maxconn.amount %s", label)
Expand All @@ -460,28 +440,59 @@ func (p *Provider) getMaxConnAmount(application marathon.Application) int64 {
}

func (p *Provider) getMaxConnExtractorFunc(application marathon.Application) string {
if label, err := p.getLabel(application, "traefik.backend.maxconn.extractorfunc"); err == nil {
if label, ok := p.getLabel(application, "traefik.backend.maxconn.extractorfunc"); ok {
return label
}
return "request.host"
}

func (p *Provider) getLoadBalancerMethod(application marathon.Application) string {
if label, err := p.getLabel(application, "traefik.backend.loadbalancer.method"); err == nil {
if label, ok := p.getLabel(application, "traefik.backend.loadbalancer.method"); ok {
return label
}
return "wrr"
}

func (p *Provider) getCircuitBreakerExpression(application marathon.Application) string {
if label, err := p.getLabel(application, "traefik.backend.circuitbreaker.expression"); err == nil {
if label, ok := p.getLabel(application, "traefik.backend.circuitbreaker.expression"); ok {
return label
}
return "NetworkErrorRatio() > 1"
}

func processPorts(application marathon.Application, task marathon.Task) []int {
func processPorts(application marathon.Application, task marathon.Task) (int, error) {
if portLabel, ok := (*application.Labels)[labelPort]; ok {
port, err := strconv.Atoi(portLabel)
switch {
case err != nil:
return 0, fmt.Errorf("failed to parse port label: %s", err)
case port <= 0:
return 0, fmt.Errorf("explicitly specified port %d must be larger than zero", port)
}
return port, nil
}

ports := retrieveAvailablePorts(application, task)
if len(ports) == 0 {
return 0, errors.New("no port found")
}

portIndex := 0
portIndexLabel, ok := (*application.Labels)[labelPortIndex]
if ok {
var err error
portIndex, err = strconv.Atoi(portIndexLabel)
switch {
case err != nil:
return 0, fmt.Errorf("failed to parse port index label: %s", err)
case portIndex < 0, portIndex > len(ports)-1:
return 0, fmt.Errorf("port index %d must be within port range (0, %d)", portIndex, len(ports)-1)
}
}
return ports[portIndex], nil
}

func retrieveAvailablePorts(application marathon.Application, task marathon.Task) []int {
// Using default port configuration
if task.Ports != nil && len(task.Ports) > 0 {
return task.Ports
Expand Down Expand Up @@ -520,8 +531,8 @@ func (p *Provider) getBackendServer(task marathon.Task, applications []marathon.
} else if len(task.IPAddresses) == 1 {
return task.IPAddresses[0].IPAddress
} else {
ipAddressIdxStr, err := p.getLabel(application, "traefik.ipAddressIdx")
if err != nil {
ipAddressIdxStr, ok := p.getLabel(application, "traefik.ipAddressIdx")
if !ok {
log.Errorf("Unable to get marathon IPAddress from task %s", task.AppID)
return ""
}
Expand Down
Loading

0 comments on commit 750fa22

Please sign in to comment.