1
0
mirror of https://github.com/containous/traefik.git synced 2024-12-24 21:34:39 +03:00
traefik/provider/consul_catalog.go

304 lines
8.1 KiB
Go
Raw Normal View History

2016-02-02 20:03:40 +03:00
package provider
import (
"errors"
"strconv"
2016-02-02 20:03:40 +03:00
"strings"
"text/template"
"time"
log "github.com/Sirupsen/logrus"
"github.com/cenkalti/backoff"
"github.com/containous/traefik/safe"
"github.com/containous/traefik/types"
2016-02-02 20:03:40 +03:00
"github.com/hashicorp/consul/api"
)
const (
// DefaultWatchWaitTime is the duration to wait when polling consul
DefaultWatchWaitTime = 15 * time.Second
// DefaultConsulCatalogTagPrefix is a prefix for additional service/node configurations
DefaultConsulCatalogTagPrefix = "traefik"
2016-02-02 20:03:40 +03:00
)
// ConsulCatalog holds configurations of the Consul catalog provider.
type ConsulCatalog struct {
2016-05-24 18:31:50 +03:00
BaseProvider
Endpoint string `description:"Consul server endpoint"`
Domain string `description:"Default domain used"`
client *api.Client
Prefix string
}
type serviceUpdate struct {
ServiceName string
Attributes []string
2016-02-02 20:03:40 +03:00
}
type catalogUpdate struct {
Service *serviceUpdate
2016-02-02 20:03:40 +03:00
Nodes []*api.ServiceEntry
}
func (provider *ConsulCatalog) watchServices(stopCh <-chan struct{}) <-chan map[string][]string {
watchCh := make(chan map[string][]string)
catalog := provider.client.Catalog()
safe.Go(func() {
2016-02-02 20:03:40 +03:00
defer close(watchCh)
opts := &api.QueryOptions{WaitTime: DefaultWatchWaitTime}
for {
select {
case <-stopCh:
return
default:
}
data, meta, err := catalog.Services(opts)
if err != nil {
log.WithError(err).Errorf("Failed to list services")
return
}
// If LastIndex didn't change then it means `Get` returned
// because of the WaitTime and the key didn't changed.
if opts.WaitIndex == meta.LastIndex {
continue
}
opts.WaitIndex = meta.LastIndex
if data != nil {
watchCh <- data
}
}
})
2016-02-02 20:03:40 +03:00
return watchCh
}
func (provider *ConsulCatalog) healthyNodes(service string) (catalogUpdate, error) {
health := provider.client.Health()
opts := &api.QueryOptions{}
data, _, err := health.Service(service, "", true, opts)
if err != nil {
log.WithError(err).Errorf("Failed to fetch details of " + service)
return catalogUpdate{}, err
}
set := map[string]bool{}
tags := []string{}
nodes := []*api.ServiceEntry{}
for _, node := range data {
constraintTags := provider.getContraintTags(node.Service.Tags)
if ok, failingConstraint, err := provider.MatchConstraints(constraintTags); err != nil {
return catalogUpdate{}, err
} else if ok == true {
nodes = append(nodes, node)
// merge tags of every nodes in a single slice
// only if node match constraint
for _, tag := range node.Service.Tags {
if _, ok := set[tag]; ok == false {
set[tag] = true
tags = append(tags, tag)
}
}
} else if ok == false && failingConstraint != nil {
log.Debugf("Service %v pruned by '%v' constraint", service, failingConstraint.String())
}
}
2016-02-02 20:03:40 +03:00
return catalogUpdate{
Service: &serviceUpdate{
ServiceName: service,
Attributes: tags,
},
Nodes: nodes,
2016-02-02 20:03:40 +03:00
}, nil
}
func (provider *ConsulCatalog) getEntryPoints(list string) []string {
return strings.Split(list, ",")
}
2016-02-02 20:03:40 +03:00
func (provider *ConsulCatalog) getBackend(node *api.ServiceEntry) string {
return strings.ToLower(node.Service.Service)
}
func (provider *ConsulCatalog) getFrontendRule(service serviceUpdate) string {
customFrontendRule := provider.getAttribute("frontend.rule", service.Attributes, "")
if customFrontendRule != "" {
return customFrontendRule
}
return "Host:" + service.ServiceName + "." + provider.Domain
}
func (provider *ConsulCatalog) getBackendAddress(node *api.ServiceEntry) string {
if node.Service.Address != "" {
return node.Service.Address
}
return node.Node.Address
}
func (provider *ConsulCatalog) getBackendName(node *api.ServiceEntry, index int) string {
serviceName := node.Service.Service + "--" + node.Service.Address + "--" + strconv.Itoa(node.Service.Port)
for _, tag := range node.Service.Tags {
serviceName += "--" + normalize(tag)
}
serviceName = strings.Replace(serviceName, ".", "-", -1)
serviceName = strings.Replace(serviceName, "=", "-", -1)
// unique int at the end
serviceName += "--" + strconv.Itoa(index)
return serviceName
}
func (provider *ConsulCatalog) getAttribute(name string, tags []string, defaultValue string) string {
for _, tag := range tags {
if strings.Index(strings.ToLower(tag), DefaultConsulCatalogTagPrefix+".") == 0 {
if kv := strings.SplitN(tag[len(DefaultConsulCatalogTagPrefix+"."):], "=", 2); len(kv) == 2 && strings.ToLower(kv[0]) == strings.ToLower(name) {
return kv[1]
}
}
}
return defaultValue
2016-02-02 20:03:40 +03:00
}
func (provider *ConsulCatalog) getContraintTags(tags []string) []string {
var list []string
for _, tag := range tags {
if strings.Index(strings.ToLower(tag), DefaultConsulCatalogTagPrefix+".tags=") == 0 {
splitedTags := strings.Split(tag[len(DefaultConsulCatalogTagPrefix+".tags="):], ",")
list = append(list, splitedTags...)
}
}
return list
}
2016-02-02 20:03:40 +03:00
func (provider *ConsulCatalog) buildConfig(catalog []catalogUpdate) *types.Configuration {
var FuncMap = template.FuncMap{
"getBackend": provider.getBackend,
"getFrontendRule": provider.getFrontendRule,
"getBackendName": provider.getBackendName,
"getBackendAddress": provider.getBackendAddress,
"getAttribute": provider.getAttribute,
"getEntryPoints": provider.getEntryPoints,
2016-02-02 20:03:40 +03:00
}
allNodes := []*api.ServiceEntry{}
services := []*serviceUpdate{}
2016-02-02 20:03:40 +03:00
for _, info := range catalog {
for _, node := range info.Nodes {
isEnabled := provider.getAttribute("enable", node.Service.Tags, "true")
if isEnabled != "false" && len(info.Nodes) > 0 {
services = append(services, info.Service)
allNodes = append(allNodes, info.Nodes...)
break
}
2016-02-02 20:03:40 +03:00
}
}
templateObjects := struct {
Services []*serviceUpdate
2016-02-02 20:03:40 +03:00
Nodes []*api.ServiceEntry
}{
Services: services,
2016-02-02 20:03:40 +03:00
Nodes: allNodes,
}
configuration, err := provider.getConfiguration("templates/consul_catalog.tmpl", FuncMap, templateObjects)
if err != nil {
log.WithError(err).Error("Failed to create config")
}
return configuration
}
func (provider *ConsulCatalog) 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.WithFields(log.Fields{
"service": name,
}).Debug("Fetching service")
healthy, err := provider.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)
}
2016-02-02 20:03:40 +03:00
}
}
return nodes, nil
}
func (provider *ConsulCatalog) watch(configurationChan chan<- types.ConfigMessage, stop chan bool) error {
2016-02-02 20:03:40 +03:00
stopCh := make(chan struct{})
serviceCatalog := provider.watchServices(stopCh)
defer close(stopCh)
for {
select {
case <-stop:
return nil
2016-02-02 20:03:40 +03:00
case index, ok := <-serviceCatalog:
if !ok {
return errors.New("Consul service list nil")
}
log.Debug("List of services changed")
nodes, err := provider.getNodes(index)
if err != nil {
return err
}
configuration := provider.buildConfig(nodes)
configurationChan <- types.ConfigMessage{
ProviderName: "consul_catalog",
Configuration: configuration,
}
}
}
}
// Provide allows the provider to provide configurations to traefik
// using the given configuration channel.
func (provider *ConsulCatalog) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool, constraints []*types.Constraint) error {
2016-02-02 20:03:40 +03:00
config := api.DefaultConfig()
config.Address = provider.Endpoint
client, err := api.NewClient(config)
if err != nil {
return err
}
provider.client = client
provider.Constraints = append(provider.Constraints, constraints...)
2016-02-02 20:03:40 +03:00
pool.Go(func(stop chan bool) {
2016-02-02 20:03:40 +03:00
notify := func(err error, time time.Duration) {
log.Errorf("Consul connection error %+v, retrying in %s", err, time)
}
worker := func() error {
return provider.watch(configurationChan, stop)
2016-02-02 20:03:40 +03:00
}
err := backoff.RetryNotify(worker, backoff.NewExponentialBackOff(), notify)
if err != nil {
log.Fatalf("Cannot connect to consul server %+v", err)
}
})
2016-02-02 20:03:40 +03:00
return err
}