Add hook to GetRateLimit

.github/workflows/lint.yaml

@@ -6,7 +6,7 @@ on:
   pull_request:
 
 env:
-  GOLANGCI_LINT_VERSION: v1.56.2
+  GOLANGCI_LINT_VERSION: v1.61.0
 
 jobs:
   lint:
@@ -20,7 +20,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Go
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version-file: ./go.mod
           cache: true # caching and restoring go modules and build outputs
@@ -29,7 +29,7 @@ jobs:
         run: go mod tidy && git diff --exit-code
 
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v6
         with:
           version: ${{ env.GOLANGCI_LINT_VERSION }}
           skip-cache: true # cache/restore is done by actions/setup-go@v3 step

.github/workflows/on-pull-request.yml

@@ -4,9 +4,6 @@ on:
   pull_request:
     branches: [ master ]
 
-env:
-  GOLANGCI_LINT_VERSION: v1.56.2
-
 jobs:
   test:
     name: test
@@ -40,14 +37,9 @@ jobs:
       - name: Install deps
         run: go mod download
 
-      - name: golangci-lint
-        uses: golangci/golangci-lint-action@v4
-        with:
-          version: ${{ env.GOLANGCI_LINT_VERSION }}
-          skip-cache: true
-
       - name: Test
         run: go test -v -race -p=1 -count=1 -tags holster_test_mode
+
   go-bench:
     runs-on: ubuntu-latest
     timeout-minutes: 30

Makefile

@@ -2,7 +2,7 @@
 VERSION=$(shell cat version)
 LDFLAGS="-X main.Version=$(VERSION)"
 GOLANGCI_LINT = $(GOPATH)/bin/golangci-lint
-GOLANGCI_LINT_VERSION = 1.56.2
+GOLANGCI_LINT_VERSION = v1.61.0
 
 .PHONY: help
 help:

cluster/cluster.go

@@ -148,10 +148,10 @@ func Restart(ctx context.Context) error {
 }
 
 // StartWith a local cluster with specific addresses
