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

Add mirrorBody option to HTTP mirroring

Browse source
This commit is contained in:
Matteo Paier 2024-09-02 16:36:06 +02:00 committed by GitHub
parent 51f7f610c9
commit eb99c8c785
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 165 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
docs/content/reference/dynamic-configuration/file.toml
View file

@ -82,6 +82,7 @@
[http.services.Service03]
[http.services.Service03.mirroring]
service = "foobar"
mirrorBody = true
maxBodySize = 42
[[http.services.Service03.mirroring.mirrors]]

1
docs/content/reference/dynamic-configuration/file.yaml
View file

@ -89,6 +89,7 @@ http:
Service03:
mirroring:
service: foobar
mirrorBody: true
maxBodySize: 42
mirrors:
- name: foobar

5
docs/content/reference/dynamic-configuration/kubernetes-crd-definition-v1.yml
View file

@ -2506,6 +2506,11 @@ spec:
Default value is -1, which means unlimited size.
format: int64
type: integer
mirrorBody:
description: |-
MirrorBody defines whether the body of the request should be mirrored.
Default value is true.
type: boolean
mirrors:
description: Mirrors defines the list of mirrors where Traefik
will duplicate the traffic.

1
docs/content/reference/dynamic-configuration/kubernetes-crd-resource.yml
View file

@ -63,6 +63,7 @@ spec:
mirroring:
name: wrr2
kind: TraefikService
mirrorBody: true
# Optional
maxBodySize: 2000000000
mirrors:

1
docs/content/reference/dynamic-configuration/kv-ref.md
View file

@ -264,6 +264,7 @@ THIS FILE MUST NOT BE EDITED BY HAND
| `traefik/http/services/Service02/loadBalancer/sticky/cookie/secure` | `true` |
| `traefik/http/services/Service03/mirroring/healthCheck` | `` |
| `traefik/http/services/Service03/mirroring/maxBodySize` | `42` |
| `traefik/http/services/Service03/mirroring/mirrorBody` | `true` |
| `traefik/http/services/Service03/mirroring/mirrors/0/name` | `foobar` |
| `traefik/http/services/Service03/mirroring/mirrors/0/percent` | `42` |
| `traefik/http/services/Service03/mirroring/mirrors/1/name` | `foobar` |

5
docs/content/reference/dynamic-configuration/traefik.io_traefikservices.yaml
View file

@ -121,6 +121,11 @@ spec:
Default value is -1, which means unlimited size.
format: int64
type: integer
mirrorBody:
description: |-
MirrorBody defines whether the body of the request should be mirrored.
Default value is true.
type: boolean
mirrors:
description: Mirrors defines the list of mirrors where Traefik
will duplicate the traffic.

7
docs/content/routing/services/index.md
View file

