424 lines
14 KiB
Go
424 lines
14 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"hash/fnv"
|
|
"math/rand"
|
|
"net/http"
|
|
"net/url"
|
|
"reflect"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/containous/alice"
|
|
"github.com/rs/zerolog/log"
|
|
"github.com/traefik/traefik/v3/pkg/config/dynamic"
|
|
"github.com/traefik/traefik/v3/pkg/config/runtime"
|
|
"github.com/traefik/traefik/v3/pkg/healthcheck"
|
|
"github.com/traefik/traefik/v3/pkg/logs"
|
|
"github.com/traefik/traefik/v3/pkg/middlewares/accesslog"
|
|
"github.com/traefik/traefik/v3/pkg/middlewares/capture"
|
|
metricsMiddle "github.com/traefik/traefik/v3/pkg/middlewares/metrics"
|
|
"github.com/traefik/traefik/v3/pkg/middlewares/observability"
|
|
"github.com/traefik/traefik/v3/pkg/proxy/httputil"
|
|
"github.com/traefik/traefik/v3/pkg/safe"
|
|
"github.com/traefik/traefik/v3/pkg/server/cookie"
|
|
"github.com/traefik/traefik/v3/pkg/server/middleware"
|
|
"github.com/traefik/traefik/v3/pkg/server/provider"
|
|
"github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/failover"
|
|
"github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/mirror"
|
|
"github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/wrr"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
const (
|
|
defaultMirrorBody = true
|
|
defaultMaxBodySize int64 = -1
|
|
)
|
|
|
|
// ProxyBuilder builds reverse proxy handlers.
|
|
type ProxyBuilder interface {
|
|
Build(cfgName string, targetURL *url.URL, shouldObserve, passHostHeader, preservePath bool, flushInterval time.Duration) (http.Handler, error)
|
|
Update(configs map[string]*dynamic.ServersTransport)
|
|
}
|
|
|
|
// Manager The service manager.
|
|
type Manager struct {
|
|
routinePool *safe.Pool
|
|
observabilityMgr *middleware.ObservabilityMgr
|
|
transportManager httputil.TransportManager
|
|
proxyBuilder ProxyBuilder
|
|
|
|
services map[string]http.Handler
|
|
configs map[string]*runtime.ServiceInfo
|
|
healthCheckers map[string]*healthcheck.ServiceHealthChecker
|
|
rand *rand.Rand // For the initial shuffling of load-balancers.
|
|
}
|
|
|
|
// NewManager creates a new Manager.
|
|
func NewManager(configs map[string]*runtime.ServiceInfo, observabilityMgr *middleware.ObservabilityMgr, routinePool *safe.Pool, transportManager httputil.TransportManager, proxyBuilder ProxyBuilder) *Manager {
|
|
return &Manager{
|
|
routinePool: routinePool,
|
|
observabilityMgr: observabilityMgr,
|
|
transportManager: transportManager,
|
|
proxyBuilder: proxyBuilder,
|
|
services: make(map[string]http.Handler),
|
|
configs: configs,
|
|
healthCheckers: make(map[string]*healthcheck.ServiceHealthChecker),
|
|
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
|
|
}
|
|
}
|
|
|
|
// BuildHTTP Creates a http.Handler for a service configuration.
|
|
func (m *Manager) BuildHTTP(rootCtx context.Context, serviceName string) (http.Handler, error) {
|
|
serviceName = provider.GetQualifiedName(rootCtx, serviceName)
|
|
|
|
ctx := log.Ctx(rootCtx).With().Str(logs.ServiceName, serviceName).Logger().
|
|
WithContext(provider.AddInContext(rootCtx, serviceName))
|
|
|
|
handler, ok := m.services[serviceName]
|
|
if ok {
|
|
return handler, nil
|
|
}
|
|
|
|
conf, ok := m.configs[serviceName]
|
|
if !ok {
|
|
return nil, fmt.Errorf("the service %q does not exist", serviceName)
|
|
}
|
|
|
|
if conf.Status == runtime.StatusDisabled {
|
|
return nil, errors.New(strings.Join(conf.Err, ", "))
|
|
}
|
|
|
|
value := reflect.ValueOf(*conf.Service)
|
|
var count int
|
|
for i := range value.NumField() {
|
|
if !value.Field(i).IsNil() {
|
|
count++
|
|
}
|
|
}
|
|
if count > 1 {
|
|
err := errors.New("cannot create service: multi-types service not supported, consider declaring two different pieces of service instead")
|
|
conf.AddError(err, true)
|
|
return nil, err
|
|
}
|
|
|
|
var lb http.Handler
|
|
|
|
switch {
|
|
case conf.LoadBalancer != nil:
|
|
var err error
|
|
lb, err = m.getLoadBalancerServiceHandler(ctx, serviceName, conf)
|
|
if err != nil {
|
|
conf.AddError(err, true)
|
|
return nil, err
|
|
}
|
|
case conf.Weighted != nil:
|
|
var err error
|
|
lb, err = m.getWRRServiceHandler(ctx, serviceName, conf.Weighted)
|
|
if err != nil {
|
|
conf.AddError(err, true)
|
|
return nil, err
|
|
}
|
|
case conf.Mirroring != nil:
|
|
var err error
|
|
lb, err = m.getMirrorServiceHandler(ctx, conf.Mirroring)
|
|
if err != nil {
|
|
conf.AddError(err, true)
|
|
return nil, err
|
|
}
|
|
case conf.Failover != nil:
|
|
var err error
|
|
lb, err = m.getFailoverServiceHandler(ctx, serviceName, conf.Failover)
|
|
if err != nil {
|
|
conf.AddError(err, true)
|
|
return nil, err
|
|
}
|
|
default:
|
|
sErr := fmt.Errorf("the service %q does not have any type defined", serviceName)
|
|
conf.AddError(sErr, true)
|
|
return nil, sErr
|
|
}
|
|
|
|
m.services[serviceName] = lb
|
|
|
|
return lb, nil
|
|
}
|
|
|
|
func (m *Manager) getFailoverServiceHandler(ctx context.Context, serviceName string, config *dynamic.Failover) (http.Handler, error) {
|
|
f := failover.New(config.HealthCheck)
|
|
|
|
serviceHandler, err := m.BuildHTTP(ctx, config.Service)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
f.SetHandler(serviceHandler)
|
|
|
|
updater, ok := serviceHandler.(healthcheck.StatusUpdater)
|
|
if !ok {
|
|
return nil, fmt.Errorf("child service %v of %v not a healthcheck.StatusUpdater (%T)", config.Service, serviceName, serviceHandler)
|
|
}
|
|
|
|
if err := updater.RegisterStatusUpdater(func(up bool) {
|
|
f.SetHandlerStatus(ctx, up)
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("cannot register %v as updater for %v: %w", config.Service, serviceName, err)
|
|
}
|
|
|
|
fallbackHandler, err := m.BuildHTTP(ctx, config.Fallback)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
f.SetFallbackHandler(fallbackHandler)
|
|
|
|
// Do not report the health of the fallback handler.
|
|
if config.HealthCheck == nil {
|
|
return f, nil
|
|
}
|
|
|
|
fallbackUpdater, ok := fallbackHandler.(healthcheck.StatusUpdater)
|
|
if !ok {
|
|
return nil, fmt.Errorf("child service %v of %v not a healthcheck.StatusUpdater (%T)", config.Fallback, serviceName, fallbackHandler)
|
|
}
|
|
|
|
if err := fallbackUpdater.RegisterStatusUpdater(func(up bool) {
|
|
f.SetFallbackHandlerStatus(ctx, up)
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("cannot register %v as updater for %v: %w", config.Fallback, serviceName, err)
|
|
}
|
|
|
|
return f, nil
|
|
}
|
|
|
|
func (m *Manager) getMirrorServiceHandler(ctx context.Context, config *dynamic.Mirroring) (http.Handler, error) {
|
|
serviceHandler, err := m.BuildHTTP(ctx, config.Service)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
mirrorBody := defaultMirrorBody
|
|
if config.MirrorBody != nil {
|
|
mirrorBody = *config.MirrorBody
|
|
}
|
|
|
|
maxBodySize := defaultMaxBodySize
|
|
if config.MaxBodySize != nil {
|
|
maxBodySize = *config.MaxBodySize
|
|
}
|
|
handler := mirror.New(serviceHandler, m.routinePool, mirrorBody, maxBodySize, config.HealthCheck)
|
|
for _, mirrorConfig := range config.Mirrors {
|
|
mirrorHandler, err := m.BuildHTTP(ctx, mirrorConfig.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = handler.AddMirror(mirrorHandler, mirrorConfig.Percent)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return handler, nil
|
|
}
|
|
|
|
func (m *Manager) getWRRServiceHandler(ctx context.Context, serviceName string, config *dynamic.WeightedRoundRobin) (http.Handler, error) {
|
|
// TODO Handle accesslog and metrics with multiple service name
|
|
if config.Sticky != nil && config.Sticky.Cookie != nil {
|
|
config.Sticky.Cookie.Name = cookie.GetName(config.Sticky.Cookie.Name, serviceName)
|
|
}
|
|
|
|
balancer := wrr.New(config.Sticky, config.HealthCheck != nil)
|
|
for _, service := range shuffle(config.Services, m.rand) {
|
|
serviceHandler, err := m.getServiceHandler(ctx, service)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
balancer.Add(service.Name, serviceHandler, service.Weight)
|
|
|
|
if config.HealthCheck == nil {
|
|
continue
|
|
}
|
|
|
|
childName := service.Name
|
|
updater, ok := serviceHandler.(healthcheck.StatusUpdater)
|
|
if !ok {
|
|
return nil, fmt.Errorf("child service %v of %v not a healthcheck.StatusUpdater (%T)", childName, serviceName, serviceHandler)
|
|
}
|
|
|
|
if err := updater.RegisterStatusUpdater(func(up bool) {
|
|
balancer.SetStatus(ctx, childName, up)
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("cannot register %v as updater for %v: %w", childName, serviceName, err)
|
|
}
|
|
|
|
log.Ctx(ctx).Debug().Str("parent", serviceName).Str("child", childName).
|
|
Msg("Child service will update parent on status change")
|
|
}
|
|
|
|
return balancer, nil
|
|
}
|
|
|
|
func (m *Manager) getServiceHandler(ctx context.Context, service dynamic.WRRService) (http.Handler, error) {
|
|
switch {
|
|
case service.Status != nil:
|
|
return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
|
rw.WriteHeader(*service.Status)
|
|
}), nil
|
|
|
|
case service.GRPCStatus != nil:
|
|
return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
|
|
st := status.New(service.GRPCStatus.Code, service.GRPCStatus.Msg)
|
|
|
|
body, err := json.Marshal(st.Proto())
|
|
if err != nil {
|
|
http.Error(rw, "Failed to marshal status to JSON", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
rw.Header().Set("Content-Type", "application/json")
|
|
rw.WriteHeader(http.StatusOK)
|
|
|
|
_, _ = rw.Write(body)
|
|
}), nil
|
|
|
|
default:
|
|
return m.BuildHTTP(ctx, service.Name)
|
|
}
|
|
}
|
|
|
|
func (m *Manager) getLoadBalancerServiceHandler(ctx context.Context, serviceName string, info *runtime.ServiceInfo) (http.Handler, error) {
|
|
service := info.LoadBalancer
|
|
|
|
logger := log.Ctx(ctx)
|
|
logger.Debug().Msg("Creating load-balancer")
|
|
|
|
// TODO: should we keep this config value as Go is now handling stream response correctly?
|
|
flushInterval := time.Duration(dynamic.DefaultFlushInterval)
|
|
if service.ResponseForwarding != nil {
|
|
flushInterval = time.Duration(service.ResponseForwarding.FlushInterval)
|
|
}
|
|
|
|
if len(service.ServersTransport) > 0 {
|
|
service.ServersTransport = provider.GetQualifiedName(ctx, service.ServersTransport)
|
|
}
|
|
|
|
if service.Sticky != nil && service.Sticky.Cookie != nil {
|
|
service.Sticky.Cookie.Name = cookie.GetName(service.Sticky.Cookie.Name, serviceName)
|
|
}
|
|
|
|
// We make sure that the PassHostHeader value is defined to avoid panics.
|
|
passHostHeader := dynamic.DefaultPassHostHeader
|
|
if service.PassHostHeader != nil {
|
|
passHostHeader = *service.PassHostHeader
|
|
}
|
|
|
|
lb := wrr.New(service.Sticky, service.HealthCheck != nil)
|
|
healthCheckTargets := make(map[string]*url.URL)
|
|
|
|
for _, server := range shuffle(service.Servers, m.rand) {
|
|
hasher := fnv.New64a()
|
|
_, _ = hasher.Write([]byte(server.URL)) // this will never return an error.
|
|
|
|
proxyName := hex.EncodeToString(hasher.Sum(nil))
|
|
|
|
target, err := url.Parse(server.URL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error parsing server URL %s: %w", server.URL, err)
|
|
}
|
|
|
|
logger.Debug().Str(logs.ServerName, proxyName).Stringer("target", target).
|
|
Msg("Creating server")
|
|
|
|
qualifiedSvcName := provider.GetQualifiedName(ctx, serviceName)
|
|
|
|
shouldObserve := m.observabilityMgr.ShouldAddTracing(qualifiedSvcName) || m.observabilityMgr.ShouldAddMetrics(qualifiedSvcName)
|
|
proxy, err := m.proxyBuilder.Build(service.ServersTransport, target, shouldObserve, passHostHeader, server.PreservePath, flushInterval)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error building proxy for server URL %s: %w", server.URL, err)
|
|
}
|
|
|
|
// Prevents from enabling observability for internal resources.
|
|
|
|
if m.observabilityMgr.ShouldAddAccessLogs(qualifiedSvcName) {
|
|
proxy = accesslog.NewFieldHandler(proxy, accesslog.ServiceURL, target.String(), nil)
|
|
proxy = accesslog.NewFieldHandler(proxy, accesslog.ServiceAddr, target.Host, nil)
|
|
proxy = accesslog.NewFieldHandler(proxy, accesslog.ServiceName, serviceName, accesslog.AddServiceFields)
|
|
}
|
|
|
|
if m.observabilityMgr.MetricsRegistry() != nil && m.observabilityMgr.MetricsRegistry().IsSvcEnabled() &&
|
|
m.observabilityMgr.ShouldAddMetrics(qualifiedSvcName) {
|
|
metricsHandler := metricsMiddle.WrapServiceHandler(ctx, m.observabilityMgr.MetricsRegistry(), serviceName)
|
|
|
|
proxy, err = alice.New().
|
|
Append(observability.WrapMiddleware(ctx, metricsHandler)).
|
|
Then(proxy)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error wrapping metrics handler: %w", err)
|
|
}
|
|
}
|
|
|
|
if m.observabilityMgr.ShouldAddTracing(qualifiedSvcName) {
|
|
proxy = observability.NewService(ctx, serviceName, proxy)
|
|
}
|
|
|
|
if m.observabilityMgr.ShouldAddAccessLogs(qualifiedSvcName) || m.observabilityMgr.ShouldAddMetrics(qualifiedSvcName) {
|
|
// Some piece of middleware, like the ErrorPage, are relying on this serviceBuilder to get the handler for a given service,
|
|
// to re-target the request to it.
|
|
// Those pieces of middleware can be configured on routes that expose a Traefik internal service.
|
|
// In such a case, observability for internals being optional, the capture probe could be absent from context (no wrap via the entrypoint).
|
|
// But if the service targeted by this piece of middleware is not an internal one,
|
|
// and requires observability, we still want the capture probe to be present in the request context.
|
|
// Makes sure a capture probe is in the request context.
|
|
proxy, _ = capture.Wrap(proxy)
|
|
}
|
|
|
|
lb.Add(proxyName, proxy, server.Weight)
|
|
|
|
// servers are considered UP by default.
|
|
info.UpdateServerStatus(target.String(), runtime.StatusUp)
|
|
|
|
healthCheckTargets[proxyName] = target
|
|
}
|
|
|
|
if service.HealthCheck != nil {
|
|
roundTripper, err := m.transportManager.GetRoundTripper(service.ServersTransport)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("getting RoundTripper: %w", err)
|
|
}
|
|
|
|
m.healthCheckers[serviceName] = healthcheck.NewServiceHealthChecker(
|
|
ctx,
|
|
m.observabilityMgr.MetricsRegistry(),
|
|
service.HealthCheck,
|
|
lb,
|
|
info,
|
|
roundTripper,
|
|
healthCheckTargets,
|
|
serviceName,
|
|
)
|
|
}
|
|
|
|
return lb, nil
|
|
}
|
|
|
|
// LaunchHealthCheck launches the health checks.
|
|
func (m *Manager) LaunchHealthCheck(ctx context.Context) {
|
|
for serviceName, hc := range m.healthCheckers {
|
|
logger := log.Ctx(ctx).With().Str(logs.ServiceName, serviceName).Logger()
|
|
go hc.Launch(logger.WithContext(ctx))
|
|
}
|
|
}
|
|
|
|
func shuffle[T any](values []T, r *rand.Rand) []T {
|
|
shuffled := make([]T, len(values))
|
|
copy(shuffled, values)
|
|
r.Shuffle(len(shuffled), func(i, j int) { shuffled[i], shuffled[j] = shuffled[j], shuffled[i] })
|
|
|
|
return shuffled
|
|
}
|