feat/gossip

.github/workflows/job_bazel.yaml

@@ -33,6 +33,6 @@ jobs:
       # Running containers is temporary until we moved them inside of bazel,
       # at that point they are only created if they are actually needed
       - name: Start containers
-        run: docker compose -f ./dev/docker-compose.yaml up s3 clickhouse kafka mysql vault -d --wait
+        run: docker compose -f ./dev/docker-compose.yaml up s3 clickhouse mysql vault -d --wait
       - name: Run tests
         run: bazel test //... --test_output=errors

MODULE.bazel

@@ -39,6 +39,7 @@ use_repo(
     "com_github_go_sql_driver_mysql",
     "com_github_google_go_containerregistry",
     "com_github_google_go_containerregistry_pkg_authn_k8schain",
+    "com_github_hashicorp_memberlist",
     "com_github_maypok86_otter",
     "com_github_moby_buildkit",
     "com_github_oapi_codegen_nullable",
@@ -50,7 +51,6 @@ use_repo(
     "com_github_prometheus_client_golang",
     "com_github_redis_go_redis_v9",
     "com_github_restatedev_sdk_go",
-    "com_github_segmentio_kafka_go",
     "com_github_shirou_gopsutil_v4",
     "com_github_spiffe_go_spiffe_v2",
     "com_github_sqlc_dev_plugin_sdk_go",

Makefile

@@ -61,7 +61,7 @@ pull: ## Pull latest Docker images for services
 
 .PHONY: up
 up: pull ## Start all infrastructure services
-	@docker compose -f ./dev/docker-compose.yaml up -d planetscale mysql redis clickhouse s3 otel kafka restate ctrl-api --wait
+	@docker compose -f ./dev/docker-compose.yaml up -d planetscale mysql redis clickhouse s3 otel restate ctrl-api --wait
 
 .PHONY: clean
 clean: ## Stop and remove all services with volumes
@@ -85,13 +85,14 @@ generate: generate-sql ## Generate code from protobuf and other sources
 	rm -rf ./gen || true
 	rm ./pkg/db/*_generated.go || true
 	go generate ./...
+	go run ./tools/exportoneof ./gen/proto
 	bazel run //:gazelle
 	go fmt ./...
 	pnpm --dir=web fmt
 
 .PHONY: test
 test: ## Run tests with bazel
-	docker compose -f ./dev/docker-compose.yaml up -d mysql clickhouse s3 kafka vault --wait
+	docker compose -f ./dev/docker-compose.yaml up -d mysql clickhouse s3 vault --wait
 	bazel test //...
 	make clean-docker-test
 

cmd/api/main.go

@@ -73,9 +73,21 @@ var Cmd = &cli.Command{
 		cli.String("vault-token", "Bearer token for vault service authentication",
 			cli.EnvVar("UNKEY_VAULT_TOKEN")),
 
-		// Kafka Configuration
-		cli.StringSlice("kafka-brokers", "Comma-separated list of Kafka broker addresses for distributed cache invalidation",
-			cli.EnvVar("UNKEY_KAFKA_BROKERS")),
+		// Gossip Cluster Configuration
+		cli.Bool("gossip-enabled", "Enable gossip-based distributed cache invalidation",
+			cli.Default(false), cli.EnvVar("UNKEY_GOSSIP_ENABLED")),
+		cli.String("gossip-bind-addr", "Address for gossip listeners. Default: 0.0.0.0",
+			cli.Default("0.0.0.0"), cli.EnvVar("UNKEY_GOSSIP_BIND_ADDR")),
+		cli.Int("gossip-lan-port", "LAN memberlist port. Default: 7946",
+			cli.Default(7946), cli.EnvVar("UNKEY_GOSSIP_LAN_PORT")),
+		cli.Int("gossip-wan-port", "WAN memberlist port for bridges. Default: 7947",
+			cli.Default(7947), cli.EnvVar("UNKEY_GOSSIP_WAN_PORT")),
+		cli.StringSlice("gossip-lan-seeds", "LAN seed addresses (e.g. k8s headless service DNS)",
+			cli.EnvVar("UNKEY_GOSSIP_LAN_SEEDS")),
+		cli.StringSlice("gossip-wan-seeds", "Cross-region bridge seed addresses",
+			cli.EnvVar("UNKEY_GOSSIP_WAN_SEEDS")),
+		cli.String("gossip-secret-key", "Base64-encoded AES-256 key for encrypting gossip traffic",
+			cli.EnvVar("UNKEY_GOSSIP_SECRET_KEY")),
 
 		// ClickHouse Proxy Service Configuration
 		cli.String(
@@ -142,10 +154,9 @@ func action(ctx context.Context, cmd *cli.Command) error {
 
 	config := api.Config{
 		// Basic configuration
-		CacheInvalidationTopic: "",
-		Platform:               cmd.String("platform"),
-		Image:                  cmd.String("image"),
-		Region:                 cmd.String("region"),
+		Platform: cmd.String("platform"),
+		Image:    cmd.String("image"),
+		Region:   cmd.String("region"),
 
 		// Database configuration
 		DatabasePrimary:         cmd.String("database-primary"),
@@ -176,8 +187,14 @@ func action(ctx context.Context, cmd *cli.Command) error {
 		VaultURL:   cmd.String("vault-url"),
 		VaultToken: cmd.String("vault-token"),
 
-		// Kafka configuration
-		KafkaBrokers: cmd.StringSlice("kafka-brokers"),
+		// Gossip cluster configuration
+		GossipEnabled:   cmd.Bool("gossip-enabled"),
+		GossipBindAddr:  cmd.String("gossip-bind-addr"),
+		GossipLANPort:   cmd.Int("gossip-lan-port"),
+		GossipWANPort:   cmd.Int("gossip-wan-port"),
+		GossipLANSeeds:  cmd.StringSlice("gossip-lan-seeds"),
+		GossipWANSeeds:  cmd.StringSlice("gossip-wan-seeds"),
+		GossipSecretKey: cmd.String("gossip-secret-key"),
 
 		// ClickHouse proxy configuration
 		ChproxyToken: cmd.String("chproxy-auth-token"),

cmd/frontline/main.go

@@ -75,6 +75,22 @@ var Cmd = &cli.Command{
 		cli.String("ctrl-addr", "Address of the control plane",
 			cli.Default("localhost:8080"), cli.EnvVar("UNKEY_CTRL_ADDR")),
 
+		// Gossip Cluster Configuration
+		cli.Bool("gossip-enabled", "Enable gossip-based distributed cache invalidation",
+			cli.Default(false), cli.EnvVar("UNKEY_GOSSIP_ENABLED")),
+		cli.String("gossip-bind-addr", "Address for gossip listeners. Default: 0.0.0.0",
+			cli.Default("0.0.0.0"), cli.EnvVar("UNKEY_GOSSIP_BIND_ADDR")),
+		cli.Int("gossip-lan-port", "LAN memberlist port. Default: 7946",
+			cli.Default(7946), cli.EnvVar("UNKEY_GOSSIP_LAN_PORT")),
+		cli.Int("gossip-wan-port", "WAN memberlist port for bridges. Default: 7947",
+			cli.Default(7947), cli.EnvVar("UNKEY_GOSSIP_WAN_PORT")),
+		cli.StringSlice("gossip-lan-seeds", "LAN seed addresses (e.g. k8s headless service DNS)",
+			cli.EnvVar("UNKEY_GOSSIP_LAN_SEEDS")),
+		cli.StringSlice("gossip-wan-seeds", "Cross-region bridge seed addresses",
+			cli.EnvVar("UNKEY_GOSSIP_WAN_SEEDS")),
+		cli.String("gossip-secret-key", "Base64-encoded AES-256 key for encrypting gossip traffic",
+			cli.EnvVar("UNKEY_GOSSIP_SECRET_KEY")),
+
 		// Logging Sampler Configuration
 		cli.Float("log-sample-rate", "Baseline probability (0.0-1.0) of emitting log events. Default: 1.0",
 			cli.Default(1.0), cli.EnvVar("UNKEY_LOG_SAMPLE_RATE")),
@@ -118,6 +134,15 @@ func action(ctx context.Context, cmd *cli.Command) error {
 		VaultURL:   cmd.String("vault-url"),
 		VaultToken: cmd.String("vault-token"),
 
+		// Gossip cluster configuration
+		GossipEnabled:   cmd.Bool("gossip-enabled"),
+		GossipBindAddr:  cmd.String("gossip-bind-addr"),
+		GossipLANPort:   cmd.Int("gossip-lan-port"),
+		GossipWANPort:   cmd.Int("gossip-wan-port"),
+		GossipLANSeeds:  cmd.StringSlice("gossip-lan-seeds"),
+		GossipWANSeeds:  cmd.StringSlice("gossip-wan-seeds"),
+		GossipSecretKey: cmd.String("gossip-secret-key"),
+
 		// Logging sampler configuration
 		LogSampleRate:    cmd.Float("log-sample-rate"),
 		LogSlowThreshold: cmd.Duration("log-slow-threshold"),

cmd/sentinel/main.go

@@ -53,6 +53,19 @@ var Cmd = &cli.Command{
 			cli.Default(0.25), cli.EnvVar("UNKEY_OTEL_TRACE_SAMPLING_RATE")),
 		cli.Int("prometheus-port", "Enable Prometheus /metrics endpoint on specified port. Set to 0 to disable.", cli.EnvVar("UNKEY_PROMETHEUS_PORT")),
 
+		// Gossip Cluster Configuration
+		cli.Bool("gossip-enabled", "Enable gossip-based distributed cache invalidation",
+			cli.Default(false), cli.EnvVar("UNKEY_GOSSIP_ENABLED")),
+		cli.String("gossip-bind-addr", "Address for gossip listeners. Default: 0.0.0.0",
+			cli.Default("0.0.0.0"), cli.EnvVar("UNKEY_GOSSIP_BIND_ADDR")),
+		cli.Int("gossip-lan-port", "LAN memberlist port. Default: 7946",
+			cli.Default(7946), cli.EnvVar("UNKEY_GOSSIP_LAN_PORT")),
+		cli.Int("gossip-wan-port", "WAN memberlist port for bridges. Default: 7947",
+			cli.Default(7947), cli.EnvVar("UNKEY_GOSSIP_WAN_PORT")),
+		cli.StringSlice("gossip-lan-seeds", "LAN seed addresses (e.g. k8s headless service DNS)",
+			cli.EnvVar("UNKEY_GOSSIP_LAN_SEEDS")),
+		cli.StringSlice("gossip-wan-seeds", "Cross-region bridge seed addresses",
+			cli.EnvVar("UNKEY_GOSSIP_WAN_SEEDS")),
 		// Logging Sampler Configuration
 		cli.Float("log-sample-rate", "Baseline probability (0.0-1.0) of emitting log events. Default: 1.0",
 			cli.Default(1.0), cli.EnvVar("UNKEY_LOG_SAMPLE_RATE")),
@@ -83,6 +96,14 @@ func action(ctx context.Context, cmd *cli.Command) error {
 		OtelTraceSamplingRate: cmd.Float("otel-trace-sampling-rate"),
 		PrometheusPort:        cmd.Int("prometheus-port"),
 
+		// Gossip cluster configuration
+		GossipEnabled:  cmd.Bool("gossip-enabled"),
+		GossipBindAddr: cmd.String("gossip-bind-addr"),
+		GossipLANPort:  cmd.Int("gossip-lan-port"),
+		GossipWANPort:  cmd.Int("gossip-wan-port"),
+		GossipLANSeeds: cmd.StringSlice("gossip-lan-seeds"),
+		GossipWANSeeds: cmd.StringSlice("gossip-wan-seeds"),
+
 		// Logging sampler configuration
 		LogSampleRate:    cmd.Float("log-sample-rate"),
 		LogSlowThreshold: cmd.Duration("log-slow-threshold"),

dev/Tiltfile

@@ -180,6 +180,8 @@ docker_build_with_restart(
     live_update=[sync('./bin/unkey', '/unkey')]
 )
 
+
+
 # Vault service
 k8s_yaml('k8s/manifests/vault.yaml')
 k8s_resource(

dev/docker-compose.yaml

@@ -76,8 +76,6 @@ services:
         condition: service_healthy
       clickhouse:
         condition: service_healthy
-      kafka:
-        condition: service_started
       ctrl-api:
         condition: service_started
     environment:
@@ -111,13 +109,6 @@ services:
       start_period: 10s
       interval: 5s
 
-  # The Kafka broker, available at localhost:9092
-  kafka:
-    container_name: kafka
-    image: bufbuild/bufstream:0.4.4
-    network_mode: host
-    command: ["serve", "--inmemory"]
-
   # Vault service for encryption and key management
   vault:
     networks:
@@ -438,7 +429,6 @@ volumes:
   clickhouse:
   clickhouse-keeper:
   s3:
-  kafka_data:
 
 networks:
   default:

dev/k8s/manifests/api.yaml

@@ -23,6 +23,12 @@ spec:
           imagePullPolicy: Never # Use local images
           ports:
             - containerPort: 7070
+            - containerPort: 7946
+              name: gossip-lan
+              protocol: TCP
+            - containerPort: 7946
+              name: gossip-lan-udp
+              protocol: UDP
           env:
             # Server Configuration
             - name: UNKEY_HTTP_PORT
@@ -38,8 +44,6 @@ spec:
               value: "unkey:local"
             - name: UNKEY_REGION
               value: "local"
-            - name: UNKEY_INSTANCE_ID
-              value: "api-dev"
             # Database Configuration
             - name: UNKEY_DATABASE_PRIMARY
               value: "unkey:password@tcp(mysql:3306)/unkey?parseTime=true&interpolateParams=true"
@@ -71,6 +75,13 @@ spec:
             # Request Body Configuration
             - name: UNKEY_MAX_REQUEST_BODY_SIZE
               value: "10485760"
+            # Gossip Configuration
+            - name: UNKEY_GOSSIP_ENABLED
+              value: "true"
+            - name: UNKEY_GOSSIP_LAN_PORT
+              value: "7946"
+            - name: UNKEY_GOSSIP_LAN_SEEDS
+              value: "api-gossip-lan"
           readinessProbe:
             httpGet:
               path: /health/ready
@@ -129,3 +140,23 @@ spec:
       targetPort: 7070
       protocol: TCP
   type: LoadBalancer
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: api-gossip-lan
+  namespace: unkey
+spec:
+  clusterIP: None
+  selector:
+    app: api
+  ports:
+    - name: gossip-lan
+      port: 7946
+      targetPort: 7946
+      protocol: TCP
+    - name: gossip-lan-udp
+      port: 7946
+      targetPort: 7946
+      protocol: UDP

dev/k8s/manifests/cilium-policies.yaml

@@ -7,6 +7,48 @@
 # a CiliumNetworkPolicy in the customer namespace, Cilium automatically enables
 # default deny for the selected endpoints. We don't need an explicit deny-all policy.
 ---
+# 1. Allow gossip traffic between API pods
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: api-gossip-lan
+  namespace: unkey
+spec:
+  endpointSelector:
+    matchLabels:
+      app: api
+  ingress:
+    - fromEndpoints:
+        - matchLabels:
+            app: api
+      toPorts:
+        - ports:
+            - port: "7946"
+              protocol: TCP
+            - port: "7946"
+              protocol: UDP
+---
+# 1b. Allow gossip traffic between Frontline pods
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: frontline-gossip-lan
+  namespace: unkey
+spec:
+  endpointSelector:
+    matchLabels:
+      app: frontline
+  ingress:
+    - fromEndpoints:
+        - matchLabels:
+            app: frontline
+      toPorts:
+        - ports:
+            - port: "7946"
+              protocol: TCP
+            - port: "7946"
+              protocol: UDP
+---
 # 2. Block K8s API server access from customer pods
 # Prevents customer workloads from accessing the Kubernetes API
 apiVersion: cilium.io/v2
@@ -102,6 +144,17 @@ spec:
         - ports:
             - port: "53"
               protocol: ANY
+    # Gossip between sentinel pods
+    - toEndpoints:
+        - matchLabels:
+            io.kubernetes.pod.namespace: sentinel
+            app.kubernetes.io/component: sentinel
+      toPorts:
+        - ports:
+            - port: "7946"
+              protocol: TCP
+            - port: "7946"
+              protocol: UDP
     # MySQL in unkey namespace
     - toEndpoints:
         - matchLabels:

dev/k8s/manifests/frontline.yaml

@@ -26,6 +26,12 @@ spec:
               name: http
             - containerPort: 7443
               name: https
+            - containerPort: 7946
+              name: gossip-lan
+              protocol: TCP
+            - containerPort: 7946
+              name: gossip-lan-udp
+              protocol: UDP
           env:
             - name: UNKEY_HTTP_PORT
               value: "7070"
@@ -51,6 +57,13 @@ spec:
               value: "vault-test-token-123"
             - name: UNKEY_OTEL
               value: "false"
+            # Gossip Configuration
+            - name: UNKEY_GOSSIP_ENABLED
+              value: "true"
+            - name: UNKEY_GOSSIP_LAN_PORT
+              value: "7946"
+            - name: UNKEY_GOSSIP_LAN_SEEDS
+              value: "frontline-gossip-lan"
           volumeMounts:
             - name: tls-certs
               mountPath: /certs
@@ -97,3 +110,23 @@ spec:
       port: 443
       targetPort: 7443
   type: LoadBalancer
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: frontline-gossip-lan
+  namespace: unkey
+spec:
+  clusterIP: None
+  selector:
+    app: frontline
+  ports:
+    - name: gossip-lan
+      port: 7946
+      targetPort: 7946
+      protocol: TCP
+    - name: gossip-lan-udp
+      port: 7946
+      targetPort: 7946
+      protocol: UDP