refactor(consulCatalog): review and split.
This commit is contained in:
Fernandez Ludovic
Traefiker
parent
c705d6f9b3
commit
1710800cc0
3 changed files with 544 additions and 402 deletions
|
|
@ -1,9 +1,7 @@
|
|||
package consul
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"text/template"
|
||||
|
|
@ -14,6 +12,7 @@ import (
|
|||
"github.com/containous/traefik/job"
|
||||
"github.com/containous/traefik/log"
|
||||
"github.com/containous/traefik/provider"
|
||||
"github.com/containous/traefik/provider/label"
|
||||
"github.com/containous/traefik/safe"
|
||||
"github.com/containous/traefik/types"
|
||||
"github.com/hashicorp/consul/api"
|
||||
|
|
@ -38,6 +37,13 @@ type CatalogProvider struct {
|
|||
frontEndRuleTemplate *template.Template
|
||||
}
|
||||
|
||||
// Service represent a Consul service.
|
||||
type Service struct {
|
||||
Name string
|
||||
Tags []string
|
||||
Nodes []string
|
||||
}
|
||||
|
||||
type serviceUpdate struct {
|
||||
ServiceName string
|
||||
Attributes []string
|
||||
|
|
@ -59,63 +65,149 @@ func (a nodeSorter) Swap(i int, j int) {
|
|||
}
|
||||
|
||||
func (a nodeSorter) Less(i int, j int) bool {
|
||||
lentr := a[i]
|
||||
rentr := a[j]
|
||||
lEntry := a[i]
|
||||
rEntry := a[j]
|
||||
|
||||
ls := strings.ToLower(lentr.Service.Service)
|
||||
lr := strings.ToLower(rentr.Service.Service)
|
||||
ls := strings.ToLower(lEntry.Service.Service)
|
||||
lr := strings.ToLower(rEntry.Service.Service)
|
||||
|
||||
if ls != lr {
|
||||
return ls < lr
|
||||
}
|
||||
if lentr.Service.Address != rentr.Service.Address {
|
||||
return lentr.Service.Address < rentr.Service.Address
|
||||
if lEntry.Service.Address != rEntry.Service.Address {
|
||||
return lEntry.Service.Address < rEntry.Service.Address
|
||||
}
|
||||
if lentr.Node.Address != rentr.Node.Address {
|
||||
return lentr.Node.Address < rentr.Node.Address
|
||||
if lEntry.Node.Address != rEntry.Node.Address {
|
||||
return lEntry.Node.Address < rEntry.Node.Address
|
||||
}
|
||||
return lentr.Service.Port < rentr.Service.Port
|
||||
return lEntry.Service.Port < rEntry.Service.Port
|
||||
}
|
||||
|
||||
func hasChanged(current map[string]Service, previous map[string]Service) bool {
|
||||
addedServiceKeys, removedServiceKeys := getChangedServiceKeys(current, previous)
|
||||
return len(removedServiceKeys) > 0 || len(addedServiceKeys) > 0 || hasNodeOrTagsChanged(current, previous)
|
||||
// Provide allows the consul catalog provider to provide configurations to traefik
|
||||
// using the given configuration channel.
|
||||
func (p *CatalogProvider) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool, constraints types.Constraints) error {
|
||||
config := api.DefaultConfig()
|
||||
config.Address = p.Endpoint
|
||||
client, err := api.NewClient(config)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.client = client
|
||||
p.Constraints = append(p.Constraints, constraints...)
|
||||
p.setupFrontEndTemplate()
|
||||
|
||||
pool.Go(func(stop chan bool) {
|
||||
notify := func(err error, time time.Duration) {
|
||||
log.Errorf("Consul connection error %+v, retrying in %s", err, time)
|
||||
}
|
||||
operation := func() error {
|
||||
return p.watch(configurationChan, stop)
|
||||
}
|
||||
errRetry := backoff.RetryNotify(safe.OperationWithRecover(operation), job.NewBackOff(backoff.NewExponentialBackOff()), notify)
|
||||
if errRetry != nil {
|
||||
log.Errorf("Cannot connect to consul server %+v", errRetry)
|
||||
}
|
||||
})
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func getChangedServiceKeys(current map[string]Service, previous map[string]Service) ([]string, []string) {
|
||||
currKeySet := fun.Set(fun.Keys(current).([]string)).(map[string]bool)
|
||||
prevKeySet := fun.Set(fun.Keys(previous).([]string)).(map[string]bool)
|
||||
func (p *CatalogProvider) watch(configurationChan chan<- types.ConfigMessage, stop chan bool) error {
|
||||
stopCh := make(chan struct{})
|
||||
watchCh := make(chan map[string][]string)
|
||||
errorCh := make(chan error)
|
||||
|
||||
addedKeys := fun.Difference(currKeySet, prevKeySet).(map[string]bool)
|
||||
removedKeys := fun.Difference(prevKeySet, currKeySet).(map[string]bool)
|
||||
p.watchHealthState(stopCh, watchCh, errorCh)
|
||||
p.watchCatalogServices(stopCh, watchCh, errorCh)
|
||||
|
||||
return fun.Keys(addedKeys).([]string), fun.Keys(removedKeys).([]string)
|
||||
}
|
||||
defer close(stopCh)
|
||||
defer close(watchCh)
|
||||
|
||||
func hasNodeOrTagsChanged(current map[string]Service, previous map[string]Service) bool {
|
||||
var added []string
|
||||
var removed []string
|
||||
for key, value := range current {
|
||||
if prevValue, ok := previous[key]; ok {
|
||||
addedNodesKeys, removedNodesKeys := getChangedStringKeys(value.Nodes, prevValue.Nodes)
|
||||
added = append(added, addedNodesKeys...)
|
||||
removed = append(removed, removedNodesKeys...)
|
||||
addedTagsKeys, removedTagsKeys := getChangedStringKeys(value.Tags, prevValue.Tags)
|
||||
added = append(added, addedTagsKeys...)
|
||||
removed = append(removed, removedTagsKeys...)
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return nil
|
||||
case index, ok := <-watchCh:
|
||||
if !ok {
|
||||
return errors.New("Consul service list nil")
|
||||
}
|
||||
log.Debug("List of services changed")
|
||||
nodes, err := p.getNodes(index)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
configuration := p.buildConfiguration(nodes)
|
||||
configurationChan <- types.ConfigMessage{
|
||||
ProviderName: "consul_catalog",
|
||||
Configuration: configuration,
|
||||
}
|
||||
case err := <-errorCh:
|
||||
return err
|
||||
}
|
||||
}
|
||||
return len(added) > 0 || len(removed) > 0
|
||||
}
|
||||
|
||||
func getChangedStringKeys(currState []string, prevState []string) ([]string, []string) {
|
||||
currKeySet := fun.Set(currState).(map[string]bool)
|
||||
prevKeySet := fun.Set(prevState).(map[string]bool)
|
||||
func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh chan<- map[string][]string, errorCh chan<- error) {
|
||||
catalog := p.client.Catalog()
|
||||
|
||||
addedKeys := fun.Difference(currKeySet, prevKeySet).(map[string]bool)
|
||||
removedKeys := fun.Difference(prevKeySet, currKeySet).(map[string]bool)
|
||||
safe.Go(func() {
|
||||
// variable to hold previous state
|
||||
var flashback map[string]Service
|
||||
|
||||
return fun.Keys(addedKeys).([]string), fun.Keys(removedKeys).([]string)
|
||||
options := &api.QueryOptions{WaitTime: DefaultWatchWaitTime}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-stopCh:
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
data, meta, err := catalog.Services(options)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to list services: %v", err)
|
||||
errorCh <- err
|
||||
return
|
||||
}
|
||||
|
||||
if options.WaitIndex == meta.LastIndex {
|
||||
continue
|
||||
}
|
||||
|
||||
options.WaitIndex = meta.LastIndex
|
||||
|
||||
if data != nil {
|
||||
current := make(map[string]Service)
|
||||
for key, value := range data {
|
||||
nodes, _, err := catalog.Service(key, "", &api.QueryOptions{})
|
||||
if err != nil {
|
||||
log.Errorf("Failed to get detail of service %s: %v", key, err)
|
||||
errorCh <- err
|
||||
return
|
||||
}
|
||||
nodesID := getServiceIds(nodes)
|
||||
if service, ok := current[key]; ok {
|
||||
service.Tags = value
|
||||
service.Nodes = nodesID
|
||||
} else {
|
||||
service := Service{
|
||||
Name: key,
|
||||
Tags: value,
|
||||
Nodes: nodesID,
|
||||
}
|
||||
current[key] = service
|
||||
}
|
||||
}
|
||||
// A critical note is that the return of a blocking request is no guarantee of a change.
|
||||
// It is possible that there was an idempotent write that does not affect the result of the query.
|
||||
// Thus it is required to do extra check for changes...
|
||||
if hasChanged(current, flashback) {
|
||||
watchCh <- data
|
||||
flashback = current
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<- map[string][]string, errorCh chan<- error) {
|
||||
|
|
@ -162,7 +254,7 @@ func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<
|
|||
// The response should be unified with watchCatalogServices
|
||||
data, _, err := catalog.Services(&api.QueryOptions{})
|
||||
if err != nil {
|
||||
log.Errorf("Failed to list services: %s", err)
|
||||
log.Errorf("Failed to list services: %v", err)
|
||||
errorCh <- err
|
||||
return
|
||||
}
|
||||
|
|
@ -189,74 +281,67 @@ func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<
|
|||
})
|
||||
}
|
||||
|
||||
// Service represent a Consul service.
|
||||
type Service struct {
|
||||
Name string
|
||||
Tags []string
|
||||
Nodes []string
|
||||
}
|
||||
func (p *CatalogProvider) getNodes(index map[string][]string) ([]catalogUpdate, error) {
|
||||
visited := make(map[string]bool)
|
||||
|
||||
func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh chan<- map[string][]string, errorCh chan<- error) {
|
||||
catalog := p.client.Catalog()
|
||||
|
||||
safe.Go(func() {
|
||||
// variable to hold previous state
|
||||
var flashback map[string]Service
|
||||
|
||||
options := &api.QueryOptions{WaitTime: DefaultWatchWaitTime}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-stopCh:
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
data, meta, err := catalog.Services(options)
|
||||
var nodes []catalogUpdate
|
||||
for service := range index {
|
||||
name := strings.ToLower(service)
|
||||
if !strings.Contains(name, " ") && !visited[name] {
|
||||
visited[name] = true
|
||||
log.WithField("service", name).Debug("Fetching service")
|
||||
healthy, err := p.healthyNodes(name)
|
||||
if err != nil {
|
||||
log.Errorf("Failed to list services: %s", err)
|
||||
errorCh <- err
|
||||
return
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if options.WaitIndex == meta.LastIndex {
|
||||
continue
|
||||
}
|
||||
|
||||
options.WaitIndex = meta.LastIndex
|
||||
|
||||
if data != nil {
|
||||
current := make(map[string]Service)
|
||||
for key, value := range data {
|
||||
nodes, _, err := catalog.Service(key, "", &api.QueryOptions{})
|
||||
if err != nil {
|
||||
log.Errorf("Failed to get detail of service %s: %s", key, err)
|
||||
errorCh <- err
|
||||
return
|
||||
}
|
||||
nodesID := getServiceIds(nodes)
|
||||
if service, ok := current[key]; ok {
|
||||
service.Tags = value
|
||||
service.Nodes = nodesID
|
||||
} else {
|
||||
service := Service{
|
||||
Name: key,
|
||||
Tags: value,
|
||||
Nodes: nodesID,
|
||||
}
|
||||
current[key] = service
|
||||
}
|
||||
}
|
||||
// A critical note is that the return of a blocking request is no guarantee of a change.
|
||||
// It is possible that there was an idempotent write that does not affect the result of the query.
|
||||
// Thus it is required to do extra check for changes...
|
||||
if hasChanged(current, flashback) {
|
||||
watchCh <- data
|
||||
flashback = current
|
||||
}
|
||||
// healthy.Nodes can be empty if constraints do not match, without throwing error
|
||||
if healthy.Service != nil && len(healthy.Nodes) > 0 {
|
||||
nodes = append(nodes, healthy)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
return nodes, nil
|
||||
}
|
||||
|
||||
func hasChanged(current map[string]Service, previous map[string]Service) bool {
|
||||
addedServiceKeys, removedServiceKeys := getChangedServiceKeys(current, previous)
|
||||
return len(removedServiceKeys) > 0 || len(addedServiceKeys) > 0 || hasNodeOrTagsChanged(current, previous)
|
||||
}
|
||||
|
||||
func getChangedServiceKeys(current map[string]Service, previous map[string]Service) ([]string, []string) {
|
||||
currKeySet := fun.Set(fun.Keys(current).([]string)).(map[string]bool)
|
||||
prevKeySet := fun.Set(fun.Keys(previous).([]string)).(map[string]bool)
|
||||
|
||||
addedKeys := fun.Difference(currKeySet, prevKeySet).(map[string]bool)
|
||||
removedKeys := fun.Difference(prevKeySet, currKeySet).(map[string]bool)
|
||||
|
||||
return fun.Keys(addedKeys).([]string), fun.Keys(removedKeys).([]string)
|
||||
}
|
||||
|
||||
func hasNodeOrTagsChanged(current map[string]Service, previous map[string]Service) bool {
|
||||
var added []string
|
||||
var removed []string
|
||||
for key, value := range current {
|
||||
if prevValue, ok := previous[key]; ok {
|
||||
addedNodesKeys, removedNodesKeys := getChangedStringKeys(value.Nodes, prevValue.Nodes)
|
||||
added = append(added, addedNodesKeys...)
|
||||
removed = append(removed, removedNodesKeys...)
|
||||
addedTagsKeys, removedTagsKeys := getChangedStringKeys(value.Tags, prevValue.Tags)
|
||||
added = append(added, addedTagsKeys...)
|
||||
removed = append(removed, removedTagsKeys...)
|
||||
}
|
||||
}
|
||||
return len(added) > 0 || len(removed) > 0
|
||||
}
|
||||
|
||||
func getChangedStringKeys(currState []string, prevState []string) ([]string, []string) {
|
||||
currKeySet := fun.Set(currState).(map[string]bool)
|
||||
prevKeySet := fun.Set(prevState).(map[string]bool)
|
||||
|
||||
addedKeys := fun.Difference(currKeySet, prevKeySet).(map[string]bool)
|
||||
removedKeys := fun.Difference(prevKeySet, currKeySet).(map[string]bool)
|
||||
|
||||
return fun.Keys(addedKeys).([]string), fun.Keys(removedKeys).([]string)
|
||||
}
|
||||
|
||||
func getServiceIds(services []*api.CatalogService) []string {
|
||||
|
|
@ -314,7 +399,7 @@ func (p *CatalogProvider) nodeFilter(service string, node *api.ServiceEntry) boo
|
|||
}
|
||||
|
||||
func (p *CatalogProvider) isServiceEnabled(node *api.ServiceEntry) bool {
|
||||
enable, err := strconv.ParseBool(p.getAttribute("enable", node.Service.Tags, strconv.FormatBool(p.ExposedByDefault)))
|
||||
enable, err := strconv.ParseBool(p.getAttribute(label.SuffixEnable, node.Service.Tags, strconv.FormatBool(p.ExposedByDefault)))
|
||||
if err != nil {
|
||||
log.Debugf("Invalid value for enable, set to %b", p.ExposedByDefault)
|
||||
return p.ExposedByDefault
|
||||
|
|
@ -323,116 +408,26 @@ func (p *CatalogProvider) isServiceEnabled(node *api.ServiceEntry) bool {
|
|||
}
|
||||
|
||||
func (p *CatalogProvider) getPrefixedName(name string) string {
|
||||
if len(p.Prefix) > 0 {
|
||||
if len(p.Prefix) > 0 && len(name) > 0 {
|
||||
return p.Prefix + "." + name
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getEntryPoints(list string) []string {
|
||||
return strings.Split(list, ",")
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getBackend(node *api.ServiceEntry) string {
|
||||
return strings.ToLower(node.Service.Service)
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getFrontendRule(service serviceUpdate) string {
|
||||
customFrontendRule := p.getAttribute("frontend.rule", service.Attributes, "")
|
||||
if customFrontendRule == "" {
|
||||
customFrontendRule = p.FrontEndRule
|
||||
}
|
||||
|
||||
t := p.frontEndRuleTemplate
|
||||
t, err := t.Parse(customFrontendRule)
|
||||
if err != nil {
|
||||
log.Errorf("failed to parse Consul Catalog custom frontend rule: %s", err)
|
||||
return ""
|
||||
}
|
||||
|
||||
templateObjects := struct {
|
||||
ServiceName string
|
||||
Domain string
|
||||
Attributes []string
|
||||
}{
|
||||
ServiceName: service.ServiceName,
|
||||
Domain: p.Domain,
|
||||
Attributes: service.Attributes,
|
||||
}
|
||||
|
||||
var buffer bytes.Buffer
|
||||
err = t.Execute(&buffer, templateObjects)
|
||||
if err != nil {
|
||||
log.Errorf("failed to execute Consul Catalog custom frontend rule template: %s", err)
|
||||
return ""
|
||||
}
|
||||
|
||||
return buffer.String()
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getBackendAddress(node *api.ServiceEntry) string {
|
||||
if node.Service.Address != "" {
|
||||
return node.Service.Address
|
||||
}
|
||||
return node.Node.Address
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getBackendName(node *api.ServiceEntry, index int) string {
|
||||
serviceName := strings.ToLower(node.Service.Service) + "--" + node.Service.Address + "--" + strconv.Itoa(node.Service.Port)
|
||||
|
||||
for _, tag := range node.Service.Tags {
|
||||
serviceName += "--" + provider.Normalize(tag)
|
||||
}
|
||||
|
||||
serviceName = strings.Replace(serviceName, ".", "-", -1)
|
||||
serviceName = strings.Replace(serviceName, "=", "-", -1)
|
||||
|
||||
// unique int at the end
|
||||
serviceName += "--" + strconv.Itoa(index)
|
||||
return serviceName
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getBasicAuth(tags []string) []string {
|
||||
list := p.getAttribute("frontend.auth.basic", tags, "")
|
||||
if list != "" {
|
||||
return strings.Split(list, ",")
|
||||
}
|
||||
return []string{}
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getSticky(tags []string) string {
|
||||
stickyTag := p.getTag(types.LabelBackendLoadbalancerSticky, tags, "")
|
||||
if len(stickyTag) > 0 {
|
||||
log.Warnf("Deprecated configuration found: %s. Please use %s.", types.LabelBackendLoadbalancerSticky, types.LabelBackendLoadbalancerStickiness)
|
||||
} else {
|
||||
stickyTag = "false"
|
||||
}
|
||||
return stickyTag
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) hasStickinessLabel(tags []string) bool {
|
||||
stickinessTag := p.getTag(types.LabelBackendLoadbalancerStickiness, tags, "")
|
||||
return len(stickinessTag) > 0 && strings.EqualFold(strings.TrimSpace(stickinessTag), "true")
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getStickinessCookieName(tags []string) string {
|
||||
return p.getTag(types.LabelBackendLoadbalancerStickinessCookieName, tags, "")
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getAttribute(name string, tags []string, defaultValue string) string {
|
||||
return p.getTag(p.getPrefixedName(name), tags, defaultValue)
|
||||
return getTag(p.getPrefixedName(name), tags, defaultValue)
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) hasTag(name string, tags []string) bool {
|
||||
func hasTag(name string, tags []string) bool {
|
||||
// Very-very unlikely that a Consul tag would ever start with '=!='
|
||||
tag := p.getTag(name, tags, "=!=")
|
||||
tag := getTag(name, tags, "=!=")
|
||||
return tag != "=!="
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getTag(name string, tags []string, defaultValue string) string {
|
||||
func getTag(name string, tags []string, defaultValue string) string {
|
||||
for _, tag := range tags {
|
||||
// Given the nature of Consul tags, which could be either singular markers, or key=value pairs, we check if the consul tag starts with 'name'
|
||||
if strings.Index(strings.ToLower(tag), strings.ToLower(name)) == 0 {
|
||||
if strings.HasPrefix(strings.ToLower(tag), strings.ToLower(name)) {
|
||||
// In case, where a tag might be a key=value, try to split it by the first '='
|
||||
// - If the first element (which would always be there, even if the tag is a singular marker without '=' in it
|
||||
if kv := strings.SplitN(tag, "=", 2); strings.ToLower(kv[0]) == strings.ToLower(name) {
|
||||
|
|
@ -449,165 +444,17 @@ func (p *CatalogProvider) getTag(name string, tags []string, defaultValue string
|
|||
}
|
||||
|
||||
func (p *CatalogProvider) getConstraintTags(tags []string) []string {
|
||||
var list []string
|
||||
var values []string
|
||||
|
||||
prefix := p.getPrefixedName("tags=")
|
||||
for _, tag := range tags {
|
||||
// We look for a Consul tag named 'traefik.tags' (unless different 'prefix' is configured)
|
||||
if strings.Index(strings.ToLower(tag), p.getPrefixedName("tags=")) == 0 {
|
||||
if strings.HasPrefix(strings.ToLower(tag), prefix) {
|
||||
// If 'traefik.tags=' tag is found, take the tag value and split by ',' adding the result to the list to be returned
|
||||
splitedTags := strings.Split(tag[len(p.getPrefixedName("tags=")):], ",")
|
||||
list = append(list, splitedTags...)
|
||||
splitedTags := label.SplitAndTrimString(tag[len(prefix):], ",")
|
||||
values = append(values, splitedTags...)
|
||||
}
|
||||
}
|
||||
|
||||
return list
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) buildConfig(catalog []catalogUpdate) *types.Configuration {
|
||||
var FuncMap = template.FuncMap{
|
||||
"getBackend": p.getBackend,
|
||||
"getFrontendRule": p.getFrontendRule,
|
||||
"getBackendName": p.getBackendName,
|
||||
"getBackendAddress": p.getBackendAddress,
|
||||
"getBasicAuth": p.getBasicAuth,
|
||||
"getSticky": p.getSticky,
|
||||
"hasStickinessLabel": p.hasStickinessLabel,
|
||||
"getStickinessCookieName": p.getStickinessCookieName,
|
||||
"getAttribute": p.getAttribute,
|
||||
"getTag": p.getTag,
|
||||
"hasTag": p.hasTag,
|
||||
"getEntryPoints": p.getEntryPoints,
|
||||
"hasMaxconnAttributes": p.hasMaxconnAttributes,
|
||||
}
|
||||
|
||||
allNodes := []*api.ServiceEntry{}
|
||||
services := []*serviceUpdate{}
|
||||
for _, info := range catalog {
|
||||
if len(info.Nodes) > 0 {
|
||||
services = append(services, info.Service)
|
||||
allNodes = append(allNodes, info.Nodes...)
|
||||
}
|
||||
}
|
||||
// Ensure a stable ordering of nodes so that identical configurations may be detected
|
||||
sort.Sort(nodeSorter(allNodes))
|
||||
|
||||
templateObjects := struct {
|
||||
Services []*serviceUpdate
|
||||
Nodes []*api.ServiceEntry
|
||||
}{
|
||||
Services: services,
|
||||
Nodes: allNodes,
|
||||
}
|
||||
|
||||
configuration, err := p.GetConfiguration("templates/consul_catalog.tmpl", FuncMap, templateObjects)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("Failed to create config")
|
||||
}
|
||||
|
||||
return configuration
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) hasMaxconnAttributes(attributes []string) bool {
|
||||
amount := p.getAttribute("backend.maxconn.amount", attributes, "")
|
||||
extractorfunc := p.getAttribute("backend.maxconn.extractorfunc", attributes, "")
|
||||
if amount != "" && extractorfunc != "" {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) getNodes(index map[string][]string) ([]catalogUpdate, error) {
|
||||
visited := make(map[string]bool)
|
||||
|
||||
nodes := []catalogUpdate{}
|
||||
for service := range index {
|
||||
name := strings.ToLower(service)
|
||||
if !strings.Contains(name, " ") && !visited[name] {
|
||||
visited[name] = true
|
||||
log.WithField("service", name).Debug("Fetching service")
|
||||
healthy, err := p.healthyNodes(name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// healthy.Nodes can be empty if constraints do not match, without throwing error
|
||||
if healthy.Service != nil && len(healthy.Nodes) > 0 {
|
||||
nodes = append(nodes, healthy)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nodes, nil
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) watch(configurationChan chan<- types.ConfigMessage, stop chan bool) error {
|
||||
stopCh := make(chan struct{})
|
||||
watchCh := make(chan map[string][]string)
|
||||
errorCh := make(chan error)
|
||||
|
||||
p.watchHealthState(stopCh, watchCh, errorCh)
|
||||
p.watchCatalogServices(stopCh, watchCh, errorCh)
|
||||
|
||||
defer close(stopCh)
|
||||
defer close(watchCh)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return nil
|
||||
case index, ok := <-watchCh:
|
||||
if !ok {
|
||||
return errors.New("Consul service list nil")
|
||||
}
|
||||
log.Debug("List of services changed")
|
||||
nodes, err := p.getNodes(index)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
configuration := p.buildConfig(nodes)
|
||||
configurationChan <- types.ConfigMessage{
|
||||
ProviderName: "consul_catalog",
|
||||
Configuration: configuration,
|
||||
}
|
||||
case err := <-errorCh:
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *CatalogProvider) setupFrontEndTemplate() {
|
||||
var FuncMap = template.FuncMap{
|
||||
"getAttribute": p.getAttribute,
|
||||
"getTag": p.getTag,
|
||||
"hasTag": p.hasTag,
|
||||
}
|
||||
t := template.New("consul catalog frontend rule").Funcs(FuncMap)
|
||||
p.frontEndRuleTemplate = t
|
||||
}
|
||||
|
||||
// Provide allows the consul catalog provider to provide configurations to traefik
|
||||
// using the given configuration channel.
|
||||
func (p *CatalogProvider) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool, constraints types.Constraints) error {
|
||||
config := api.DefaultConfig()
|
||||
config.Address = p.Endpoint
|
||||
client, err := api.NewClient(config)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.client = client
|
||||
p.Constraints = append(p.Constraints, constraints...)
|
||||
p.setupFrontEndTemplate()
|
||||
|
||||
pool.Go(func(stop chan bool) {
|
||||
notify := func(err error, time time.Duration) {
|
||||
log.Errorf("Consul connection error %+v, retrying in %s", err, time)
|
||||
}
|
||||
operation := func() error {
|
||||
return p.watch(configurationChan, stop)
|
||||
}
|
||||
err := backoff.RetryNotify(safe.OperationWithRecover(operation), job.NewBackOff(backoff.NewExponentialBackOff()), notify)
|
||||
if err != nil {
|
||||
log.Errorf("Cannot connect to consul server %+v", err)
|
||||
}
|
||||
})
|
||||
|
||||
return err
|
||||
return values
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue