diff --git a/docs/content/migrate/v3.md b/docs/content/migrate/v3.md index 58fd04352..304287398 100644 --- a/docs/content/migrate/v3.md +++ b/docs/content/migrate/v3.md @@ -515,3 +515,13 @@ For the experimental channel: ```shell kubectl apply -f https://github.com/kubernetes-sigs/gateway-api/releases/download/v1.4.0/experimental-install.yaml ``` + +### Kubernetes CRD Provider + +To use the new `leastime` load-balancer algorithm with the Kubernetes CRD provider, you need to update your CRDs. + +**Apply Updated CRDs:** + +```shell +kubectl apply -f https://raw.githubusercontent.com/traefik/traefik/v3.6/docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml +``` diff --git a/docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml b/docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml index bc667c19a..8847677d7 100644 --- a/docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml +++ b/docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml @@ -351,12 +351,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -1313,12 +1314,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3031,12 +3033,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3359,12 +3362,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3506,12 +3510,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3740,12 +3745,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: diff --git a/docs/content/reference/dynamic-configuration/traefik.io_ingressroutes.yaml b/docs/content/reference/dynamic-configuration/traefik.io_ingressroutes.yaml index a99934f69..0a8247d64 100644 --- a/docs/content/reference/dynamic-configuration/traefik.io_ingressroutes.yaml +++ b/docs/content/reference/dynamic-configuration/traefik.io_ingressroutes.yaml @@ -351,12 +351,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: diff --git a/docs/content/reference/dynamic-configuration/traefik.io_middlewares.yaml b/docs/content/reference/dynamic-configuration/traefik.io_middlewares.yaml index 627bc5250..9abd70ea4 100644 --- a/docs/content/reference/dynamic-configuration/traefik.io_middlewares.yaml +++ b/docs/content/reference/dynamic-configuration/traefik.io_middlewares.yaml @@ -484,12 +484,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: diff --git a/docs/content/reference/dynamic-configuration/traefik.io_traefikservices.yaml b/docs/content/reference/dynamic-configuration/traefik.io_traefikservices.yaml index e3ff0ce56..f0f44c1e0 100644 --- a/docs/content/reference/dynamic-configuration/traefik.io_traefikservices.yaml +++ b/docs/content/reference/dynamic-configuration/traefik.io_traefikservices.yaml @@ -262,12 +262,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -590,12 +591,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -737,12 +739,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -971,12 +974,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: diff --git a/docs/content/reference/routing-configuration/http/load-balancing/service.md b/docs/content/reference/routing-configuration/http/load-balancing/service.md index c74d5184c..a79326d48 100644 --- a/docs/content/reference/routing-configuration/http/load-balancing/service.md +++ b/docs/content/reference/routing-configuration/http/load-balancing/service.md @@ -485,6 +485,48 @@ Power of two choices algorithm is a load balancing strategy that selects two ser url = "http://private-ip-server-3/" ``` +## Least-Time + +The Least-Time load balancing algorithm selects the server with the lowest average response time (Time To First Byte - TTFB), +combined with the fewest active connections, weighted by server capacity. +This strategy is ideal for heterogeneous backend environments where servers have varying performance characteristics, +different hardware capabilities, or varying network latency. + +The algorithm continuously measures each backend's response time and tracks active connection counts. +When routing a request, +it calculates a score for each healthy server using the formula: `(avg_response_time × (1 + active_connections)) / weight`. +The server with the lowest score receives the request. +When multiple servers have identical scores, +Weighted Round Robin (WRR) with Earliest Deadline First (EDF) scheduling is used as a tie-breaker to ensure fair distribution. + +??? example "Basic Least-Time Load Balancing -- Using the [File Provider](../../../install-configuration/providers/others/file.md)" + + ```yaml tab="YAML" + ## Dynamic configuration + http: + services: + my-service: + loadBalancer: + strategy: "leasttime" + servers: + - url: "http://private-ip-server-1/" + - url: "http://private-ip-server-2/" + - url: "http://private-ip-server-3/" + ``` + + ```toml tab="TOML" + ## Dynamic configuration + [http.services] + [http.services.my-service.loadBalancer] + strategy = "leasttime" + [[http.services.my-service.loadBalancer.servers]] + url = "http://private-ip-server-1/" + [[http.services.my-service.loadBalancer.servers]] + url = "http://private-ip-server-2/" + [[http.services.my-service.loadBalancer.servers]] + url = "http://private-ip-server-3/" + ``` + ## Mirroring The mirroring is able to mirror requests sent to a service to other services. Please note that by default the whole request is buffered in memory while it is being mirrored. See the `maxBodySize` option in the example below for how to modify this behaviour. You can also omit the request body by setting the `mirrorBody` option to false. diff --git a/docs/content/reference/routing-configuration/kubernetes/crd/http/ingressroute.md b/docs/content/reference/routing-configuration/kubernetes/crd/http/ingressroute.md index a6c437090..cc658fb2a 100644 --- a/docs/content/reference/routing-configuration/kubernetes/crd/http/ingressroute.md +++ b/docs/content/reference/routing-configuration/kubernetes/crd/http/ingressroute.md @@ -57,7 +57,7 @@ spec: httpOnly: true name: cookie secure: true - strategy: RoundRobin + strategy: wrr weight: 10 tls: # Generate a TLS certificate using a certificate resolver diff --git a/docs/content/reference/routing-configuration/kubernetes/crd/http/service.md b/docs/content/reference/routing-configuration/kubernetes/crd/http/service.md index b69515b25..429b654c1 100644 --- a/docs/content/reference/routing-configuration/kubernetes/crd/http/service.md +++ b/docs/content/reference/routing-configuration/kubernetes/crd/http/service.md @@ -47,7 +47,7 @@ spec: httpOnly: true name: cookie secure: true - strategy: RoundRobin + strategy: wrr ``` ```yaml tab="TraefikService" @@ -75,41 +75,41 @@ spec: httpOnly: true name: cookie secure: true - strategy: RoundRobin + strategy: wrr ``` ## Configuration Options -| Field | Description | Default | Required | -|:---------------------------------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:---------------------------------------------------------------------|:---------| -| `kind` | Kind of the service targeted.
Two values allowed:
- **Service**: Kubernetes Service
**TraefikService**: Traefik Service.
More information [here](#externalname-service). | "Service" | No | -| `name` | Service name.
The character `@` is not authorized.
More information [here](#middleware). | | Yes | -| `namespace` | Service namespace.
Can be empty if the service belongs to the same namespace as the IngressRoute.
More information [here](#externalname-service). | | No | -| `port` | Service port (number or port name).
Evaluated only if the kind is **Service**. | | No | -| `responseForwarding.`
`flushInterval` | Interval, in milliseconds, in between flushes to the client while copying the response body.
A negative value means to flush immediately after each write to the client.
This configuration is ignored when a response is a streaming response; for such responses, writes are flushed to the client immediately.
Evaluated only if the kind is **Service**. | 100ms | No | -| `scheme` | Scheme to use for the request to the upstream Kubernetes Service.
Evaluated only if the kind is **Service**. | "http"
"https" if `port` is 443 or contains the string *https*. | No | -| `serversTransport` | Name of ServersTransport resource to use to configure the transport between Traefik and your servers.
Evaluated only if the kind is **Service**. | "" | No | -| `passHostHeader` | Forward client Host header to server.
Evaluated only if the kind is **Service**. | true | No | -| `healthCheck.scheme` | Server URL scheme for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "" | No | -| `healthCheck.mode` | Health check mode.
If defined to grpc, will use the gRPC health check protocol to probe the server.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "http" | No | -| `healthCheck.path` | Server URL path for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "" | No | -| `healthCheck.interval` | Frequency of the health check calls for healthy targets.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "100ms" | No | -| `healthCheck.unhealthyInterval` | Frequency of the health check calls for unhealthy targets.
When not defined, it defaults to the `interval` value.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "100ms" | No | -| `healthCheck.method` | HTTP method for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "GET" | No | -| `healthCheck.status` | Expected HTTP status code of the response to the health check request.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type ExternalName.
If not set, expect a status between 200 and 399.
Evaluated only if the kind is **Service**. | | No | -| `healthCheck.port` | URL port for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | | No | -| `healthCheck.timeout` | Maximum duration to wait before considering the server unhealthy.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "5s" | No | -| `healthCheck.hostname` | Value in the Host header of the health check request.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "" | No | -| `healthCheck.`
`followRedirect` | Follow the redirections during the healtchcheck.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | true | No | -| `healthCheck.headers` | Map of header to send to the health check endpoint
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service)). | | No | -| `sticky.`
`cookie.name` | Name of the cookie used for the stickiness.
When sticky sessions are enabled, a `Set-Cookie` header is set on the initial response to let the client know which server handles the first response.
On subsequent requests, to keep the session alive with the same server, the client should send the cookie with the value set.
If the server pecified in the cookie becomes unhealthy, the request will be forwarded to a new server (and the cookie will keep track of the new server).
Evaluated only if the kind is **Service**. | "" | No | -| `sticky.`
`cookie.httpOnly` | Allow the cookie can be accessed by client-side APIs, such as JavaScript.
Evaluated only if the kind is **Service**. | false | No | -| `sticky.`
`cookie.secure` | Allow the cookie can only be transmitted over an encrypted connection (i.e. HTTPS).
Evaluated only if the kind is **Service**. | false | No | -| `sticky.`
`cookie.sameSite` | [SameSite](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Set-Cookie/SameSite) policy
Allowed values:
-`none`
-`lax`
`strict`
Evaluated only if the kind is **Service**. | "" | No | -| `sticky.`
`cookie.maxAge` | Number of seconds until the cookie expires.
Negative number, the cookie expires immediately.
0, the cookie never expires.
Evaluated only if the kind is **Service**. | 0 | No | -| `strategy` | Load balancing strategy between the servers.
RoundRobin is the only supported value yet.
Evaluated only if the kind is **Service**. | "RoundRobin" | No | -| `nativeLB` | Allow using the Kubernetes Service load balancing between the pods instead of the one provided by Traefik.
Evaluated only if the kind is **Service**. | false | No | -| `nodePortLB` | Use the nodePort IP address when the service type is NodePort.
It allows services to be reachable when Traefik runs externally from the Kubernetes cluster but within the same network of the nodes.
Evaluated only if the kind is **Service**. | false | No | +| Field | Description | Default | Required | +|:---------------------------------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:---------------------------------------------------------------------|:---------| +| `kind` | Kind of the service targeted.
Two values allowed:
- **Service**: Kubernetes Service
**TraefikService**: Traefik Service.
More information [here](#externalname-service). | "Service" | No | +| `name` | Service name.
The character `@` is not authorized.
More information [here](#middleware). | | Yes | +| `namespace` | Service namespace.
Can be empty if the service belongs to the same namespace as the IngressRoute.
More information [here](#externalname-service). | | No | +| `port` | Service port (number or port name).
Evaluated only if the kind is **Service**. | | No | +| `responseForwarding.`
`flushInterval` | Interval, in milliseconds, in between flushes to the client while copying the response body.
A negative value means to flush immediately after each write to the client.
This configuration is ignored when a response is a streaming response; for such responses, writes are flushed to the client immediately.
Evaluated only if the kind is **Service**. | 100ms | No | +| `scheme` | Scheme to use for the request to the upstream Kubernetes Service.
Evaluated only if the kind is **Service**. | "http"
"https" if `port` is 443 or contains the string *https*. | No | +| `serversTransport` | Name of ServersTransport resource to use to configure the transport between Traefik and your servers.
Evaluated only if the kind is **Service**. | "" | No | +| `passHostHeader` | Forward client Host header to server.
Evaluated only if the kind is **Service**. | true | No | +| `healthCheck.scheme` | Server URL scheme for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "" | No | +| `healthCheck.mode` | Health check mode.
If defined to grpc, will use the gRPC health check protocol to probe the server.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "http" | No | +| `healthCheck.path` | Server URL path for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "" | No | +| `healthCheck.interval` | Frequency of the health check calls for healthy targets.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "100ms" | No | +| `healthCheck.unhealthyInterval` | Frequency of the health check calls for unhealthy targets.
When not defined, it defaults to the `interval` value.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "100ms" | No | +| `healthCheck.method` | HTTP method for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "GET" | No | +| `healthCheck.status` | Expected HTTP status code of the response to the health check request.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type ExternalName.
If not set, expect a status between 200 and 399.
Evaluated only if the kind is **Service**. | | No | +| `healthCheck.port` | URL port for the health check endpoint.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | | No | +| `healthCheck.timeout` | Maximum duration to wait before considering the server unhealthy.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "5s" | No | +| `healthCheck.hostname` | Value in the Host header of the health check request.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | "" | No | +| `healthCheck.`
`followRedirect` | Follow the redirections during the healtchcheck.
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service). | true | No | +| `healthCheck.headers` | Map of header to send to the health check endpoint
Evaluated only if the kind is **Service**.
Only for [Kubernetes service](https://kubernetes.io/docs/concepts/services-networking/service/) of type [ExternalName](#externalname-service)). | | No | +| `sticky.`
`cookie.name` | Name of the cookie used for the stickiness.
When sticky sessions are enabled, a `Set-Cookie` header is set on the initial response to let the client know which server handles the first response.
On subsequent requests, to keep the session alive with the same server, the client should send the cookie with the value set.
If the server pecified in the cookie becomes unhealthy, the request will be forwarded to a new server (and the cookie will keep track of the new server).
Evaluated only if the kind is **Service**. | "" | No | +| `sticky.`
`cookie.httpOnly` | Allow the cookie can be accessed by client-side APIs, such as JavaScript.
Evaluated only if the kind is **Service**. | false | No | +| `sticky.`
`cookie.secure` | Allow the cookie can only be transmitted over an encrypted connection (i.e. HTTPS).
Evaluated only if the kind is **Service**. | false | No | +| `sticky.`
`cookie.sameSite` | [SameSite](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Set-Cookie/SameSite) policy
Allowed values:
-`none`
-`lax`
`strict`
Evaluated only if the kind is **Service**. | "" | No | +| `sticky.`
`cookie.maxAge` | Number of seconds until the cookie expires.
Negative number, the cookie expires immediately.
0, the cookie never expires.
Evaluated only if the kind is **Service**. | 0 | No | +| `strategy` | Strategy defines the load balancing strategy between the servers.
Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time).
Evaluated only if the kind is **Service**. | "RoundRobin" | No | +| `nativeLB` | Allow using the Kubernetes Service load balancing between the pods instead of the one provided by Traefik.
Evaluated only if the kind is **Service**. | false | No | +| `nodePortLB` | Use the nodePort IP address when the service type is NodePort.
It allows services to be reachable when Traefik runs externally from the Kubernetes cluster but within the same network of the nodes.
Evaluated only if the kind is **Service**. | false | No | ### ExternalName Service diff --git a/integration/docker_test.go b/integration/docker_test.go index d6a6819ff..1a767ff6f 100644 --- a/integration/docker_test.go +++ b/integration/docker_test.go @@ -4,7 +4,6 @@ import ( "encoding/json" "io" "net/http" - "strings" "testing" "time" @@ -56,56 +55,6 @@ func (s *DockerSuite) TestSimpleConfiguration() { require.NoError(s.T(), err) } -func (s *DockerSuite) TestWRRServer() { - tempObjects := struct { - DockerHost string - DefaultRule string - }{ - DockerHost: s.getDockerHost(), - DefaultRule: "Host(`{{ normalize .Name }}.docker.localhost`)", - } - - file := s.adaptFile("fixtures/docker/simple.toml", tempObjects) - - s.composeUp() - - s.traefikCmd(withConfigFile(file)) - - whoami1IP := s.getComposeServiceIP("wrr-server") - whoami2IP := s.getComposeServiceIP("wrr-server2") - - // Expected a 404 as we did not configure anything - err := try.GetRequest("http://127.0.0.1:8000/", 500*time.Millisecond, try.StatusCodeIs(http.StatusNotFound)) - require.NoError(s.T(), err) - - err = try.GetRequest("http://127.0.0.1:8080/api/http/services", 1000*time.Millisecond, try.BodyContains("wrr-server")) - require.NoError(s.T(), err) - - repartition := map[string]int{} - for range 4 { - req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8000/whoami", nil) - req.Host = "my.wrr.host" - require.NoError(s.T(), err) - - response, err := http.DefaultClient.Do(req) - require.NoError(s.T(), err) - assert.Equal(s.T(), http.StatusOK, response.StatusCode) - - body, err := io.ReadAll(response.Body) - require.NoError(s.T(), err) - - if strings.Contains(string(body), whoami1IP) { - repartition[whoami1IP]++ - } - if strings.Contains(string(body), whoami2IP) { - repartition[whoami2IP]++ - } - } - - assert.Equal(s.T(), 3, repartition[whoami1IP]) - assert.Equal(s.T(), 1, repartition[whoami2IP]) -} - func (s *DockerSuite) TestDefaultDockerContainers() { tempObjects := struct { DockerHost string diff --git a/integration/fixtures/k8s/01-traefik-crd.yml b/integration/fixtures/k8s/01-traefik-crd.yml index bc667c19a..8847677d7 100644 --- a/integration/fixtures/k8s/01-traefik-crd.yml +++ b/integration/fixtures/k8s/01-traefik-crd.yml @@ -351,12 +351,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -1313,12 +1314,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3031,12 +3033,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3359,12 +3362,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3506,12 +3510,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: @@ -3740,12 +3745,13 @@ spec: strategy: description: |- Strategy defines the load balancing strategy between the servers. - Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). RoundRobin value is deprecated and supported for backward compatibility. enum: - wrr - p2c - hrw + - leasttime - RoundRobin type: string weight: diff --git a/integration/fixtures/leasttime_server.toml b/integration/fixtures/leasttime_server.toml new file mode 100644 index 000000000..263fe906a --- /dev/null +++ b/integration/fixtures/leasttime_server.toml @@ -0,0 +1,36 @@ +[global] + checkNewVersion = false + sendAnonymousUsage = false + +[api] + insecure = true + +[log] + level = "DEBUG" + noColor = true + +[entryPoints] + + [entryPoints.web] + address = ":8000" + +[providers.file] + filename = "{{ .SelfFilename }}" + +## dynamic configuration ## + +[http.routers] + [http.routers.router] + service = "service1" + rule = "Path(`/whoami`)" + +[http.services] + + [http.services.service1.loadBalancer] + strategy = "leasttime" + [[http.services.service1.loadBalancer.servers]] + url = "{{ .Server1 }}" + weight = 1 + [[http.services.service1.loadBalancer.servers]] + url = "{{ .Server2 }}" + weight = 1 diff --git a/integration/resources/compose/docker.yml b/integration/resources/compose/docker.yml index 7ee4492cf..788866e34 100644 --- a/integration/resources/compose/docker.yml +++ b/integration/resources/compose/docker.yml @@ -35,14 +35,3 @@ services: labels: traefik.http.Routers.Super.Rule: Host(`my.super.host`) traefik.http.Services.powpow.LoadBalancer.server.Port: 2375 - - wrr-server: - image: traefik/whoami - labels: - traefik.http.Routers.wrr-server.Rule: Host(`my.wrr.host`) - traefik.http.Services.wrr-server.LoadBalancer.server.Weight: 4 - wrr-server2: - image: traefik/whoami - labels: - traefik.http.Routers.wrr-server.Rule: Host(`my.wrr.host`) - traefik.http.Services.wrr-server.LoadBalancer.server.Weight: 1 diff --git a/integration/simple_test.go b/integration/simple_test.go index 2be9f7f06..ed30d7428 100644 --- a/integration/simple_test.go +++ b/integration/simple_test.go @@ -885,6 +885,109 @@ func (s *SimpleSuite) TestWRRServer() { assert.Equal(s.T(), 1, repartition[whoami2IP]) } +func (s *SimpleSuite) TestLeastTimeServer() { + s.createComposeProject("base") + + s.composeUp() + defer s.composeDown() + + whoami1IP := s.getComposeServiceIP("whoami1") + whoami2IP := s.getComposeServiceIP("whoami2") + + file := s.adaptFile("fixtures/leasttime_server.toml", struct { + Server1 string + Server2 string + }{Server1: "http://" + whoami1IP, Server2: "http://" + whoami2IP}) + + s.traefikCmd(withConfigFile(file)) + + err := try.GetRequest("http://127.0.0.1:8080/api/http/services", 1000*time.Millisecond, try.BodyContains("service1")) + require.NoError(s.T(), err) + + // Verify leasttime strategy is configured + err = try.GetRequest("http://127.0.0.1:8080/api/http/services", 1000*time.Millisecond, try.BodyContains("leasttime")) + require.NoError(s.T(), err) + + // Make requests and verify both servers respond + repartition := map[string]int{} + for range 10 { + req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8000/whoami", nil) + require.NoError(s.T(), err) + + response, err := http.DefaultClient.Do(req) + require.NoError(s.T(), err) + assert.Equal(s.T(), http.StatusOK, response.StatusCode) + + body, err := io.ReadAll(response.Body) + require.NoError(s.T(), err) + + if strings.Contains(string(body), whoami1IP) { + repartition[whoami1IP]++ + } + if strings.Contains(string(body), whoami2IP) { + repartition[whoami2IP]++ + } + } + + // Both servers should have received requests + assert.Positive(s.T(), repartition[whoami1IP]) + assert.Positive(s.T(), repartition[whoami2IP]) +} + +func (s *SimpleSuite) TestLeastTimeHeterogeneousPerformance() { + // Create test servers with different response times + var fastServerCalls, slowServerCalls atomic.Int32 + + fastServer := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + fastServerCalls.Add(1) + time.Sleep(10 * time.Millisecond) // Fast server + rw.WriteHeader(http.StatusOK) + _, _ = rw.Write([]byte("fast-server")) + })) + defer fastServer.Close() + + slowServer := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + slowServerCalls.Add(1) + time.Sleep(100 * time.Millisecond) // Slow server + rw.WriteHeader(http.StatusOK) + _, _ = rw.Write([]byte("slow-server")) + })) + defer slowServer.Close() + + file := s.adaptFile("fixtures/leasttime_server.toml", struct { + Server1 string + Server2 string + }{Server1: fastServer.URL, Server2: slowServer.URL}) + + s.traefikCmd(withConfigFile(file)) + + err := try.GetRequest("http://127.0.0.1:8080/api/http/services", 1000*time.Millisecond, try.BodyContains("service1")) + require.NoError(s.T(), err) + + // Make 20 requests to build up response time statistics + for range 20 { + req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8000/whoami", nil) + require.NoError(s.T(), err) + + response, err := http.DefaultClient.Do(req) + require.NoError(s.T(), err) + assert.Equal(s.T(), http.StatusOK, response.StatusCode) + _, _ = io.ReadAll(response.Body) + response.Body.Close() + } + + // Verify that the fast server received significantly more requests (>70%) + fastCalls := fastServerCalls.Load() + slowCalls := slowServerCalls.Load() + totalCalls := fastCalls + slowCalls + + assert.Equal(s.T(), int32(20), totalCalls) + + // Fast server should get >70% of traffic due to lower response time + fastPercentage := float64(fastCalls) / float64(totalCalls) * 100 + assert.Greater(s.T(), fastPercentage, 70.0) +} + func (s *SimpleSuite) TestWRR() { s.createComposeProject("base") diff --git a/pkg/config/dynamic/http_config.go b/pkg/config/dynamic/http_config.go index 7ea0ac3c7..f9022c19d 100644 --- a/pkg/config/dynamic/http_config.go +++ b/pkg/config/dynamic/http_config.go @@ -190,6 +190,12 @@ type WRRService struct { GRPCStatus *GRPCStatus `json:"-" toml:"-" yaml:"-" label:"-" file:"-"` } +// SetDefaults Default values for a WRRService. +func (w *WRRService) SetDefaults() { + defaultWeight := 1 + w.Weight = &defaultWeight +} + // +k8s:deepcopy-gen=true // HRWService is a reference to a service load-balanced with highest random weight. @@ -204,12 +210,6 @@ func (w *HRWService) SetDefaults() { w.Weight = &defaultWeight } -// SetDefaults Default values for a WRRService. -func (w *WRRService) SetDefaults() { - defaultWeight := 1 - w.Weight = &defaultWeight -} - // +k8s:deepcopy-gen=true type GRPCStatus struct { @@ -267,6 +267,8 @@ const ( BalancerStrategyP2C BalancerStrategy = "p2c" // BalancerStrategyHRW is the highest random weight strategy. BalancerStrategyHRW BalancerStrategy = "hrw" + // BalancerStrategyLeastTime is the least-time strategy. + BalancerStrategyLeastTime BalancerStrategy = "leasttime" ) // +k8s:deepcopy-gen=true diff --git a/pkg/provider/kubernetes/crd/fixtures/with_leasttime_strategy.yml b/pkg/provider/kubernetes/crd/fixtures/with_leasttime_strategy.yml new file mode 100644 index 000000000..4f47fd4bb --- /dev/null +++ b/pkg/provider/kubernetes/crd/fixtures/with_leasttime_strategy.yml @@ -0,0 +1,16 @@ +apiVersion: traefik.io/v1alpha1 +kind: IngressRoute +metadata: + name: test.route + namespace: default +spec: + entryPoints: + - web + routes: + - match: Host(`foo.com`) && PathPrefix(`/leasttime`) + kind: Rule + priority: 12 + services: + - name: whoami2 + port: 8080 + strategy: leasttime diff --git a/pkg/provider/kubernetes/crd/kubernetes_http.go b/pkg/provider/kubernetes/crd/kubernetes_http.go index 346752c42..26f8b1bfd 100644 --- a/pkg/provider/kubernetes/crd/kubernetes_http.go +++ b/pkg/provider/kubernetes/crd/kubernetes_http.go @@ -384,7 +384,7 @@ func (c configBuilder) buildServersLB(namespace string, svc traefikv1alpha1.Load // TODO: remove this when the fake client apply default values. if svc.Strategy != "" { switch svc.Strategy { - case dynamic.BalancerStrategyWRR, dynamic.BalancerStrategyP2C, dynamic.BalancerStrategyHRW: + case dynamic.BalancerStrategyWRR, dynamic.BalancerStrategyP2C, dynamic.BalancerStrategyHRW, dynamic.BalancerStrategyLeastTime: lb.Strategy = svc.Strategy // Here we are just logging a warning as the default value is already applied. diff --git a/pkg/provider/kubernetes/crd/kubernetes_test.go b/pkg/provider/kubernetes/crd/kubernetes_test.go index 124fd87c9..eac9b5fc1 100644 --- a/pkg/provider/kubernetes/crd/kubernetes_test.go +++ b/pkg/provider/kubernetes/crd/kubernetes_test.go @@ -5667,6 +5667,54 @@ func TestLoadIngressRoutes(t *testing.T) { TLS: &dynamic.TLSConfiguration{}, }, }, + { + desc: "Simple Ingress Route with leasttime strategy", + paths: []string{"services.yml", "with_leasttime_strategy.yml"}, + expected: &dynamic.Configuration{ + UDP: &dynamic.UDPConfiguration{ + Routers: map[string]*dynamic.UDPRouter{}, + Services: map[string]*dynamic.UDPService{}, + }, + TCP: &dynamic.TCPConfiguration{ + Routers: map[string]*dynamic.TCPRouter{}, + Middlewares: map[string]*dynamic.TCPMiddleware{}, + Services: map[string]*dynamic.TCPService{}, + ServersTransports: map[string]*dynamic.TCPServersTransport{}, + }, + HTTP: &dynamic.HTTPConfiguration{ + Routers: map[string]*dynamic.Router{ + "default-test-route-55869f6407935ccfa805": { + EntryPoints: []string{"web"}, + Service: "default-test-route-55869f6407935ccfa805", + Rule: "Host(`foo.com`) && PathPrefix(`/leasttime`)", + Priority: 12, + }, + }, + Middlewares: map[string]*dynamic.Middleware{}, + Services: map[string]*dynamic.Service{ + "default-test-route-55869f6407935ccfa805": { + LoadBalancer: &dynamic.ServersLoadBalancer{ + Strategy: dynamic.BalancerStrategyLeastTime, + Servers: []dynamic.Server{ + { + URL: "http://10.10.0.3:8080", + }, + { + URL: "http://10.10.0.4:8080", + }, + }, + PassHostHeader: pointer(true), + ResponseForwarding: &dynamic.ResponseForwarding{ + FlushInterval: ptypes.Duration(100 * time.Millisecond), + }, + }, + }, + }, + ServersTransports: map[string]*dynamic.ServersTransport{}, + }, + TLS: &dynamic.TLSConfiguration{}, + }, + }, } for _, test := range testCases { diff --git a/pkg/provider/kubernetes/crd/traefikio/v1alpha1/ingressroute.go b/pkg/provider/kubernetes/crd/traefikio/v1alpha1/ingressroute.go index d7b7af331..c3978e94a 100644 --- a/pkg/provider/kubernetes/crd/traefikio/v1alpha1/ingressroute.go +++ b/pkg/provider/kubernetes/crd/traefikio/v1alpha1/ingressroute.go @@ -118,10 +118,10 @@ type LoadBalancerSpec struct { // It defaults to https when Kubernetes Service port is 443, http otherwise. Scheme string `json:"scheme,omitempty"` // Strategy defines the load balancing strategy between the servers. - // Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), and hrw (Highest Random Weight). + // Supported values are: wrr (Weighed round-robin), p2c (Power of two choices), hrw (Highest Random Weight), and leasttime (Least-Time). // RoundRobin value is deprecated and supported for backward compatibility. - // TODO: when the deprecated RoundRobin value will be removed, set the default value to wrr. - // +kubebuilder:validation:Enum=wrr;p2c;hrw;RoundRobin + // TODO: when the deprecated RoundRobin value will be removed, set the default kubebuilder value to wrr. + // +kubebuilder:validation:Enum=wrr;p2c;hrw;leasttime;RoundRobin Strategy dynamic.BalancerStrategy `json:"strategy,omitempty"` // PassHostHeader defines whether the client Host header is forwarded to the upstream Kubernetes Service. // By default, passHostHeader is true. diff --git a/pkg/server/service/loadbalancer/hrw/hrw.go b/pkg/server/service/loadbalancer/hrw/hrw.go index f213b326a..51ca397fd 100644 --- a/pkg/server/service/loadbalancer/hrw/hrw.go +++ b/pkg/server/service/loadbalancer/hrw/hrw.go @@ -13,6 +13,8 @@ import ( "github.com/traefik/traefik/v3/pkg/ip" ) +var errNoAvailableServer = errors.New("no available server") + type namedHandler struct { http.Handler name string @@ -114,15 +116,13 @@ func (b *Balancer) SetStatus(ctx context.Context, childName string, up bool) { // Not thread safe. func (b *Balancer) RegisterStatusUpdater(fn func(up bool)) error { if !b.wantsHealthCheck { - return errors.New("healthCheck not enabled in config for this weighted service") + return errors.New("healthCheck not enabled in config for this HRW service") } b.updaters = append(b.updaters, fn) return nil } -var errNoAvailableServer = errors.New("no available server") - func (b *Balancer) nextServer(ip string) (*namedHandler, error) { b.handlersMu.RLock() var healthy []*namedHandler diff --git a/pkg/server/service/loadbalancer/leasttime/leasttime.go b/pkg/server/service/loadbalancer/leasttime/leasttime.go new file mode 100644 index 000000000..41f4f5218 --- /dev/null +++ b/pkg/server/service/loadbalancer/leasttime/leasttime.go @@ -0,0 +1,373 @@ +package leasttime + +import ( + "context" + "errors" + "math" + "net/http" + "net/http/httptrace" + "sync" + "sync/atomic" + "time" + + "github.com/rs/zerolog/log" + "github.com/traefik/traefik/v3/pkg/config/dynamic" + "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer" +) + +const sampleSize = 100 // Number of response time samples to track. + +var errNoAvailableServer = errors.New("no available server") + +// namedHandler wraps an HTTP handler with metrics and server information. +// Tracks response time (TTFB) and inflight request count for load balancing decisions. +type namedHandler struct { + http.Handler + name string + weight float64 + + deadlineMu sync.RWMutex + deadline float64 // WRR tie-breaking (EDF scheduling). + + inflightCount atomic.Int64 // Number of inflight requests. + + responseTimeMu sync.RWMutex + responseTimes [sampleSize]float64 // Fixed-size ring buffer (TTFB measurements in ms). + responseTimeIdx int // Current position in ring buffer. + responseTimeSum float64 // Sum of all values in buffer. + sampleCount int // Number of samples collected so far. +} + +// updateResponseTime updates the average response time for this server using a ring buffer. +func (s *namedHandler) updateResponseTime(elapsed time.Duration) { + s.responseTimeMu.Lock() + defer s.responseTimeMu.Unlock() + + ms := float64(elapsed.Milliseconds()) + + if s.sampleCount < sampleSize { + // Still filling the buffer. + s.responseTimes[s.responseTimeIdx] = ms + s.responseTimeSum += ms + s.sampleCount++ + } else { + // Buffer is full, replace oldest value. + oldValue := s.responseTimes[s.responseTimeIdx] + s.responseTimes[s.responseTimeIdx] = ms + s.responseTimeSum = s.responseTimeSum - oldValue + ms + } + + s.responseTimeIdx = (s.responseTimeIdx + 1) % sampleSize +} + +// getAvgResponseTime returns the average response time in milliseconds. +// Returns 0 if no samples have been collected yet (cold start). +func (s *namedHandler) getAvgResponseTime() float64 { + s.responseTimeMu.RLock() + defer s.responseTimeMu.RUnlock() + + if s.sampleCount == 0 { + return 0 + } + return s.responseTimeSum / float64(s.sampleCount) +} + +func (s *namedHandler) getDeadline() float64 { + s.deadlineMu.RLock() + defer s.deadlineMu.RUnlock() + return s.deadline +} + +func (s *namedHandler) setDeadline(deadline float64) { + s.deadlineMu.Lock() + defer s.deadlineMu.Unlock() + s.deadline = deadline +} + +// Balancer implements the least-time load balancing algorithm. +// It selects the server with the lowest average response time (TTFB) and fewest active connections. +type Balancer struct { + wantsHealthCheck bool + + // handlersMu protects the handlers slice, the status and the fenced maps. + handlersMu sync.RWMutex + handlers []*namedHandler + // status is a record of which child services of the Balancer are healthy, keyed + // by name of child service. A service is initially added to the map when it is + // created via Add, and it is later removed or added to the map as needed, + // through the SetStatus method. + status map[string]struct{} + // fenced is the list of terminating yet still serving child services. + fenced map[string]struct{} + + // updaters is the list of hooks that are run (to update the Balancer + // parent(s)), whenever the Balancer status changes. + // No mutex is needed, as it is modified only during the configuration build. + updaters []func(bool) + + sticky *loadbalancer.Sticky + + // deadlineMu protects EDF scheduling state (curDeadline and all handler deadline fields). + // Separate from handlersMu to reduce lock contention during tie-breaking. + curDeadlineMu sync.RWMutex + // curDeadline is used for WRR tie-breaking (EDF scheduling). + curDeadline float64 +} + +// New creates a new least-time load balancer. +func New(stickyConfig *dynamic.Sticky, wantsHealthCheck bool) *Balancer { + balancer := &Balancer{ + status: make(map[string]struct{}), + fenced: make(map[string]struct{}), + wantsHealthCheck: wantsHealthCheck, + } + if stickyConfig != nil && stickyConfig.Cookie != nil { + balancer.sticky = loadbalancer.NewSticky(*stickyConfig.Cookie) + } + + return balancer +} + +// SetStatus sets on the balancer that its given child is now of the given +// status. childName is only needed for logging purposes. +func (b *Balancer) SetStatus(ctx context.Context, childName string, up bool) { + b.handlersMu.Lock() + defer b.handlersMu.Unlock() + + upBefore := len(b.status) > 0 + + status := "DOWN" + if up { + status = "UP" + } + + log.Ctx(ctx).Debug().Msgf("Setting status of %s to %v", childName, status) + + if up { + b.status[childName] = struct{}{} + } else { + delete(b.status, childName) + } + + upAfter := len(b.status) > 0 + status = "DOWN" + if upAfter { + status = "UP" + } + + // No Status Change. + if upBefore == upAfter { + // We're still with the same status, no need to propagate. + log.Ctx(ctx).Debug().Msgf("Still %s, no need to propagate", status) + return + } + + // Status Change. + log.Ctx(ctx).Debug().Msgf("Propagating new %s status", status) + for _, fn := range b.updaters { + fn(upAfter) + } +} + +// RegisterStatusUpdater adds fn to the list of hooks that are run when the +// status of the Balancer changes. +// Not thread safe. +func (b *Balancer) RegisterStatusUpdater(fn func(up bool)) error { + if !b.wantsHealthCheck { + return errors.New("healthCheck not enabled in config for this LeastTime service") + } + b.updaters = append(b.updaters, fn) + return nil +} + +// getHealthyServers returns the list of healthy, non-fenced servers. +func (b *Balancer) getHealthyServers() []*namedHandler { + b.handlersMu.RLock() + defer b.handlersMu.RUnlock() + + var healthy []*namedHandler + for _, h := range b.handlers { + if _, ok := b.status[h.name]; ok { + if _, fenced := b.fenced[h.name]; !fenced { + healthy = append(healthy, h) + } + } + } + return healthy +} + +// selectWRR selects a server from candidates using Weighted Round Robin (EDF scheduling). +// This is used for tie-breaking when multiple servers have identical scores. +func (b *Balancer) selectWRR(candidates []*namedHandler) *namedHandler { + if len(candidates) == 0 { + return nil + } + + selected := candidates[0] + minDeadline := math.MaxFloat64 + + // Find handler with earliest deadline. + for _, h := range candidates { + handlerDeadline := h.getDeadline() + if handlerDeadline < minDeadline { + minDeadline = handlerDeadline + selected = h + } + } + + // Update deadline based on when this server was selected (minDeadline), + // not the global curDeadline. This ensures proper weighted distribution. + newDeadline := minDeadline + 1/selected.weight + selected.setDeadline(newDeadline) + + // Track the maximum deadline assigned for initializing new servers. + b.curDeadlineMu.Lock() + if newDeadline > b.curDeadline { + b.curDeadline = newDeadline + } + b.curDeadlineMu.Unlock() + + return selected +} + +// Score = (avgResponseTime × (1 + inflightCount)) / weight. +func (b *Balancer) nextServer() (*namedHandler, error) { + healthy := b.getHealthyServers() + + if len(healthy) == 0 { + return nil, errNoAvailableServer + } + + if len(healthy) == 1 { + return healthy[0], nil + } + + // Calculate scores and find minimum. + minScore := math.MaxFloat64 + var candidates []*namedHandler + + for _, h := range healthy { + avgRT := h.getAvgResponseTime() + inflight := float64(h.inflightCount.Load()) + score := (avgRT * (1 + inflight)) / h.weight + + if score < minScore { + minScore = score + candidates = []*namedHandler{h} + } else if score == minScore { + candidates = append(candidates, h) + } + } + + if len(candidates) == 1 { + return candidates[0], nil + } + + // Multiple servers with same score: use WRR (EDF) tie-breaking. + selected := b.selectWRR(candidates) + if selected == nil { + return nil, errNoAvailableServer + } + + return selected, nil +} + +func (b *Balancer) ServeHTTP(rw http.ResponseWriter, req *http.Request) { + // Handle sticky sessions first. + if b.sticky != nil { + h, rewrite, err := b.sticky.StickyHandler(req) + if err != nil { + log.Error().Err(err).Msg("Error while getting sticky handler") + } else if h != nil { + b.handlersMu.RLock() + _, ok := b.status[h.Name] + b.handlersMu.RUnlock() + if ok { + if rewrite { + if err := b.sticky.WriteStickyCookie(rw, h.Name); err != nil { + log.Error().Err(err).Msg("Writing sticky cookie") + } + } + + h.ServeHTTP(rw, req) + return + } + } + } + + server, err := b.nextServer() + if err != nil { + if errors.Is(err, errNoAvailableServer) { + http.Error(rw, errNoAvailableServer.Error(), http.StatusServiceUnavailable) + } else { + http.Error(rw, err.Error(), http.StatusInternalServerError) + } + return + } + + if b.sticky != nil { + if err := b.sticky.WriteStickyCookie(rw, server.name); err != nil { + log.Error().Err(err).Msg("Error while writing sticky cookie") + } + } + + // Track inflight requests. + server.inflightCount.Add(1) + defer server.inflightCount.Add(-1) + + startTime := time.Now() + trace := &httptrace.ClientTrace{ + GotFirstResponseByte: func() { + // Update average response time (TTFB). + server.updateResponseTime(time.Since(startTime)) + }, + } + traceCtx := httptrace.WithClientTrace(req.Context(), trace) + server.ServeHTTP(rw, req.WithContext(traceCtx)) +} + +// AddServer adds a handler with a server. +func (b *Balancer) AddServer(name string, handler http.Handler, server dynamic.Server) { + b.Add(name, handler, server.Weight, server.Fenced) +} + +// Add adds a handler. +// A handler with a non-positive weight is ignored. +func (b *Balancer) Add(name string, handler http.Handler, weight *int, fenced bool) { + w := 1 + if weight != nil { + w = *weight + } + + if w <= 0 { // non-positive weight is meaningless. + return + } + + h := &namedHandler{Handler: handler, name: name, weight: float64(w)} + + // Initialize deadline by adding 1/weight to current deadline. + // This staggers servers to prevent all starting at the same time. + var deadline float64 + b.curDeadlineMu.RLock() + deadline = b.curDeadline + 1/h.weight + b.curDeadlineMu.RUnlock() + + h.setDeadline(deadline) + + // Update balancer's current deadline with the new server's deadline. + b.curDeadlineMu.Lock() + b.curDeadline = deadline + b.curDeadlineMu.Unlock() + + b.handlersMu.Lock() + b.handlers = append(b.handlers, h) + b.status[name] = struct{}{} + if fenced { + b.fenced[name] = struct{}{} + } + b.handlersMu.Unlock() + + if b.sticky != nil { + b.sticky.AddHandler(name, handler) + } +} diff --git a/pkg/server/service/loadbalancer/leasttime/leasttime_test.go b/pkg/server/service/loadbalancer/leasttime/leasttime_test.go new file mode 100644 index 000000000..aae0d8240 --- /dev/null +++ b/pkg/server/service/loadbalancer/leasttime/leasttime_test.go @@ -0,0 +1,1093 @@ +package leasttime + +import ( + "context" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/traefik/traefik/v3/pkg/config/dynamic" +) + +type key string + +const serviceName key = "serviceName" + +func pointer[T any](v T) *T { return &v } + +// responseRecorder tracks which servers handled requests. +type responseRecorder struct { + *httptest.ResponseRecorder + save map[string]int +} + +func (r *responseRecorder) WriteHeader(statusCode int) { + server := r.Header().Get("server") + if server != "" { + r.save[server]++ + } + r.ResponseRecorder.WriteHeader(statusCode) +} + +// TestBalancer tests basic server addition and least-time selection. +func TestBalancer(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "second") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 10 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // With least-time and equal response times, both servers should get some traffic. + assert.Positive(t, recorder.save["first"]) + assert.Positive(t, recorder.save["second"]) + assert.Equal(t, 10, recorder.save["first"]+recorder.save["second"]) +} + +// TestBalancerNoService tests behavior when no servers are configured. +func TestBalancerNoService(t *testing.T) { + balancer := New(nil, false) + + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + + assert.Equal(t, http.StatusServiceUnavailable, recorder.Result().StatusCode) +} + +// TestBalancerNoServiceUp tests behavior when all servers are marked down. +func TestBalancerNoServiceUp(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.WriteHeader(http.StatusInternalServerError) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.WriteHeader(http.StatusInternalServerError) + }), pointer(1), false) + + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "parent"), "first", false) + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "parent"), "second", false) + + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + + assert.Equal(t, http.StatusServiceUnavailable, recorder.Result().StatusCode) +} + +// TestBalancerOneServerDown tests that down servers are excluded from selection. +func TestBalancerOneServerDown(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.WriteHeader(http.StatusInternalServerError) + }), pointer(1), false) + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "parent"), "second", false) + + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 3 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + assert.Equal(t, 3, recorder.save["first"]) + assert.Equal(t, 0, recorder.save["second"]) +} + +// TestBalancerOneServerDownThenUp tests server status transitions. +func TestBalancerOneServerDownThenUp(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "second") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "parent"), "second", false) + + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 3 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + assert.Equal(t, 3, recorder.save["first"]) + assert.Equal(t, 0, recorder.save["second"]) + + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "parent"), "second", true) + recorder = &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 20 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + // Both servers should get some traffic. + assert.Positive(t, recorder.save["first"]) + assert.Positive(t, recorder.save["second"]) + assert.Equal(t, 20, recorder.save["first"]+recorder.save["second"]) +} + +// TestBalancerAllServersZeroWeight tests that all zero-weight servers result in no available server. +func TestBalancerAllServersZeroWeight(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), pointer(0), false) + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), pointer(0), false) + + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + + assert.Equal(t, http.StatusServiceUnavailable, recorder.Result().StatusCode) +} + +// TestBalancerOneServerZeroWeight tests that zero-weight servers are ignored. +func TestBalancerOneServerZeroWeight(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), pointer(0), false) + + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 3 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Only first server should receive traffic. + assert.Equal(t, 3, recorder.save["first"]) + assert.Equal(t, 0, recorder.save["second"]) +} + +// TestBalancerPropagate tests status propagation to parent balancers. +func TestBalancerPropagate(t *testing.T) { + balancer1 := New(nil, true) + + balancer1.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + balancer1.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "second") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer2 := New(nil, true) + balancer2.Add("third", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "third") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + balancer2.Add("fourth", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "fourth") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + topBalancer := New(nil, true) + topBalancer.Add("balancer1", balancer1, pointer(1), false) + topBalancer.Add("balancer2", balancer2, pointer(1), false) + err := balancer1.RegisterStatusUpdater(func(up bool) { + topBalancer.SetStatus(context.WithValue(t.Context(), serviceName, "top"), "balancer1", up) + }) + assert.NoError(t, err) + err = balancer2.RegisterStatusUpdater(func(up bool) { + topBalancer.SetStatus(context.WithValue(t.Context(), serviceName, "top"), "balancer2", up) + }) + assert.NoError(t, err) + + // Set all children of balancer1 to down, should propagate to top. + balancer1.SetStatus(context.WithValue(t.Context(), serviceName, "top"), "first", false) + balancer1.SetStatus(context.WithValue(t.Context(), serviceName, "top"), "second", false) + + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 4 { + topBalancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Only balancer2 should receive traffic. + assert.Equal(t, 0, recorder.save["first"]) + assert.Equal(t, 0, recorder.save["second"]) + assert.Equal(t, 4, recorder.save["third"]+recorder.save["fourth"]) +} + +// TestBalancerOneServerFenced tests that fenced servers are excluded from selection. +func TestBalancerOneServerFenced(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "second") + rw.WriteHeader(http.StatusOK) + }), pointer(1), true) // fenced + + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 3 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Only first server should receive traffic. + assert.Equal(t, 3, recorder.save["first"]) + assert.Equal(t, 0, recorder.save["second"]) +} + +// TestBalancerAllFencedServers tests that all fenced servers result in no available server. +func TestBalancerAllFencedServers(t *testing.T) { + balancer := New(nil, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), pointer(1), true) + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), pointer(1), true) + + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + + assert.Equal(t, http.StatusServiceUnavailable, recorder.Result().StatusCode) +} + +// TestBalancerRegisterStatusUpdaterWithoutHealthCheck tests error when registering updater without health check. +func TestBalancerRegisterStatusUpdaterWithoutHealthCheck(t *testing.T) { + balancer := New(nil, false) + + err := balancer.RegisterStatusUpdater(func(up bool) {}) + assert.Error(t, err) + assert.Contains(t, err.Error(), "healthCheck not enabled") +} + +// TestBalancerSticky tests sticky session support. +func TestBalancerSticky(t *testing.T) { + balancer := New(&dynamic.Sticky{ + Cookie: &dynamic.Cookie{ + Name: "test", + }, + }, false) + + balancer.Add("first", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "first") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("second", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "second") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + // First request should set cookie. + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + firstServer := recorder.Header().Get("server") + assert.NotEmpty(t, firstServer) + + // Extract cookie from first response. + cookies := recorder.Result().Cookies() + assert.NotEmpty(t, cookies) + + // Second request with cookie should hit same server. + req := httptest.NewRequest(http.MethodGet, "/", nil) + for _, cookie := range cookies { + req.AddCookie(cookie) + } + + recorder2 := httptest.NewRecorder() + balancer.ServeHTTP(recorder2, req) + secondServer := recorder2.Header().Get("server") + + assert.Equal(t, firstServer, secondServer) +} + +// TestBalancerStickyFallback tests that sticky sessions fallback to least-time when sticky server is down. +func TestBalancerStickyFallback(t *testing.T) { + balancer := New(&dynamic.Sticky{ + Cookie: &dynamic.Cookie{ + Name: "test", + }, + }, false) + + balancer.Add("server1", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(50 * time.Millisecond) + rw.Header().Set("server", "server1") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("server2", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(50 * time.Millisecond) + rw.Header().Set("server", "server2") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + // Make initial request to establish sticky session with server1. + recorder1 := httptest.NewRecorder() + balancer.ServeHTTP(recorder1, httptest.NewRequest(http.MethodGet, "/", nil)) + firstServer := recorder1.Header().Get("server") + assert.NotEmpty(t, firstServer) + + // Extract cookie from first response. + cookies := recorder1.Result().Cookies() + assert.NotEmpty(t, cookies) + + // Mark the sticky server as DOWN + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "test"), firstServer, false) + + // Request with sticky cookie should fallback to the other server + req2 := httptest.NewRequest(http.MethodGet, "/", nil) + for _, cookie := range cookies { + req2.AddCookie(cookie) + } + recorder2 := httptest.NewRecorder() + balancer.ServeHTTP(recorder2, req2) + fallbackServer := recorder2.Header().Get("server") + assert.NotEqual(t, firstServer, fallbackServer) + assert.NotEmpty(t, fallbackServer) + + // New sticky cookie should be written for the fallback server + newCookies := recorder2.Result().Cookies() + assert.NotEmpty(t, newCookies) + + // Verify sticky session persists with the fallback server + req3 := httptest.NewRequest(http.MethodGet, "/", nil) + for _, cookie := range newCookies { + req3.AddCookie(cookie) + } + recorder3 := httptest.NewRecorder() + balancer.ServeHTTP(recorder3, req3) + assert.Equal(t, fallbackServer, recorder3.Header().Get("server")) + + // Bring original server back UP + balancer.SetStatus(context.WithValue(t.Context(), serviceName, "test"), firstServer, true) + + // Request with fallback server cookie should still stick to fallback server + req4 := httptest.NewRequest(http.MethodGet, "/", nil) + for _, cookie := range newCookies { + req4.AddCookie(cookie) + } + recorder4 := httptest.NewRecorder() + balancer.ServeHTTP(recorder4, req4) + assert.Equal(t, fallbackServer, recorder4.Header().Get("server")) +} + +// TestBalancerStickyFenced tests that sticky sessions persist to fenced servers (graceful shutdown) +// Fencing enables zero-downtime deployments: fenced servers reject NEW connections +// but continue serving EXISTING sticky sessions until they complete. +func TestBalancerStickyFenced(t *testing.T) { + balancer := New(&dynamic.Sticky{ + Cookie: &dynamic.Cookie{ + Name: "test", + }, + }, false) + + balancer.Add("server1", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "server1") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("server2", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + rw.Header().Set("server", "server2") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + // Establish sticky session with any server. + recorder1 := httptest.NewRecorder() + balancer.ServeHTTP(recorder1, httptest.NewRequest(http.MethodGet, "/", nil)) + stickyServer := recorder1.Header().Get("server") + assert.NotEmpty(t, stickyServer) + + cookies := recorder1.Result().Cookies() + assert.NotEmpty(t, cookies) + + // Fence the sticky server (simulate graceful shutdown). + balancer.handlersMu.Lock() + balancer.fenced[stickyServer] = struct{}{} + balancer.handlersMu.Unlock() + + // Existing sticky session should STILL work (graceful draining). + req := httptest.NewRequest(http.MethodGet, "/", nil) + for _, cookie := range cookies { + req.AddCookie(cookie) + } + recorder2 := httptest.NewRecorder() + balancer.ServeHTTP(recorder2, req) + assert.Equal(t, stickyServer, recorder2.Header().Get("server")) + + // But NEW requests should NOT go to the fenced server. + recorder3 := httptest.NewRecorder() + balancer.ServeHTTP(recorder3, httptest.NewRequest(http.MethodGet, "/", nil)) + newServer := recorder3.Header().Get("server") + assert.NotEqual(t, stickyServer, newServer) + assert.NotEmpty(t, newServer) +} + +// TestRingBufferBasic tests basic ring buffer functionality with few samples. +func TestRingBufferBasic(t *testing.T) { + handler := &namedHandler{ + Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), + name: "test", + weight: 1, + } + + // Test cold start - no samples. + avg := handler.getAvgResponseTime() + assert.InDelta(t, 0.0, avg, 0) + + // Add one sample. + handler.updateResponseTime(10 * time.Millisecond) + avg = handler.getAvgResponseTime() + assert.InDelta(t, 10.0, avg, 0) + + // Add more samples. + handler.updateResponseTime(20 * time.Millisecond) + handler.updateResponseTime(30 * time.Millisecond) + avg = handler.getAvgResponseTime() + assert.InDelta(t, 20.0, avg, 0) // (10 + 20 + 30) / 3 = 20 +} + +// TestRingBufferWraparound tests ring buffer behavior when it wraps around +func TestRingBufferWraparound(t *testing.T) { + handler := &namedHandler{ + Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), + name: "test", + weight: 1, + } + + // Fill the buffer with 100 samples of 10ms each. + for range sampleSize { + handler.updateResponseTime(10 * time.Millisecond) + } + avg := handler.getAvgResponseTime() + assert.InDelta(t, 10.0, avg, 0) + + // Add one more sample (should replace oldest). + handler.updateResponseTime(20 * time.Millisecond) + avg = handler.getAvgResponseTime() + // Sum: 99*10 + 1*20 = 1010, avg = 1010/100 = 10.1 + assert.InDelta(t, 10.1, avg, 0) + + // Add 10 more samples of 30ms. + for range 10 { + handler.updateResponseTime(30 * time.Millisecond) + } + avg = handler.getAvgResponseTime() + // Sum: 89*10 + 1*20 + 10*30 = 890 + 20 + 300 = 1210, avg = 1210/100 = 12.1 + assert.InDelta(t, 12.1, avg, 0) +} + +// TestRingBufferLarge tests ring buffer with many samples (> 100). +func TestRingBufferLarge(t *testing.T) { + handler := &namedHandler{ + Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), + name: "test", + weight: 1, + } + + // Add 150 samples. + for i := range 150 { + handler.updateResponseTime(time.Duration(i+1) * time.Millisecond) + } + + // Should only track last 100 samples: 51, 52, ..., 150 + // Sum = (51 + 150) * 100 / 2 = 10050 + // Avg = 10050 / 100 = 100.5 + avg := handler.getAvgResponseTime() + assert.InDelta(t, 100.5, avg, 0) +} + +// TestInflightCounter tests inflight request tracking. +func TestInflightCounter(t *testing.T) { + balancer := New(nil, false) + + var inflightAtRequest atomic.Int64 + + balancer.Add("test", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + inflightAtRequest.Store(balancer.handlers[0].inflightCount.Load()) + rw.Header().Set("server", "test") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + // Check that inflight count is 0 initially. + balancer.handlersMu.RLock() + handler := balancer.handlers[0] + balancer.handlersMu.RUnlock() + assert.Equal(t, int64(0), handler.inflightCount.Load()) + + // Make a request. + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + + // During request, inflight should have been 1. + assert.Equal(t, int64(1), inflightAtRequest.Load()) + + // After request completes, inflight should be back to 0. + assert.Equal(t, int64(0), handler.inflightCount.Load()) +} + +// TestConcurrentResponseTimeUpdates tests thread safety of response time updates. +func TestConcurrentResponseTimeUpdates(t *testing.T) { + handler := &namedHandler{ + Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), + name: "test", + weight: 1, + } + + // Concurrently update response times. + var wg sync.WaitGroup + numGoroutines := 10 + updatesPerGoroutine := 20 + + for i := range numGoroutines { + wg.Add(1) + go func(id int) { + defer wg.Done() + for range updatesPerGoroutine { + handler.updateResponseTime(time.Duration(id+1) * time.Millisecond) + } + }(i) + } + + wg.Wait() + + // Should have exactly 100 samples (buffer size). + assert.Equal(t, sampleSize, handler.sampleCount) +} + +// TestConcurrentInflightTracking tests thread safety of inflight counter. +func TestConcurrentInflightTracking(t *testing.T) { + handler := &namedHandler{ + Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(10 * time.Millisecond) + rw.WriteHeader(http.StatusOK) + }), + name: "test", + weight: 1, + } + + var maxInflight atomic.Int64 + + var wg sync.WaitGroup + numRequests := 50 + + for range numRequests { + wg.Add(1) + go func() { + defer wg.Done() + handler.inflightCount.Add(1) + defer handler.inflightCount.Add(-1) + + // Track maximum inflight count. + current := handler.inflightCount.Load() + for { + maxLoad := maxInflight.Load() + if current <= maxLoad || maxInflight.CompareAndSwap(maxLoad, current) { + break + } + } + + time.Sleep(1 * time.Millisecond) + }() + } + + wg.Wait() + + // All requests completed, inflight should be 0. + assert.Equal(t, int64(0), handler.inflightCount.Load()) + // Max inflight should be > 1 (concurrent requests). + assert.Greater(t, maxInflight.Load(), int64(1)) +} + +// TestConcurrentRequestsRespectInflight tests that the load balancer dynamically +// adapts to inflight request counts during concurrent request processing. +func TestConcurrentRequestsRespectInflight(t *testing.T) { + balancer := New(nil, false) + + // Use a channel to control when handlers start sleeping. + // This ensures we can fill one server with inflight requests before routing new ones. + blockChan := make(chan struct{}) + + // Add two servers with equal response times and weights. + balancer.Add("server1", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + <-blockChan // Wait for signal to proceed. + time.Sleep(10 * time.Millisecond) + rw.Header().Set("server", "server1") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("server2", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + <-blockChan // Wait for signal to proceed. + time.Sleep(10 * time.Millisecond) + rw.Header().Set("server", "server2") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + // Pre-warm both servers to establish equal average response times. + for i := range sampleSize { + balancer.handlers[0].responseTimes[i] = 10.0 + } + balancer.handlers[0].responseTimeSum = 10.0 * sampleSize + balancer.handlers[0].sampleCount = sampleSize + + for i := range sampleSize { + balancer.handlers[1].responseTimes[i] = 10.0 + } + balancer.handlers[1].responseTimeSum = 10.0 * sampleSize + balancer.handlers[1].sampleCount = sampleSize + + // Phase 1: Launch concurrent requests to server1 that will block. + var wg sync.WaitGroup + inflightRequests := 5 + + for range inflightRequests { + wg.Add(1) + go func() { + defer wg.Done() + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + }() + } + + // Wait for goroutines to start and increment inflight counters. + // They will block on the channel, keeping inflight count high. + time.Sleep(50 * time.Millisecond) + + // Verify inflight counts before making new requests. + server1Inflight := balancer.handlers[0].inflightCount.Load() + server2Inflight := balancer.handlers[1].inflightCount.Load() + assert.Equal(t, int64(5), server1Inflight+server2Inflight) + + // Phase 2: Make new requests while the initial requests are blocked. + // These should see the high inflight counts and route to the less-loaded server. + var saveMu sync.Mutex + save := map[string]int{} + newRequests := 50 + + // Launch new requests in background so they don't block. + var newWg sync.WaitGroup + for range newRequests { + newWg.Add(1) + go func() { + defer newWg.Done() + rec := httptest.NewRecorder() + balancer.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil)) + server := rec.Header().Get("server") + if server != "" { + saveMu.Lock() + save[server]++ + saveMu.Unlock() + } + }() + } + + // Wait for new requests to start and see the inflight counts. + time.Sleep(50 * time.Millisecond) + + close(blockChan) + + wg.Wait() + newWg.Wait() + + saveMu.Lock() + total := save["server1"] + save["server2"] + server1Count := save["server1"] + server2Count := save["server2"] + saveMu.Unlock() + + assert.Equal(t, newRequests, total) + + // With inflight tracking, load should naturally balance toward equal distribution. + // We allow variance due to concurrent execution and race windows in server selection. + assert.InDelta(t, 25.0, float64(server1Count), 5.0) // 20-30 requests + assert.InDelta(t, 25.0, float64(server2Count), 5.0) // 20-30 requests +} + +// TestTTFBMeasurement tests TTFB measurement accuracy. +func TestTTFBMeasurement(t *testing.T) { + balancer := New(nil, false) + + // Add server with known delay. + delay := 50 * time.Millisecond + balancer.Add("slow", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(delay) + rw.Header().Set("server", "slow") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + // Make multiple requests to build average. + for range 5 { + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Check that average response time is approximately the delay. + avg := balancer.handlers[0].getAvgResponseTime() + + // Allow 5ms tolerance for Go timing jitter and test environment variations. + assert.InDelta(t, float64(delay.Milliseconds()), avg, 5.0) +} + +// TestZeroSamplesReturnsZero tests that getAvgResponseTime returns 0 when no samples. +func TestZeroSamplesReturnsZero(t *testing.T) { + handler := &namedHandler{ + Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), + name: "test", + weight: 1, + } + + avg := handler.getAvgResponseTime() + assert.InDelta(t, 0.0, avg, 0) +} + +// TestScoreCalculationWithWeights tests that weights are properly considered in score calculation. +func TestScoreCalculationWithWeights(t *testing.T) { + balancer := New(nil, false) + + // Add two servers with same response time but different weights. + // Server with higher weight should be preferred. + balancer.Add("weighted", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(50 * time.Millisecond) + rw.Header().Set("server", "weighted") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(3), false) // Weight 3 + + balancer.Add("normal", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(50 * time.Millisecond) + rw.Header().Set("server", "normal") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) // Weight 1 + + // Make requests to build up response time averages. + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 2 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Score for weighted: (50 × (1 + 0)) / 3 = 16.67 + // Score for normal: (50 × (1 + 0)) / 1 = 50 + // After warmup, weighted server has 3x better score (16.67 vs 50) and should receive nearly all requests. + recorder = &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 10 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + assert.Equal(t, 10, recorder.save["weighted"]) + assert.Zero(t, recorder.save["normal"]) +} + +// TestScoreCalculationWithInflight tests that inflight requests are considered in score calculation. +func TestScoreCalculationWithInflight(t *testing.T) { + balancer := New(nil, false) + + // We'll manually control the inflight counters to test the score calculation. + // Add two servers with same response time. + balancer.Add("server1", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(10 * time.Millisecond) + rw.Header().Set("server", "server1") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + balancer.Add("server2", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(10 * time.Millisecond) + rw.Header().Set("server", "server2") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + // Build up response time averages for both servers. + for range 2 { + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Now manually set server1 to have high inflight count. + balancer.handlers[0].inflightCount.Store(5) + + // Make requests - they should prefer server2 because: + // Score for server1: (10 × (1 + 5)) / 1 = 60 + // Score for server2: (10 × (1 + 0)) / 1 = 10 + recorder := &responseRecorder{save: map[string]int{}} + for range 5 { + // Manually increment to simulate the ServeHTTP behavior. + server, _ := balancer.nextServer() + server.inflightCount.Add(1) + + if server.name == "server1" { + recorder.save["server1"]++ + } else { + recorder.save["server2"]++ + } + } + + // Server2 should get all requests + assert.Equal(t, 5, recorder.save["server2"]) + assert.Zero(t, recorder.save["server1"]) +} + +// TestScoreCalculationColdStart tests that new servers (0ms avg) get fair selection +func TestScoreCalculationColdStart(t *testing.T) { + balancer := New(nil, false) + + // Add a warm server with established response time + balancer.Add("warm", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(50 * time.Millisecond) + rw.Header().Set("server", "warm") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + // Warm up the first server + for range 10 { + recorder := httptest.NewRecorder() + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Now add a cold server (new, no response time data) + balancer.Add("cold", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(10 * time.Millisecond) // Actually faster + rw.Header().Set("server", "cold") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + // Cold server should get selected because: + // Score for warm: (50 × (1 + 0)) / 1 = 50 + // Score for cold: (0 × (1 + 0)) / 1 = 0 + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 20 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // Cold server should get all or most requests initially due to 0ms average + assert.Greater(t, recorder.save["cold"], 10) + + // After cold server builds up its average, it should continue to get more traffic + // because it's actually faster (10ms vs 50ms) + recorder = &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 20 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + assert.Greater(t, recorder.save["cold"], recorder.save["warm"]) +} + +// TestFastServerGetsMoreTraffic verifies that servers with lower response times +// receive proportionally more traffic in steady state (after cold start). +// This tests the core selection bias of the least-time algorithm. +func TestFastServerGetsMoreTraffic(t *testing.T) { + balancer := New(nil, false) + + // Add two servers with different static response times. + balancer.Add("fast", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(20 * time.Millisecond) + rw.Header().Set("server", "fast") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + balancer.Add("slow", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(100 * time.Millisecond) + rw.Header().Set("server", "slow") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + // After just 1 request to each server, the algorithm identifies the fastest server + // and routes nearly all subsequent traffic there (converges in ~2 requests). + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 50 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + assert.Greater(t, recorder.save["fast"], recorder.save["slow"]) + assert.Greater(t, recorder.save["fast"], 48) // Expect ~96-98% to fast server (48-49/50). +} + +// TestTrafficShiftsWhenPerformanceDegrades verifies that the load balancer +// adapts to changing server performance by shifting traffic away from degraded servers. +// This tests the adaptive behavior - the core value proposition of least-time load balancing. +func TestTrafficShiftsWhenPerformanceDegrades(t *testing.T) { + balancer := New(nil, false) + + // Use atomic to dynamically control server1's response time. + server1Delay := atomic.Int64{} + server1Delay.Store(5) // Start with 5ms + + balancer.Add("server1", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(time.Duration(server1Delay.Load()) * time.Millisecond) + rw.Header().Set("server", "server1") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + balancer.Add("server2", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) // Static 5ms + rw.Header().Set("server", "server2") + rw.WriteHeader(http.StatusOK) + httptrace.ContextClientTrace(req.Context()).GotFirstResponseByte() + }), pointer(1), false) + + // Pre-fill ring buffers to eliminate cold start effects and ensure deterministic equal performance state. + for _, h := range balancer.handlers { + for i := range sampleSize { + h.responseTimes[i] = 5.0 + } + h.responseTimeSum = 5.0 * sampleSize + h.sampleCount = sampleSize + } + + // Phase 1: Both servers perform equally (5ms each). + recorder := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 50 { + balancer.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // With equal performance and pre-filled buffers, distribution should be balanced via WRR tie-breaking. + total := recorder.save["server1"] + recorder.save["server2"] + assert.Equal(t, 50, total) + assert.InDelta(t, 25, recorder.save["server1"], 10) // 25 ± 10 requests + assert.InDelta(t, 25, recorder.save["server2"], 10) // 25 ± 10 requests + + // Phase 2: server1 degrades (simulating GC pause, CPU spike, or network latency). + server1Delay.Store(15) // Now 15ms (3x slower) + + // Make more requests to shift the moving average. + // Ring buffer has 100 samples, need significant new samples to shift average. + // server1's average will climb from ~5ms toward 15ms. + recorder2 := &responseRecorder{ResponseRecorder: httptest.NewRecorder(), save: map[string]int{}} + for range 60 { + balancer.ServeHTTP(recorder2, httptest.NewRequest(http.MethodGet, "/", nil)) + } + + // server2 should get significantly more traffic (>75%) + // Score for server1: (~10-15ms × 1) / 1 = 10-15 (as average climbs) + // Score for server2: (5ms × 1) / 1 = 5 + total2 := recorder2.save["server1"] + recorder2.save["server2"] + assert.Equal(t, 60, total2) + assert.Greater(t, recorder2.save["server2"], 45) // At least 75% (45/60) + assert.Less(t, recorder2.save["server1"], 15) // At most 25% (15/60) +} + +// TestMultipleServersWithSameScore tests WRR tie-breaking when multiple servers have identical scores. +// Uses nextServer() directly to avoid timing variations in the test. +func TestMultipleServersWithSameScore(t *testing.T) { + balancer := New(nil, false) + + // Add three servers with identical response times and weights. + balancer.Add("server1", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "server1") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("server2", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "server2") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + balancer.Add("server3", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "server3") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) + + // Set all servers to identical response times to trigger tie-breaking. + for _, h := range balancer.handlers { + for i := range sampleSize { + h.responseTimes[i] = 5.0 + } + h.responseTimeSum = 5.0 * sampleSize + h.sampleCount = sampleSize + } + + // With all servers having identical scores, WRR tie-breaking should distribute fairly. + // Test the selection logic directly without actual HTTP requests to avoid timing variations. + counts := map[string]int{"server1": 0, "server2": 0, "server3": 0} + for range 90 { + server, err := balancer.nextServer() + assert.NoError(t, err) + counts[server.name]++ + } + + total := counts["server1"] + counts["server2"] + counts["server3"] + assert.Equal(t, 90, total) + + // With WRR and 90 requests, each server should get ~30 requests (±1 due to initialization). + assert.InDelta(t, 30, counts["server1"], 1) + assert.InDelta(t, 30, counts["server2"], 1) + assert.InDelta(t, 30, counts["server3"], 1) +} + +// TestWRRTieBreakingWeightedDistribution tests weighted distribution among tied servers. +// Uses nextServer() directly to avoid timing variations in the test. +func TestWRRTieBreakingWeightedDistribution(t *testing.T) { + balancer := New(nil, false) + + // Add two servers with different weights. + // To create equal scores, response times must be proportional to weights. + balancer.Add("weighted", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(15 * time.Millisecond) // 3x longer due to 3x weight + rw.Header().Set("server", "weighted") + rw.WriteHeader(http.StatusOK) + }), pointer(3), false) // Weight 3 + + balancer.Add("normal", http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + time.Sleep(5 * time.Millisecond) + rw.Header().Set("server", "normal") + rw.WriteHeader(http.StatusOK) + }), pointer(1), false) // Weight 1 + + // Since response times is proportional to weights, both scores are equal, so WRR tie-breaking will apply. + // weighted: score = (15 * 1) / 3 = 5 + // normal: score = (5 * 1) / 1 = 5 + for i := range sampleSize { + balancer.handlers[0].responseTimes[i] = 15.0 + } + balancer.handlers[0].responseTimeSum = 15.0 * sampleSize + balancer.handlers[0].sampleCount = sampleSize + + for i := range sampleSize { + balancer.handlers[1].responseTimes[i] = 5.0 + } + balancer.handlers[1].responseTimeSum = 5.0 * sampleSize + balancer.handlers[1].sampleCount = sampleSize + + // Test the selection logic directly without actual HTTP requests to avoid timing variations. + counts := map[string]int{"weighted": 0, "normal": 0} + for range 80 { + server, err := balancer.nextServer() + assert.NoError(t, err) + counts[server.name]++ + } + + total := counts["weighted"] + counts["normal"] + assert.Equal(t, 80, total) + + // With 3:1 weight ratio, distribution should be ~75%/25% (60/80 and 20/80), ±1 due to initialization. + assert.InDelta(t, 60, counts["weighted"], 1) + assert.InDelta(t, 20, counts["normal"], 1) +} diff --git a/pkg/server/service/loadbalancer/p2c/p2c.go b/pkg/server/service/loadbalancer/p2c/p2c.go index d41de0a2e..948a7d5a1 100644 --- a/pkg/server/service/loadbalancer/p2c/p2c.go +++ b/pkg/server/service/loadbalancer/p2c/p2c.go @@ -14,6 +14,8 @@ import ( "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer" ) +var errNoAvailableServer = errors.New("no available server") + type namedHandler struct { http.Handler @@ -81,7 +83,7 @@ func New(stickyConfig *dynamic.Sticky, wantsHealthCheck bool) *Balancer { } // SetStatus sets on the balancer that its given child is now of the given -// status. balancerName is only needed for logging purposes. +// status. childName is only needed for logging purposes. func (b *Balancer) SetStatus(ctx context.Context, childName string, up bool) { b.handlersMu.Lock() defer b.handlersMu.Unlock() @@ -126,14 +128,12 @@ func (b *Balancer) SetStatus(ctx context.Context, childName string, up bool) { // Not thread safe. func (b *Balancer) RegisterStatusUpdater(fn func(up bool)) error { if !b.wantsHealthCheck { - return errors.New("healthCheck not enabled in config for this weighted service") + return errors.New("healthCheck not enabled in config for this P2C service") } b.updaters = append(b.updaters, fn) return nil } -var errNoAvailableServer = errors.New("no available server") - func (b *Balancer) nextServer() (*namedHandler, error) { // We kept the same representation (map) as in the WRR strategy to improve maintainability. // However, with the P2C strategy, we only need a slice of healthy servers. diff --git a/pkg/server/service/loadbalancer/wrr/wrr.go b/pkg/server/service/loadbalancer/wrr/wrr.go index b2665b56d..69bc2c498 100644 --- a/pkg/server/service/loadbalancer/wrr/wrr.go +++ b/pkg/server/service/loadbalancer/wrr/wrr.go @@ -12,6 +12,8 @@ import ( "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer" ) +var errNoAvailableServer = errors.New("no available server") + type namedHandler struct { http.Handler name string @@ -94,7 +96,7 @@ func (b *Balancer) Pop() interface{} { } // SetStatus sets on the balancer that its given child is now of the given -// status. balancerName is only needed for logging purposes. +// status. childName is only needed for logging purposes. func (b *Balancer) SetStatus(ctx context.Context, childName string, up bool) { b.handlersMu.Lock() defer b.handlersMu.Unlock() @@ -139,14 +141,12 @@ func (b *Balancer) SetStatus(ctx context.Context, childName string, up bool) { // Not thread safe. func (b *Balancer) RegisterStatusUpdater(fn func(up bool)) error { if !b.wantsHealthCheck { - return errors.New("healthCheck not enabled in config for this weighted service") + return errors.New("healthCheck not enabled in config for this WRR service") } b.updaters = append(b.updaters, fn) return nil } -var errNoAvailableServer = errors.New("no available server") - func (b *Balancer) nextServer() (*namedHandler, error) { b.handlersMu.Lock() defer b.handlersMu.Unlock() diff --git a/pkg/server/service/service.go b/pkg/server/service/service.go index c49bd8809..e61820f3d 100644 --- a/pkg/server/service/service.go +++ b/pkg/server/service/service.go @@ -29,6 +29,7 @@ import ( "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/hrw" + "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/leasttime" "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/mirror" "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/p2c" "github.com/traefik/traefik/v3/pkg/server/service/loadbalancer/wrr" @@ -402,6 +403,8 @@ func (m *Manager) getLoadBalancerServiceHandler(ctx context.Context, serviceName lb = p2c.New(service.Sticky, service.HealthCheck != nil) case dynamic.BalancerStrategyHRW: lb = hrw.New(service.HealthCheck != nil) + case dynamic.BalancerStrategyLeastTime: + lb = leasttime.New(service.Sticky, service.HealthCheck != nil) default: return nil, fmt.Errorf("unsupported load-balancer strategy %q", service.Strategy) } diff --git a/pkg/server/service/service_test.go b/pkg/server/service/service_test.go index 05d674e16..574dd1a5b 100644 --- a/pkg/server/service/service_test.go +++ b/pkg/server/service/service_test.go @@ -82,6 +82,20 @@ func TestGetLoadBalancer(t *testing.T) { fwd: &forwarderMock{}, expectError: false, }, + { + desc: "Fails when unsupported strategy is set", + serviceName: "test", + service: &dynamic.ServersLoadBalancer{ + Strategy: "invalid", + Servers: []dynamic.Server{ + { + URL: "http://localhost:8080", + }, + }, + }, + fwd: &forwarderMock{}, + expectError: true, + }, } for _, test := range testCases {