Repository: containous/maesh
Pull request #531 (2020-04-30 13:27:26)

Multiple middleware support & annotations refactoring
### What does this PR do?

This PR removes code duplication around annotations evaluation. It also fixes a bug in the ShadowServiceManager, when the service has an invalid `traffic-type` value. Before this PR, if a service had an invalid traffic-type, it was considered to be an HTTP service. Since the provider doesn't handle it this way, a shadow service was created but not used.
This PR also moves the middleware builder in the annotations package and the provider now accepts it as a dependency. This will allow external projects to reuse the provider while allowing a customization of the middlewares.

This PR fixes a bug which prevented the use of multiple middlewares on the same service.

### How to test it

- `make test-integration`
- `make test`
- Manually: 
  - Deploy a service with a valid traffic-type
  - Deploy a service with no traffic-type
  - Deploy a service with an invalid traffic-type
  - Apply multiple middleware on the same service


Patch:
diff --git a/pkg/annotations/annotations.go b/pkg/annotations/annotations.go
new file mode 100644
index 0000000..10a38bf
--- /dev/null
+++ b/pkg/annotations/annotations.go
@@ -0,0 +1,127 @@
+package annotations
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+)
+
+const (
+	// ServiceTypeHTTP HTTP service type.
+	ServiceTypeHTTP string = "http"
+	// ServiceTypeTCP TCP service type.
+	ServiceTypeTCP string = "tcp"
+	// ServiceTypeUDP UDP service type.
+	ServiceTypeUDP string = "udp"
+
+	// SchemeHTTP HTTP scheme.
+	SchemeHTTP string = "http"
+	// SchemeH2C h2c scheme.
+	SchemeH2C string = "h2c"
+	// SchemeHTTPS HTTPS scheme.
+	SchemeHTTPS string = "https"
+)
+
+const (
+	baseAnnotation                     = "maesh.containo.us/"
+	annotationServiceType              = baseAnnotation + "traffic-type"
+	annotationScheme                   = baseAnnotation + "scheme"
+	annotationRetryAttempts            = baseAnnotation + "retry-attempts"
+	annotationCircuitBreakerExpression = baseAnnotation + "circuit-breaker-expression"
+	annotationRateLimitAverage         = baseAnnotation + "ratelimit-average"
+	annotationRateLimitBurst           = baseAnnotation + "ratelimit-burst"
+)
+
+// ErrNotFound indicates that the annotation hasn't been found.
+var ErrNotFound = errors.New("annotation not found")
+
+// GetTrafficType returns the value of the traffic-type annotation.
+func GetTrafficType(defaultTrafficType string, annotations map[string]string) (string, error) {
+	trafficType, ok := annotations[annotationServiceType]
+	if !ok {
+		return defaultTrafficType, nil
+	}
+
+	switch trafficType {
+	case ServiceTypeHTTP:
+	case ServiceTypeTCP:
+	case ServiceTypeUDP:
+	default:
+		return trafficType, fmt.Errorf("unsupported traffic type %q: %q", annotationServiceType, trafficType)
+	}
+
+	return trafficType, nil
+}
+
+// GetScheme returns the value of the scheme annotation.
+func GetScheme(annotations map[string]string) (string, error) {
+	scheme, ok := annotations[annotationScheme]
+	if !ok {
+		return SchemeHTTP, nil
+	}
+
+	switch scheme {
+	case SchemeHTTP:
+	case SchemeH2C:
+	case SchemeHTTPS:
+	default:
+		return scheme, fmt.Errorf("unsupported scheme %q: %q", annotationScheme, scheme)
+	}
+
+	return scheme, nil
+}
+
+// GetRetryAttempts returns the value of the retry-attempts annotation.
+func GetRetryAttempts(annotations map[string]string) (int, error) {
+	retryAttempts, ok := annotations[annotationRetryAttempts]
+	if !ok {
+		return 0, ErrNotFound
+	}
+
+	attempts, err := strconv.Atoi(retryAttempts)
+	if err != nil {
+		return 0, fmt.Errorf("invalid value %q: %w", annotationRetryAttempts, err)
+	}
+
+	return attempts, nil
+}
+
+// GetCircuitBreakerExpression returns the value of the circuit-breaker-expression annotation.
+func GetCircuitBreakerExpression(annotations map[string]string) (string, error) {
+	circuitBreakerExpression, ok := annotations[annotationCircuitBreakerExpression]
+	if !ok {
+		return "", ErrNotFound
+	}
+
+	return circuitBreakerExpression, nil
+}
+
+// GetRateLimitBurst returns the value of the rate-limit-burst annotation.
+func GetRateLimitBurst(annotations map[string]string) (int, error) {
+	rateLimitBurst, ok := annotations[annotationRateLimitBurst]
+	if !ok {
+		return 0, ErrNotFound
+	}
+
+	burst, err := strconv.Atoi(rateLimitBurst)
+	if err != nil {
+		return 0, fmt.Errorf("invalid value %q: %w", annotationRateLimitBurst, err)
+	}
+
+	return burst, nil
+}
+
+// GetRateLimitAverage returns the value of the rate-limit-average annotation.
+func GetRateLimitAverage(annotations map[string]string) (int, error) {
+	rateLimitAverage, ok := annotations[annotationRateLimitAverage]
+	if !ok {
+		return 0, ErrNotFound
+	}
+
+	average, err := strconv.Atoi(rateLimitAverage)
+	if err != nil {
+		return 0, fmt.Errorf("invalid value %q: %w", annotationRateLimitAverage, err)
+	}
+
+	return average, nil
+}
diff --git a/pkg/annotations/middleware.go b/pkg/annotations/middleware.go
new file mode 100644
index 0000000..ba5e387
--- /dev/null
+++ b/pkg/annotations/middleware.go
@@ -0,0 +1,111 @@
+package annotations
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/containous/traefik/v2/pkg/config/dynamic"
+)
+
+type middlewareBuilder func(annotations map[string]string) (middleware *dynamic.Middleware, name string, err error)
+
+// BuildMiddlewares builds middlewares from the given annotations.
+func BuildMiddlewares(annotations map[string]string) (map[string]*dynamic.Middleware, error) {
+	builders := []middlewareBuilder{
+		buildRetryMiddleware,
+		buildRateLimitMiddleware,
+		buildCircuitBreakerMiddleware,
+	}
+
+	middlewares := map[string]*dynamic.Middleware{}
+
+	for _, builder := range builders {
+		middleware, name, err := builder(annotations)
+		if err != nil {
+			return nil, err
+		}
+
+		if middleware != nil {
+			middlewares[name] = middleware
+		}
+	}
+
+	return middlewares, nil
+}
+
+func buildRetryMiddleware(annotations map[string]string) (middleware *dynamic.Middleware, name string, err error) {
+	var retryAttempts int
+
+	retryAttempts, err = GetRetryAttempts(annotations)
+	if err != nil {
+		if err == ErrNotFound {
+			return nil, "", nil
+		}
+
+		return nil, "", fmt.Errorf("unable to build retry middleware: %w", err)
+	}
+
+	name = "retry"
+	middleware = &dynamic.Middleware{
+		Retry: &dynamic.Retry{Attempts: retryAttempts},
+	}
+
+	return middleware, name, nil
+}
+
+func buildRateLimitMiddleware(annotations map[string]string) (middleware *dynamic.Middleware, name string, err error) {
+	var (
+		rateLimitBurst   int
+		rateLimitAverage int
+	)
+
+	rateLimitBurst, err = GetRateLimitBurst(annotations)
+	if err == ErrNotFound {
+		return nil, "", nil
+	} else if err != nil {
+		return nil, "", fmt.Errorf("unable to build rate-limit middleware: %w", err)
+	}
+
+	rateLimitAverage, err = GetRateLimitAverage(annotations)
+	if err == ErrNotFound {
+		return nil, "", nil
+	} else if err != nil {
+		return nil, "", fmt.Errorf("unable to build rate-limit middleware: %w", err)
+	}
+
+	if rateLimitBurst <= 0 || rateLimitAverage <= 0 {
+		return nil, "", errors.New("unable to build rate-limit middleware: burst and average must be greater than 0")
+	}
+
+	name = "rate-limit"
+	middleware = &dynamic.Middleware{
+		RateLimit: &dynamic.RateLimit{
+			Burst:   int64(rateLimitBurst),
+			Average: int64(rateLimitAverage),
+		},
+	}
+
+	return middleware, name, nil
+}
+
+func buildCircuitBreakerMiddleware(annotations map[string]string) (middleware *dynamic.Middleware, name string, err error) {
+	var circuitBreakerExpression string
+
+	circuitBreakerExpression, err = GetCircuitBreakerExpression(annotations)
+	if err != nil {
+		if err == ErrNotFound {
+			return nil, "", nil
+		}
+
+		return nil, "", fmt.Errorf("unable to build circuit-breaker middleware: %w", err)
+	}
+
+	name = "circuit-breaker"
+	middleware = &dynamic.Middleware{
+		CircuitBreaker: &dynamic.CircuitBreaker{
+			Expression: circuitBreakerExpression,
+		},
+	}
+
+	return middleware, name, nil
+}
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 35b1a75..085d2df 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -6,6 +6,7 @@ import (
 	"reflect"
 	"time"
 
+	"github.com/containous/maesh/pkg/annotations"
 	"github.com/containous/maesh/pkg/api"
 	"github.com/containous/maesh/pkg/k8s"
 	"github.com/containous/maesh/pkg/provider"
@@ -184,7 +185,7 @@ func (c *Controller) init() {
 		MaeshNamespace:     c.cfg.Namespace,
 	}
 
-	c.provider = provider.New(c.tcpStateTable, c.udpStateTable, providerCfg, c.logger)
+	c.provider = provider.New(c.tcpStateTable, c.udpStateTable, annotations.BuildMiddlewares, providerCfg, c.logger)
 }
 
 // Run is the main entrypoint for the controller.
