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

Make the loadbalancers servers order random

Browse source 
Co-authored-by: Mathieu Lonjaret <mathieu.lonjaret@gmail.com>
Co-authored-by: Romain <rtribotte@users.noreply.github.com>
This commit is contained in:
Qi 2022-09-14 20:42:08 +08:00 • committed by GitHub
parent 89dc466b23
commit 788f8fa951
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 177 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

15
pkg/server/service/service.go
View file

@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"math/rand"
"net/http"
"net/http/httputil"
"net/url"
@ -51,6 +52,7 @@ func NewManager(configs map[string]*runtime.ServiceInfo, metricsRegistry metrics
roundTripperManager: roundTripperManager,
balancers: make(map[string]healthcheck.Balancers),
configs: configs,
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
}
}
@ -66,6 +68,7 @@ type Manager struct {
// which is why there is not just one Balancer per service name.
balancers map[string]healthcheck.Balancers
configs map[string]*runtime.ServiceInfo
rand *rand.Rand // For the initial shuffling of load-balancers.
}
// BuildHTTP Creates a http.Handler for a service configuration.
@ -212,7 +215,7 @@ func (m *Manager) getWRRServiceHandler(ctx context.Context, serviceName string,
}
balancer := wrr.New(config.Sticky, config.HealthCheck)
for _, service := range config.Services {
for _, service := range shuffle(config.Services, m.rand) {
serviceHandler, err := m.BuildHTTP(ctx, service.Name)
if err != nil {
return nil, err
@ -414,7 +417,7 @@ func (m *Manager) getLoadBalancer(ctx context.Context, serviceName string, servi
func (m *Manager) upsertServers(ctx context.Context, lb healthcheck.BalancerHandler, servers []dynamic.Server) error {
logger := log.FromContext(ctx)
for name, srv := range servers {
for name, srv := range shuffle(servers, m.rand) {
u, err := url.Parse(srv.URL)
if err != nil {
return fmt.Errorf("error parsing server URL %s: %w", srv.URL, err)
@ -443,3 +446,11 @@ func convertSameSite(sameSite string) http.SameSite {
return 0
}
}
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
}

27
pkg/server/service/service_test.go
View file

@ -111,6 +111,7 @@ func TestGetLoadBalancerServiceHandler(t *testing.T) {
type ExpectedResult struct {
StatusCode int
XFrom string
LoadBalanced bool
SecureCookie bool
HTTPOnlyCookie bool
}
@ -139,12 +140,12 @@ func TestGetLoadBalancerServiceHandler(t *testing.T) {
},
expected: []ExpectedResult{
{
StatusCode: http.StatusOK,
XFrom: "first",
StatusCode: http.StatusOK,
LoadBalanced: true,
},
{
StatusCode: http.StatusOK,
XFrom: "second",
StatusCode: http.StatusOK,
LoadBalanced: true,
},
},
},
@ -193,11 +194,9 @@ func TestGetLoadBalancerServiceHandler(t *testing.T) {
expected: []ExpectedResult{
{
StatusCode: http.StatusOK,
XFrom: "first",
},
{
StatusCode: http.StatusOK,
XFrom: "first",
},
},
},
@ -302,13 +301,27 @@ func TestGetLoadBalancerServiceHandler(t *testing.T) {
req.Header.Set("Cookie", test.cookieRawValue)
}
var prevXFrom string
for _, expected := range test.expected {
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, req)
assert.Equal(t, expected.StatusCode, recorder.Code)
assert.Equal(t, expected.XFrom, recorder.Header().Get("X-From"))
if expected.XFrom != "" {
assert.Equal(t, expected.XFrom, recorder.Header().Get("X-From"))
}
xFrom := recorder.Header().Get("X-From")
if prevXFrom != "" {
if expected.LoadBalanced {
assert.NotEqual(t, prevXFrom, xFrom)
} else {
assert.Equal(t, prevXFrom, xFrom)
}
}
prevXFrom = xFrom
cookieHeader := recorder.Header().Get("Set-Cookie")
if len(cookieHeader) > 0 {

16
pkg/server/service/tcp/service.go
View file

@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"math/rand"
"net"
"time"
@ -16,12 +17,14 @@ import (
// Manager is the TCPHandlers factory.
type Manager struct {
configs map[string]*runtime.TCPServiceInfo
rand *rand.Rand // For the initial shuffling of load-balancers.
}
// NewManager creates a new manager.
func NewManager(conf *runtime.Configuration) *Manager {
return &Manager{
configs: conf.TCPServices,
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
}
}
@ -53,7 +56,7 @@ func (m *Manager) BuildTCP(rootCtx context.Context, serviceName string) (tcp.Han
}
duration := time.Duration(*conf.LoadBalancer.TerminationDelay) * time.Millisecond
for name, server := range conf.LoadBalancer.Servers {
for name, server := range shuffle(conf.LoadBalancer.Servers, m.rand) {
if _, _, err := net.SplitHostPort(server.Address); err != nil {
logger.Errorf("In service %q: %v", serviceQualifiedName, err)
continue
@ -71,7 +74,8 @@ func (m *Manager) BuildTCP(rootCtx context.Context, serviceName string) (tcp.Han
return loadBalancer, nil
case conf.Weighted != nil:
loadBalancer := tcp.NewWRRLoadBalancer()
for _, service := range conf.Weighted.Services {
for _, service := range shuffle(conf.Weighted.Services, m.rand) {
handler, err := m.BuildTCP(rootCtx, service.Name)
if err != nil {
logger.Errorf("In service %q: %v", serviceQualifiedName, err)
@ -86,3 +90,11 @@ func (m *Manager) BuildTCP(rootCtx context.Context, serviceName string) (tcp.Han
return nil, err
}
}
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
}

17
pkg/server/service/udp/service.go
View file

@ -4,7 +4,9 @@ import (
"context"
"errors"
"fmt"
"math/rand"
"net"
"time"
"github.com/traefik/traefik/v2/pkg/config/runtime"
"github.com/traefik/traefik/v2/pkg/log"
@ -15,12 +17,14 @@ import (
// Manager handles UDP services creation.
type Manager struct {
configs map[string]*runtime.UDPServiceInfo
rand *rand.Rand // For the initial shuffling of load-balancers.
}
// NewManager creates a new manager.
func NewManager(conf *runtime.Configuration) *Manager {
return &Manager{
configs: conf.UDPServices,
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
}
}
@ -46,7 +50,7 @@ func (m *Manager) BuildUDP(rootCtx context.Context, serviceName string) (udp.Han
case conf.LoadBalancer != nil:
loadBalancer := udp.NewWRRLoadBalancer()
for name, server := range conf.LoadBalancer.Servers {
for name, server := range shuffle(conf.LoadBalancer.Servers, m.rand) {
if _, _, err := net.SplitHostPort(server.Address); err != nil {
logger.Errorf("In udp service %q: %v", serviceQualifiedName, err)
continue
@ -64,7 +68,8 @@ func (m *Manager) BuildUDP(rootCtx context.Context, serviceName string) (udp.Han
return loadBalancer, nil
case conf.Weighted != nil:
loadBalancer := udp.NewWRRLoadBalancer()
for _, service := range conf.Weighted.Services {
for _, service := range shuffle(conf.Weighted.Services, m.rand) {
handler, err := m.BuildUDP(rootCtx, service.Name)
if err != nil {
logger.Errorf("In udp service %q: %v", serviceQualifiedName, err)
@ -79,3 +84,11 @@ func (m *Manager) BuildUDP(rootCtx context.Context, serviceName string) (udp.Han
return nil, err
}
}
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
}