traefik/server/server.go

314 lines
9.5 KiB
Go
Raw Normal View History

package server
import (
2016-08-16 15:26:10 +00:00
"context"
"encoding/json"
"net/http"
"os"
"os/signal"
"sync"
"time"
"github.com/containous/traefik/cluster"
2018-11-14 09:18:03 +00:00
"github.com/containous/traefik/config"
"github.com/containous/traefik/config/static"
"github.com/containous/traefik/log"
"github.com/containous/traefik/metrics"
"github.com/containous/traefik/middlewares/accesslog"
2018-11-14 09:18:03 +00:00
"github.com/containous/traefik/middlewares/requestdecorator"
"github.com/containous/traefik/provider"
"github.com/containous/traefik/safe"
2018-11-14 09:18:03 +00:00
"github.com/containous/traefik/server/middleware"
"github.com/containous/traefik/tracing"
"github.com/containous/traefik/tracing/datadog"
"github.com/containous/traefik/tracing/jaeger"
"github.com/containous/traefik/tracing/zipkin"
"github.com/containous/traefik/types"
)
// Server is the reverse-proxy/load-balancer engine
type Server struct {
entryPoints EntryPoints
2018-11-14 09:18:03 +00:00
configurationChan chan config.Message
configurationValidatedChan chan config.Message
signals chan os.Signal
stopChan chan bool
currentConfigurations safe.Safe
providerConfigUpdateMap map[string]chan config.Message
accessLoggerMiddleware *accesslog.Handler
tracer *tracing.Tracing
routinesPool *safe.Pool
leadership *cluster.Leadership //FIXME Cluster
2018-11-14 09:18:03 +00:00
defaultRoundTripper http.RoundTripper
metricsRegistry metrics.Registry
provider provider.Provider
configurationListeners []func(config.Configuration)
requestDecorator *requestdecorator.RequestDecorator
providersThrottleDuration time.Duration
2018-11-14 09:18:03 +00:00
}
// RouteAppenderFactory the route appender factory interface
type RouteAppenderFactory interface {
NewAppender(ctx context.Context, middlewaresBuilder *middleware.Builder, currentConfigurations *safe.Safe) types.RouteAppender
}
2018-11-14 09:18:03 +00:00
func setupTracing(conf *static.Tracing) tracing.TrackingBackend {
switch conf.Backend {
case jaeger.Name:
return conf.Jaeger
case zipkin.Name:
return conf.Zipkin
case datadog.Name:
return conf.DataDog
default:
log.WithoutContext().Warnf("Could not initialize tracing: unknown tracer %q", conf.Backend)
return nil
}
}
// NewServer returns an initialized Server.
func NewServer(staticConfiguration static.Configuration, provider provider.Provider, entryPoints EntryPoints) *Server {
2018-06-11 09:36:03 +00:00
server := &Server{}
server.provider = provider
server.entryPoints = entryPoints
2018-11-14 09:18:03 +00:00
server.configurationChan = make(chan config.Message, 100)
server.configurationValidatedChan = make(chan config.Message, 100)
server.signals = make(chan os.Signal, 1)
server.stopChan = make(chan bool, 1)
server.configureSignals()
2018-11-14 09:18:03 +00:00
currentConfigurations := make(config.Configurations)
server.currentConfigurations.Set(currentConfigurations)
2018-11-14 09:18:03 +00:00
server.providerConfigUpdateMap = make(map[string]chan config.Message)
if staticConfiguration.Providers != nil {
server.providersThrottleDuration = time.Duration(staticConfiguration.Providers.ProvidersThrottleDuration)
}
transport, err := createHTTPTransport(staticConfiguration.ServersTransport)
2018-11-14 09:18:03 +00:00
if err != nil {
log.WithoutContext().Errorf("Could not configure HTTP Transport, fallbacking on default transport: %v", err)
2018-11-14 09:18:03 +00:00
server.defaultRoundTripper = http.DefaultTransport
} else {
server.defaultRoundTripper = transport
}
2018-06-11 09:36:03 +00:00
server.routinesPool = safe.NewPool(context.Background())
2018-06-11 09:36:03 +00:00
if staticConfiguration.Tracing != nil {
trackingBackend := setupTracing(staticConfiguration.Tracing)
2018-11-14 09:18:03 +00:00
var err error
server.tracer, err = tracing.NewTracing(staticConfiguration.Tracing.ServiceName, staticConfiguration.Tracing.SpanNameLimit, trackingBackend)
2018-11-14 09:18:03 +00:00
if err != nil {
log.WithoutContext().Warnf("Unable to create tracer: %v", err)
}
2018-06-11 09:36:03 +00:00
}
server.requestDecorator = requestdecorator.New(staticConfiguration.HostResolver)
server.metricsRegistry = registerMetricClients(staticConfiguration.Metrics)
if staticConfiguration.AccessLog != nil {
2017-05-25 11:25:53 +00:00
var err error
server.accessLoggerMiddleware, err = accesslog.NewHandler(staticConfiguration.AccessLog)
2017-05-25 11:25:53 +00:00
if err != nil {
2018-11-14 09:18:03 +00:00
log.WithoutContext().Warnf("Unable to create access logger : %v", err)
2017-05-25 11:25:53 +00:00
}
2017-05-22 19:39:29 +00:00
}
return server
}
// Start starts the server and Stop/Close it when context is Done
func (s *Server) Start(ctx context.Context) {
go func() {
defer s.Close()
<-ctx.Done()
logger := log.FromContext(ctx)
logger.Info("I have to go...")
logger.Info("Stopping server gracefully")
s.Stop()
}()
2017-11-24 18:18:03 +00:00
s.startHTTPServers()
s.startLeadership()
s.routinesPool.Go(func(stop chan bool) {
s.listenProviders(stop)
})
2017-11-24 18:18:03 +00:00
s.routinesPool.Go(func(stop chan bool) {
s.listenConfigurations(stop)
})
s.startProvider()
2018-09-06 12:24:03 +00:00
s.routinesPool.Go(func(stop chan bool) {
s.listenSignals(stop)
})
}
// Wait blocks until server is shutted down.
2017-11-24 18:18:03 +00:00
func (s *Server) Wait() {
<-s.stopChan
}
// Stop stops the server
2017-11-24 18:18:03 +00:00
func (s *Server) Stop() {
2018-11-14 09:18:03 +00:00
defer log.WithoutContext().Info("Server stopped")
var wg sync.WaitGroup
for epn, ep := range s.entryPoints {
wg.Add(1)
go func(entryPointName string, entryPoint *EntryPoint) {
ctx := log.With(context.Background(), log.Str(log.EntryPointName, entryPointName))
defer wg.Done()
2018-11-14 09:18:03 +00:00
entryPoint.Shutdown(ctx)
2018-11-14 09:18:03 +00:00
log.FromContext(ctx).Debugf("Entry point %s closed", entryPointName)
}(epn, ep)
}
wg.Wait()
2017-11-24 18:18:03 +00:00
s.stopChan <- true
}
// Close destroys the server
2017-11-24 18:18:03 +00:00
func (s *Server) Close() {
2018-03-14 12:14:03 +00:00
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
go func(ctx context.Context) {
<-ctx.Done()
if ctx.Err() == context.Canceled {
return
} else if ctx.Err() == context.DeadlineExceeded {
2018-03-14 12:14:03 +00:00
panic("Timeout while stopping traefik, killing instance ✝")
}
}(ctx)
2018-11-14 09:18:03 +00:00
stopMetricsClients()
2017-11-24 18:18:03 +00:00
s.stopLeadership()
s.routinesPool.Cleanup()
close(s.configurationChan)
close(s.configurationValidatedChan)
signal.Stop(s.signals)
close(s.signals)
close(s.stopChan)
2018-11-14 09:18:03 +00:00
2017-11-24 18:18:03 +00:00
if s.accessLoggerMiddleware != nil {
if err := s.accessLoggerMiddleware.Close(); err != nil {
2018-11-14 09:18:03 +00:00
log.WithoutContext().Errorf("Could not close the access log file: %s", err)
2017-05-22 19:39:29 +00:00
}
}
2018-11-14 09:18:03 +00:00
if s.tracer != nil {
s.tracer.Close()
}
cancel()
}
2017-11-24 18:18:03 +00:00
func (s *Server) startLeadership() {
if s.leadership != nil {
s.leadership.Participate(s.routinesPool)
}
}
2017-11-24 18:18:03 +00:00
func (s *Server) stopLeadership() {
if s.leadership != nil {
s.leadership.Stop()
}
}
2017-11-24 18:18:03 +00:00
func (s *Server) startHTTPServers() {
// Use an empty configuration in order to initialize the default handlers with internal routes
handlers := s.applyConfiguration(context.Background(), config.Configuration{})
for entryPointName, handler := range handlers {
s.entryPoints[entryPointName].httpRouter.UpdateHandler(handler)
}
2017-01-12 13:34:54 +00:00
for entryPointName, entryPoint := range s.entryPoints {
ctx := log.With(context.Background(), log.Str(log.EntryPointName, entryPointName))
go entryPoint.Start(ctx)
}
}
2017-11-24 18:18:03 +00:00
func (s *Server) listenProviders(stop chan bool) {
for {
select {
case <-stop:
return
2017-11-24 18:18:03 +00:00
case configMsg, ok := <-s.configurationChan:
if !ok || configMsg.Configuration == nil {
return
}
2017-11-24 18:18:03 +00:00
s.preLoadConfiguration(configMsg)
}
}
}
2018-03-05 19:54:04 +00:00
// AddListener adds a new listener function used when new configuration is provided
2018-11-14 09:18:03 +00:00
func (s *Server) AddListener(listener func(config.Configuration)) {
2018-03-05 19:54:04 +00:00
if s.configurationListeners == nil {
2018-11-14 09:18:03 +00:00
s.configurationListeners = make([]func(config.Configuration), 0)
2018-03-05 19:54:04 +00:00
}
s.configurationListeners = append(s.configurationListeners, listener)
}
func (s *Server) startProvider() {
jsonConf, err := json.Marshal(s.provider)
if err != nil {
2018-11-14 09:18:03 +00:00
log.WithoutContext().Debugf("Unable to marshal provider configuration %T: %v", s.provider, err)
}
2018-11-14 09:18:03 +00:00
log.WithoutContext().Infof("Starting provider %T %s", s.provider, jsonConf)
currentProvider := s.provider
2018-11-14 09:18:03 +00:00
safe.Go(func() {
err := currentProvider.Provide(s.configurationChan, s.routinesPool)
if err != nil {
2018-11-14 09:18:03 +00:00
log.WithoutContext().Errorf("Error starting provider %T: %s", s.provider, err)
}
})
}
func registerMetricClients(metricsConfig *types.Metrics) metrics.Registry {
if metricsConfig == nil {
return metrics.NewVoidRegistry()
}
2018-01-31 18:10:04 +00:00
var registries []metrics.Registry
2018-11-14 09:18:03 +00:00
if metricsConfig.Prometheus != nil {
2018-11-14 09:18:03 +00:00
ctx := log.With(context.Background(), log.Str(log.MetricsProviderName, "prometheus"))
prometheusRegister := metrics.RegisterPrometheus(ctx, metricsConfig.Prometheus)
2018-08-06 12:58:03 +00:00
if prometheusRegister != nil {
registries = append(registries, prometheusRegister)
2018-11-14 09:18:03 +00:00
log.FromContext(ctx).Debug("Configured Prometheus metrics")
2018-08-06 12:58:03 +00:00
}
}
2018-11-14 09:18:03 +00:00
if metricsConfig.Datadog != nil {
2018-11-14 09:18:03 +00:00
ctx := log.With(context.Background(), log.Str(log.MetricsProviderName, "datadog"))
registries = append(registries, metrics.RegisterDatadog(ctx, metricsConfig.Datadog))
log.FromContext(ctx).Debugf("Configured DataDog metrics: pushing to %s once every %s",
metricsConfig.Datadog.Address, metricsConfig.Datadog.PushInterval)
}
2018-11-14 09:18:03 +00:00
if metricsConfig.StatsD != nil {
2018-11-14 09:18:03 +00:00
ctx := log.With(context.Background(), log.Str(log.MetricsProviderName, "statsd"))
registries = append(registries, metrics.RegisterStatsd(ctx, metricsConfig.StatsD))
log.FromContext(ctx).Debugf("Configured StatsD metrics: pushing to %s once every %s",
metricsConfig.StatsD.Address, metricsConfig.StatsD.PushInterval)
2017-04-18 06:22:06 +00:00
}
2018-11-14 09:18:03 +00:00
if metricsConfig.InfluxDB != nil {
2018-11-14 09:18:03 +00:00
ctx := log.With(context.Background(), log.Str(log.MetricsProviderName, "influxdb"))
registries = append(registries, metrics.RegisterInfluxDB(ctx, metricsConfig.InfluxDB))
log.FromContext(ctx).Debugf("Configured InfluxDB metrics: pushing to %s once every %s",
metricsConfig.InfluxDB.Address, metricsConfig.InfluxDB.PushInterval)
}
2017-04-18 06:22:06 +00:00
return metrics.NewMultiRegistry(registries)
}
func stopMetricsClients() {
metrics.StopDatadog()
metrics.StopStatsd()
metrics.StopInfluxDB()
}