(feat) add redis mutually exclusive distributed locks

.github/workflows/ci.yaml

@@ -18,7 +18,13 @@ jobs:
       - uses: actions/setup-go@v4
         with:
           go-version: ">=1.21.0"
-      - name: Install
+      - name: Install redis server package
+        run: |
+          curl -fsSL https://packages.redis.io/gpg | sudo gpg --dearmor -o /usr/share/keyrings/redis-archive-keyring.gpg
+          sudo chmod 644 /usr/share/keyrings/redis-archive-keyring.gpg
+          echo "deb [signed-by=/usr/share/keyrings/redis-archive-keyring.gpg] https://packages.redis.io/deb $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/redis.list
+          sudo apt-get update && sudo apt-get install redis-stack-server
+      - name: Install dev binaries
         run: make install
       - name: Build
         run: make build

.gitignore

@@ -27,6 +27,7 @@ testbin/
 # data directories
 
 default.etcd/
+*.rdb
 
 # dev configs
 .vscode

Makefile

@@ -3,14 +3,16 @@
 
 GOCMD=go
 GO_BUILD_FLAGS=
-GOBUILDSERVER=$(GOCMD) build $(GO_BUILD_FLAGS) -o ./bin/dlock ./cmd/dlock 
-GOBUILDCLI=$(GOCMD) build $(GO_BUILD_FLAGS) -o ./bin/dlockctl ./cmd/dlockctl
+GOBUILDSERVER=$(GOCMD) build $(GO_BUILD_FLAGS) -ldflags "-w -s" -o ./bin/dlock ./cmd/dlock 
+GOBUILDCLI=$(GOCMD) build $(GO_BUILD_FLAGS) -ldflags "-w -s" -o ./bin/dlockctl ./cmd/dlockctl
 GOOS=$(shell go env GOOS)
 GOARCH=$(shell go env GOARCH)
 NATS_VERSION=v2.10.9
 NATS_BIN=nats-server-$(NATS_VERSION)-$(GOOS)-$(GOARCH)
 ETCD_VERSION=v3.5.11
 ETCD_BIN=etcd-$(ETCD_VERSION)-$(GOOS)-$(GOARCH)
+REDIS_VERSION=7.2.0
+REDIS_BIN=redis-stack-server-$(REDIS_VERSION)-v8-x86_64.AppImage
 
 install:
 	go install github.com/bufbuild/buf/cmd/[email protected]
@@ -34,6 +36,9 @@ testbin:
 	wget https://github.com/etcd-io/etcd/releases/download/$(ETCD_VERSION)/$(ETCD_BIN).tar.gz
 	tar -C ./testbin -zxvf $(ETCD_BIN).tar.gz --strip-components=1 $(ETCD_BIN)/etcd 
 	rm $(ETCD_BIN).tar.gz
+	wget https://packages.redis.io/redis-stack/$(REDIS_BIN)
+	chmod +x $(REDIS_BIN)
+	mv $(REDIS_BIN) ./testbin/redis-server
 
 clean:
 	rm -rf ./bin

README.md

