baalajimaestro/traefik
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Fix goroutine leak in consulcatalog when consul is down

Browse source
This commit is contained in:
yutopp 2018-03-20 22:36:03 +09:00 committed by Traefiker Bot
parent 7afa33dfa1
commit 43a510c046
1 changed files with 16 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
provider/consul/consul_catalog.go
View file

@ -6,6 +6,7 @@ import (
"sort"
"strconv"
"strings"
"sync"
"text/template"
"time"
@ -135,7 +136,7 @@ func getChangedIntKeys(currState []int, prevState []int) ([]int, []int) {
return fun.Keys(addedKeys).([]int), fun.Keys(removedKeys).([]int)
}
func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<- map[string][]string, errorCh chan<- error) {
func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<- map[string][]string, notifyError func(error)) {
health := p.client.Health()
catalog := p.client.Catalog()
@ -156,7 +157,7 @@ func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<
healthyState, meta, err := health.State("passing", options)
if err != nil {
log.WithError(err).Error("Failed to retrieve health checks")
errorCh <- err
notifyError(err)
return
}
@ -180,7 +181,7 @@ func (p *CatalogProvider) watchHealthState(stopCh <-chan struct{}, watchCh chan<
data, _, err := catalog.Services(&api.QueryOptions{})
if err != nil {
log.Errorf("Failed to list services: %s", err)
errorCh <- err
notifyError(err)
return
}
@ -214,7 +215,7 @@ type Service struct {
Ports []int
}
func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh chan<- map[string][]string, errorCh chan<- error) {
func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh chan<- map[string][]string, notifyError func(error)) {
catalog := p.client.Catalog()
safe.Go(func() {
@ -233,7 +234,7 @@ func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh c
data, meta, err := catalog.Services(options)
if err != nil {
log.Errorf("Failed to list services: %s", err)
errorCh <- err
notifyError(err)
return
}
@ -249,7 +250,7 @@ func (p *CatalogProvider) watchCatalogServices(stopCh <-chan struct{}, watchCh c
nodes, _, err := catalog.Service(key, "", &api.QueryOptions{})
if err != nil {
log.Errorf("Failed to get detail of service %s: %s", key, err)
errorCh <- err
notifyError(err)
return
}
nodesID := getServiceIds(nodes)
@ -572,8 +573,15 @@ func (p *CatalogProvider) watch(configurationChan chan<- types.ConfigMessage, st
watchCh := make(chan map[string][]string)
errorCh := make(chan error)
p.watchHealthState(stopCh, watchCh, errorCh)
p.watchCatalogServices(stopCh, watchCh, errorCh)
var errorOnce sync.Once
notifyError := func(err error) {
errorOnce.Do(func() {
errorCh <- err
})
}
p.watchHealthState(stopCh, watchCh, notifyError)
p.watchCatalogServices(stopCh, watchCh, notifyError)
defer close(stopCh)
defer close(watchCh)