Global tests pre-PR219

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

@@ -50,7 +50,7 @@ jobs:
           skip-cache: true
 
       - name: Test
-        run: go test -v -race -p=1 -count=1 -tags holster_test_mode
+        run: go test -v -race -p=1 -count=1
   go-bench:
     runs-on: ubuntu-latest
     timeout-minutes: 30

Makefile

@@ -1,57 +1,47 @@
-.DEFAULT_GOAL := build
+.DEFAULT_GOAL := release
 VERSION=$(shell cat version)
 LDFLAGS="-X main.Version=$(VERSION)"
 GOLANGCI_LINT = $(GOPATH)/bin/golangci-lint
 GOLANGCI_LINT_VERSION = 1.56.2
 
-.PHONY: help
-help:
-	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
-
-$(GOLANGCI_LINT): ## Download Go linter
+$(GOLANGCI_LINT):
 	curl -sfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(GOPATH)/bin $(GOLANGCI_LINT_VERSION)
 
 .PHONY: lint
-lint: $(GOLANGCI_LINT) ## Run Go linter
-	$(GOLANGCI_LINT) run -v --fix -c .golangci.yml ./...
+lint: $(GOLANGCI_LINT)
+	$(GOLANGCI_LINT) run
 
 .PHONY: test
-test: ## Run unit tests and measure code coverage
-	(go test -v -race -p=1 -count=1 -tags holster_test_mode -coverprofile coverage.out ./...; ret=$$?; \
+test:
+	(go test -v -race -p=1 -count=1 -coverprofile coverage.out ./...; ret=$$?; \
 		go tool cover -func coverage.out; \
 		go tool cover -html coverage.out -o coverage.html; \
 		exit $$ret)
 
 .PHONY: bench
-bench: ## Run Go benchmarks
+bench:
 	go test ./... -bench . -benchtime 5s -timeout 0 -run=XXX -benchmem
 
 .PHONY: docker
-docker: ## Build Docker image
+docker:
 	docker build --build-arg VERSION=$(VERSION) -t ghcr.io/mailgun/gubernator:$(VERSION) .
 	docker tag ghcr.io/mailgun/gubernator:$(VERSION) ghcr.io/mailgun/gubernator:latest
 
-.PHONY: build
-build: proto ## Build binary
+.PHONY: release
+release:
 	go build -v -ldflags $(LDFLAGS) -o gubernator ./cmd/gubernator/main.go
 
 .PHONY: clean
-clean: ## Clean binaries
+clean:
 	rm -f gubernator gubernator-cli
 
-.PHONY: clean-proto
-clean-proto: ## Clean the generated source files from the protobuf sources
-	@echo "==> Cleaning up the go generated files from proto"
-	@find . -name "*.pb.go" -type f -delete
-	@find . -name "*.pb.*.go" -type f -delete
-
-
 .PHONY: proto
-proto: ## Build protos
-	./buf.gen.yaml
+proto:
+	# Install buf: https://buf.build/docs/installation
+	buf generate
 
 .PHONY: certs
-certs: ## Generate SSL certificates
+certs:
 	rm certs/*.key || rm certs/*.srl || rm certs/*.csr || rm certs/*.pem || rm certs/*.cert || true
 	openssl genrsa -out certs/ca.key 4096
 	openssl req -new -x509 -key certs/ca.key -sha256 -subj "/C=US/ST=TX/O=Mailgun Technologies, Inc." -days 3650 -out certs/ca.cert

algorithms.go

@@ -26,16 +26,8 @@ import (
 	"go.opentelemetry.io/otel/trace"
 )
 
-// ### NOTE ###
-// The both token and leaky follow the same semantic which allows for requests of more than the limit
-// to be rejected, but subsequent requests within the same window that are under the limit to succeed.
-// IE: client attempts to send 1000 emails but 100 is their limit. The request is rejected as over the
-// limit, but we do not set the remainder to 0 in the cache. The client can retry within the same window
-// with 100 emails and the request will succeed. You can override this default behavior with `DRAIN_OVER_LIMIT`
-
 // Implements token bucket algorithm for rate limiting. https://en.wikipedia.org/wiki/Token_bucket
 func tokenBucket(ctx context.Context, s Store, c Cache, r *RateLimitReq) (resp *RateLimitResp, err error) {
-
 	tokenBucketTimer := prometheus.NewTimer(metricFuncTimeDuration.WithLabelValues("tokenBucket"))
 	defer tokenBucketTimer.ObserveDuration()
 
@@ -89,6 +81,12 @@ func tokenBucket(ctx context.Context, s Store, c Cache, r *RateLimitReq) (resp *
 				ResetTime: 0,
 			}, nil
 		}
+
+		// The following semantic allows for requests of more than the limit to be rejected, but subsequent
+		// requests within the same duration that are under the limit to succeed. IE: client attempts to
+		// send 1000 emails but 100 is their limit. The request is rejected as over the limit, but since we
+		// don't store OVER_LIMIT in the cache the client can retry within the same rate limit duration with
+		// 100 emails and the request will succeed.
 		t, ok := item.Value.(*TokenBucketItem)
 		if !ok {
 			// Client switched algorithms; perhaps due to a migration?
@@ -389,24 +387,22 @@ func leakyBucket(ctx context.Context, s Store, c Cache, r *RateLimitReq) (resp *
 
 		// If requested hits takes the remainder
 		if int64(b.Remaining) == r.Hits {
-			b.Remaining = 0
-			rl.Remaining = int64(b.Remaining)
+			b.Remaining -= float64(r.Hits)
+			rl.Remaining = 0
 			rl.ResetTime = now + (rl.Limit-rl.Remaining)*int64(rate)
 			return rl, nil
 		}
 
 		// If requested is more than available, then return over the limit
-		// without updating the bucket, unless `DRAIN_OVER_LIMIT` is set.
+		// without updating the bucket.
 		if r.Hits > int64(b.Remaining) {
 			metricOverLimitCounter.Add(1)
 			rl.Status = Status_OVER_LIMIT
-
-			// DRAIN_OVER_LIMIT behavior drains the remaining counter.
 			if HasBehavior(r.Behavior, Behavior_DRAIN_OVER_LIMIT) {
+				// DRAIN_OVER_LIMIT behavior drains the remaining counter.
 				b.Remaining = 0
 				rl.Remaining = 0
 			}
-
 			return rl, nil
 		}
 

benchmark_test.go

@@ -33,13 +33,10 @@ func BenchmarkServer(b *testing.B) {
 	require.NoError(b, err, "Error in conf.SetDefaults")
 
 	b.Run("GetPeerRateLimit() with no batching", func(b *testing.B) {
-		client, err := guber.NewPeerClient(guber.PeerConfig{
+		client := guber.NewPeerClient(guber.PeerConfig{
 			Info:     cluster.GetRandomPeer(cluster.DataCenterNone),
 			Behavior: conf.Behaviors,
 		})
-		if err != nil {
-			b.Errorf("Error building client: %s", err)
-		}
 
 		b.ResetTimer()
 

buf.gen.yaml

@@ -1,21 +1,13 @@
-#!/usr/bin/env -S buf generate --debug --template
----
 version: v1
 plugins:
-  - plugin: buf.build/protocolbuffers/go:v1.32.0
+  - name: go
     out: ./
     opt: paths=source_relative
   - plugin: buf.build/grpc/go:v1.3.0
     out: ./
     opt:
       - paths=source_relative
       - require_unimplemented_servers=false
-  - plugin: buf.build/grpc-ecosystem/gateway:v2.18.0 # same version in go.mod
-    out: ./
-    opt:
-      - paths=source_relative
-      - logtostderr=true
-      - generate_unbound_methods=true
   - plugin: buf.build/grpc/python:v1.57.0
     out: ./python/gubernator
   - plugin: buf.build/protocolbuffers/python

cluster/cluster_test.go

@@ -23,16 +23,12 @@ import (
 	"github.com/mailgun/gubernator/v2/cluster"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/goleak"
 )
 
 func TestStartMultipleInstances(t *testing.T) {
-	t.Cleanup(func() {
-		goleak.VerifyNone(t)
-	})
 	err := cluster.Start(2)
 	require.NoError(t, err)
-	t.Cleanup(cluster.Stop)
+	defer cluster.Stop()
 
 	assert.Equal(t, 2, len(cluster.GetPeers()))
 	assert.Equal(t, 2, len(cluster.GetDaemons()))

cmd/gubernator/main.go

@@ -31,7 +31,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"go.opentelemetry.io/otel/sdk/resource"
 	semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
-	"k8s.io/klog/v2"
+	"k8s.io/klog"
 )
 
 var log = logrus.WithField("category", "gubernator")