wip

apps/api/Dockerfile.dev

@@ -1,7 +1,9 @@
 FROM node:lts AS base
 
 RUN npm i -g pnpm turbo
-RUN corepack enable
+# I don't know why, but this caused signature mismatch errors
+# and doesn't seem to be needed for dev
+# RUN corepack enable
 
 
 FROM base AS builder
@@ -36,4 +38,4 @@ COPY --from=installer /unkey .
 WORKDIR /unkey/apps/api
 ENV WRANGLER_SEND_METRICS=false
 RUN rm .dev.vars || true
-CMD pnpm wrangler dev 
+CMD pnpm wrangler dev

deployment/docker-compose.yaml

@@ -31,9 +31,18 @@ services:
     ports:
       - 3900:3900
 
+  api_lb:
+    container_name: api_lb
+    image: nginx:latest
+    volumes:
+      - ./nginx.apiv2.conf:/etc/nginx/nginx.conf:ro
+    depends_on:
+      - apiv2
+    ports:
+      - 7070:7070
   apiv2:
     deploy:
-      replicas: 1
+      replicas: 3
       endpoint_mode: vip
 
     command:  [ "api", "--config", "config.docker.json"]
@@ -43,10 +52,12 @@ services:
     depends_on:
       - mysql
       - clickhouse
+      - redis
     environment:
-      PORT: 8080
+      PORT: 7070
       DATABASE_PRIMARY_DSN: "mysql://unkey:password@tcp(mysql:3900)/unkey"
       CLICKHOUSE_URL: "clickhouse://default:password@clickhouse:9000"
+      REDIS_URL: "redis://redis:6379"
 
   agent:
     deploy:
@@ -134,6 +145,10 @@ services:
     volumes:
       - s3:/data
 
+  redis:
+    image: redis:latest
+    ports:
+      - 6379:6379
 
   api:
     # deploy:

deployment/nginx.apiv2.conf

@@ -0,0 +1,21 @@
+user  nginx;
+
+events {
+    worker_connections   1000;
+}
+http {
+        server {
+              listen 7070;
+              location / {
+                proxy_pass http://apiv2:7070;
+              }
+        }
+         server {
+              listen 9090;
+              location / {
+                proxy_pass http://apiv2:9095;
+              }
+        }
+
+
+}

go/Dockerfile

@@ -8,7 +8,7 @@ RUN go mod download
 
 COPY . .
 ARG VERSION
-RUN go build -o bin/unkey -ldflags "-X 'github.com/unkeyed/unkey/go/pkg/version.Version=${VERSION}'"  ./cmd/main.go
+RUN go build -o bin/unkey -ldflags "-X 'github.com/unkeyed/unkey/go/pkg/version.Version=${VERSION}'"  ./main.go
 
 FROM golang:1.23-alpine
 WORKDIR  /usr/local/bin

go/cmd/api/config.go

@@ -13,6 +13,7 @@ type nodeConfig struct {
 		Interval int    `json:"interval" min:"1" description:"Interval in seconds to send heartbeat"`
 	} `json:"heartbeat,omitempty" description:"Send heartbeat to a URL"`
 
+	RedisUrl   string `json:"redisUrl"`
 	Clickhouse *struct {
 		Url string `json:"url" minLength:"1"`
 	} `json:"clickhouse,omitempty"`

go/cmd/api/main.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
 	"os"
