fix

.deploy/deploy-dev.yaml

@@ -6,14 +6,17 @@ services:
     image: mathwave/sprint-repo:queues
     networks:
       - queues-development
+      - monitoring
+      - mongo-development
     environment:
-      MONGO_HOST: "mongo.develop.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_DEV
       STAGE: "development"
     deploy:
       mode: replicated
       restart_policy:
         condition: any
+      placement:
+        constraints: [node.labels.stage == development]
       update_config:
         parallelism: 1
         order: start-first
@@ -21,3 +24,7 @@ services:
 networks:
   queues-development:
     external: true
+  monitoring:
+    external: true
+  mongo-development:
+    external: true

.deploy/deploy-prod.yaml

@@ -6,14 +6,17 @@ services:
     image: mathwave/sprint-repo:queues
     networks:
       - queues
+      - monitoring
+      - mongo
     environment:
-      MONGO_HOST: "mongo.sprinthub.ru"
       MONGO_PASSWORD: $MONGO_PASSWORD_PROD
       STAGE: "production"
     deploy:
       mode: replicated
       restart_policy:
         condition: any
+      placement:
+        constraints: [node.labels.stage == production]
       update_config:
         parallelism: 1
         order: start-first
@@ -21,3 +24,7 @@ services:
 networks:
   queues:
     external: true
+  monitoring:
+    external: true
+  mongo:
+    external: true

.gitea/workflows/deploy-dev.yaml

@@ -9,7 +9,7 @@ on:
 jobs:
   build:
     name: Build
-    runs-on: [ dev ]
+    runs-on: [dev]
     steps:
       - name: login
         run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
@@ -21,7 +21,7 @@ jobs:
         run: docker build -t mathwave/sprint-repo:queues .
   push:
     name: Push
-    runs-on: [ dev ]
+    runs-on: [dev]
     needs: build
     steps:
       - name: push
@@ -40,4 +40,5 @@ jobs:
       - name: deploy
         env:
           MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
-        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml infra-development
+          REDIS_PASSWORD_DEV: ${{ secrets.REDIS_PASSWORD_DEV }}
+        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml queues-development

.gitea/workflows/deploy-prod.yaml

@@ -9,7 +9,7 @@ on:
 jobs:
   build:
     name: Build
-    runs-on: [ dev ]
+    runs-on: [dev]
     steps:
       - name: login
         run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
@@ -21,7 +21,7 @@ jobs:
         run: docker build -t mathwave/sprint-repo:queues .
   push:
     name: Push
-    runs-on: [ dev ]
+    runs-on: [dev]
     needs: build
     steps:
       - name: push
@@ -40,4 +40,5 @@ jobs:
       - name: deploy
         env:
           MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
-        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml infra
+          REDIS_PASSWORD_PROD: ${{ secrets.REDIS_PASSWORD_PROD }}
+        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml queues

app/routers/finish.go

@@ -10,20 +10,18 @@ type FinishRequestBody struct {
 	Id string `json:"id"`
 }
 
-func Finish(w http.ResponseWriter, r *http.Request) {
+func Finish(r *http.Request) (interface{}, int) {
 	d := json.NewDecoder(r.Body)
 	body := FinishRequestBody{}
 	err := d.Decode(&body)
 	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
+		return nil, http.StatusBadRequest
-		return
 	}
 
 	err = tasks.Finish(body.Id)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		return nil, http.StatusInternalServerError
-		return
 	}
 
-	w.WriteHeader(http.StatusAccepted)
+	return nil, http.StatusAccepted
 }

app/routers/put.go

@@ -15,13 +15,12 @@ type PutRequestBody struct {
 	Delay            *int   `json:"delay"`
 }
 
-func Put(w http.ResponseWriter, r *http.Request) {
+func Put(r *http.Request) (interface{}, int) {
 	d := json.NewDecoder(r.Body)
 	body := PutRequestBody{}
 	err := d.Decode(&body)
 	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
+		return nil, http.StatusBadRequest
-		return
 	}
 
 	queue := r.Header.Get("queue")
@@ -44,9 +43,8 @@ func Put(w http.ResponseWriter, r *http.Request) {
 
 	err = tasks.Add(task)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		return nil, http.StatusInternalServerError
-		return
 	}
 
-	w.WriteHeader(http.StatusAccepted)
+	return nil, http.StatusAccepted
 }

app/routers/take.go

