traefik/pkg/provider/aggregator/aggregator.go

139 lines
3.1 KiB
Go
Raw Normal View History

2018-11-14 09:18:03 +00:00
package aggregator
import (
"encoding/json"
"github.com/traefik/traefik/v2/pkg/config/dynamic"
"github.com/traefik/traefik/v2/pkg/config/static"
"github.com/traefik/traefik/v2/pkg/log"
"github.com/traefik/traefik/v2/pkg/provider"
"github.com/traefik/traefik/v2/pkg/provider/file"
"github.com/traefik/traefik/v2/pkg/safe"
2018-11-14 09:18:03 +00:00
)
// ProviderAggregator aggregates providers.
type ProviderAggregator struct {
fileProvider *file.Provider
providers []provider.Provider
2018-11-14 09:18:03 +00:00
}
// NewProviderAggregator returns an aggregate of all the providers configured in the static configuration.
func NewProviderAggregator(conf static.Providers) ProviderAggregator {
p := ProviderAggregator{}
2018-11-14 09:18:03 +00:00
if conf.File != nil {
p.quietAddProvider(conf.File)
}
if conf.Docker != nil {
p.quietAddProvider(conf.Docker)
}
if conf.Marathon != nil {
p.quietAddProvider(conf.Marathon)
}
2018-12-03 10:32:05 +00:00
if conf.Rest != nil {
p.quietAddProvider(conf.Rest)
}
if conf.KubernetesIngress != nil {
p.quietAddProvider(conf.KubernetesIngress)
}
if conf.KubernetesCRD != nil {
p.quietAddProvider(conf.KubernetesCRD)
}
2019-04-05 10:22:04 +00:00
if conf.Rancher != nil {
p.quietAddProvider(conf.Rancher)
}
if conf.Ecs != nil {
p.quietAddProvider(conf.Ecs)
}
2019-10-15 15:34:08 +00:00
if conf.ConsulCatalog != nil {
p.quietAddProvider(conf.ConsulCatalog)
}
if conf.Consul != nil {
p.quietAddProvider(conf.Consul)
}
if conf.Etcd != nil {
p.quietAddProvider(conf.Etcd)
}
if conf.ZooKeeper != nil {
p.quietAddProvider(conf.ZooKeeper)
}
if conf.Redis != nil {
p.quietAddProvider(conf.Redis)
}
if conf.HTTP != nil {
p.quietAddProvider(conf.HTTP)
}
2018-11-14 09:18:03 +00:00
return p
}
func (p *ProviderAggregator) quietAddProvider(provider provider.Provider) {
err := p.AddProvider(provider)
if err != nil {
log.WithoutContext().Errorf("Error while initializing provider %T: %v", provider, err)
}
}
// AddProvider adds a provider in the providers map.
func (p *ProviderAggregator) AddProvider(provider provider.Provider) error {
err := provider.Init()
2018-11-14 09:18:03 +00:00
if err != nil {
return err
}
if fileProvider, ok := provider.(*file.Provider); ok {
p.fileProvider = fileProvider
} else {
p.providers = append(p.providers, provider)
}
2018-11-14 09:18:03 +00:00
return nil
}
2020-05-11 10:06:07 +00:00
// Init the provider.
func (p ProviderAggregator) Init() error {
2018-11-14 09:18:03 +00:00
return nil
}
2020-05-11 10:06:07 +00:00
// Provide calls the provide method of every providers.
func (p ProviderAggregator) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
if p.fileProvider != nil {
launchProvider(configurationChan, pool, p.fileProvider)
}
2018-11-14 09:18:03 +00:00
for _, prd := range p.providers {
2019-03-19 09:04:04 +00:00
prd := prd
2018-11-14 09:18:03 +00:00
safe.Go(func() {
launchProvider(configurationChan, pool, prd)
2018-11-14 09:18:03 +00:00
})
}
return nil
}
func launchProvider(configurationChan chan<- dynamic.Message, pool *safe.Pool, prd provider.Provider) {
jsonConf, err := json.Marshal(prd)
if err != nil {
log.WithoutContext().Debugf("Cannot marshal the provider configuration %T: %v", prd, err)
}
log.WithoutContext().Infof("Starting provider %T %s", prd, jsonConf)
currentProvider := prd
err = currentProvider.Provide(configurationChan, pool)
if err != nil {
log.WithoutContext().Errorf("Cannot start the provider %T: %v", prd, err)
}
}