TRQ-002: Integrate service with ZooKeeper (#4)

* TRQ-002: Setup zookeeper via docker

* TRQ-002: Integrate zookeeper with service

* TRQ-002: Move service to docker

* TRQ-002: Add retry for connecting to zookeeper

* TRQ-002: (minor) remove comment

.env

@@ -1,2 +1,3 @@
 PORT=8081
 ENV=release
+ZOOKEEPER_SERVERS=127.0.0.1:2181

Dockerfile

@@ -0,0 +1,15 @@
+FROM golang:alpine
+
+WORKDIR /app
+
+COPY go.mod go.sum ./
+
+RUN go mod download
+
+COPY . .
+
+RUN go build -o main ./cmd/main.go
+
+ENV PORT=8080
+
+CMD ["./main"]

cmd/main.go

@@ -2,10 +2,11 @@ package main
 
 import (
 	"log"
-	"net/http"
 	"os"
 
 	"github.com/gin-gonic/gin"
+	"github.com/godcrampy/torquay/pkg/counter"
+	"github.com/godcrampy/torquay/pkg/handlers"
 	"github.com/joho/godotenv"
 )
 
@@ -18,17 +19,19 @@ func main() {
 	mode := os.Getenv("ENV")
 	gin.SetMode(mode)
 
-	r := gin.Default()
+	servers := []string{os.Getenv("ZOOKEEPER_SERVERS")}
+	zkPath := "/counter"
 
-	token := 1
+	c, err := counter.NewCounterWithRetry(servers, zkPath)
+	if err != nil {
+		log.Fatalf("Unable to connect to ZooKeeper: %v", err)
+	}
+	defer c.Close()
 
-	r.GET("/api/v1/token", func(ctx *gin.Context) {
-		ctx.JSON(http.StatusOK, gin.H{
-			"token": token,
-		})
+	h := handlers.NewHandler(c)
 
-		token += 1
-	})
+	r := gin.Default()
+	r.GET("/api/v1/token", h.GetToken)
 
 	port := os.Getenv("PORT")
 	log.Printf("INFO: Starting server on port %s\n", port)

docker-compose.yml

@@ -0,0 +1,31 @@
+version: '3.8'
+
+services:
+  go-server-8080:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    ports:
+      - "8080:8080"
+    environment:
+      - PORT=8080
+      - ZOOKEEPER_SERVERS=zookeeper:2181
+    depends_on:
+      - zookeeper
+  go-server-8081:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    ports:
+      - "8081:8081"
+    environment:
+      - PORT=8081
+      - ZOOKEEPER_SERVERS=zookeeper:2181
+    depends_on:
+      - zookeeper
+  zookeeper:
+    image: 'bitnami/zookeeper:latest'
+    ports:
+      - '2181:2181'
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.20.0 // indirect
+	github.com/go-zookeeper/zk v1.0.3
 	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/joho/godotenv v1.5.1 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect

go.sum

@@ -20,6 +20,8 @@ github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJn
 github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
 github.com/go-playground/validator/v10 v10.20.0 h1:K9ISHbSaI0lyB2eWMPJo+kOS/FBExVwjEviJTixqxL8=
 github.com/go-playground/validator/v10 v10.20.0/go.mod h1:dbuPbCMFw/DrkbEynArYaCwl3amGuJotoKCe95atGMM=
+github.com/go-zookeeper/zk v1.0.3 h1:7M2kwOsc//9VeeFiPtf+uSJlVpU66x9Ba5+8XK7/TDg=
+github.com/go-zookeeper/zk v1.0.3/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=

pkg/counter/counter.go

@@ -0,0 +1,92 @@
+package counter
+
+import (
+	"fmt"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/go-zookeeper/zk"
+)
+
+type Counter struct {
+	zkConn *zk.Conn
+	zkPath string
+}
+
+func NewCounterWithRetry(servers []string, path string) (*Counter, error) {
+	const RetryCount = 5
+	const RetryDelayS = 5
+
+	var (
+		c   *Counter
+		err error
+	)
+
+	for attempt := 1; attempt <= RetryCount; attempt++ {
+		c, err = NewCounter(servers, path)
+
+		if err != nil {
+			log.Printf("ERROR: Failed to connect to ZooKeeper (attempt %d/%d): %v", attempt, RetryCount, err)
+			time.Sleep(time.Second * RetryDelayS)
+			continue
+		}
+
+		return c, nil
+	}
+
+	return nil, err
+}
+
+func NewCounter(servers []string, path string) (*Counter, error) {
+	conn, _, err := zk.Connect(servers, time.Second*5)
+	if err != nil {
+		return nil, err
+	}
+
+	c := &Counter{
+		zkConn: conn,
+		zkPath: path,
+	}
+
+	if exists, _, err := conn.Exists(path); err != nil {
+		return nil, err
+	} else if !exists {
+		_, err := conn.Create(path, []byte("0"), 0, zk.WorldACL(zk.PermAll))
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return c, nil
+}
+
+func (c *Counter) GetAndIncrement() (int, error) {
+	for {
+		data, stat, err := c.zkConn.Get(c.zkPath)
+		if err != nil {
+			return 0, err
+		}
+
+		currentValue, err := strconv.Atoi(string(data))
+		if err != nil {
+			return 0, err
+		}
+
+		newValue := currentValue + 1
+		newData := []byte(fmt.Sprintf("%d", newValue))
+
+		_, err = c.zkConn.Set(c.zkPath, newData, stat.Version)
+		if err == zk.ErrBadVersion {
+			continue
+		} else if err != nil {
+			return 0, err
+		}
+
+		return newValue, nil
+	}
+}
+
+func (c *Counter) Close() {
+	c.zkConn.Close()
+}

pkg/handlers/token_handler.go

@@ -0,0 +1,26 @@
+package handlers
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/godcrampy/torquay/pkg/counter"
+)
+
+type Handler struct {
+	Counter *counter.Counter
+}
+
+func NewHandler(c *counter.Counter) *Handler {
+	return &Handler{Counter: c}
+}
+
+func (h *Handler) GetToken(c *gin.Context) {
+	newValue, err := h.Counter.GetAndIncrement()
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{"token": newValue})
+}