@@ -102,6 +102,7 @@ As apposed to standard single-host locking mechanisms, distributed locks can be
 | :-------------------------------------------------------: | :----------------: | :-: | :-: | :-: | :-: | :-: |
 | [Jetstream](https://docs.nats.io/nats-concepts/jetstream) | :white_check_mark: | :x: | :x: | :x: | :x: | :x: |
 |                 [Etcd ](https://etcd.io/)                 | :white_check_mark: | :x: | :x: | :x: | :x: | :x: |
+|                [Redis ](https://redis.io/)                | :white_check_mark: | :x: | :x: | :x: | :x: | :x: |
 
 ## Dlock specific guarantees
 
@@ -119,3 +120,4 @@ As apposed to standard single-host locking mechanisms, distributed locks can be
 
 - [Distributed Lock Manager](https://en.wikipedia.org/wiki/Distributed_lock_manager). (n.d.). In Wikipedia. Retrieved from https://en.wikipedia.org/wiki/Distributed_lock_manager
 - Kleppmann, Martin. "Designing Data-Intensive Applications." (2019).
+- [Redis redlock algorithm](https://redis.io/docs/manual/patterns/distributed-locks/) from https://redis.io/docs/manual/patterns/distributed-locks/

docs/TODO.md

@@ -16,7 +16,7 @@
 ## Backends
 
 - [ ] Filesystem-based distributed locks using `flock`
-- [ ] Redis support
+- [x] Redis support
 - [ ] Embedded raft / mini-raft support
 
 ## Testing
@@ -27,4 +27,4 @@
 ## Miscellaneous
 
 - [ ] Examples folder for distributed lock usecases
-- [ ] Internal metrics & traces
+- [x] Server & internal metrics & traces instrumentation

go.mod

@@ -6,7 +6,10 @@ replace github.com/coreos/bbolt => go.etcd.io/bbolt v1.3.8
 
 replace google.golang.org/grpc => google.golang.org/grpc v1.61.0
 
+replace github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203 => github.com/alexandreLamarre/tempredis v0.0.0-20240129193023-7f411f64c2c7
+
 require (
+	github.com/go-redsync/redsync/v4 v4.11.0
 	github.com/google/uuid v1.6.0
 	github.com/jwalton/go-supportscolor v1.2.0
 	github.com/kralicky/gpkg v0.0.0-20240119195700-64f32830b14f
@@ -16,10 +19,13 @@ require (
 	github.com/onsi/ginkgo/v2 v2.15.0
 	github.com/onsi/gomega v1.31.1
 	github.com/prometheus/client_golang v1.18.0
+	github.com/redis/go-redis/v9 v9.0.2
+	github.com/redis/rueidis v1.0.19
 	github.com/samber/lo v1.39.0
 	github.com/samber/slog-multi v1.0.2
 	github.com/samber/slog-sampling v1.4.0
 	github.com/spf13/cobra v1.8.0
+	github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203
 	github.com/ttacon/chalk v0.0.0-20160626202418-22c06c80ed31
 	go.etcd.io/etcd/client/v3 v3.5.11
 	go.etcd.io/etcd/server/v3 v3.5.11
@@ -47,6 +53,7 @@ require (
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cornelk/hashmap v1.0.8 // 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/dustin/go-humanize v1.0.1 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect

go.sum

@@ -1142,6 +1142,8 @@ github.com/ajstarks/deck v0.0.0-20200831202436-30c9fc6549a9/go.mod h1:JynElWSGnm
 github.com/ajstarks/deck/generate v0.0.0-20210309230005-c3f852c02e19/go.mod h1:T13YZdzov6OU0A1+RfKZiZN9ca6VeKdBdyDV+BY97Tk=
 github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
 github.com/ajstarks/svgo v0.0.0-20211024235047-1546f124cd8b/go.mod h1:1KcenG0jGWcpt8ov532z81sp/kMMUG485J2InIOyADM=
+github.com/alexandreLamarre/tempredis v0.0.0-20240129193023-7f411f64c2c7 h1:jTMPs0rYyrP5xVUuX0iZpYu9UICZKdPnZ45xw8gHiu4=
+github.com/alexandreLamarre/tempredis v0.0.0-20240129193023-7f411f64c2c7/go.mod h1:mpTEMdh0NBnmDxuJ8UACOTeLZjX+yG7vGwF2dTfmj0g=
 github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/apache/arrow/go/v10 v10.0.1/go.mod h1:YvhnlEePVnBS4+0z3fhPfUy7W1Ikj0Ih0vcRo/gZ1M0=
@@ -1156,6 +1158,10 @@ github.com/bluele/gcache v0.0.2 h1:WcbfdXICg7G/DGBh1PFfcirkWOQV+v077yF1pSy3DGw=
 github.com/bluele/gcache v0.0.2/go.mod h1:m15KV+ECjptwSPxKhOhQoAFQVtUFjTVkc3H8o0t/fp0=
 github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/boombuler/barcode v1.0.1/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
+github.com/bsm/ginkgo/v2 v2.5.0 h1:aOAnND1T40wEdAtkGSkvSICWeQ8L3UASX7YVCqQx+eQ=
+github.com/bsm/ginkgo/v2 v2.5.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
+github.com/bsm/gomega v1.20.0 h1:JhAwLmtRzXFTx2AkALSLa8ijZafntmhSoU63Ok18Uq8=
+github.com/bsm/gomega v1.20.0/go.mod h1:JifAceMQ4crZIWYUKrlGcmbN3bqHogVTADMD2ATsbwk=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -1193,6 +1199,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/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
@@ -1232,6 +1240,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-pdf/fpdf v0.5.0/go.mod h1:HzcnA+A23uwogo0tp9yU+l3V+KXhiESpt1PMayhOh5M=
 github.com/go-pdf/fpdf v0.6.0/go.mod h1:HzcnA+A23uwogo0tp9yU+l3V+KXhiESpt1PMayhOh5M=
+github.com/go-redsync/redsync/v4 v4.11.0 h1:OPEcAxHBb95EzfwCKWM93ksOwHd5bTce2BD4+R14N6k=
+github.com/go-redsync/redsync/v4 v4.11.0/go.mod h1:ZfayzutkgeBmEmBlUR3j+rF6kN44UUGtEdfzhBFZTPc=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls=
@@ -1279,6 +1289,8 @@ github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg
 github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/gomodule/redigo v1.8.9 h1:Sl3u+2BI/kk+VEatbj0scLdrFhjPmbxOc1myhDP41ws=
+github.com/gomodule/redigo v1.8.9/go.mod h1:7ArFNvsTjH8GMMzB4uy1snslv2BwmginuMs06a1uzZE=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.1 h1:gK4Kx5IaGY9CD5sPJ36FHiBJ6ZXl0kilRiiCj+jdYp4=
@@ -1477,6 +1489,10 @@ github.com/prometheus/common v0.45.0 h1:2BGz0eBc2hdMDLnO/8n0jeB3oPrt2D08CekT0lne
 github.com/prometheus/common v0.45.0/go.mod h1:YJmSTw9BoKxJplESWWxlbyttQR4uaEcGyv9MZjVOJsY=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
+github.com/redis/go-redis/v9 v9.0.2 h1:BA426Zqe/7r56kCcvxYLWe1mkaz71LKF77GwgFzSxfE=
+github.com/redis/go-redis/v9 v9.0.2/go.mod h1:/xDTe9EF1LM61hek62Poq2nzQSGj0xSrEtEHbBQevps=
+github.com/redis/rueidis v1.0.19 h1:s65oWtotzlIFN8eMPhyYwxlwLR1lUdhza2KtWprKYSo=
+github.com/redis/rueidis v1.0.19/go.mod h1:8B+r5wdnjwK3lTFml5VtxjzGOQAC+5UmujoD12pDrEo=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=

pkg/config/v1alpha1/config.go

@@ -1,8 +1,9 @@
 package v1alpha1
 
 type LockServerConfig struct {
-	EtcdStorageSpec      *EtcdStorageSpec      `json:"etcd,omitempty"`
-	JetStreamStorageSpec *JetStreamStorageSpec `json:"jetstream,omitempty"`
+	EtcdClientSpec      *EtcdClientSpec      `json:"etcd,omitempty"`
+	JetstreamClientSpec *JetstreamClientSpec `json:"jetstream,omitempty"`
+	RedisClientSpec     *RedisClientSpec     `json:"redis,omitempty"`
 }
 
 type TracesConfig struct {

pkg/config/v1alpha1/etcd.go

@@ -1,6 +1,6 @@
 package v1alpha1
 
-type EtcdStorageSpec struct {
+type EtcdClientSpec struct {
 	// List of etcd endpoints to connect to.
 	Endpoints []string `json:"endpoints,omitempty"`
 	// Configuration for etcd client-cert auth.

pkg/config/v1alpha1/jetstream.go

@@ -1,6 +1,6 @@
 package v1alpha1
 
-type JetStreamStorageSpec struct {
+type JetstreamClientSpec struct {
 	Endpoint     string `json:"endpoint,omitempty"`
 	NkeySeedPath string `json:"nkeySeedPath,omitempty"`
 }

pkg/config/v1alpha1/redis.go

@@ -0,0 +1,6 @@
+package v1alpha1
+
+type RedisClientSpec struct {
+	Network string `json:"network,omitempty"`
+	Addr    string `json:"addr,omitempty"`
+}

pkg/lock/backend/etcd/util.go

@@ -10,7 +10,7 @@ import (
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
-func NewEtcdClient(ctx context.Context, conf *v1alpha1.EtcdStorageSpec) (*clientv3.Client, error) {
+func NewEtcdClient(ctx context.Context, conf *v1alpha1.EtcdClientSpec) (*clientv3.Client, error) {
 	var tlsConfig *tls.Config
 	if conf.Certs != nil {
 		var err error

pkg/lock/backend/jetstream/jetstream_suite_test.go

@@ -45,7 +45,7 @@ var _ = BeforeSuite(func() {
 			js,
 			"test",
 			nil,
-			logger.New().WithGroup("js-lock"),
+			logger.NewNop(),
 		)
 		lmF.Set(lm)
 
@@ -66,13 +66,13 @@ var _ = BeforeSuite(func() {
 		js3, err := jetstream.AcquireJetstreamConn(
 			context.Background(),
 			conf,
-			logger.New(),
+			logger.NewNop(),
 		)
 		Expect(err).NotTo(HaveOccurred())
 
-		x := jetstream.NewLockManager(context.Background(), js1, "test", nil, logger.New())
-		y := jetstream.NewLockManager(context.Background(), js2, "test", nil, logger.New())
-		z := jetstream.NewLockManager(context.Background(), js3, "test", nil, logger.New())
+		x := jetstream.NewLockManager(context.Background(), js1, "test", nil, logger.NewNop())
+		y := jetstream.NewLockManager(context.Background(), js2, "test", nil, logger.NewNop())
+		z := jetstream.NewLockManager(context.Background(), js3, "test", nil, logger.NewNop())
 
 		lmSetF.Set(lo.Tuple3[lock.LockManager, lock.LockManager, lock.LockManager]{
 			A: x, B: y, C: z,

pkg/lock/backend/jetstream/util.go

@@ -17,7 +17,7 @@ func sanitizePrefix(prefix string) string {
 	return strings.ReplaceAll(strings.ReplaceAll(prefix, "/", "-"), ".", "_")
 }
 
-func AcquireJetstreamConn(ctx context.Context, conf *v1alpha1.JetStreamStorageSpec, lg *slog.Logger) (nats.JetStreamContext, error) {
+func AcquireJetstreamConn(ctx context.Context, conf *v1alpha1.JetstreamClientSpec, lg *slog.Logger) (nats.JetStreamContext, error) {
 	options := []nats.Option{
 		nats.MaxReconnects(-1),
 		nats.RetryOnFailedConnect(true),

pkg/lock/backend/redis/errors.go

@@ -0,0 +1,30 @@
+package redis
+
+import (
+	"errors"
+	"fmt"
+)
+
+var ErrExtendFailed = errors.New("redsync: failed to extend lock")
+
+// A RedisError is an error communicating with one of the Redis nodes.
+type RedisError struct {
+	Node int
+	Err  error
+}
+
+func (err RedisError) Error() string {
+	return fmt.Sprintf("node %d: %v", err.Node, err.Err)
+}
+
+// ErrNodeTaken is the error resulting if the lock is already taken in one of
+// the cluster's nodes
+type ErrNodeTaken struct {
+	Node int
+}
+
+func (err ErrNodeTaken) Error() string {
+	return fmt.Sprintf("node #%d: lock already taken", err.Node)
+}
+
+var ErrTaken = errors.New("lock already taken")