522 lines
15 KiB
Go
522 lines
15 KiB
Go
package consul
|
|
|
|
import (
|
|
"bytes"
|
|
"errors"
|
|
"sort"
|
|
"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/provider"
|
|
"github.com/containous/traefik/safe"
|
|
"github.com/containous/traefik/types"
|
|
"github.com/hashicorp/consul/api"
|
|
)
|
|
|
|
const (
|
|
// DefaultWatchWaitTime is the duration to wait when polling consul
|
|
DefaultWatchWaitTime = 15 * time.Second
|
|
)
|
|
|
|
var _ provider.Provider = (*CatalogProvider)(nil)
|
|
|
|
// CatalogProvider holds configurations of the Consul catalog provider.
|
|
type CatalogProvider struct {
|
|
provider.BaseProvider `mapstructure:",squash"`
|
|
Endpoint string `description:"Consul server endpoint"`
|
|
Domain string `description:"Default domain used"`
|
|
ExposedByDefault bool `description:"Expose Consul services by default"`
|
|
Prefix string `description:"Prefix used for Consul catalog tags"`
|
|
FrontEndRule string `description:"Frontend rule used for Consul services"`
|
|
client *api.Client
|
|
frontEndRuleTemplate *template.Template
|
|
}
|
|
|
|
type serviceUpdate struct {
|
|
ServiceName string
|
|
Attributes []string
|
|
}
|
|
|
|
type catalogUpdate struct {
|
|
Service *serviceUpdate
|
|
Nodes []*api.ServiceEntry
|
|
}
|
|
|
|
type nodeSorter []*api.ServiceEntry
|
|
|
|
func (a nodeSorter) Len() int {
|
|
return len(a)
|
|
}
|
|
|
|
func (a nodeSorter) Swap(i int, j int) {
|
|
a[i], a[j] = a[j], a[i]
|
|
}
|
|
|
|
func (a nodeSorter) Less(i int, j int) bool {
|
|
lentr := a[i]
|
|
rentr := a[j]
|
|
|
|
ls := strings.ToLower(lentr.Service.Service)
|
|
lr := strings.ToLower(rentr.Service.Service)
|
|
|
|
if ls != lr {
|
|
return ls < lr
|
|
}
|
|
if lentr.Service.Address != rentr.Service.Address {
|
|
return lentr.Service.Address < rentr.Service.Address
|
|
}
|
|
if lentr.Node.Address != rentr.Node.Address {
|
|
return lentr.Node.Address < rentr.Node.Address
|
|
}
|
|
return lentr.Service.Port < rentr.Service.Port
|
|
}
|
|
|
|
func getChangedKeys(currState map[string][]string, prevState map[string][]string) ([]string, []string) {
|
|
currKeySet := fun.Set(fun.Keys(currState).([]string)).(map[string]bool)
|
|
prevKeySet := fun.Set(fun.Keys(prevState).([]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 (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<- map[string][]string) {
|
|
health := p.client.Health()
|
|
catalog := p.client.Catalog()
|
|
|
|
safe.Go(func() {
|
|
// variable to hold previous state
|
|
var flashback map[string][]string
|
|
|
|
options := &api.QueryOptions{WaitTime: DefaultWatchWaitTime}
|
|
|
|
for {
|
|
select {
|
|
case <-stopCh:
|
|
return
|
|
default:
|
|
}
|
|
|
|
// Listening to changes that leads to `passing` state or degrades from it.
|
|
// The call is used just as a trigger for further actions
|
|
// (intentionally there is no interest in the received data).
|
|
_, meta, err := health.State("passing", options)
|
|
if err != nil {
|
|
log.WithError(err).Error("Failed to retrieve health checks")
|
|
return
|
|
}
|
|
|
|
// If LastIndex didn't change then it means `Get` returned
|
|
// because of the WaitTime and the key didn't changed.
|
|
if options.WaitIndex == meta.LastIndex {
|
|
continue
|
|
}
|
|
|
|
options.WaitIndex = meta.LastIndex
|
|
|
|
// The response should be unified with watchCatalogServices
|
|
data, _, err := catalog.Services(&api.QueryOptions{})
|
|
if err != nil {
|
|
log.Errorf("Failed to list services: %s", err)
|
|
return
|
|
}
|
|
|
|
if data != nil {
|
|
// 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...
|
|
addedKeys, removedKeys := getChangedKeys(data, flashback)
|
|
|
|
if len(addedKeys) > 0 {
|
|
log.WithField("DiscoveredServices", addedKeys).Debug("Health State change detected.")
|
|
watchCh <- data
|
|
flashback = data
|
|
}
|
|
|
|
if len(removedKeys) > 0 {
|
|
log.WithField("MissingServices", removedKeys).Debug("Health State change detected.")
|
|
watchCh <- data
|
|
flashback = data
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh chan<- map[string][]string) {
|
|
catalog := p.client.Catalog()
|
|
|
|
safe.Go(func() {
|
|
// variable to hold previous state
|
|
var flashback map[string][]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: %s", err)
|
|
return
|
|
}
|
|
|
|
if options.WaitIndex == meta.LastIndex {
|
|
continue
|
|
}
|
|
|
|
options.WaitIndex = meta.LastIndex
|
|
|
|
if data != nil {
|
|
// 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...
|
|
addedKeys, removedKeys := getChangedKeys(data, flashback)
|
|
|
|
if len(addedKeys) > 0 {
|
|
log.WithField("DiscoveredServices", addedKeys).Debug("Catalog Services change detected.")
|
|
watchCh <- data
|
|
flashback = data
|
|
}
|
|
|
|
if len(removedKeys) > 0 {
|
|
log.WithField("MissingServices", removedKeys).Debug("Catalog Services change detected.")
|
|
watchCh <- data
|
|
flashback = data
|
|
}
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
func (p *CatalogProvider) healthyNodes(service string) (catalogUpdate, error) {
|
|
health := p.client.Health()
|
|
opts := &api.QueryOptions{}
|
|
data, _, err := health.Service(service, "", true, opts)
|
|
if err != nil {
|
|
log.WithError(err).Errorf("Failed to fetch details of %s", service)
|
|
return catalogUpdate{}, err
|
|
}
|
|
|
|
nodes := fun.Filter(func(node *api.ServiceEntry) bool {
|
|
return p.nodeFilter(service, node)
|
|
}, data).([]*api.ServiceEntry)
|
|
|
|
//Merge tags of nodes matching constraints, in a single slice.
|
|
tags := fun.Foldl(func(node *api.ServiceEntry, set []string) []string {
|
|
return fun.Keys(fun.Union(
|
|
fun.Set(set),
|
|
fun.Set(node.Service.Tags),
|
|
).(map[string]bool)).([]string)
|
|
}, []string{}, nodes).([]string)
|
|
|
|
return catalogUpdate{
|
|
Service: &serviceUpdate{
|
|
ServiceName: service,
|
|
Attributes: tags,
|
|
},
|
|
Nodes: nodes,
|
|
}, nil
|
|
}
|
|
|
|
func (p *CatalogProvider) nodeFilter(service string, node *api.ServiceEntry) bool {
|
|
// Filter disabled application.
|
|
if !p.isServiceEnabled(node) {
|
|
log.Debugf("Filtering disabled Consul service %s", service)
|
|
return false
|
|
}
|
|
|
|
// Filter by constraints.
|
|
constraintTags := p.getConstraintTags(node.Service.Tags)
|
|
ok, failingConstraint := p.MatchConstraints(constraintTags)
|
|
if !ok && failingConstraint != nil {
|
|
log.Debugf("Service %v pruned by '%v' constraint", service, failingConstraint.String())
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func (p *CatalogProvider) isServiceEnabled(node *api.ServiceEntry) bool {
|
|
enable, err := strconv.ParseBool(p.getAttribute("enable", node.Service.Tags, strconv.FormatBool(p.ExposedByDefault)))
|
|
if err != nil {
|
|
log.Debugf("Invalid value for enable, set to %b", p.ExposedByDefault)
|
|
return p.ExposedByDefault
|
|
}
|
|
return enable
|
|
}
|
|
|
|
func (p *CatalogProvider) getPrefixedName(name string) string {
|
|
if len(p.Prefix) > 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) getAttribute(name string, tags []string, defaultValue string) string {
|
|
return p.getTag(p.getPrefixedName(name), tags, defaultValue)
|
|
}
|
|
|
|
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) hasTag(name string, tags []string) bool {
|
|
// Very-very unlikely that a Consul tag would ever start with '=!='
|
|
tag := p.getTag(name, tags, "=!=")
|
|
return tag != "=!="
|
|
}
|
|
|
|
func (p *CatalogProvider) 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 {
|
|
// 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) {
|
|
// If the returned result is a key=value pair, return the 'value' component
|
|
if len(kv) == 2 {
|
|
return kv[1]
|
|
}
|
|
// If the returned result is a singular marker, return the 'key' component
|
|
return kv[0]
|
|
}
|
|
}
|
|
}
|
|
return defaultValue
|
|
}
|
|
|
|
func (p *CatalogProvider) getConstraintTags(tags []string) []string {
|
|
var list []string
|
|
|
|
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 '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...)
|
|
}
|
|
}
|
|
|
|
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,
|
|
"getAttribute": p.getAttribute,
|
|
"getBasicAuth": p.getBasicAuth,
|
|
"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)
|
|
|
|
p.watchHealthState(stopCh, watchCh)
|
|
p.watchCatalogServices(stopCh, watchCh)
|
|
|
|
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,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|