traefik/provider/eureka/eureka.go

93 lines
2.6 KiB
Go
Raw Normal View History

package eureka
2016-08-31 20:43:05 +00:00
import (
2018-02-13 08:20:04 +00:00
"io/ioutil"
"time"
2018-02-13 08:20:04 +00:00
"github.com/ArthurHlt/go-eureka-client/eureka"
2016-08-31 20:43:05 +00:00
"github.com/cenk/backoff"
2018-07-31 17:28:03 +00:00
"github.com/containous/flaeg/parse"
2016-08-31 20:43:05 +00:00
"github.com/containous/traefik/job"
2017-06-13 23:22:14 +00:00
"github.com/containous/traefik/log"
"github.com/containous/traefik/provider"
2016-08-31 20:43:05 +00:00
"github.com/containous/traefik/safe"
"github.com/containous/traefik/types"
)
// Provider holds configuration of the Provider provider.
type Provider struct {
2017-10-02 08:32:02 +00:00
provider.BaseProvider `mapstructure:",squash" export:"true"`
2018-02-13 08:20:04 +00:00
Endpoint string `description:"Eureka server endpoint"`
2018-07-31 17:28:03 +00:00
RefreshSeconds parse.Duration `description:"Override default configuration time between refresh" export:"true"`
2016-08-31 20:43:05 +00:00
}
// Init the provider
func (p *Provider) Init(constraints types.Constraints) error {
return p.BaseProvider.Init(constraints)
}
// Provide allows the eureka provider to provide configurations to traefik
2016-08-31 20:43:05 +00:00
// using the given configuration channel.
func (p *Provider) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool) error {
2018-02-13 08:20:04 +00:00
eureka.GetLogger().SetOutput(ioutil.Discard)
2016-08-31 20:43:05 +00:00
operation := func() error {
2018-02-13 08:20:04 +00:00
client := eureka.NewClient([]string{p.Endpoint})
applications, err := client.GetApplications()
if err != nil {
log.Errorf("Failed to retrieve applications, error: %s", err)
return err
}
configuration, err := p.buildConfiguration(applications)
2016-08-31 20:43:05 +00:00
if err != nil {
log.Errorf("Failed to build configuration for Provider, error: %s", err)
2016-08-31 20:43:05 +00:00
return err
}
configurationChan <- types.ConfigMessage{
ProviderName: "eureka",
Configuration: configuration,
}
ticker := time.NewTicker(time.Duration(p.RefreshSeconds))
pool.Go(func(stop chan bool) {
for {
select {
case t := <-ticker.C:
log.Debugf("Refreshing Provider %s", t.String())
applications, err := client.GetApplications()
if err != nil {
log.Errorf("Failed to retrieve applications, error: %s", err)
continue
}
configuration, err := p.buildConfiguration(applications)
if err != nil {
log.Errorf("Failed to refresh Provider configuration, error: %s", err)
continue
}
configurationChan <- types.ConfigMessage{
ProviderName: "eureka",
Configuration: configuration,
}
case <-stop:
return
2016-08-31 20:43:05 +00:00
}
}
})
2016-08-31 20:43:05 +00:00
return nil
}
err := backoff.RetryNotify(operation, job.NewBackOff(backoff.NewExponentialBackOff()), notify)
if err != nil {
log.Errorf("Cannot connect to Provider server %+v", err)
2016-08-31 20:43:05 +00:00
return err
}
return nil
}
2018-02-13 08:20:04 +00:00
func notify(err error, time time.Duration) {
log.Errorf("Provider connection error %+v, retrying in %s", err, time)
}