Implemented healthz endpoint in http and grpc

benchmark_test.go

@@ -63,10 +63,10 @@ func BenchmarkServer_NoOp(b *testing.B) {
 	//total := time.Second / dur
 	//fmt.Printf("Total: %d\n", total)
 
-	b.Run("Ping", func(b *testing.B) {
+	b.Run("HealthCheck", func(b *testing.B) {
 		for n := 0; n < b.N; n++ {
 			if err := client.Ping(context.Background()); err != nil {
-				b.Errorf("client.Ping() err: %s", err)
+				b.Errorf("client.HealthCheck() err: %s", err)
 			}
 		}
 	})

cache/lru.go

@@ -20,12 +20,13 @@ package cache
 
 import (
 	"container/list"
-	"github.com/mailgun/gubernator/pb"
-	"github.com/sirupsen/logrus"
 	"sync"
 	"time"
 
+	"github.com/mailgun/gubernator/pb"
 	"github.com/mailgun/holster"
+	"github.com/mailgun/holster/clock"
+	"github.com/sirupsen/logrus"
 )
 
 type LRUCacheConfig struct {
@@ -41,6 +42,9 @@ type LRUCacheConfig struct {
 
 	// The initial cache size. If not provided defaults to 30% of the max cache size.
 	InitialCacheSize int
+
+	// Interval at which the cache should check if it needs to shrink or grow
+	InspectInterval clock.DurationJSON
 }
 
 // Cache is an thread unsafe LRU cache that supports expiration
@@ -66,6 +70,8 @@ func NewLRUCache(conf LRUCacheConfig) *LRUCache {
 	holster.SetDefault(&conf.MaxCacheSize, 50000)
 	// If not provided init cache with 30 percent of the max cache size
 	holster.SetDefault(&conf.InitialCacheSize, int(float32(conf.MaxCacheSize)*0.30))
+	// Inspect the cache for possible resize every 30 seconds
+	holster.SetDefault(&conf.InspectInterval.Duration, time.Second*30)
 
 	return &LRUCache{
 		log:       logrus.WithField("category", "lru-cache"),
@@ -82,23 +88,18 @@ func (c *LRUCache) inspectAndResize() {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
-	// If we have NOT reached the size of our cache
-	if c.cacheSize != c.Size() {
-		return
-	}
-
 	// Inspect the bottom 20% of the cache for expired items
 	inspectSize := int(float32(c.cacheSize) * 0.20)
 	ele := c.ll.Back()
 	if ele == nil {
-		// Not sure how this would happened
+		// return if the cache is empty
 		return
 	}
 
 	var prev *list.Element
 	var count, expired = 0, 0
 	for {
-		if count == inspectSize {
+		if count == inspectSize || ele == nil {
 			break
 		}
 
@@ -119,8 +120,7 @@ func (c *LRUCache) inspectAndResize() {
 
 	// If all the elements expired, we can shrink the cache size
 	if expired == inspectSize {
-		// TODO: Will never be called, since this code doesn't execute unless the cache is at capacity
-		// Increase the cache size by 30%
+		// Decrease the cache size by 30%
 		newSize := c.cacheSize - int(float32(c.cacheSize)*0.30)
 		// Don't shrink beyond the initial cache size
 		if newSize < c.conf.InitialCacheSize {
@@ -148,7 +148,7 @@ func (c *LRUCache) inspectAndResize() {
 }
 
 func (c *LRUCache) Start() error {
-	tick := time.NewTicker(time.Second * 5)
+	tick := time.NewTicker(c.conf.InspectInterval.Duration)
 	c.wg.Until(func(done chan struct{}) bool {
 		select {
 		case <-tick.C:

client.go

@@ -73,7 +73,7 @@ func (c *Client) GetClient() pb.RateLimitServiceClient {
 }
 
 func (c *Client) Ping(ctx context.Context) error {
-	_, err := c.client.Ping(ctx, &pb.PingRequest{})
+	_, err := c.client.HealthCheck(ctx, &pb.HealthCheckRequest{})
 	return err
 }
 

cmd/gubernator-server/main.go

@@ -2,13 +2,16 @@ package main
 
 import (
 	"context"
+	"net/http"
 	"os"
 
 	"github.com/mailgun/gubernator"
 	"github.com/mailgun/gubernator/cache"
 	"github.com/mailgun/gubernator/metrics"
+	"github.com/mailgun/gubernator/pb"
 	"github.com/mailgun/gubernator/sync"
 	"github.com/mailgun/service"
+	"github.com/mailgun/service/httpapi"
 )
 
 var Version = "dev-build"
@@ -43,6 +46,18 @@ func (s *Service) Start(ctx context.Context) error {
 		return err
 	}
 
+	err = s.AddHandler(httpapi.Spec{
+		Method: "GET",
+		Path:   "/healthz",
+		Scope:  httpapi.ScopeProtected,
+		Handler: func(w http.ResponseWriter, r *http.Request, p map[string]string) (interface{}, error) {
+			return s.srv.HealthCheck(r.Context(), &pb.HealthCheckRequest{})
+		},
+	})
+	if err != nil {
+		return err
+	}
+
 	return s.srv.Start()
 }
 

cmd/gubernator/main.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"time"
 
-	"github.com/davecgh/go-spew/spew"
 	"github.com/mailgun/gubernator"
 	"github.com/mailgun/holster"
 )
@@ -47,12 +46,12 @@ func main() {
 			fan.Run(func(obj interface{}) error {
 				r := obj.(*gubernator.Request)
 				// Now hit our cluster with the rate limits
-				resp, err := client.GetRateLimit(context.Background(), r)
+				_, err := client.GetRateLimit(context.Background(), r)
 				checkErr(err)
 
-				if resp.Status == gubernator.OverLimit {
+				/*if resp.Status == gubernator.OverLimit {
 					spew.Dump(resp)
-				}
+				}*/
 				return nil
 			}, rateLimit)
 		}

config.yaml

@@ -29,10 +29,15 @@ AdvertiseAddress: localhost:9040
 LRUCache:
   # Max size of the cache; The cache size will never grow beyond
   # this size.
-  MaxCacheSize: 100
+  MaxCacheSize: 100000
   # The initial size of the cache; The cache may grow or shrink
   # depending on the number of rate limits the server is handling
   # currently. However the cache will never shrink lower than the
   # initial size.
-  InitialCacheSize: 50
+  InitialCacheSize: 10000
+  # Interval at which gubernator should check if the cache size
+  # needs to shrink or grow. Lower the inspection interval to
+  # make the cache more responsive to high volume bursts, raise
+  # the interval to conserve more memory during bursts.
+  InspectInterval: 30s
 

etc/server-1.yaml

@@ -23,5 +23,5 @@ statsd:
 # Gubernator config
 ListenAddress: localhost:9040
 LRUCache:
-  MaxCacheSize: 1000
+  MaxCacheSize: 3000
   InitialCacheSize: 50

etc/server-2.yaml

@@ -23,5 +23,5 @@ statsd:
 # Gubernator config
 ListenAddress: localhost:9041
 LRUCache:
-  MaxCacheSize: 1000
+  MaxCacheSize: 3000
   InitialCacheSize: 50

etc/server-3.yaml

@@ -23,5 +23,5 @@ statsd:
 # Gubernator config
 ListenAddress: localhost:9042
 LRUCache:
-  MaxCacheSize: 1000
+  MaxCacheSize: 3000
   InitialCacheSize: 50

functional_test.go

@@ -144,7 +144,6 @@ func TestTokenBucket(t *testing.T) {
 	}
 }
 
-// TODO: This test is very time (clock) sensitive, We could ignore the number remaining and increase the limit duration
 func TestLeakyBucket(t *testing.T) {
 	client, errs := gubernator.NewClient(gubernator.RandomPeer(peers))
 	require.Nil(t, errs)