tests: fix e2e tests

cmd/guardian/main.go

@@ -45,6 +45,7 @@ func main() {
 	synchronous := kingpin.Flag("synchronous", "synchronously enforce ratelimit").Default("false").OverrideDefaultFromEnvar("GUARDIAN_FLAG_SYNCHRONOUS").Bool()
 	initConfig := kingpin.Flag("init", "create missing configuration on startup if needed").Default("false").Envar("GUARDIAN_FLAG_INIT_CONFIG").Bool()
 	prisonerCacheMaxSize := kingpin.Flag("prisoner-cache-max-size", "max size of the cache of prisoner that each Guardian instance maintains").Default("1000").Envar("GUARDIAN_FLAG_PRISONER_CACHE_MAX_SIZE").Uint16()
+	internalCacheServerAddress := kingpin.Flag("internal-cache-server-address", "network address for internal cache server address").Default("").Envar("GUARDIAN_FLAG_INTERNAL_CACHE_SERVER_ADDRESS").String()
 
 	kingpin.Parse()
 
@@ -115,9 +116,20 @@ func main() {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
+		// TODO: Make prune interval configurable through the CLI
 		redisCounter.Run(30*time.Second, stop)
 	}()
 
+	if *internalCacheServerAddress != "" {
+		cacheServer := guardian.NewCacheServer(*internalCacheServerAddress, redisCounter)
+		go func() {
+			csErr := cacheServer.Run(stop)
+			if csErr != nil {
+				logger.WithError(csErr).Errorf("cache server ended with an error")
+			}
+		}()
+	}
+
 	whitelister := guardian.NewIPWhitelister(redisConfStore, logger.WithField("context", "ip-whitelister"), reporter)
 	blacklister := guardian.NewIPBlacklister(redisConfStore, logger.WithField("context", "ip-blacklister"), reporter)
 	jailer := guardian.NewGenericJailer(redisConfStore, logger.WithField("context", "route-jailer"), redisCounter, redisConfStore, reporter)

e2e/docker-compose-async.yml

@@ -10,11 +10,13 @@ services:
     ports:
     - "3000:3000"
     - "6060:6060"
+    - "4343:4343"
     environment:
       GUARDIAN_FLAG_REDIS_ADDRESS: "redis:6379"
       GUARDIAN_FLAG_LOG_LEVEL: "debug"
       GUARDIAN_FLAG_CONF_UPDATE_INTERVAL: "500ms"
       GUARDIAN_FLAG_SYNCHRONOUS: "false"
+      GUARDIAN_FLAG_INTERNAL_CACHE_SERVER_ADDRESS: "0.0.0.0:4343"
     depends_on:
     - redis
   upstream:

e2e/docker-compose-sync.yml

@@ -9,11 +9,13 @@ services:
     ports:
     - "3000:3000"
     - "6060:6060"
+    - "4343:4343"
     environment:
       GUARDIAN_FLAG_REDIS_ADDRESS: "redis:6379"
       GUARDIAN_FLAG_LOG_LEVEL: "debug"
       GUARDIAN_FLAG_CONF_UPDATE_INTERVAL: "500ms"
       GUARDIAN_FLAG_SYNCHRONOUS: "true"
+      GUARDIAN_FLAG_INTERNAL_CACHE_SERVER_ADDRESS: "0.0.0.0:4343"
     depends_on:
     - redis
   upstream:

e2e/e2e_test.go

@@ -26,11 +26,12 @@ import (
 
 var redisAddr = flag.String("redis-addr", "localhost:6379", "redis address")
 var envoyAddr = flag.String("envoy-addr", "localhost:8080", "envoy address")
+var internalCacheAddr = flag.String("internal-cache-addr", "localhost:4343", "internal cache address")
 
-const defaultAsyncCounterTimeout = 300 * time.Millisecond
+const defaultAsyncCounterTimeout = 25 * time.Millisecond
 
 func TestWhitelist(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	IP := pseudoRandomIPV4Address()
 	CIDR := fmt.Sprintf("%v/32", IP)
 	config := guardianConfig{
@@ -53,7 +54,7 @@ func TestWhitelist(t *testing.T) {
 }
 
 func TestBlacklist(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	IP := pseudoRandomIPV4Address()
 	CIDR := fmt.Sprintf("%v/32", IP)
 	config := guardianConfig{
@@ -76,7 +77,7 @@ func TestBlacklist(t *testing.T) {
 }
 
 func TestGlobalRateLimit(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	IP := pseudoRandomIPV4Address()
 	guardianConfig := guardianConfig{
 		whitelist:                 []string{},
@@ -98,12 +99,11 @@ func TestGlobalRateLimit(t *testing.T) {
 	}
 
 	for i := uint64(0); i < 10; i++ {
-		if len(os.Getenv("SYNC")) == 0 {
-			time.Sleep(defaultAsyncCounterTimeout) // helps prevents races due asynchronous rate limiting
-		}
-
 		res := GET(t, IP, "/")
 		res.Body.Close()
+		if len(os.Getenv("SYNC")) == 0 {
+			time.Sleep(defaultAsyncCounterTimeout) // helps prevent races due to asynchronous rate limiting
+		}
 
 		want := 200
 		if i >= config.Spec.Limit.Count {
@@ -117,7 +117,7 @@ func TestGlobalRateLimit(t *testing.T) {
 }
 
 func TestRateLimit(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	IP := pseudoRandomIPV4Address()
 	guardianConfig := guardianConfig{
 		whitelist:                 []string{},
@@ -143,13 +143,11 @@ func TestRateLimit(t *testing.T) {
 			t.Fatalf("error decoding yaml: %v", err)
 		}
 		for i := uint64(0); i < config.Spec.Limit.Count+5; i++ {
-			if len(os.Getenv("SYNC")) == 0 {
-				time.Sleep(defaultAsyncCounterTimeout) // helps prevents races due asynchronous rate limiting
-			}
-
 			res := GET(t, IP, config.Spec.Conditions.Path)
 			res.Body.Close()
-
+			if len(os.Getenv("SYNC")) == 0 {
+				time.Sleep(defaultAsyncCounterTimeout) // helps prevent races due to asynchronous rate limiting
+			}
 			want := 200
 			if i >= config.Spec.Limit.Count && config.Spec.Limit.Enabled {
 				want = 429
@@ -163,7 +161,7 @@ func TestRateLimit(t *testing.T) {
 }
 
 func TestJails(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	whitelistedIP := "192.168.1.1"
 
 	guardianConfig := guardianConfig{
@@ -195,13 +193,9 @@ func TestJails(t *testing.T) {
 			t.Fatalf("error decoding yaml: %v", err)
 		}
 		banned := false
-		resetRedis(*redisAddr)
+		resetStores(t, *redisAddr, *internalCacheAddr)
 		applyGuardianConfig(t, *redisAddr, guardianConfig)
 		for i := 0; uint64(i) <= config.Spec.Limit.Count; i++ {
-			if os.Getenv("SYNC") != "" {
-				time.Sleep(defaultAsyncCounterTimeout) // helps prevents races due asynchronous rate limiting
-			}
-
 			res := GET(t, "192.168.1.43", config.Spec.Conditions.Path)
 			whitelistedRes := GET(t, whitelistedIP, config.Spec.Conditions.Path)
 			res.Body.Close()
@@ -234,7 +228,7 @@ func TestJails(t *testing.T) {
 }
 
 func TestDeleteRateLimit(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 
 	config := guardianConfig{
 		whitelist:                 []string{},
@@ -257,7 +251,7 @@ func TestDeleteRateLimit(t *testing.T) {
 }
 
 func TestDeleteJail(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 
 	config := guardianConfig{
 		whitelist:                 []string{},
@@ -280,7 +274,7 @@ func TestDeleteJail(t *testing.T) {
 }
 
 func TestRateLimitDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	IP := pseudoRandomIPV4Address()
 	config := guardianConfig{
 		whitelist:               []string{},
@@ -293,13 +287,11 @@ func TestRateLimitDeprecated(t *testing.T) {
 	applyGuardianConfigDeprecated(t, *redisAddr, config)
 
 	for i := 0; i < 10; i++ {
-		if len(os.Getenv("SYNC")) == 0 {
-			time.Sleep(defaultAsyncCounterTimeout) // helps prevents races due asynchronous rate limiting
-		}
-
 		res := GET(t, IP, "/")
 		res.Body.Close()
-
+		if len(os.Getenv("SYNC")) == 0 {
+			time.Sleep(defaultAsyncCounterTimeout) // helps prevent races due to asynchronous rate limiting
+		}
 		want := 200
 		if i >= config.limitCountDeprecated {
 			want = 429
@@ -312,7 +304,7 @@ func TestRateLimitDeprecated(t *testing.T) {
 }
 
 func TestRouteRateLimitDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	IP := pseudoRandomIPV4Address()
 	config := guardianConfig{
 		whitelist:                          []string{},
@@ -337,13 +329,11 @@ func TestRouteRateLimitDeprecated(t *testing.T) {
 
 	for _, routeRateLimit := range rrlConfig.RouteRateLimits {
 		for i := uint64(0); i < routeRateLimit.Limit.Count+5; i++ {
-			if len(os.Getenv("SYNC")) == 0 {
-				time.Sleep(defaultAsyncCounterTimeout) // helps prevents races due asynchronous rate limiting
-			}
-
 			res := GET(t, IP, routeRateLimit.Route)
 			res.Body.Close()
-
+			if len(os.Getenv("SYNC")) == 0 {
+				time.Sleep(defaultAsyncCounterTimeout) // helps prevent races due to asynchronous rate limiting
+			}
 			want := 200
 			if i >= routeRateLimit.Limit.Count && routeRateLimit.Limit.Enabled {
 				want = 429
@@ -357,7 +347,7 @@ func TestRouteRateLimitDeprecated(t *testing.T) {
 }
 
 func TestJailsDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	whitelistedIP := "192.168.1.1"
 
 	config := guardianConfig{
@@ -385,18 +375,16 @@ func TestJailsDeprecated(t *testing.T) {
 	// to execute this particular test.
 	for _, j := range jailConfig.Jails {
 		banned := false
-		resetRedis(*redisAddr)
+		resetStores(t, *redisAddr, *internalCacheAddr)
 		applyGuardianConfigDeprecated(t, *redisAddr, config)
 		for i := uint64(0); i < j.Jail.Limit.Count+1; i++ {
-			if len(os.Getenv("SYNC")) == 0 {
-				time.Sleep(defaultAsyncCounterTimeout) // helps prevents races due asynchronous rate limiting
-			}
-
 			res := GET(t, "192.168.1.43", j.Route)
 			whitelistedRes := GET(t, whitelistedIP, j.Route)
 			res.Body.Close()
 			whitelistedRes.Body.Close()
-
+			if len(os.Getenv("SYNC")) == 0 {
+				time.Sleep(defaultAsyncCounterTimeout) // helps prevent races due to asynchronous rate limiting
+			}
 			want := 200
 			if (i >= j.Jail.Limit.Count && j.Jail.Limit.Enabled) || banned {
 				banned = true
@@ -440,7 +428,7 @@ func routeRateLimitsEqual(rrl1, rrl2 []guardian.RouteRateLimitConfigEntryDepreca
 }
 
 func TestSetRouteRateLimitsDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	config := guardianConfig{
 		whitelist:                          []string{},
 		blacklist:                          []string{},
@@ -477,7 +465,7 @@ func TestSetRouteRateLimitsDeprecated(t *testing.T) {
 }
 
 func TestRemoveRouteRateLimitsDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	config := guardianConfig{
 		whitelist:                          []string{},
 		blacklist:                          []string{},
@@ -506,7 +494,7 @@ func TestRemoveRouteRateLimitsDeprecated(t *testing.T) {
 }
 
 func TestSetJailsDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 	config := guardianConfig{
 		whitelist:                []string{},
 		blacklist:                []string{},
@@ -551,7 +539,7 @@ func TestSetJailsDeprecated(t *testing.T) {
 }
 
 func TestRemoveJailDeprecated(t *testing.T) {
-	resetRedis(*redisAddr)
+	resetStores(t, *redisAddr, *internalCacheAddr)
 
 	config := guardianConfig{
 		whitelist:                []string{},
@@ -609,6 +597,28 @@ var currentRedisDBIndex = redisDBIndex{
 	Index: 0,
 }
 
+func resetStores(t *testing.T, redisAddress string, internalCacheAddress string) {
+	t.Helper()
+	resetCounters(t, internalCacheAddress)
+	resetRedis(redisAddress)
+}
+
+func resetCounters(t *testing.T, address string) {
+	t.Helper()
+	client := &http.Client{
+		Timeout: 5 * time.Second,
+	}
+	req, _ := http.NewRequest("DELETE", "http://"+address+"/v0/counters", nil)
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Logf("error resetting internal cache: %v", err)
+		return
+	}
+	if resp.StatusCode != 200 {
+		t.Logf("error response from internal cache server: %v", resp.StatusCode)
+	}
+}
+
 func resetRedis(redisAddr string) {
 	currentRedisDBIndex.Lock()
 	redisOpts := &redis.Options{

pkg/guardian/cach_server_test.go

@@ -0,0 +1,49 @@
+package guardian
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestCacheServerDeleteCounters(t *testing.T) {
+	c, s := newTestRedisCounter(t)
+	defer s.Close()
+	cacheServer := NewCacheServer("", c)
+	testServer := httptest.NewServer(cacheServer.server.Handler)
+	defer testServer.Close()
+
+	key := "test_key"
+	incrBy := uint(5)
+	expire := 1 * time.Second
+	maxBlock := uint64(15)
+
+	c.Incr(context.Background(), key, incrBy, maxBlock, expire)
+	c.Incr(context.Background(), key, incrBy, maxBlock, expire)
+
+	// sleep required because newTestRedisCounter returns an async counter
+	time.Sleep(100 * time.Millisecond)
+
+	req, err := http.NewRequest("DELETE", testServer.URL+"/v0/counters", nil)
+	if err != nil {
+		t.Fatalf("could not create request: %v", err)
+	}
+	client := &http.Client{Timeout: time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatalf("error from cache server: %v", err)
+	}
+
+	if resp.StatusCode != 200 {
+		t.Fatalf("unexpected status code from cache server: %v", resp.StatusCode)
+	}
+
+	c.cache.RLock()
+	defer c.cache.RUnlock()
+	_, ok := c.cache.m[key]
+	if ok {
+		t.Fatalf("key exists in cache when it should not")
+	}
+}