Adds mirroring service

Author: juliens

docs/content/routing/services/index.md

@@ -353,6 +353,53 @@ http:
         - url: "http://private-ip-server-2/"
 ```
 
+### Mirroring (service)
+
+The mirroring is able to mirror requests sent to a service to other services.
+
+This strategy can be defined only with [File](../../providers/file.md).
+
+```toml tab="TOML"
+[http.services]
+  [http.services.mirroring]
+    [http.services.mirroring.mirroring]
+      service = "app"
+    [[http.services.mirroring.mirroring.mirrors]]
+      name = "mirror"
+      percent = 10
+
+  [http.services.app]
+    [http.services.app.loadBalancer]
+      [[http.services.appv1.loadBalancer.servers]]
+        url = "http://private-ip-server-1/"
+
+  [http.services.mirror]
+    [http.services.mirror.loadBalancer]
+      [[http.services.mirror.loadBalancer.servers]]
+        url = "http://private-ip-server-2/"
+```
+
+```yaml tab="YAML"
+http:
+  services:
+    mirroring:
+      mirroring:
+        service: app
+        mirrors:
+        - name: mirror
+          percent: 10
+
+    app:
+      loadBalancer:
+        servers:
+        - url: "http://private-ip-server-1/"
+
+    mirror:
+      loadBalancer:
+        servers:
+        - url: "http://private-ip-server-2/"
+```
+
 ## Configuring TCP Services
 
 ### General

integration/fixtures/mirror.toml

@@ -0,0 +1,44 @@
+[global]
+  checkNewVersion = false
+  sendAnonymousUsage = false
+
+[api]
+
+[log]
+  level = "DEBUG"
+
+[entryPoints]
+
+  [entryPoints.web]
+    address = ":8000"
+
+[providers.file]
+  filename = "{{ .SelfFilename }}"
+
+## dynamic configuration ##
+
+[http.routers]
+  [http.routers.router]
+    service = "mirror"
+    rule = "Path(`/whoami`)"
+
+[http.services]
+  [http.services.mirror.mirroring]
+    service = "service1"
+  [[http.services.mirror.mirroring.mirrors]]
+    name = "mirror1"
+    percent = 10
+  [[http.services.mirror.mirroring.mirrors]]
+    name = "mirror2"
+    percent = 50
+
+  [http.services.service1.loadBalancer]
+    [[http.services.service1.loadBalancer.servers]]
+      url = "{{ .MainServer }}"
+  [http.services.mirror1.loadBalancer]
+    [[http.services.mirror1.loadBalancer.servers]]
+      url = "{{ .Mirror1Server }}"
+  [http.services.mirror2.loadBalancer]
+      [[http.services.mirror2.loadBalancer.servers]]
+        url = "{{ .Mirror2Server }}"
+

integration/simple_test.go

@@ -2,13 +2,15 @@ package integration
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync/atomic"
 	"syscall"
 	"time"
 
@@ -671,3 +673,111 @@ func (s *SimpleSuite) TestWRRSticky(c *check.C) {
 	c.Assert(repartition[server1], checker.Equals, 4)
 	c.Assert(repartition[server2], checker.Equals, 0)
 }
+
+func (s *SimpleSuite) TestMirror(c *check.C) {
+	var count, countMirror1, countMirror2 int32
+
+	main := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(&count, 1)
+	}))
+
+	mirror1 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(&countMirror1, 1)
+	}))
+
+	mirror2 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(&countMirror2, 1)
+	}))
+
+	mainServer := main.URL
+	mirror1Server := mirror1.URL
+	mirror2Server := mirror2.URL
+
+	file := s.adaptFile(c, "fixtures/mirror.toml", struct {
+		MainServer    string
+		Mirror1Server string
+		Mirror2Server string
+	}{MainServer: mainServer, Mirror1Server: mirror1Server, Mirror2Server: mirror2Server})
+	defer os.Remove(file)
+
+	cmd, output := s.traefikCmd(withConfigFile(file))
+	defer output(c)
+
+	err := cmd.Start()
+	c.Assert(err, checker.IsNil)
+	defer cmd.Process.Kill()
+
+	err = try.GetRequest("http://127.0.0.1:8080/api/http/services", 1000*time.Millisecond, try.BodyContains("mirror1", "mirror2", "service1"))
+	c.Assert(err, checker.IsNil)
+
+	req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8000/whoami", nil)
+	c.Assert(err, checker.IsNil)
+	for i := 0; i < 10; i++ {
+		response, err := http.DefaultClient.Do(req)
+		c.Assert(err, checker.IsNil)
+		c.Assert(response.StatusCode, checker.Equals, http.StatusOK)
+	}
+
+	countTotal := atomic.LoadInt32(&count)
+	val1 := atomic.LoadInt32(&countMirror1)
+	val2 := atomic.LoadInt32(&countMirror2)
+
+	c.Assert(countTotal, checker.Equals, int32(10))
+	c.Assert(val1, checker.Equals, int32(1))
+	c.Assert(val2, checker.Equals, int32(5))
+}
+
+func (s *SimpleSuite) TestMirrorCanceled(c *check.C) {
+	var count, countMirror1, countMirror2 int32
+
+	main := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(&count, 1)
+		time.Sleep(time.Second * 2)
+	}))
+
+	mirror1 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(&countMirror1, 1)
+	}))
+
+	mirror2 := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		atomic.AddInt32(&countMirror2, 1)
+	}))
+
+	mainServer := main.URL
+	mirror1Server := mirror1.URL
+	mirror2Server := mirror2.URL
+
+	file := s.adaptFile(c, "fixtures/mirror.toml", struct {
+		MainServer    string
+		Mirror1Server string
+		Mirror2Server string
+	}{MainServer: mainServer, Mirror1Server: mirror1Server, Mirror2Server: mirror2Server})
+	defer os.Remove(file)
+
+	cmd, output := s.traefikCmd(withConfigFile(file))
+	defer output(c)
+
+	err := cmd.Start()
+	c.Assert(err, checker.IsNil)
+	defer cmd.Process.Kill()
+
+	err = try.GetRequest("http://127.0.0.1:8080/api/http/services", 1000*time.Millisecond, try.BodyContains("mirror1", "mirror2", "service1"))
+	c.Assert(err, checker.IsNil)
+
+	for i := 0; i < 5; i++ {
+		req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1:8000/whoami", nil)
+		c.Assert(err, checker.IsNil)
+
+		newCtx, _ := context.WithTimeout(req.Context(), time.Second)
+		req = req.WithContext(newCtx)
+		http.DefaultClient.Do(req)
+	}
+
+	countTotal := atomic.LoadInt32(&count)
+	val1 := atomic.LoadInt32(&countMirror1)
+	val2 := atomic.LoadInt32(&countMirror2)
+
+	c.Assert(countTotal, checker.Equals, int32(5))
+	c.Assert(val1, checker.Equals, int32(0))
+	c.Assert(val2, checker.Equals, int32(0))
+}

pkg/config/dynamic/http_config.go

@@ -21,6 +21,7 @@ type HTTPConfiguration struct {
 type Service struct {
 	LoadBalancer *ServersLoadBalancer `json:"loadBalancer,omitempty" toml:"loadBalancer,omitempty" yaml:"loadBalancer,omitempty"`
 	Weighted     *WeightedRoundRobin  `json:"weighted,omitempty" toml:"weighted,omitempty" yaml:"weighted,omitempty" label:"-"`
+	Mirroring    *Mirroring           `json:"mirroring,omitempty" toml:"mirroring,omitempty" yaml:"mirroring,omitempty" label:"-"`
 }
 
 // +k8s:deepcopy-gen=true
@@ -46,6 +47,22 @@ type RouterTLSConfig struct {
 
 // +k8s:deepcopy-gen=true
 
+// Mirroring holds the Mirroring configuration.
+type Mirroring struct {
+	Service string          `json:"service,omitempty" toml:"service,omitempty" yaml:"service,omitempty"`
+	Mirrors []MirrorService `json:"mirrors,omitempty" toml:"mirrors,omitempty" yaml:"mirrors,omitempty"`
+}
+
+// +k8s:deepcopy-gen=true
+
+// MirrorService holds the MirrorService configuration.
+type MirrorService struct {
+	Name    string `json:"name,omitempty" toml:"name,omitempty" yaml:"name,omitempty"`
+	Percent int    `json:"percent,omitempty" toml:"percent,omitempty" yaml:"percent,omitempty"`
+}
+
+// +k8s:deepcopy-gen=true
+
 // WeightedRoundRobin is a weighted round robin load-balancer of services.
 type WeightedRoundRobin struct {
 	Services []WRRService `json:"services,omitempty" toml:"services,omitempty" yaml:"services,omitempty"`

pkg/server/router/router_test.go

@@ -306,7 +306,7 @@ func TestRouterManager_Get(t *testing.T) {
 					Middlewares: test.middlewaresConfig,
 				},
 			})
-			serviceManager := service.NewManager(rtConf.Services, http.DefaultTransport, nil)
+			serviceManager := service.NewManager(rtConf.Services, http.DefaultTransport, nil, nil)
 			middlewaresBuilder := middleware.NewBuilder(rtConf.Middlewares, serviceManager)
 			responseModifierFactory := responsemodifiers.NewBuilder(rtConf.Middlewares)
 			routerManager := NewManager(rtConf, serviceManager, middlewaresBuilder, responseModifierFactory)
@@ -407,7 +407,7 @@ func TestAccessLog(t *testing.T) {
 					Middlewares: test.middlewaresConfig,
 				},
 			})
-			serviceManager := service.NewManager(rtConf.Services, http.DefaultTransport, nil)
+			serviceManager := service.NewManager(rtConf.Services, http.DefaultTransport, nil, nil)
 			middlewaresBuilder := middleware.NewBuilder(rtConf.Middlewares, serviceManager)
 			responseModifierFactory := responsemodifiers.NewBuilder(rtConf.Middlewares)
 			routerManager := NewManager(rtConf, serviceManager, middlewaresBuilder, responseModifierFactory)
@@ -693,7 +693,7 @@ func TestRuntimeConfiguration(t *testing.T) {
 					Middlewares: test.middlewareConfig,
 				},
 			})
-			serviceManager := service.NewManager(rtConf.Services, http.DefaultTransport, nil)
+			serviceManager := service.NewManager(rtConf.Services, http.DefaultTransport, nil, nil)
 			middlewaresBuilder := middleware.NewBuilder(rtConf.Middlewares, serviceManager)
 			responseModifierFactory := responsemodifiers.NewBuilder(map[string]*runtime.MiddlewareInfo{})
 			routerManager := NewManager(rtConf, serviceManager, middlewaresBuilder, responseModifierFactory)
@@ -767,7 +767,7 @@ func BenchmarkRouterServe(b *testing.B) {
 			Middlewares: map[string]*dynamic.Middleware{},
 		},
 	})
-	serviceManager := service.NewManager(rtConf.Services, &staticTransport{res}, nil)
+	serviceManager := service.NewManager(rtConf.Services, &staticTransport{res}, nil, nil)
 	middlewaresBuilder := middleware.NewBuilder(rtConf.Middlewares, serviceManager)
 	responseModifierFactory := responsemodifiers.NewBuilder(rtConf.Middlewares)
 	routerManager := NewManager(rtConf, serviceManager, middlewaresBuilder, responseModifierFactory)
@@ -808,7 +808,7 @@ func BenchmarkService(b *testing.B) {
 			Services: serviceConfig,
 		},
 	})
-	serviceManager := service.NewManager(rtConf.Services, &staticTransport{res}, nil)
+	serviceManager := service.NewManager(rtConf.Services, &staticTransport{res}, nil, nil)
 	w := httptest.NewRecorder()
 	req := testhelpers.MustNewRequest(http.MethodGet, "http://foo.bar/", nil)
 

pkg/server/server_configuration.go

@@ -97,7 +97,7 @@ func (s *Server) createTCPRouters(ctx context.Context, configuration *runtime.Co
 
 // createHTTPHandlers returns, for the given configuration and entryPoints, the HTTP handlers for non-TLS connections, and for the TLS ones. the given configuration must not be nil. its fields will get mutated.
 func (s *Server) createHTTPHandlers(ctx context.Context, configuration *runtime.Configuration, entryPoints []string) (map[string]http.Handler, map[string]http.Handler) {
-	serviceManager := service.NewManager(configuration.Services, s.defaultRoundTripper, s.metricsRegistry)
+	serviceManager := service.NewManager(configuration.Services, s.defaultRoundTripper, s.metricsRegistry, s.routinesPool)
 	middlewaresBuilder := middleware.NewBuilder(configuration.Middlewares, serviceManager)
 	responseModifierFactory := responsemodifiers.NewBuilder(configuration.Middlewares)
 	routerManager := router.NewManager(configuration, serviceManager, middlewaresBuilder, responseModifierFactory)