diff --git a/pkg/controller/service.go b/pkg/controller/service.go
index dc891e0..019ef67 100644
--- a/pkg/controller/service.go
+++ b/pkg/controller/service.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/containous/maesh/pkg/annotations"
 	"github.com/containous/maesh/pkg/k8s"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
@@ -18,29 +19,29 @@ import (
 
 // ShadowServiceManager manages shadow services.
 type ShadowServiceManager struct {
-	log           logrus.FieldLogger
-	lister        listers.ServiceLister
-	namespace     string
-	tcpStateTable PortMapper
-	udpStateTable PortMapper
-	defaultMode   string
-	minHTTPPort   int32
-	maxHTTPPort   int32
-	kubeClient    kubernetes.Interface
+	log                logrus.FieldLogger
+	lister             listers.ServiceLister
+	namespace          string
+	tcpStateTable      PortMapper
+	udpStateTable      PortMapper
+	defaultTrafficType string
+	minHTTPPort        int32
+	maxHTTPPort        int32
+	kubeClient         kubernetes.Interface
 }
 
 // NewShadowServiceManager returns new shadow service manager.
-func NewShadowServiceManager(log logrus.FieldLogger, lister listers.ServiceLister, namespace string, tcpStateTable PortMapper, udpStateTable PortMapper, defaultMode string, minHTTPPort, maxHTTPPort int32, kubeClient kubernetes.Interface) *ShadowServiceManager {
+func NewShadowServiceManager(log logrus.FieldLogger, lister listers.ServiceLister, namespace string, tcpStateTable PortMapper, udpStateTable PortMapper, defaultTrafficType string, minHTTPPort, maxHTTPPort int32, kubeClient kubernetes.Interface) *ShadowServiceManager {
 	return &ShadowServiceManager{
-		log:           log,
-		lister:        lister,
-		namespace:     namespace,
-		tcpStateTable: tcpStateTable,
-		udpStateTable: udpStateTable,
-		defaultMode:   defaultMode,
-		minHTTPPort:   minHTTPPort,
-		maxHTTPPort:   maxHTTPPort,
-		kubeClient:    kubeClient,
+		log:                log,
+		lister:             lister,
+		namespace:          namespace,
+		tcpStateTable:      tcpStateTable,
+		udpStateTable:      udpStateTable,
+		defaultTrafficType: defaultTrafficType,
+		minHTTPPort:        minHTTPPort,
+		maxHTTPPort:        maxHTTPPort,
+		kubeClient:         kubeClient,
 	}
 }
 
@@ -59,7 +60,10 @@ func (s *ShadowServiceManager) Create(userSvc *corev1.Service) error {
 		return fmt.Errorf("unable to get shadow service %q: %w", name, err)
 	}
 
-	ports := s.getShadowServicePorts(userSvc)
+	ports, err := s.getShadowServicePorts(userSvc)
+	if err != nil {
+		return fmt.Errorf("unable to get ports for service %s/%s: %w", userSvc.Namespace, userSvc.Name, err)
+	}
 
 	svc := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
@@ -100,7 +104,15 @@ func (s *ShadowServiceManager) Create(userSvc *corev1.Service) error {
 func (s *ShadowServiceManager) Update(oldUserSvc *corev1.Service, newUserSvc *corev1.Service) (*corev1.Service, error) {
 	name := s.getShadowServiceName(newUserSvc.Name, newUserSvc.Namespace)
 
-	s.cleanupPortMapping(oldUserSvc, newUserSvc)
+	ports, err := s.getShadowServicePorts(newUserSvc)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get ports for service %s/%s: %w", newUserSvc.Namespace, newUserSvc.Name, err)
+	}
+
+	err = s.cleanupPortMapping(oldUserSvc, newUserSvc)
+	if err != nil {
+		return nil, fmt.Errorf("unable to cleanup port mapping for service %s/%s: %w", oldUserSvc.Namespace, oldUserSvc.Name, err)
+	}
 
 	var updatedSvc *corev1.Service
 
@@ -111,7 +123,7 @@ func (s *ShadowServiceManager) Update(oldUserSvc *corev1.Service, newUserSvc *co
 		}
 
 		newSvc := svc.DeepCopy()
-		newSvc.Spec.Ports = s.getShadowServicePorts(newUserSvc)
+		newSvc.Spec.Ports = ports
 
 		if updatedSvc, err = s.kubeClient.CoreV1().Services(s.namespace).Update(newSvc); err != nil {
 			return fmt.Errorf("unable to update kubernetes service: %w", err)
@@ -133,7 +145,9 @@ func (s *ShadowServiceManager) Update(oldUserSvc *corev1.Service, newUserSvc *co
 func (s *ShadowServiceManager) Delete(userSvc *corev1.Service) error {
 	name := s.getShadowServiceName(userSvc.Name, userSvc.Namespace)
 
-	s.cleanupPortMapping(userSvc, nil)
+	if err := s.cleanupPortMapping(userSvc, nil); err != nil {
+		return fmt.Errorf("unable to cleanup port mapping for service %s/%s: %w", userSvc.Namespace, userSvc.Name, err)
+	}
 
 	_, err := s.lister.Services(s.namespace).Get(name)
 	if err != nil {
@@ -149,16 +163,21 @@ func (s *ShadowServiceManager) Delete(userSvc *corev1.Service) error {
 	return nil
 }
 
-func (s *ShadowServiceManager) cleanupPortMapping(oldUserSvc *corev1.Service, newUserSvc *corev1.Service) {
+func (s *ShadowServiceManager) cleanupPortMapping(oldUserSvc *corev1.Service, newUserSvc *corev1.Service) error {
 	var stateTable PortMapper
 
-	switch svcMode := s.getServiceMode(oldUserSvc); svcMode {
-	case k8s.ServiceTypeTCP:
+	trafficType, err := annotations.GetTrafficType(s.defaultTrafficType, oldUserSvc.Annotations)
+	if err != nil {
+		return fmt.Errorf("unable to get service traffic type: %w", err)
+	}
+
+	switch trafficType {
+	case annotations.ServiceTypeTCP:
 		stateTable = s.tcpStateTable
-	case k8s.ServiceTypeUDP:
+	case annotations.ServiceTypeUDP:
 		stateTable = s.udpStateTable
 	default:
-		return
+		return nil
 	}
 
 	for _, old := range oldUserSvc.Spec.Ports {
@@ -185,20 +204,24 @@ func (s *ShadowServiceManager) cleanupPortMapping(oldUserSvc *corev1.Service, ne
 			}
 		}
 	}
+
+	return nil
 }
 
-func (s *ShadowServiceManager) getShadowServicePorts(svc *corev1.Service) []corev1.ServicePort {
+func (s *ShadowServiceManager) getShadowServicePorts(svc *corev1.Service) ([]corev1.ServicePort, error) {
 	var ports []corev1.ServicePort
 
-	svcMode := s.getServiceMode(svc)
+	trafficType, err := annotations.GetTrafficType(s.defaultTrafficType, svc.Annotations)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get service traffic-type: %w", err)
+	}
 
 	for i, sp := range svc.Spec.Ports {
-		if !(svcMode == k8s.ServiceTypeUDP && sp.Protocol == corev1.ProtocolUDP) && !(svcMode != k8s.ServiceTypeUDP && sp.Protocol == corev1.ProtocolTCP) {
-			s.log.Warnf("Unsupported port type: %s, skipping port %s on service %s/%s", sp.Protocol, sp.Name, svc.Namespace, svc.Name)
-			continue
+		if !isPortSuitable(trafficType, sp) {
+			s.log.Warnf("Unsupported port type %q on %q service %s/%s, skipping port %q", sp.Protocol, trafficType, svc.Namespace, svc.Name, sp.Name)
 		}
 
-		targetPort, err := s.getTargetPort(svcMode, i, svc.Name, svc.Namespace, sp.Port)
+		targetPort, err := s.getTargetPort(trafficType, i, svc.Name, svc.Namespace, sp.Port)
 		if err != nil {
 			s.log.Errorf("Unable to find available %s port: %v, skipping port %s on service %s/%s", sp.Name, err, sp.Name, svc.Namespace, svc.Name)
 			continue
@@ -212,15 +235,7 @@ func (s *ShadowServiceManager) getShadowServicePorts(svc *corev1.Service) []core
 		})
 	}
 
-	return ports
-}
-
-func (s *ShadowServiceManager) getServiceMode(svc *corev1.Service) string {
-	if svcMode, ok := svc.Annotations[k8s.AnnotationServiceType]; ok {
-		return svcMode
-	}
-
-	return s.defaultMode
+	return ports, nil
 }
 
 // getShadowServiceName converts a User service with a namespace to a mesh service name.
@@ -228,13 +243,13 @@ func (s *ShadowServiceManager) getShadowServiceName(name string, namespace strin
 	return fmt.Sprintf("%s-%s-6d61657368-%s", s.namespace, name, namespace)
 }
 
-func (s *ShadowServiceManager) getTargetPort(svcMode string, portID int, name, namespace string, port int32) (int32, error) {
-	switch svcMode {
-	case k8s.ServiceTypeHTTP:
+func (s *ShadowServiceManager) getTargetPort(trafficType string, portID int, name, namespace string, port int32) (int32, error) {
+	switch trafficType {
+	case annotations.ServiceTypeHTTP:
 		return s.getHTTPPort(portID)
-	case k8s.ServiceTypeTCP:
+	case annotations.ServiceTypeTCP:
 		return s.getMappedPort(s.tcpStateTable, name, namespace, port)
-	case k8s.ServiceTypeUDP:
+	case annotations.ServiceTypeUDP:
 		return s.getMappedPort(s.udpStateTable, name, namespace, port)
 	default:
 		return 0, errors.New("unknown service mode")
@@ -273,6 +288,18 @@ func (s *ShadowServiceManager) getMappedPort(stateTable PortMapper, svcName, svc
 	return port, nil
 }
 
+func isPortSuitable(trafficType string, sp corev1.ServicePort) bool {
+	if trafficType == annotations.ServiceTypeUDP {
+		return sp.Protocol == corev1.ProtocolUDP
+	}
+
+	if trafficType == annotations.ServiceTypeTCP || trafficType == annotations.ServiceTypeHTTP {
+		return sp.Protocol == corev1.ProtocolTCP
+	}
+
+	return false
+}
+
 func parseKubernetesServerVersion(kubeClient kubernetes.Interface) (major, minor int) {
 	kubeVersion, err := kubeClient.Discovery().ServerVersion()
 	if err != nil {
diff --git a/pkg/k8s/constants.go b/pkg/k8s/constants.go
index 11b03e1..08328c1 100644
--- a/pkg/k8s/constants.go
+++ b/pkg/k8s/constants.go
@@ -6,8 +6,7 @@ import (
 
 const (
 	// ResyncPeriod set the resync period.
-	ResyncPeriod          = 5 * time.Minute
-	baseAnnotation string = "maesh.containo.us/"
+	ResyncPeriod = 5 * time.Minute
 
 	// CoreObjectKinds is a filter for objects to process by the core client.
 	CoreObjectKinds = "Deployment|Endpoints|Service|Ingress|Secret|Namespace|Pod|ConfigMap"
@@ -18,33 +17,6 @@ const (
 	// SplitObjectKinds is a filter for objects to process by the split client.
 	SplitObjectKinds = "TrafficSplit"
 
-	// AnnotationServiceType service type annotation.
-	AnnotationServiceType = baseAnnotation + "traffic-type"
-	// AnnotationScheme scheme.
-	AnnotationScheme = baseAnnotation + "scheme"
-	// AnnotationRetryAttempts retry attempts annotation.
-	AnnotationRetryAttempts = baseAnnotation + "retry-attempts"
-	// AnnotationCircuitBreakerExpression circuit breaker expression annotation.
-	AnnotationCircuitBreakerExpression = baseAnnotation + "circuit-breaker-expression"
-	// AnnotationRateLimitAverage sets the average value for rate limiting.
-	AnnotationRateLimitAverage = baseAnnotation + "ratelimit-average"
-	// AnnotationRateLimitBurst sets the burst value for rate limiting.
-	AnnotationRateLimitBurst = baseAnnotation + "ratelimit-burst"
-
-	// ServiceTypeHTTP HTTP service type.
-	ServiceTypeHTTP string = "http"
-	// ServiceTypeTCP TCP service type.
-	ServiceTypeTCP string = "tcp"
-	// ServiceTypeUDP UDP service type.
-	ServiceTypeUDP string = "udp"
-
-	// SchemeHTTP HTTP scheme.
-	SchemeHTTP string = "http"
-	// SchemeH2c h2c scheme.
-	SchemeH2c string = "h2c"
-	// SchemeHTTPS HTTPS scheme.
-	SchemeHTTPS string = "https"
-
 	// TCPStateConfigMapName TCP config map name.
 	TCPStateConfigMapName string = "tcp-state-table"
 	// UDPStateConfigMapName UDP config map name.
diff --git a/pkg/provider/key.go b/pkg/provider/key.go
index 7c290fa..c644739 100644
--- a/pkg/provider/key.go
+++ b/pkg/provider/key.go
@@ -11,8 +11,8 @@ const (
 	blockAllServiceKey    = "block-all-service"
 )
 
-func getMiddlewareKey(svc *topology.Service) string {
-	return fmt.Sprintf("%s-%s", svc.Namespace, svc.Name)
+func getMiddlewareKey(svc *topology.Service, name string) string {
+	return fmt.Sprintf("%s-%s-%s", svc.Namespace, svc.Name, name)
 }
 
 func getServiceRouterKeyFromService(svc *topology.Service, port int32) string {
diff --git a/pkg/provider/middleware.go b/pkg/provider/middleware.go
deleted file mode 100644
index d86527d..0000000
--- a/pkg/provider/middleware.go
+++ /dev/null
@@ -1,65 +0,0 @@
-package provider
-
-import (
-	"errors"
-	"fmt"
-	"strconv"
-
-	"github.com/containous/maesh/pkg/k8s"
-	"github.com/containous/maesh/pkg/topology"
-	"github.com/containous/traefik/v2/pkg/config/dynamic"
-)
-
-// MiddlewareBuilder builds middlewares of a service.
-type MiddlewareBuilder func(svc *topology.Service) (*dynamic.Middleware, error)
-
-// Build builds middlewares of the given service using annotations.
-func buildMiddlewareFromAnnotations(svc *topology.Service) (*dynamic.Middleware, error) {
-	var middleware dynamic.Middleware
-
-	// Build circuit-breaker middleware.
-	if circuitBreakerExpression, ok := svc.Annotations[k8s.AnnotationCircuitBreakerExpression]; ok {
-		middleware.CircuitBreaker = &dynamic.CircuitBreaker{Expression: circuitBreakerExpression}
-	}
-
-	// Build retry middleware.
-	if retryAttempts, ok := svc.Annotations[k8s.AnnotationRetryAttempts]; ok {
-		attempts, err := strconv.Atoi(retryAttempts)
-		if err != nil {
-			return nil, fmt.Errorf("unable to build retry middleware, %q annotation is invalid: %w", k8s.AnnotationRetryAttempts, err)
-		}
-
-		middleware.Retry = &dynamic.Retry{Attempts: attempts}
-	}
-
-	// Build rate-limit middleware.
-	rateLimitAverage, hasRateLimitAverage := svc.Annotations[k8s.AnnotationRateLimitAverage]
-	rateLimitBurst, hasRateLimitBurst := svc.Annotations[k8s.AnnotationRateLimitBurst]
-
-	if hasRateLimitAverage && hasRateLimitBurst {
-		average, err := strconv.Atoi(rateLimitAverage)
-		if err != nil {
-			return nil, fmt.Errorf("unable to build rate-limit middleware, %q annotation is invalid: %w", k8s.AnnotationRateLimitAverage, err)
-		}
-
-		burst, err := strconv.Atoi(rateLimitBurst)
-		if err != nil {
-			return nil, fmt.Errorf("unable to build rate-limit middleware, %q annotation is invalid: %w", k8s.AnnotationRateLimitBurst, err)
-		}
-
-		if burst <= 0 || average <= 0 {
-			return nil, errors.New("unable to build rate-limit middleware, burst and average must be greater than 0")
-		}
-
-		middleware.RateLimit = &dynamic.RateLimit{
-			Average: int64(average),
-			Burst:   int64(burst),
-		}
-	}
-
-	if middleware.CircuitBreaker == nil && middleware.Retry == nil && middleware.RateLimit == nil {
-		return nil, nil
-	}
-
-	return &middleware, nil
-}
diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go
index 8fd7398..a9b506c 100644
--- a/pkg/provider/provider.go
+++ b/pkg/provider/provider.go
@@ -6,6 +6,7 @@ import (
 	"net"
 	"strconv"
 
+	"github.com/containous/maesh/pkg/annotations"
 	"github.com/containous/maesh/pkg/k8s"
 	"github.com/containous/maesh/pkg/topology"
 	"github.com/containous/traefik/v2/pkg/config/dynamic"
@@ -13,6 +14,9 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
+// MiddlewareBuilder is capable of building a middleware from service annotations.
+type MiddlewareBuilder func(annotations map[string]string) (map[string]*dynamic.Middleware, error)
+
 // PortFinder finds service port mappings.
 type PortFinder interface {
 	Find(k8s.ServiceWithPort) (int32, bool)
@@ -55,13 +59,13 @@ type Provider struct {
 }
 
 // New creates a new Provider.
-func New(tcpStateTable, udpStateTable PortFinder, cfg Config, logger logrus.FieldLogger) *Provider {
+func New(tcpStateTable, udpStateTable PortFinder, middlewareBuilder MiddlewareBuilder, cfg Config, logger logrus.FieldLogger) *Provider {
 	return &Provider{
 		config:                 cfg,
 		tcpStateTable:          tcpStateTable,
 		udpStateTable:          udpStateTable,
 		logger:                 logger,
-		buildServiceMiddleware: buildMiddlewareFromAnnotations,
+		buildServiceMiddleware: middlewareBuilder,
 	}
 }
 
@@ -82,41 +86,41 @@ func (p *Provider) BuildConfig(t *topology.Topology) *dynamic.Configuration {
 
 // buildConfigForService builds the dynamic configuration for the given service.
 func (p *Provider) buildConfigForService(t *topology.Topology, cfg *dynamic.Configuration, svc *topology.Service) error {
-	trafficType, err := p.getTrafficTypeAnnotation(svc)
+	trafficType, err := annotations.GetTrafficType(p.config.DefaultTrafficType, svc.Annotations)
 	if err != nil {
 		return fmt.Errorf("unable to evaluate traffic-type annotation: %w", err)
 	}
 
-	scheme, err := getSchemeAnnotation(svc)
+	scheme, err := annotations.GetScheme(svc.Annotations)
 	if err != nil {
 		return fmt.Errorf("unable to evaluate scheme annotation: %w", err)
 	}
 
-	var middlewares []string
+	var middlewareKeys []string
 
 	// Middlewares are currently supported only for HTTP services.
-	if trafficType == k8s.ServiceTypeHTTP {
-		middleware, err := p.buildServiceMiddleware(svc)
+	if trafficType == annotations.ServiceTypeHTTP {
+		middlewares, err := p.buildServiceMiddleware(svc.Annotations)
 		if err != nil {
 			return fmt.Errorf("unable to build middlewares: %w", err)
 		}
 
-		if middleware != nil {
-			middlewareKey := getMiddlewareKey(svc)
+		for name, middleware := range middlewares {
+			middlewareKey := getMiddlewareKey(svc, name)
 			cfg.HTTP.Middlewares[middlewareKey] = middleware
 
-			middlewares = append(middlewares, middlewareKey)
+			middlewareKeys = append(middlewareKeys, middlewareKey)
 		}
 	}
 
 	// When ACL mode is on, all traffic must be forbidden unless explicitly authorized via a TrafficTarget.
 	if p.config.ACL {
-		if trafficType == k8s.ServiceTypeHTTP {
+		if trafficType == annotations.ServiceTypeHTTP {
 			p.buildBlockAllRouters(cfg, svc)
 		}
 
 		for _, ttKey := range svc.TrafficTargets {
-			if err := p.buildServicesAndRoutersForTrafficTarget(t, cfg, ttKey, scheme, trafficType, middlewares); err != nil {
+			if err := p.buildServicesAndRoutersForTrafficTarget(t, cfg, ttKey, scheme, trafficType, middlewareKeys); err != nil {
 				err = fmt.Errorf("unable to build routers and services: %v", err)
 				t.ServiceTrafficTargets[ttKey].AddError(err)
 				p.logger.Errorf("Error building dynamic configuration for TrafficTarget %q: %v", ttKey, err)
@@ -125,14 +129,14 @@ func (p *Provider) buildConfigForService(t *topology.Topology, cfg *dynamic.Conf
 			}
 		}
 	} else {
-		err := p.buildServicesAndRoutersForService(t, cfg, svc, scheme, trafficType, middlewares)
+		err := p.buildServicesAndRoutersForService(t, cfg, svc, scheme, trafficType, middlewareKeys)
 		if err != nil {
 			return fmt.Errorf("unable to build routers and services: %w", err)
 		}
 	}
 
 	for _, tsKey := range svc.TrafficSplits {
-		if err := p.buildServiceAndRoutersForTrafficSplit(t, cfg, tsKey, scheme, trafficType, middlewares); err != nil {
+		if err := p.buildServiceAndRoutersForTrafficSplit(t, cfg, tsKey, scheme, trafficType, middlewareKeys); err != nil {
 			err = fmt.Errorf("unable to build routers and services : %v", err)
 			t.TrafficSplits[tsKey].AddError(err)
 			p.logger.Errorf("Error building dynamic configuration for TrafficSplit %q: %v", tsKey, err)
@@ -148,7 +152,7 @@ func (p *Provider) buildServicesAndRoutersForService(t *topology.Topology, cfg *
 	svcKey := topology.Key{Name: svc.Name, Namespace: svc.Namespace}
 
 	switch trafficType {
-	case k8s.ServiceTypeHTTP:
+	case annotations.ServiceTypeHTTP:
 		httpRule := buildHTTPRuleFromService(svc)
 
 		for portID, svcPort := range svc.Ports {
@@ -166,7 +170,7 @@ func (p *Provider) buildServicesAndRoutersForService(t *topology.Topology, cfg *
 			cfg.HTTP.Routers[key] = buildHTTPRouter(httpRule, entrypoint, middlewares, key, priorityService)
 		}
 
-	case k8s.ServiceTypeTCP:
+	case annotations.ServiceTypeTCP:
 		rule := buildTCPRouterRule()
 
 		for _, svcPort := range svc.Ports {
@@ -184,7 +188,7 @@ func (p *Provider) buildServicesAndRoutersForService(t *topology.Topology, cfg *
 			cfg.TCP.Routers[key] = buildTCPRouter(rule, entrypoint, key)
 		}
 
-	case k8s.ServiceTypeUDP:
+	case annotations.ServiceTypeUDP:
 		for _, svcPort := range svc.Ports {
 			entrypoint, err := p.buildUDPEntrypoint(svc, svcPort.Port)
 			if err != nil {
@@ -219,7 +223,7 @@ func (p *Provider) buildServicesAndRoutersForTrafficTarget(t *topology.Topology,
 	}
 
 	switch trafficType {
-	case k8s.ServiceTypeHTTP:
+	case annotations.ServiceTypeHTTP:
 		whitelistDirect := p.buildWhitelistMiddlewareFromTrafficTargetDirect(t, tt)
 		whitelistDirectKey := getWhitelistMiddlewareKeyFromTrafficTargetDirect(tt)
 		cfg.HTTP.Middlewares[whitelistDirectKey] = whitelistDirect
@@ -258,7 +262,7 @@ func (p *Provider) buildServicesAndRoutersForTrafficTarget(t *topology.Topology,
 				cfg.HTTP.Routers[indirectRtrKey] = buildHTTPRouter(rule, entrypoint, rtrMiddlewares, svcKey, priorityTrafficTargetIndirect)
 			}
 		}
-	case k8s.ServiceTypeTCP:
+	case annotations.ServiceTypeTCP:
 		if !hasTrafficTargetSpecTCPRoute(tt) {
 			return nil
 		}
@@ -298,13 +302,13 @@ func (p *Provider) buildServiceAndRoutersForTrafficSplit(t *topology.Topology, c
 	}
 
 	switch trafficType {
-	case k8s.ServiceTypeHTTP:
+	case annotations.ServiceTypeHTTP:
 		p.buildHTTPServiceAndRoutersForTrafficSplit(t, cfg, tsKey, scheme, ts, tsSvc, middlewares)
 
-	case k8s.ServiceTypeTCP:
+	case annotations.ServiceTypeTCP:
 		p.buildTCPServiceAndRoutersForTrafficSplit(cfg, tsKey, ts, tsSvc)
 
-	case k8s.ServiceTypeUDP:
+	case annotations.ServiceTypeUDP:
 		p.buildUDPServiceAndRoutersForTrafficSplit(cfg, tsKey, ts, tsSvc)
 
 	default:
@@ -514,20 +518,6 @@ func (p Provider) buildUDPEntrypoint(svc *topology.Service, port int32) (string,
 	return fmt.Sprintf("udp-%d", meshPort), nil
 }
 
-func (p *Provider) getTrafficTypeAnnotation(svc *topology.Service) (string, error) {
-	trafficType, ok := svc.Annotations[k8s.AnnotationServiceType]
-
-	if !ok {
-		return p.config.DefaultTrafficType, nil
-	}
-
-	if trafficType != k8s.ServiceTypeHTTP && trafficType != k8s.ServiceTypeTCP && trafficType != k8s.ServiceTypeUDP {
-		return "", fmt.Errorf("traffic-type annotation references an unsupported traffic type %q", trafficType)
-	}
-
-	return trafficType, nil
-}
-
 func (p *Provider) buildHTTPServiceFromService(t *topology.Topology, svc *topology.Service, scheme string, port int32) *dynamic.Service {
 	var servers []dynamic.Server
 
@@ -716,20 +706,6 @@ func (p *Provider) buildWhitelistMiddlewareFromTrafficSplitIndirect(t *topology.
 	return whitelist
 }
 
-func getSchemeAnnotation(svc *topology.Service) (string, error) {
-	scheme, ok := svc.Annotations[k8s.AnnotationScheme]
-
-	if !ok {
-		return k8s.SchemeHTTP, nil
-	}
-
-	if scheme != k8s.SchemeHTTP && scheme != k8s.SchemeH2c && scheme != k8s.SchemeHTTPS {
-		return "", fmt.Errorf("scheme annotation references an unknown scheme %q", scheme)
-	}
-
-	return scheme, nil
-}
-
 func buildHTTPServiceFromTrafficSplit(backendSvc []dynamic.WRRService) *dynamic.Service {
 	return &dynamic.Service{
 		Weighted: &dynamic.WeightedRoundRobin{


Test patch:
diff --git a/pkg/annotations/annotations_test.go b/pkg/annotations/annotations_test.go
new file mode 100644
index 0000000..df07cce
--- /dev/null
+++ b/pkg/annotations/annotations_test.go
@@ -0,0 +1,203 @@
+package annotations_test
+
+import (
+	"testing"
+
+	"github.com/containous/maesh/pkg/annotations"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetTrafficType(t *testing.T) {
+	tests := []struct {
+		desc        string
+		annotations map[string]string
+		want        string
+		err         bool
+	}{
+		{
+			desc: "unknown service type",
+			annotations: map[string]string{
+				"maesh.containo.us/traffic-type": "hello",
+			},
+			err: true,
+		},
+		{
+			desc:        "returns the default traffic-type if not set",
+			annotations: map[string]string{},
+			want:        annotations.ServiceTypeHTTP,
+		},
+		{
+			desc: "http",
+			annotations: map[string]string{
+				"maesh.containo.us/traffic-type": "http",
+			},
+			want: annotations.ServiceTypeHTTP,
+		},
+		{
+			desc: "tcp",
+			annotations: map[string]string{
+				"maesh.containo.us/traffic-type": "tcp",
+			},
+			want: annotations.ServiceTypeTCP,
+		},
+		{
+			desc: "udp",
+			annotations: map[string]string{
+				"maesh.containo.us/traffic-type": "udp",
+			},
+			want: annotations.ServiceTypeUDP,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			tt, err := annotations.GetTrafficType(annotations.ServiceTypeHTTP, test.annotations)
+			if test.err {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, test.want, tt)
+			}
+		})
+	}
+}
+
+func TestGetScheme(t *testing.T) {
+	tests := []struct {
+		desc        string
+		annotations map[string]string
+		want        string
+		err         bool
+	}{
+		{
+			desc: "unknown service type",
+			annotations: map[string]string{
+				"maesh.containo.us/scheme": "hello",
+			},
+			err: true,
+		},
+		{
+			desc:        "returns the default scheme if not set",
+			annotations: map[string]string{},
+			want:        annotations.SchemeHTTP,
+		},
+		{
+			desc: "http",
+			annotations: map[string]string{
+				"maesh.containo.us/scheme": "http",
+			},
+			want: annotations.SchemeHTTP,
+		},
+		{
+			desc: "https",
+			annotations: map[string]string{
+				"maesh.containo.us/scheme": "https",
+			},
+			want: annotations.SchemeHTTPS,
+		},
+		{
+			desc: "h2c",
+			annotations: map[string]string{
+				"maesh.containo.us/scheme": "h2c",
+			},
+			want: annotations.SchemeH2C,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			tt, err := annotations.GetScheme(test.annotations)
+			if test.err {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, test.want, tt)
+			}
+		})
+	}
+}
+
+func TestGetRetryAttempts_Valid(t *testing.T) {
+	attempts, err := annotations.GetRetryAttempts(map[string]string{
+		"maesh.containo.us/retry-attempts": "2",
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, 2, attempts)
+}
+
+func TestGetRetryAttempts_NotSet(t *testing.T) {
+	_, err := annotations.GetRetryAttempts(map[string]string{})
+
+	assert.Equal(t, annotations.ErrNotFound, err)
+}
+
+func TestGetRetryAttempts_Invalid(t *testing.T) {
+	_, err := annotations.GetRetryAttempts(map[string]string{
+		"maesh.containo.us/retry-attempts": "hello",
+	})
+
+	assert.Error(t, err)
+}
+
+func TestGetCircuitBreakerExpression_Valid(t *testing.T) {
+	expression, err := annotations.GetCircuitBreakerExpression(map[string]string{
+		"maesh.containo.us/circuit-breaker-expression": "LatencyAtQuantileMS(50.0) > 100",
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, "LatencyAtQuantileMS(50.0) > 100", expression)
+}
+
+func TestGetCircuitBreakerExpression_NotSet(t *testing.T) {
+	_, err := annotations.GetCircuitBreakerExpression(map[string]string{})
+
+	assert.Equal(t, annotations.ErrNotFound, err)
+}
+
+func TestGetRateLimitBurst_Valid(t *testing.T) {
+	attempts, err := annotations.GetRateLimitBurst(map[string]string{
+		"maesh.containo.us/ratelimit-burst": "200",
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, 200, attempts)
+}
+
+func TestGetRateLimitBurst_NotSet(t *testing.T) {
+	_, err := annotations.GetRateLimitBurst(map[string]string{})
+
+	assert.Equal(t, annotations.ErrNotFound, err)
+}
+
+func TestGetRateLimitBurst_Invalid(t *testing.T) {
+	_, err := annotations.GetRateLimitBurst(map[string]string{
+		"maesh.containo.us/ratelimit-burst": "hello",
+	})
+
+	assert.Error(t, err)
+}
+
+func TestGetRateLimitAverage_Valid(t *testing.T) {
+	attempts, err := annotations.GetRateLimitAverage(map[string]string{
+		"maesh.containo.us/ratelimit-average": "100",
+	})
+
+	require.NoError(t, err)
+	assert.Equal(t, 100, attempts)
+}
+
+func TestGetRateLimitAverage_NotSet(t *testing.T) {
+	_, err := annotations.GetRateLimitAverage(map[string]string{})
+
+	assert.Equal(t, annotations.ErrNotFound, err)
+}
+
+func TestGetRateLimitAverage_Invalid(t *testing.T) {
+	_, err := annotations.GetRateLimitAverage(map[string]string{
+		"maesh.containo.us/ratelimit-average": "hello",
+	})
+
+	assert.Error(t, err)
+}
diff --git a/pkg/annotations/middleware_test.go b/pkg/annotations/middleware_test.go
new file mode 100644
index 0000000..1cdabe5
--- /dev/null
+++ b/pkg/annotations/middleware_test.go
@@ -0,0 +1,140 @@
+package annotations
+
+import (
+	"testing"
+
+	"github.com/containous/traefik/v2/pkg/config/dynamic"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildMiddleware(t *testing.T) {
+	tests := []struct {
+		desc        string
+		annotations map[string]string
+		want        map[string]*dynamic.Middleware
+		err         bool
+	}{
+		{
+			desc:        "nil when no middleware have been created",
+			annotations: map[string]string{},
+			want:        map[string]*dynamic.Middleware{},
+		},
+		{
+			desc: "retry-attempts annotation is valid",
+			annotations: map[string]string{
+				"maesh.containo.us/retry-attempts": "5",
+			},
+			want: map[string]*dynamic.Middleware{
+				"retry": {
+					Retry: &dynamic.Retry{
+						Attempts: 5,
+					},
+				},
+			},
+		},
+		{
+			desc: "retry-attempts annotation is invalid",
+			annotations: map[string]string{
+				"maesh.containo.us/retry-attempts": "hello",
+			},
+			err: true,
+		},
+		{
+			desc: "circuit-breaker-expression",
+			annotations: map[string]string{
+				"maesh.containo.us/circuit-breaker-expression": "LatencyAtQuantileMS(50.0) > 100",
+			},
+			want: map[string]*dynamic.Middleware{
+				"circuit-breaker": {
+					CircuitBreaker: &dynamic.CircuitBreaker{
+						Expression: "LatencyAtQuantileMS(50.0) > 100",
+					},
+				},
+			},
+		},
+		{
+			desc: "ratelimit-average and ratelimit-burst are both valid",
+			annotations: map[string]string{
+				"maesh.containo.us/ratelimit-average": "200",
+				"maesh.containo.us/ratelimit-burst":   "100",
+			},
+			want: map[string]*dynamic.Middleware{
+				"rate-limit": {
+					RateLimit: &dynamic.RateLimit{
+						Average: 200,
+						Burst:   100,
+					},
+				},
+			},
+		},
+		{
+			desc: "ratelimit-average is valid but ratelimit-burst is invalid",
+			annotations: map[string]string{
+				"maesh.containo.us/ratelimit-average": "200",
+				"maesh.containo.us/ratelimit-burst":   "hello",
+			},
+			err: true,
+		},
+		{
+			desc: "ratelimit-burst is valid but ratelimit-average is invalid",
+			annotations: map[string]string{
+				"maesh.containo.us/ratelimit-burst":   "200",
+				"maesh.containo.us/ratelimit-average": "hello",
+			},
+			err: true,
+		},
+		{
+			desc: "ratelimit-average is set but ratelimit-burst is not",
+			annotations: map[string]string{
+				"maesh.containo.us/ratelimit-average": "200",
+			},
+			want: map[string]*dynamic.Middleware{},
+		},
+		{
+			desc: "ratelimit-burst is set but ratelimit-average is not",
+			annotations: map[string]string{
+				"maesh.containo.us/ratelimit-burst": "200",
+			},
+			want: map[string]*dynamic.Middleware{},
+		},
+		{
+			desc: "multiple middlewares",
+			annotations: map[string]string{
+				"maesh.containo.us/retry-attempts":             "5",
+				"maesh.containo.us/ratelimit-average":          "200",
+				"maesh.containo.us/ratelimit-burst":            "100",
+				"maesh.containo.us/circuit-breaker-expression": "LatencyAtQuantileMS(50.0) > 100",
+			},
+			want: map[string]*dynamic.Middleware{
+				"retry": {
+					Retry: &dynamic.Retry{
+						Attempts: 5,
+					},
+				},
+				"rate-limit": {
+					RateLimit: &dynamic.RateLimit{
+						Average: 200,
+						Burst:   100,
+					},
+				},
+				"circuit-breaker": {
+					CircuitBreaker: &dynamic.CircuitBreaker{
+						Expression: "LatencyAtQuantileMS(50.0) > 100",
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			got, err := BuildMiddlewares(test.annotations)
+			if test.err {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, test.want, got)
+			}
+		})
+	}
+}
diff --git a/pkg/provider/provider_test.go b/pkg/provider/provider_test.go
index efddc06..d23e817 100644
--- a/pkg/provider/provider_test.go
+++ b/pkg/provider/provider_test.go
@@ -145,8 +145,11 @@ func TestProvider_BuildConfig(t *testing.T) {
 				p, ok := test.udpStateTable[port]
 				return p, ok
 			}
+			middlewareBuilder := func(a map[string]string) (map[string]*dynamic.Middleware, error) {
+				return nil, nil
+			}
 
-			p := provider.New(stateTableMock(tcpStateTable), stateTableMock(udpStateTable), cfg, logger)
+			p := provider.New(stateTableMock(tcpStateTable), stateTableMock(udpStateTable), middlewareBuilder, cfg, logger)
 
 			topo, err := loadTopology(test.topology)
 			require.NoError(t, err)