@@ -11,6 +12,7 @@ import (
 	"github.com/unkeyed/unkey/go/pkg/clickhouse"
 	"github.com/unkeyed/unkey/go/pkg/config"
 	"github.com/unkeyed/unkey/go/pkg/logging"
+	"github.com/unkeyed/unkey/go/pkg/membership"
 	"github.com/unkeyed/unkey/go/pkg/uid"
 	"github.com/unkeyed/unkey/go/pkg/version"
 	"github.com/unkeyed/unkey/go/pkg/zen"
@@ -70,12 +72,46 @@ func run(c *cli.Context) error {
 
 	logger.Info(c.Context, "configration loaded", slog.String("file", configFile))
 
+	m, err := membership.New(membership.Config{
+		RedisUrl: cfg.RedisUrl,
+		NodeID:   cfg.NodeId,
+		RpcAddr:  "",
+		Logger:   logger,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to create membership")
+	}
+
+	go func() {
+		joins := m.SubscribeJoinEvents()
+		leaves := m.SubscribeLeaveEvents()
+
+		for {
+			select {
+			case j := <-joins:
+				{
+					logger.Info(c.Context, "node joined", slog.String("nodeID", j.ID))
+
+				}
+			case l := <-leaves:
+				{
+					logger.Info(c.Context, "node left", slog.String("nodeID", l.ID))
+
+				}
+			}
+
+		}
+	}()
+
 	var ch clickhouse.Bufferer = clickhouse.NewNoop()
 	if cfg.Clickhouse != nil {
 		ch, err = clickhouse.New(clickhouse.Config{
 			URL:    cfg.Clickhouse.Url,
 			Logger: logger,
 		})
+		if err != nil {
+			return fmt.Errorf("unable to create clickhouse: %w", err)
+		}
 	}
 
 	srv, err := zen.New(zen.Config{
@@ -110,8 +146,12 @@ func run(c *cli.Context) error {
 	signal.Notify(cShutdown, os.Interrupt, syscall.SIGTERM)
 
 	<-cShutdown
+	shutdownCtx := context.Background()
 	logger.Info(c.Context, "shutting down")
-
+	err = m.Leave(shutdownCtx)
+	if err != nil {
+		return fmt.Errorf("unable to leave cluster: %w", err)
+	}
 	return nil
 }
 

go/cmd/api/routes/v2_ratelimit_set_override/happy_test.go

@@ -43,7 +43,10 @@ func TestCreateNewOverride(t *testing.T) {
 		DeletedAt:   time.Time{},
 	})
 
-	route := handler.New(handler.Services{})
+	route := handler.New(handler.Services{
+		DB:   db,
+		Keys: nil,
+	})
 
 	h.Register(route)
 

go/config.docker.json

@@ -2,6 +2,7 @@
   "$schema": "schema.json",
   "port": "8080",
   "nodeId": "node_1",
+  "redisUrl": "${REDIS_URL}",
   "database": {
     "primary": "mysql://"
   }

go/go.mod

@@ -15,6 +15,8 @@ require (
 	github.com/ory/dockertest/v3 v3.11.0
 	github.com/pb33f/libopenapi v0.16.5
 	github.com/pb33f/libopenapi-validator v0.1.0
+	github.com/prometheus/client_golang v1.20.2
+	github.com/redis/go-redis/v9 v9.6.1
 	github.com/segmentio/ksuid v1.0.4
 	github.com/sqlc-dev/sqlc v1.28.0
 	github.com/stretchr/testify v1.10.0
@@ -45,6 +47,7 @@ require (
 	github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect
 	github.com/cubicdaiya/gonp v1.0.4 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/docker/cli v27.2.0+incompatible // indirect
 	github.com/docker/docker v27.2.0+incompatible // indirect
 	github.com/docker/go-connections v0.5.0 // indirect
@@ -101,7 +104,6 @@ require (
 	github.com/pingcap/tidb/pkg/parser v0.0.0-20241203170126-9812d85d0d25 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_golang v1.20.2 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.57.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect

go/go.sum

@@ -98,6 +98,10 @@ github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+Ce
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
 github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f/go.mod h1:TdznJufoqS23FtqVCzL0ZqgP5MqXbb4fg/WgDys70nA=
 github.com/btcsuite/btcutil v0.0.0-20190425235716-9e5f4b9a998d/go.mod h1:+5NJ2+qvTyV9exUAL/rxXi3DcLg2Ts+ymUAY5y4NvMg=
@@ -155,6 +159,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/docker/cli v27.2.0+incompatible h1:yHD1QEB1/0vr5eBNpu8tncu8gWxg8EydFPOSKHzXSMM=
 github.com/docker/cli v27.2.0+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
 github.com/docker/docker v27.2.0+incompatible h1:Rk9nIVdfH3+Vz4cyI/uhbINhEZ/oLmc+CBXmH6fbNk4=
@@ -555,6 +561,8 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/redis/go-redis/v9 v9.6.1 h1:HHDteefn6ZkTtY5fGUE8tj8uy85AHk6zP7CpzIAM0y4=
+github.com/redis/go-redis/v9 v9.6.1/go.mod h1:0C0c6ycQsdpVNQpxb1njEQIqkx5UcsM8FJCQLgE9+RA=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/riza-io/grpc-go v0.2.0 h1:2HxQKFVE7VuYstcJ8zqpN84VnAoJ4dCL6YFhJewNcHQ=

go/pkg/batch/process.go

@@ -41,7 +41,7 @@ func New[T any](config Config[T]) *BatchProcessor[T] {
 		config: config,
 	}
 
-	for _ = range bp.config.Consumers {
+	for range bp.config.Consumers {
 		go bp.process()
 	}
 
@@ -70,7 +70,7 @@ func (bp *BatchProcessor[T]) process() {
 				return
 			}
 			bp.batch = append(bp.batch, e)
-			if len(bp.batch) >= int(bp.config.BatchSize) {
+			if len(bp.batch) >= bp.config.BatchSize {
 				flushAndReset()
 
 			}
@@ -90,7 +90,8 @@ func (bp *BatchProcessor[T]) Buffer(t T) {
 		select {
 		case bp.buffer <- t:
 		default:
-			droppedMessages.WithLabelValues(bp.name).Inc()
+			// Emit a metric to signal we dropped a message
+
 		}
 	} else {
 		bp.buffer <- t

go/pkg/clickhouse/client.go

@@ -44,11 +44,15 @@ func New(config Config) (*Clickhouse, error) {
 		return nil, fault.Wrap(err, fault.WithDesc("opening clickhouse failed", ""))
 	}
 
-	err = retry.Retry(func() error {
-		return conn.Ping(context.Background())
-	}, 10, func(n int) time.Duration {
-		return time.Duration(n) * time.Second
-	})
+	err = retry.New(
+		retry.Attempts(10),
+		retry.Backoff(func(n int) time.Duration {
+			return time.Duration(n) * time.Second
+		}),
+	).
+		Do(func() error {
+			return conn.Ping(context.Background())
+		})
 	if err != nil {
 		return nil, fault.Wrap(err, fault.WithDesc("pinging clickhouse failed", ""))
 	}