✨ add one instance of cache for all the service (#19)

* ✨ add one instance of cache for all the service

* 🐛 fix the healthy issue

* 🐛 fix lint

* 🍱 upgrade codebase

* 🍱 fix lint

* 🐛 when there is master node shut down

* Set live mode as default, clean docker and recipes

* Restore live mode in readme as default

* 🍱 add channel management

* 🍱 order funcs

* 🍱 revert service in docker compose

* ✨ logLevel added + cache put out the bouncer

* 🍱 fix

* 🍱 fix

* 🍱 fix

Co-authored-by: MathieuHa <mathieu@hanotaux.fr>

bouncer.go

@@ -7,16 +7,14 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"log"
-	"math/rand"
-	"net"
 	"net/http"
 	"net/url"
 	"text/template"
 	"time"
 
-	ttl_map "github.com/leprosus/golang-ttl-map"
-	ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/ip"
+	cache "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/cache"
+	ip "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/ip"
+	logger "github.com/maxlerebourg/crowdsec-bouncer-traefik-plugin/pkg/logger"
 )
 
 const (
@@ -26,28 +24,33 @@ const (
 	crowdsecLapiHeader      = "X-Api-Key"
 	crowdsecLapiRoute       = "v1/decisions"
 	crowdsecLapiStreamRoute = "v1/decisions/stream"
-	cacheBannedValue        = "t"
-	cacheNoBannedValue      = "f"
-	xForwardedFor           = "X-Forwarded-For"
+	cacheTimeoutKey         = "updated"
+)
+
+var (
+	crowdsecStreamHealthy = false
+	ticker                chan bool
 )
 
 // Config the plugin configuration.
 type Config struct {
 	Enabled                    bool     `json:"enabled,omitempty"`
+	LogLevel                   string   `json:"logLevel,omitempty"`
 	CrowdsecMode               string   `json:"crowdsecMode,omitempty"`
 	CrowdsecLapiScheme         string   `json:"crowdsecLapiScheme,omitempty"`
 	CrowdsecLapiHost           string   `json:"crowdsecLapiHost,omitempty"`
 	CrowdsecLapiKey            string   `json:"crowdsecLapiKey,omitempty"`
+	ForwardedHeadersCustomName string   `json:"forwardedheaderscustomheader,omitempty"`
 	UpdateIntervalSeconds      int64    `json:"updateIntervalSeconds,omitempty"`
 	DefaultDecisionSeconds     int64    `json:"defaultDecisionSeconds,omitempty"`
 	ForwardedHeadersTrustedIPs []string `json:"forwardedheaderstrustedips,omitempty"`
-	ForwardedHeadersCustomName string   `json:"forwardedheaderscustomheader,omitempty"`
 }
 
 // CreateConfig creates the default plugin configuration.
 func CreateConfig() *Config {
 	return &Config{
 		Enabled:                    false,
+		LogLevel:                   "INFO",
 		CrowdsecMode:               liveMode,
 		CrowdsecLapiScheme:         "http",
 		CrowdsecLapiHost:           "crowdsec:8080",
@@ -55,18 +58,17 @@ func CreateConfig() *Config {
 		UpdateIntervalSeconds:      60,
 		DefaultDecisionSeconds:     60,
 		ForwardedHeadersTrustedIPs: []string{},
-		ForwardedHeadersCustomName: xForwardedFor,
+		ForwardedHeadersCustomName: "X-Forwarded-For",
 	}
 }
 
-// Bouncer a Bouncer plugin.
+// Bouncer a Bouncer struct.
 type Bouncer struct {
 	next     http.Handler
 	name     string
 	template *template.Template
 
 	enabled                bool
-	crowdsecStreamHealthy  bool
 	crowdsecScheme         string
 	crowdsecHost           string
 	crowdsecKey            string
@@ -76,13 +78,14 @@ type Bouncer struct {
 	customHeader           string
 	poolStrategy           *ip.PoolStrategy
 	client                 *http.Client
-	cache                  *ttl_map.Heap
 }
 
 // New creates the crowdsec bouncer plugin.
 func New(ctx context.Context, next http.Handler, config *Config, name string) (http.Handler, error) {
+	logger.Init(config.LogLevel)
 	err := validateParams(config)
 	if err != nil {
+		logger.Info(fmt.Sprintf("%w", err))
 		return nil, err
 	}
 
@@ -94,7 +97,6 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
 		template: template.New("CrowdsecBouncer").Delims("[[", "]]"),
 
 		enabled:                config.Enabled,
-		crowdsecStreamHealthy:  false,
 		crowdsecMode:           config.CrowdsecMode,
 		crowdsecScheme:         config.CrowdsecLapiScheme,
 		crowdsecHost:           config.CrowdsecLapiHost,
@@ -112,18 +114,14 @@ func New(ctx context.Context, next http.Handler, config *Config, name string) (h
 			},
 			Timeout: 5 * time.Second,
 		},
-		cache: ttl_map.New(),
 	}
 	if config.CrowdsecMode == streamMode {
 		go func() {
-			rand.Seed(time.Now().UnixNano())
-			timeout := rand.Int63n(30)
-			logger(fmt.Sprintf("Wait: %v", timeout))
-			time.Sleep(time.Duration(timeout) * time.Second)
-			go handleStreamCache(bouncer)
-			ticker := time.NewTicker(time.Duration(config.UpdateIntervalSeconds) * time.Second)
-			for range ticker.C {
+			if ticker == nil {
 				go handleStreamCache(bouncer)
+				ticker = startTicker(config, func() {
+					handleStreamCache(bouncer)
+				})
 			}
 		}()
 	}
@@ -137,15 +135,18 @@ func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	remoteHost, err := getRemoteIP(bouncer, req)
+	remoteHost, err := ip.GetRemoteIP(req, bouncer.poolStrategy, bouncer.customHeader)
 	if err != nil {
+		logger.Info(fmt.Sprintf("%w", err))
 		bouncer.next.ServeHTTP(rw, req)
 		return
 	}
+	logger.Debug(fmt.Sprintf("ServeHTTP ip:%v", remoteHost))
 
 	if bouncer.crowdsecMode != noneMode {
-		isBanned, err := getDecision(bouncer.cache, remoteHost)
+		isBanned, err := cache.GetDecision(remoteHost)
 		if err == nil {
+			logger.Debug(fmt.Sprintf("ServeHTTP cacheHit isBanned:%v", isBanned))
 			if isBanned {
 				rw.WriteHeader(http.StatusForbidden)
 			} else {
@@ -157,7 +158,7 @@ func (bouncer *Bouncer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 
 	// Right here if we cannot join the stream we forbid the request to go on.
 	if bouncer.crowdsecMode == streamMode {
-		if bouncer.crowdsecStreamHealthy {
+		if crowdsecStreamHealthy {
 			bouncer.next.ServeHTTP(rw, req)
 		} else {
 			rw.WriteHeader(http.StatusForbidden)
@@ -188,10 +189,6 @@ type Stream struct {
 	New     []Decision `json:"new"`
 }
 
-func logger(str string) {
-	log.Printf("Crowdsec Bouncer Traefik Plugin - %s", str)
-}
-
 func contains(source []string, target string) bool {
 	for _, item := range source {
 		if item == target {
@@ -201,42 +198,21 @@ func contains(source []string, target string) bool {
 	return false
 }
 
-// It returns the first IP that is not in the pool, or the empty string otherwise.
-func getRemoteIP(bouncer *Bouncer, req *http.Request) (string, error) {
-	remoteIP := bouncer.poolStrategy.GetIP(req, bouncer.customHeader)
-	if len(remoteIP) != 0 {
-		return remoteIP, nil
-	}
-	remoteIP, _, err := net.SplitHostPort(req.RemoteAddr)
-	if err != nil {
-		logger(fmt.Sprintf("failed to extract ip from remote address: %v", err))
-		return "", err
-	}
-	return remoteIP, nil
-}
-
-// Get Decision check in the cache if the IP has the banned / not banned value.
-// Otherwise return with an error to add the IP in cache if we are on.
-func getDecision(cache *ttl_map.Heap, clientIP string) (bool, error) {
-	banned, isCached := cache.Get(clientIP)
-	bannedString, isValid := banned.(string)
-	if isCached && isValid && len(bannedString) > 0 {
-		return bannedString == cacheBannedValue, nil
-	}
-	return false, fmt.Errorf("no cache data")
-}
-
-func setDecision(cache *ttl_map.Heap, clientIP string, isBanned bool, duration int64) {
-	if isBanned {
-		logger(fmt.Sprintf("%v banned", clientIP))
-		cache.Set(clientIP, cacheBannedValue, duration)
-	} else {
-		cache.Set(clientIP, cacheNoBannedValue, duration)
-	}
-}
-
-func deleteDecision(cache *ttl_map.Heap, clientIP string) {
-	cache.Del(clientIP)
+func startTicker(config *Config, work func()) chan bool {
+	ticker := time.NewTicker(time.Duration(config.UpdateIntervalSeconds) * time.Second)
+	stop := make(chan bool, 1)
+	go func() {
+		defer logger.Debug("ticker:stopped")
+		for {
+			select {
+			case <-ticker.C:
+				go work()
+			case <-stop:
+				return
+			}
+		}
+	}()
+	return stop
 }
 
 // We are now in none or live mode.
@@ -247,26 +223,31 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req
 		Path:     crowdsecLapiRoute,
 		RawQuery: fmt.Sprintf("ip=%v&banned=true", remoteHost),
 	}
-	body := crowdsecQuery(bouncer, routeURL.String())
+	body, err := crowdsecQuery(bouncer, routeURL.String())
+	if err != nil {
+		logger.Info(fmt.Sprintf("%w", err))
+		rw.WriteHeader(http.StatusForbidden)
+		return
+	}
 
 	if bytes.Equal(body, []byte("null")) {
 		if bouncer.crowdsecMode == liveMode {
-			setDecision(bouncer.cache, remoteHost, false, bouncer.defaultDecisionTimeout)
+			cache.SetDecision(remoteHost, false, bouncer.defaultDecisionTimeout)
 		}
 		bouncer.next.ServeHTTP(rw, req)
 		return
 	}
 
 	var decisions []Decision
-	err := json.Unmarshal(body, &decisions)
+	err = json.Unmarshal(body, &decisions)
 	if err != nil {
-		logger(fmt.Sprintf("failed to parse body: %s", err))
+		logger.Info(fmt.Sprintf("failed to parse body: %s", err))
 		rw.WriteHeader(http.StatusForbidden)
 		return
 	}
 	if len(decisions) == 0 {
 		if bouncer.crowdsecMode == liveMode {
-			setDecision(bouncer.cache, remoteHost, false, bouncer.defaultDecisionTimeout)
+			cache.SetDecision(remoteHost, false, bouncer.defaultDecisionTimeout)
 		}
 		bouncer.next.ServeHTTP(rw, req)
 		return
@@ -274,70 +255,78 @@ func handleNoStreamCache(bouncer *Bouncer, rw http.ResponseWriter, req *http.Req
 	rw.WriteHeader(http.StatusForbidden)
 	duration, err := time.ParseDuration(decisions[0].Duration)
 	if err != nil {
-		logger(fmt.Sprintf("failed to parse duration: %s", err))
+		logger.Info(fmt.Sprintf("failed to parse duration: %s", err))
 		return
 	}
 	if bouncer.crowdsecMode == liveMode {
-		setDecision(bouncer.cache, remoteHost, true, int64(duration.Seconds()))
+		cache.SetDecision(remoteHost, true, int64(duration.Seconds()))
 	}
 }
 
 func handleStreamCache(bouncer *Bouncer) {
 	// TODO clean properly on exit.
+	// Instead of blocking the goroutine interval for all the secondary node,
+	// if the master service is shut down, other goroutine can take the lead
+	// because updated routine information is in the cache
+	logger.Debug("handleStreamCache")
+	_, err := cache.GetDecision(cacheTimeoutKey)
+	if err == nil {
+		return
+	}
+	cache.SetDecision(cacheTimeoutKey, false, bouncer.updateInterval-1)
 	streamRouteURL := url.URL{
 		Scheme:   bouncer.crowdsecScheme,
 		Host:     bouncer.crowdsecHost,
 		Path:     crowdsecLapiStreamRoute,
-		RawQuery: fmt.Sprintf("startup=%t", !bouncer.crowdsecStreamHealthy),
+		RawQuery: fmt.Sprintf("startup=%t", !crowdsecStreamHealthy),
 	}
-	body := crowdsecQuery(bouncer, streamRouteURL.String())
-	var stream Stream
-	err := json.Unmarshal(body, &stream)
+	body, err := crowdsecQuery(bouncer, streamRouteURL.String())
 	if err != nil {
-		logger(fmt.Sprintf("error while parsing body: %s", err))
-		bouncer.crowdsecStreamHealthy = false
+		logger.Info(fmt.Sprintf("%w", err))
+		crowdsecStreamHealthy = false
+		return
+	}
+	var stream Stream
+	err = json.Unmarshal(body, &stream)
+	if err != nil {
+		logger.Info(fmt.Sprintf("error while parsing body: %s", err))
+		crowdsecStreamHealthy = false
 		return
 	}
 	for _, decision := range stream.New {
 		duration, err := time.ParseDuration(decision.Duration)
 		if err == nil {
-			setDecision(bouncer.cache, decision.Value, true, int64(duration.Seconds()))
+			cache.SetDecision(decision.Value, true, int64(duration.Seconds()))
 		}
 	}
 	for _, decision := range stream.Deleted {
-		deleteDecision(bouncer.cache, decision.Value)
+		cache.DeleteDecision(decision.Value)
 	}
-	bouncer.crowdsecStreamHealthy = true
+	crowdsecStreamHealthy = true
 }
 
-func crowdsecQuery(bouncer *Bouncer, stringURL string) []byte {
+func crowdsecQuery(bouncer *Bouncer, stringURL string) ([]byte, error) {
 	var req *http.Request
 	req, _ = http.NewRequest(http.MethodGet, stringURL, nil)
 	req.Header.Add(crowdsecLapiHeader, bouncer.crowdsecKey)
 	res, err := bouncer.client.Do(req)
 	if err != nil {
-		logger(fmt.Sprintf("error while fetching %v: %s", stringURL, err))
-		bouncer.crowdsecStreamHealthy = false
-		return nil
+		return nil, fmt.Errorf("error while fetching %v: %s", stringURL, err)
 	}
 	if res.StatusCode != http.StatusOK {
-		logger(fmt.Sprintf("error while fetching %v, status code: %d", stringURL, res.StatusCode))
-		bouncer.crowdsecStreamHealthy = false
-		return nil
+		return nil, fmt.Errorf("error while fetching %v, status code: %d", stringURL, res.StatusCode)
 	}
 	defer func(body io.ReadCloser) {
 		err = body.Close()
 		if err != nil {
-			logger(fmt.Sprintf("failed to close body reader: %s", err))
+			logger.Info(fmt.Sprintf("failed to close body reader: %s", err))
 		}
 	}(res.Body)
 	body, err := ioutil.ReadAll(res.Body)
 	if err != nil {
-		logger(fmt.Sprintf("error while reading body: %s", err))
-		bouncer.crowdsecStreamHealthy = false
-		return nil
+		return nil, fmt.Errorf("error while reading body: %s", err)
 	}
-	return body
+	return body, nil
 }
 
 func validateParams(config *Config) error {
@@ -378,10 +367,10 @@ func validateParams(config *Config) error {
 	if len(config.ForwardedHeadersTrustedIPs) > 0 {
 		_, err = ip.NewChecker(config.ForwardedHeadersTrustedIPs)
 		if err != nil {
-			return fmt.Errorf("ForwardedHeadersTrustedIPs must be a list of IP/CIDR :%v", err)
+			return fmt.Errorf("ForwardedHeadersTrustedIPs must be a list of IP/CIDR :%w", err)
 		}
 	} else {
-		logger("No IP provided for ForwardedHeadersTrustedIPs")
+		logger.Debug("No IP provided for ForwardedHeadersTrustedIPs")
 	}
 
 	return nil