@@ -1,9 +1,15 @@
 package routers
 
 import (
-	"encoding/json"
+	"bytes"
+	"fmt"
+	"log"
 	"net/http"
+	"os"
 	tasks "queues-go/app/storage/mongo/collections"
+	"strconv"
+	"sync"
+	"time"
 )
 
 type TaskResponse struct {
@@ -16,13 +22,38 @@ type TakeResponse struct {
 	Task *TaskResponse `json:"task"`
 }
 
-func Take(w http.ResponseWriter, r *http.Request) {
+var MutexMap map[string]*sync.Mutex
-	queue := r.Header.Get("queue")
+
-	task, err := tasks.Take(queue)
+func sendLatency(timestamp time.Time, latency int) error {
+	loc, _ := time.LoadLocation("Europe/Moscow")
+
+	s := fmt.Sprintf(
+		`{"timestamp":"%s","service":"queues","environment":"%s","name":"latency","count":%s}`,
+		timestamp.In(loc).Format("2006-01-02T15:04:05Z"),
+		os.Getenv("STAGE"),
+		strconv.Itoa(latency),
+	)
+	data := []byte(s)
+	r := bytes.NewReader(data)
+	_, err := http.Post("http://monitoring:1237/api/v1/metrics/increment", "application/json", r)
 	if err != nil {
-		println("Error taking")
+		log.Printf("ERROR %s", err.Error())
-		w.WriteHeader(http.StatusInternalServerError)
+	}
-		return
+	return err
+}
+
+func Take(r *http.Request) (interface{}, int) {
+	queue := r.Header.Get("queue")
+	mutex, ok := MutexMap[queue]
+	if !ok {
+		mutex = &sync.Mutex{}
+		MutexMap[queue] = mutex
+	}
+	mutex.Lock()
+	task, err := tasks.Take(queue)
+	mutex.Unlock()
+	if err != nil {
+		return nil, http.StatusInternalServerError
 	}
 
 	var response TakeResponse
@@ -34,14 +65,9 @@ func Take(w http.ResponseWriter, r *http.Request) {
 			Attempt: task.Attempts,
 			Payload: task.Payload,
 		}
+		now := time.Now()
+		go sendLatency(now, int(now.Sub(task.PutAt).Milliseconds()))
 	}
 
-	data, err := json.Marshal(response)
+	return response, http.StatusOK
-	if err != nil {
-		println("Error marshalling")
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	w.Write(data)
 }

app/storage/mongo/client.go

@@ -12,7 +12,7 @@ import (
 var Database mongo.Database
 
 func Connect() {
-	connectionString := fmt.Sprintf("mongodb://mongo:%s@%s:27017/", utils.GetEnv("MONGO_PASSWORD", "password"), utils.GetEnv("MONGO_HOST", "localhost"))
+	connectionString := fmt.Sprintf("mongodb://mongo:%s@mongo:27017/", utils.GetEnv("MONGO_PASSWORD", "password"))
 	client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(connectionString))
 	if err != nil {
 		panic(err)

app/storage/mongo/collections/tasks.go

@@ -9,6 +9,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
 type Task struct {
@@ -32,6 +33,14 @@ type InsertedTask struct {
 	Attempts         int        `bson:"attempts"`
 }
 
+func Count() (*int64, error) {
+	count, err := collection().CountDocuments(context.TODO(), bson.M{})
+	if err != nil {
+		return nil, err
+	}
+	return &count, nil
+}
+
 func Add(task InsertedTask) error {
 	_, err := collection().InsertOne(context.TODO(), task)
 	if err != nil {
@@ -62,7 +71,19 @@ func Take(queue string) (*Task, error) {
 	if task == nil {
 		return nil, nil
 	}
-	_, err = collection().UpdateByID(context.TODO(), task.Id, bson.M{"$set": bson.M{"taken_at": now, "attempts": task.Attempts + 1}})
+	_, err = collection().UpdateByID(
+		context.TODO(),
+		task.Id,
+		bson.M{
+			"$set": bson.M{
+				"taken_at": now,
+				"attempts": task.Attempts + 1,
+				"available_from": now.Add(
+					time.Duration(task.SecondsToExecute+task.Attempts) * time.Second,
+				),
+			},
+		},
+	)
 	if err != nil {
 		println("Error updaing")
 		println(err.Error())
@@ -78,6 +99,7 @@ func findTask(queue string, now time.Time) (*Task, error) {
 			"queue":          queue,
 			"available_from": bson.M{"$lte": now},
 		},
+		options.Find().SetSort(bson.D{{Key: "available_from", Value: 1}}).SetLimit(1),
 	)
 	if err != nil {
 		println("Error find")
@@ -94,14 +116,7 @@ func findTask(queue string, now time.Time) (*Task, error) {
 	}
 
 	for _, task := range results {
-		if task.TakenAt == nil {
+		return &task, nil
-			return &task, nil
-		}
-		takenAt := *task.TakenAt
-
-		if takenAt.Add(time.Second * time.Duration(task.SecondsToExecute)).Before(now) {
-			return &task, nil
-		}
 	}
 	return nil, nil
 }

app/storage/redis/locks.go

@@ -0,0 +1,28 @@
+package redis
+
+import (
+	"fmt"
+	"queues-go/app/utils"
+
+	"github.com/go-redsync/redsync/v4"
+	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
+	goredislib "github.com/redis/go-redis/v9"
+)
+
+var Sync redsync.Redsync
+
+func Connect() {
+	client := goredislib.NewClient(getOptions())
+	pool := goredis.NewPool(client)
+	Sync = *redsync.New(pool)
+}
+
+func getOptions() *goredislib.Options {
+	addr := fmt.Sprintf("%s:6379", utils.GetEnv("REDIS_HOST", "localhost"))
+	password := utils.GetEnv("REDIS_PASSWORD", "password")
+
+	return &goredislib.Options{
+		Addr:     addr,
+		Password: password,
+	}
+}

go.mod

@@ -2,10 +2,17 @@ module queues-go
 
 go 1.23.4
 
+require github.com/go-redsync/redsync/v4 v4.13.0
+
 require (
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/golang/snappy v0.0.4 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/montanaflynn/stats v0.7.1 // indirect
+	github.com/redis/go-redis/v9 v9.5.1 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect

go.sum

@@ -1,9 +1,22 @@
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+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/go-redsync/redsync/v4 v4.13.0 h1:49X6GJfnbLGaIpBBREM/zA4uIMDXKAh1NDkvQ1EkZKA=
+github.com/go-redsync/redsync/v4 v4.13.0/go.mod h1:HMW4Q224GZQz6x1Xc7040Yfgacukdzu7ifTDAKiyErQ=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
+github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
 github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
+github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8=
+github.com/redis/go-redis/v9 v9.5.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
 github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=

main.go

@@ -1,15 +1,96 @@
 package main
 
 import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
 	"net/http"
+	"os"
 	"queues-go/app/routers"
 	client "queues-go/app/storage/mongo"
+	tasks "queues-go/app/storage/mongo/collections"
+	"strconv"
+	"sync"
+	"time"
 )
 
+func writeMetric(timestamp time.Time, endpoint string, statusCode int, responseTime int, method string) {
+	loc, _ := time.LoadLocation("Europe/Moscow")
+	s := fmt.Sprintf(
+		`{"timestamp":"%s","service":"queues","environment":"%s","endpoint":"%s","status_code":%s,"response_time":%s,"method":"%s"}`,
+		timestamp.In(loc).Format("2006-01-02T15:04:05Z"),
+		os.Getenv("STAGE"),
+		endpoint,
+		strconv.Itoa(statusCode),
+		strconv.Itoa(responseTime),
+		method,
+	)
+	data := []byte(s)
+	r := bytes.NewReader(data)
+	_, err := http.Post("http://monitoring:1237/api/v1/metrics/endpoint", "application/json", r)
+	if err != nil {
+		log.Printf("Error sending metrics %s", err.Error())
+	}
+}
+
+func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		start := time.Now()
+		response, status := f(r)
+		if response != nil {
+			data, err := json.Marshal(response)
+			if err != nil {
+				log.Printf("Error marshalling response")
+				status = http.StatusInternalServerError
+			}
+			w.WriteHeader(status)
+			w.Write(data)
+		} else {
+			w.WriteHeader(status)
+		}
+		go writeMetric(start, r.URL.Path, status, int(time.Since(start).Milliseconds()), r.Method)
+		log.Printf("%s %d %s", r.URL, status, time.Since(start))
+	}
+}
+
+func metricProxy(w http.ResponseWriter, r *http.Request) {
+	http.Post("http://monitoring:1237/api/v1/metrics/task", "application/json", r.Body)
+	w.WriteHeader(202)
+}
+
+func writeCount() {
+	for {
+		count, err := tasks.Count()
+		if err != nil {
+			log.Printf("Failed getting docs count: %s", err.Error())
+		} else {
+			loc, _ := time.LoadLocation("Europe/Moscow")
+			s := fmt.Sprintf(
+				`{"timestamp":"%s","service":"queues","environment":"%s","name":"tasks","count":%s}`,
+				time.Now().In(loc).Format("2006-01-02T15:04:05Z"),
+				os.Getenv("STAGE"),
+				strconv.Itoa(int(*count)),
+			)
+			data := []byte(s)
+			r := bytes.NewReader(data)
+			_, err := http.Post("http://monitoring:1237/api/v1/metrics/increment", "application/json", r)
+			if err != nil {
+				log.Printf("ERROR %s", err.Error())
+			}
+		}
+		time.Sleep(time.Second)
+	}
+}
+
 func main() {
 	client.Connect()
-	http.HandleFunc("/api/v1/take", routers.Take)
+	routers.MutexMap = make(map[string]*sync.Mutex)
-	http.HandleFunc("/api/v1/finish", routers.Finish)
+	http.HandleFunc("/api/v1/take", handlerWrapper(routers.Take))
-	http.HandleFunc("/api/v1/put", routers.Put)
+	http.HandleFunc("/api/v1/finish", handlerWrapper(routers.Finish))
+	http.HandleFunc("/api/v1/put", handlerWrapper(routers.Put))
+	http.HandleFunc("/api/v1/metric", metricProxy)
+	log.Printf("Server started")
+	go writeCount()
 	http.ListenAndServe(":1239", nil)
 }