-func StartWith(localPeers []gubernator.PeerInfo) error {
+func StartWith(localPeers []gubernator.PeerInfo, opts ...option) error {
 	for _, peer := range localPeers {
 		ctx, cancel := context.WithTimeout(context.Background(), clock.Second*10)
-		d, err := gubernator.SpawnDaemon(ctx, gubernator.DaemonConfig{
+		cfg := gubernator.DaemonConfig{
 			Logger:            logrus.WithField("instance", peer.GRPCAddress),
 			InstanceID:        peer.GRPCAddress,
 			GRPCListenAddress: peer.GRPCAddress,
@@ -163,7 +163,11 @@ func StartWith(localPeers []gubernator.PeerInfo) error {
 				GlobalTimeout:  clock.Second * 5,
 				BatchTimeout:   clock.Second * 5,
 			},
-		})
+		}
+		for _, opt := range opts {
+			opt.Apply(&cfg)
+		}
+		d, err := gubernator.SpawnDaemon(ctx, cfg)
 		cancel()
 		if err != nil {
 			return errors.Wrapf(err, "while starting server for addr '%s'", peer.GRPCAddress)
@@ -196,3 +200,20 @@ func Stop() {
 	peers = nil
 	daemons = nil
 }
+
+type option interface {
+	Apply(cfg *gubernator.DaemonConfig)
+}
+
+type eventChannelOption struct {
+	eventChannel chan<- gubernator.HitEvent
+}
+
+func (o *eventChannelOption) Apply(cfg *gubernator.DaemonConfig) {
+	cfg.EventChannel = o.eventChannel
+}
+
+// WithEventChannel sets EventChannel to Gubernator config.
+func WithEventChannel(eventChannel chan<- gubernator.HitEvent) option {
+	return &eventChannelOption{eventChannel: eventChannel}
+}

cmd/gubernator-cli/main.go

@@ -170,7 +170,7 @@ func min(a, b int) int {
 
 func checkErr(err error) {
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatal(err.Error())
 	}
 }
 

config.go

@@ -119,6 +119,14 @@ type Config struct {
 
 	// (Optional) The total size of the cache used to store rate limits. Defaults to 50,000
 	CacheSize int
+
+	// (Optional) EventChannel receives hit events
+	EventChannel chan<- HitEvent
+}
+
+type HitEvent struct {
+	Request  *RateLimitReq
+	Response *RateLimitResp
 }
 
 func (c *Config) SetDefaults() error {
@@ -248,6 +256,9 @@ type DaemonConfig struct {
 	// (Optional) TraceLevel sets the tracing level, this controls the number of spans included in a single trace.
 	//  Valid options are (tracing.InfoLevel, tracing.DebugLevel) Defaults to tracing.InfoLevel
 	TraceLevel tracing.Level
+
+	// (Optional) EventChannel receives hit events
+	EventChannel chan<- HitEvent
 }
 
 func (d *DaemonConfig) ClientTLS() *tls.Config {

daemon.go

@@ -156,6 +156,7 @@ func (s *Daemon) Start(ctx context.Context) error {
 		CacheSize:     s.conf.CacheSize,
 		Workers:       s.conf.Workers,
 		InstanceID:    s.conf.InstanceID,
+		EventChannel:  s.conf.EventChannel,
 	}
 
 	s.V1Server, err = NewV1Instance(s.instanceConf)

functional_test.go

@@ -2096,6 +2096,80 @@ func TestGlobalBehavior(t *testing.T) {
 	})
 }
 
+func TestEventChannel(t *testing.T) {
+	eventChannel := make(chan guber.HitEvent)
+	defer close(eventChannel)
+	hits := make(map[string]int64)
+	var mu sync.Mutex
+	sem := make(chan struct{})
+	go func() {
+		for e := range eventChannel {
+			mu.Lock()
+			key := e.Request.Name + "|" + e.Request.UniqueKey
+			hits[key] += e.Request.Hits
+			mu.Unlock()
+			sem <- struct{}{}
+		}
+	}()
+
+	// Spawn specialized Gubernator cluster with EventChannel enabled.
+	cluster.Stop()
+	defer func() {
+		err := startGubernator()
+		require.NoError(t, err)
+	}()
+	peers := []guber.PeerInfo{
+		{GRPCAddress: "127.0.0.1:10000", HTTPAddress: "127.0.0.1:10001", DataCenter: cluster.DataCenterNone},
+		{GRPCAddress: "127.0.0.1:10002", HTTPAddress: "127.0.0.1:10003", DataCenter: cluster.DataCenterNone},
+		{GRPCAddress: "127.0.0.1:10004", HTTPAddress: "127.0.0.1:10005", DataCenter: cluster.DataCenterNone},
+	}
+	err := cluster.StartWith(peers, cluster.WithEventChannel(eventChannel))
+	require.NoError(t, err)
+	defer cluster.Stop()
+
+	client, err := guber.DialV1Server(cluster.GetRandomPeer(cluster.DataCenterNone).GRPCAddress, nil)
+	require.NoError(t, err)
+	sendHit := func(key string, behavior guber.Behavior) {
+		ctx, cancel := context.WithTimeout(context.Background(), clock.Second*10)
+		defer cancel()
+		_, err = client.GetRateLimits(ctx, &guber.GetRateLimitsReq{
+			Requests: []*guber.RateLimitReq{
+				{
+					Name:      "test",
+					UniqueKey: key,
+					Algorithm: guber.Algorithm_TOKEN_BUCKET,
+					Behavior:  behavior,
+					Duration:  guber.Minute * 3,
+					Hits:      2,
+					Limit:     1000,
+				},
+			},
+		})
+		require.NoError(t, err)
+		select {
+		case <-sem:
+		case <-time.After(3 * time.Second):
+			t.Fatal("Timeout waiting for EventChannel handler")
+		}
+	}
+
+	// Send hits using all peering behaviors.
+	t.Run("Batching", func(t *testing.T) {
+		sendHit("foobar0", guber.Behavior_BATCHING)
+		assert.Equal(t, int64(2), hits["test|foobar0"])
+	})
+
+	t.Run("No batching", func(t *testing.T) {
+		sendHit("foobar1", guber.Behavior_NO_BATCHING)
+		assert.Equal(t, int64(2), hits["test|foobar1"])
+	})
+
+	t.Run("Global", func(t *testing.T) {
+		sendHit("foobar2", guber.Behavior_GLOBAL)
+		assert.Equal(t, int64(2), hits["test|foobar2"])
+	})
+}
+
 // Request metrics and parse into map.
 // Optionally pass names to filter metrics by name.
 func getMetrics(HTTPAddr string, names ...string) (map[string]*model.Sample, error) {

gubernator.go

@@ -608,6 +608,18 @@ func (s *V1Instance) getLocalRateLimit(ctx context.Context, r *RateLimitReq, req
 
 	if reqState.IsOwner {
 		metricGetRateLimitCounter.WithLabelValues("local").Inc()
+
+		// Send to event channel, if set.
+		if s.conf.EventChannel != nil {
+			e := HitEvent{
+				Request:  r,
+				Response: resp,
+			}
+			select {
+			case s.conf.EventChannel <- e:
+			case <-ctx.Done():
+			}
+		}
 	}
 	return resp, nil
 }