@ -1207,6 +1207,7 @@ http:
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`.
!!! info "Supported Providers"
@ -1219,6 +1220,9 @@ http:
mirrored-api:
mirroring:
service: appv1
# mirrorBody defines whether the request body should be mirrored.
# Default value is true.
mirrorBody: false
# maxBodySize is the maximum size allowed for the body of the request.
# If the body is larger, the request is not mirrored.
# Default value is -1, which means unlimited size.
@ -1248,6 +1252,9 @@ http:
# If the body is larger, the request is not mirrored.
# Default value is -1, which means unlimited size.
maxBodySize = 1024
# mirrorBody defines whether the request body should be mirrored.
# Default value is true.
mirrorBody = false
[[http.services.mirrored-api.mirroring.mirrors]]
name = "appv2"
percent = 10

5
integration/fixtures/k8s/01-traefik-crd.yml
View file

@ -2506,6 +2506,11 @@ spec:
Default value is -1, which means unlimited size.
format: int64
type: integer
mirrorBody:
description: |-
MirrorBody defines whether the body of the request should be mirrored.
Default value is true.
type: boolean
mirrors:
description: Mirrors defines the list of mirrors where Traefik
will duplicate the traffic.

14
integration/fixtures/mirror.toml
View file

@ -28,6 +28,10 @@
service = "mirrorWithMaxBody"
rule = "Path(`/whoamiWithMaxBody`)"
[http.routers.router3]
service = "mirrorWithoutBody"
rule = "Path(`/whoamiWithoutBody`)"
[http.services]
[http.services.mirror.mirroring]
@ -49,6 +53,16 @@
name = "mirror2"
percent = 50
[http.services.mirrorWithoutBody.mirroring]
service = "service1"
mirrorBody = false
[[http.services.mirrorWithoutBody.mirroring.mirrors]]
name = "mirror1"
percent = 10
[[http.services.mirrorWithoutBody.mirroring.mirrors]]
name = "mirror2"
percent = 50
[http.services.service1.loadBalancer]
[[http.services.service1.loadBalancer.servers]]

40
integration/simple_test.go
View file

@ -1004,8 +1004,13 @@ func (s *SimpleSuite) TestMirrorWithBody() {
_, err = rand.Read(body5)
require.NoError(s.T(), err)
verifyBody := func(req *http.Request) {
// forceOkResponse is used to avoid errors when Content-Length is set but no body is received
verifyBody := func(req *http.Request, canBodyBeEmpty bool) (forceOkResponse bool) {
b, _ := io.ReadAll(req.Body)
if canBodyBeEmpty && req.Header.Get("NoBody") == "true" {
require.Empty(s.T(), b)
return true
}
switch req.Header.Get("Size") {
case "20":
require.Equal(s.T(), body20, b)
@ -1014,20 +1019,25 @@ func (s *SimpleSuite) TestMirrorWithBody() {
default:
s.T().Fatal("Size header not present")
}
return false
}
main := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
verifyBody(req)
verifyBody(req, false)
atomic.AddInt32(&count, 1)
}))
mirror1 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
verifyBody(req)
if verifyBody(req, true) {
rw.WriteHeader(http.StatusOK)
}
atomic.AddInt32(&countMirror1, 1)
}))
mirror2 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
verifyBody(req)
if verifyBody(req, true) {
rw.WriteHeader(http.StatusOK)
}
atomic.AddInt32(&countMirror2, 1)
}))
@ -1104,6 +1114,28 @@ func (s *SimpleSuite) TestMirrorWithBody() {
assert.Equal(s.T(), int32(10), countTotal)
assert.Equal(s.T(), int32(0), val1)
assert.Equal(s.T(), int32(0), val2)
atomic.StoreInt32(&count, 0)
atomic.StoreInt32(&countMirror1, 0)
atomic.StoreInt32(&countMirror2, 0)
req, err = http.NewRequest(http.MethodGet, "http://127.0.0.1:8000/whoamiWithoutBody", bytes.NewBuffer(body20))
require.NoError(s.T(), err)
req.Header.Set("Size", "20")
req.Header.Set("NoBody", "true")
for range 10 {
response, err := http.DefaultClient.Do(req)
require.NoError(s.T(), err)
assert.Equal(s.T(), http.StatusOK, response.StatusCode)
}
countTotal = atomic.LoadInt32(&count)
val1 = atomic.LoadInt32(&countMirror1)
val2 = atomic.LoadInt32(&countMirror2)
assert.Equal(s.T(), int32(10), countTotal)
assert.Equal(s.T(), int32(1), val1)
assert.Equal(s.T(), int32(5), val2)
}
func (s *SimpleSuite) TestMirrorCanceled() {

3
pkg/config/dynamic/http_config.go
View file

@ -81,6 +81,7 @@ type RouterTLSConfig struct {
// Mirroring holds the Mirroring configuration.
type Mirroring struct {
Service string `json:"service,omitempty" toml:"service,omitempty" yaml:"service,omitempty" export:"true"`
MirrorBody *bool `json:"mirrorBody,omitempty" toml:"mirrorBody,omitempty" yaml:"mirrorBody,omitempty" export:"true"`
MaxBodySize *int64 `json:"maxBodySize,omitempty" toml:"maxBodySize,omitempty" yaml:"maxBodySize,omitempty" export:"true"`
Mirrors []MirrorService `json:"mirrors,omitempty" toml:"mirrors,omitempty" yaml:"mirrors,omitempty" export:"true"`
HealthCheck *HealthCheck `json:"healthCheck,omitempty" toml:"healthCheck,omitempty" yaml:"healthCheck,omitempty" label:"allowEmpty" file:"allowEmpty" kv:"allowEmpty" export:"true"`
@ -88,6 +89,8 @@ type Mirroring struct {
// SetDefaults Default values for a WRRService.
func (m *Mirroring) SetDefaults() {
defaultMirrorBody := true
m.MirrorBody = &defaultMirrorBody
var defaultMaxBodySize int64 = -1
m.MaxBodySize = &defaultMaxBodySize
}

5
pkg/config/dynamic/zz_generated.deepcopy.go
View file

@ -967,6 +967,11 @@ func (in *MirrorService) DeepCopy() *MirrorService {
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Mirroring) DeepCopyInto(out *Mirroring) {
*out = *in
if in.MirrorBody != nil {
in, out := &in.MirrorBody, &out.MirrorBody
*out = new(bool)
**out = **in
}
if in.MaxBodySize != nil {
in, out := &in.MaxBodySize, &out.MaxBodySize
*out = new(int64)

1
pkg/provider/kubernetes/crd/kubernetes_http.go
View file

@ -290,6 +290,7 @@ func (c configBuilder) buildMirroring(ctx context.Context, tService *traefikv1al
Mirroring: &dynamic.Mirroring{
Service: fullNameMain,
Mirrors: mirrorServices,
MirrorBody: tService.Spec.Mirroring.MirrorBody,
MaxBodySize: tService.Spec.Mirroring.MaxBodySize,
},
}

3
pkg/provider/kubernetes/crd/traefikio/v1alpha1/service.go
View file

@ -53,6 +53,9 @@ type TraefikServiceSpec struct {
type Mirroring struct {
LoadBalancerSpec `json:",inline"`
// MirrorBody defines whether the body of the request should be mirrored.
// Default value is true.
MirrorBody *bool `json:"mirrorBody,omitempty"`
// MaxBodySize defines the maximum size allowed for the body of the request.
// If the body is larger, the request is not mirrored.
// Default value is -1, which means unlimited size.

5
pkg/provider/kubernetes/crd/traefikio/v1alpha1/zz_generated.deepcopy.go
View file

@ -972,6 +972,11 @@ func (in *MirrorService) DeepCopy() *MirrorService {
func (in *Mirroring) DeepCopyInto(out *Mirroring) {
*out = *in
in.LoadBalancerSpec.DeepCopyInto(&out.LoadBalancerSpec)
if in.MirrorBody != nil {
in, out := &in.MirrorBody, &out.MirrorBody
*out = new(bool)
**out = **in
}
if in.MaxBodySize != nil {
in, out := &in.MaxBodySize, &out.MaxBodySize
*out = new(int64)

2
pkg/provider/kv/kv_test.go
View file

@ -61,6 +61,7 @@ func Test_buildConfiguration(t *testing.T) {
"traefik/http/services/Service01/loadBalancer/servers/0/url": "foobar",
"traefik/http/services/Service01/loadBalancer/servers/1/url": "foobar",
"traefik/http/services/Service02/mirroring/service": "foobar",
"traefik/http/services/Service02/mirroring/mirrorBody": "true",
"traefik/http/services/Service02/mirroring/maxBodySize": "42",
"traefik/http/services/Service02/mirroring/mirrors/0/name": "foobar",
"traefik/http/services/Service02/mirroring/mirrors/0/percent": "42",
@ -676,6 +677,7 @@ func Test_buildConfiguration(t *testing.T) {
"Service02": {
Mirroring: &dynamic.Mirroring{
Service: "foobar",
MirrorBody: func(v bool) *bool { return &v }(true),
MaxBodySize: func(v int64) *int64 { return &v }(42),
Mirrors: []dynamic.MirrorService{
{

10
pkg/server/service/loadbalancer/mirror/mirror.go
View file

@ -25,6 +25,7 @@ type Mirroring struct {
rw http.ResponseWriter
routinePool *safe.Pool
mirrorBody bool
maxBodySize int64
wantsHealthCheck bool
@ -33,11 +34,12 @@ type Mirroring struct {
}
// New returns a new instance of *Mirroring.
func New(handler http.Handler, pool *safe.Pool, maxBodySize int64, hc *dynamic.HealthCheck) *Mirroring {
func New(handler http.Handler, pool *safe.Pool, mirrorBody bool, maxBodySize int64, hc *dynamic.HealthCheck) *Mirroring {
return &Mirroring{
routinePool: pool,
handler: handler,
rw: blackHoleResponseWriter{},
mirrorBody: mirrorBody,
maxBodySize: maxBodySize,
wantsHealthCheck: hc != nil,
}
@ -83,7 +85,7 @@ func (m *Mirroring) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
}
logger := log.Ctx(req.Context())
rr, bytesRead, err := newReusableRequest(req, m.maxBodySize)
rr, bytesRead, err := newReusableRequest(req, m.mirrorBody, m.maxBodySize)
if err != nil && !errors.Is(err, errBodyTooLarge) {
http.Error(rw, fmt.Sprintf("%s: creating reusable request: %v",
http.StatusText(http.StatusInternalServerError), err), http.StatusInternalServerError)
@ -200,11 +202,11 @@ var errBodyTooLarge = errors.New("request body too large")
// if the returned error is errBodyTooLarge, newReusableRequest also returns the
// bytes that were already consumed from the request's body.
func newReusableRequest(req *http.Request, maxBodySize int64) (*reusableRequest, []byte, error) {
func newReusableRequest(req *http.Request, mirrorBody bool, maxBodySize int64) (*reusableRequest, []byte, error) {
if req == nil {
return nil, nil, errors.New("nil input request")
}
if req.Body == nil || req.ContentLength == 0 {
if req.Body == nil || req.ContentLength == 0 || !mirrorBody {
return &reusableRequest{req: req}, nil, nil
}

66
pkg/server/service/loadbalancer/mirror/mirror_test.go
View file

@ -21,7 +21,7 @@ func TestMirroringOn100(t *testing.T) {
rw.WriteHeader(http.StatusOK)
})
pool := safe.NewPool(context.Background())
mirror := New(handler, pool, defaultMaxBodySize, nil)
mirror := New(handler, pool, true, defaultMaxBodySize, nil)
err := mirror.AddMirror(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
atomic.AddInt32(&countMirror1, 1)
}), 10)
@ -50,7 +50,7 @@ func TestMirroringOn10(t *testing.T) {
rw.WriteHeader(http.StatusOK)
})
pool := safe.NewPool(context.Background())
mirror := New(handler, pool, defaultMaxBodySize, nil)
mirror := New(handler, pool, true, defaultMaxBodySize, nil)
err := mirror.AddMirror(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
atomic.AddInt32(&countMirror1, 1)
}), 10)
@ -74,7 +74,7 @@ func TestMirroringOn10(t *testing.T) {
}
func TestInvalidPercent(t *testing.T) {
mirror := New(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), safe.NewPool(context.Background()), defaultMaxBodySize, nil)
mirror := New(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}), safe.NewPool(context.Background()), true, defaultMaxBodySize, nil)
err := mirror.AddMirror(nil, -1)
assert.Error(t, err)
@ -93,7 +93,7 @@ func TestHijack(t *testing.T) {
rw.WriteHeader(http.StatusOK)
})
pool := safe.NewPool(context.Background())
mirror := New(handler, pool, defaultMaxBodySize, nil)
mirror := New(handler, pool, true, defaultMaxBodySize, nil)
var mirrorRequest bool
err := mirror.AddMirror(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
@ -117,7 +117,7 @@ func TestFlush(t *testing.T) {
rw.WriteHeader(http.StatusOK)
})
pool := safe.NewPool(context.Background())
mirror := New(handler, pool, defaultMaxBodySize, nil)
mirror := New(handler, pool, true, defaultMaxBodySize, nil)
var mirrorRequest bool
err := mirror.AddMirror(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
@ -154,7 +154,7 @@ func TestMirroringWithBody(t *testing.T) {
rw.WriteHeader(http.StatusOK)
})
mirror := New(handler, pool, defaultMaxBodySize, nil)
mirror := New(handler, pool, true, defaultMaxBodySize, nil)
for range numMirrors {
err := mirror.AddMirror(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
@ -177,13 +177,55 @@ func TestMirroringWithBody(t *testing.T) {
assert.Equal(t, numMirrors, int(val))
}
func TestMirroringWithIgnoredBody(t *testing.T) {
const numMirrors = 10
var (
countMirror int32
body = []byte(`body`)
emptyBody = []byte(``)
)
pool := safe.NewPool(context.Background())
handler := http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
assert.NotNil(t, r.Body)
bb, err := io.ReadAll(r.Body)
assert.NoError(t, err)
assert.Equal(t, body, bb)
rw.WriteHeader(http.StatusOK)
})
mirror := New(handler, pool, false, defaultMaxBodySize, nil)
for range numMirrors {
err := mirror.AddMirror(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
assert.NotNil(t, r.Body)
bb, err := io.ReadAll(r.Body)
assert.NoError(t, err)
assert.Equal(t, emptyBody, bb)
atomic.AddInt32(&countMirror, 1)
}), 100)
assert.NoError(t, err)
}
req := httptest.NewRequest(http.MethodPost, "/", bytes.NewBuffer(body))
mirror.ServeHTTP(httptest.NewRecorder(), req)
pool.Stop()
val := atomic.LoadInt32(&countMirror)
assert.Equal(t, numMirrors, int(val))
}
func TestCloneRequest(t *testing.T) {
t.Run("http request body is nil", func(t *testing.T) {
req, err := http.NewRequest(http.MethodPost, "/", nil)
assert.NoError(t, err)
ctx := req.Context()
rr, _, err := newReusableRequest(req, defaultMaxBodySize)
rr, _, err := newReusableRequest(req, true, defaultMaxBodySize)
assert.NoError(t, err)
// first call
@ -208,7 +250,7 @@ func TestCloneRequest(t *testing.T) {
ctx := req.Context()
req.ContentLength = int64(contentLength)
rr, _, err := newReusableRequest(req, defaultMaxBodySize)
rr, _, err := newReusableRequest(req, true, defaultMaxBodySize)
assert.NoError(t, err)
// first call
@ -231,7 +273,7 @@ func TestCloneRequest(t *testing.T) {
req, err := http.NewRequest(http.MethodPost, "/", buf)
assert.NoError(t, err)
_, expectedBytes, err := newReusableRequest(req, 2)
_, expectedBytes, err := newReusableRequest(req, true, 2)
assert.Error(t, err)
assert.Equal(t, expectedBytes, bb[:3])
})
@ -243,7 +285,7 @@ func TestCloneRequest(t *testing.T) {
req, err := http.NewRequest(http.MethodPost, "/", buf)
assert.NoError(t, err)
rr, expectedBytes, err := newReusableRequest(req, 20)
rr, expectedBytes, err := newReusableRequest(req, true, 20)
assert.NoError(t, err)
assert.Nil(t, expectedBytes)
assert.Len(t, rr.body, 10)
@ -255,14 +297,14 @@ func TestCloneRequest(t *testing.T) {
req, err := http.NewRequest(http.MethodGet, "/", buf)
assert.NoError(t, err)
rr, expectedBytes, err := newReusableRequest(req, 20)
rr, expectedBytes, err := newReusableRequest(req, true, 20)
assert.NoError(t, err)
assert.Nil(t, expectedBytes)
assert.Empty(t, rr.body)
})
t.Run("no request given", func(t *testing.T) {
_, _, err := newReusableRequest(nil, defaultMaxBodySize)
_, _, err := newReusableRequest(nil, true, defaultMaxBodySize)
assert.Error(t, err)
})
}

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

@ -34,7 +34,10 @@ import (
"google.golang.org/grpc/status"
)
const defaultMaxBodySize int64 = -1
const (
defaultMirrorBody = true
defaultMaxBodySize int64 = -1
)
// RoundTripperGetter is a roundtripper getter interface.
type RoundTripperGetter interface {
@ -197,11 +200,16 @@ func (m *Manager) getMirrorServiceHandler(ctx context.Context, config *dynamic.M
return nil, err
}
mirrorBody := defaultMirrorBody
if config.MirrorBody != nil {
mirrorBody = *config.MirrorBody
}
maxBodySize := defaultMaxBodySize
if config.MaxBodySize != nil {
maxBodySize = *config.MaxBodySize
}
handler := mirror.New(serviceHandler, m.routinePool, maxBodySize, config.HealthCheck)
handler := mirror.New(serviceHandler, m.routinePool, mirrorBody, maxBodySize, config.HealthCheck)
for _, mirrorConfig := range config.Mirrors {
mirrorHandler, err := m.BuildHTTP(ctx, mirrorConfig.Name)
if err != nil {