Merge pull request 'fix' (#9) from master into dev

Reviewed-on: #9

.deploy/deploy-dev.yaml

@@ -8,7 +8,6 @@ services:
       - clickhouse-development
     environment:
       CLICKHOUSE_PASSWORD: $CLICKHOUSE_PASSWORD_DEV
-      STAGE: "development"
     deploy:
       mode: replicated
       restart_policy:

.deploy/deploy-prod.yaml

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

.gitea/workflows/deploy-prod.yaml

@@ -1,43 +0,0 @@
-name: Deploy Prod
-
-on:
-  pull_request:
-    branches:
-      - prod
-    types: [closed]
-
-jobs:
-  build:
-    name: Build
-    runs-on: [dev]
-    steps:
-      - name: login
-        run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
-      - name: checkout
-        uses: actions/checkout@v4
-        with:
-          ref: prod
-      - name: build
-        run: docker build -t mathwave/sprint-repo:monitoring .
-  push:
-    name: Push
-    runs-on: [dev]
-    needs: build
-    steps:
-      - name: push
-        run: docker push mathwave/sprint-repo:monitoring
-  deploy-prod:
-    name: Deploy prod
-    runs-on: [prod]
-    needs: push
-    steps:
-      - name: login
-        run: docker login -u mathwave -p ${{ secrets.DOCKERHUB_PASSWORD }}
-      - name: checkout
-        uses: actions/checkout@v4
-        with:
-            ref: prod
-      - name: deploy
-        env:
-          CLICKHOUSE_PASSWORD_PROD: ${{ secrets.CLICKHOUSE_PASSWORD_PROD }}
-        run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml infra

app/routers/metrics/endpoint.go

@@ -3,22 +3,21 @@ package routers
 import (
 	"encoding/json"
 	"log"
-	client "monitoring/app/storage/clickhouse"
 	endpoints "monitoring/app/storage/clickhouse/tables"
 	"net/http"
 )
 
-func AddEndpointMetric(r *http.Request) (interface{}, int) {
+func AddEndpointMetric (r *http.Request) (interface{}, int) {
 	d := json.NewDecoder(r.Body)
-	body := client.EndpointMetric{}
+	body := endpoints.EndpointMetric{}
 	err := d.Decode(&body)
 	if err != nil {
 		return nil, http.StatusBadRequest
 	}
 
-	endpoints.AddEndpointMetric(body)
+	err = endpoints.AddEndpointMetric(body)
 	if err != nil {
-		log.Print(err.Error())
+		log.Printf(err.Error())
 		return nil, http.StatusInternalServerError
 	}
 

app/routers/metrics/increment.go

@@ -1,26 +0,0 @@
-package routers
-
-import (
-	"encoding/json"
-	"log"
-	client "monitoring/app/storage/clickhouse"
-	increments "monitoring/app/storage/clickhouse/tables"
-	"net/http"
-)
-
-func AddIncrementMetric(r *http.Request) (interface{}, int) {
-	d := json.NewDecoder(r.Body)
-	body := client.IncrementMetric{}
-	err := d.Decode(&body)
-	if err != nil {
-		return nil, http.StatusBadRequest
-	}
-
-	increments.AddIncrementMetric(body)
-	if err != nil {
-		log.Print(err.Error())
-		return nil, http.StatusInternalServerError
-	}
-
-	return nil, http.StatusAccepted
-}

app/routers/metrics/task.go

@@ -1,26 +0,0 @@
-package routers
-
-import (
-	"encoding/json"
-	"log"
-	client "monitoring/app/storage/clickhouse"
-	tasks "monitoring/app/storage/clickhouse/tables"
-	"net/http"
-)
-
-func AddTaskMetric(r *http.Request) (interface{}, int) {
-	d := json.NewDecoder(r.Body)
-	body := client.TaskMetric{}
-	err := d.Decode(&body)
-	if err != nil {
-		return nil, http.StatusBadRequest
-	}
-
-	tasks.AddTaskMetric(body)
-	if err != nil {
-		log.Print(err.Error())
-		return nil, http.StatusInternalServerError
-	}
-
-	return nil, http.StatusAccepted
-}

app/storage/clickhouse/client.go

@@ -3,49 +3,13 @@ package storage
 import (
 	"context"
 	"fmt"
-	"log"
 	"os"
-	"sync"
-	"time"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 )
 
-type EndpointMetric struct {
-	Timestamp    time.Time `json:"timestamp"`
-	Service      string    `json:"service"`
-	Environment  string    `json:"environment"`
-	Endpoint     string    `json:"endpoint"`
-	StatusCode   int       `json:"status_code"`
-	ResponseTime int       `json:"response_time"`
-	Method       string    `json:"method"`
-}
-
-type IncrementMetric struct {
-	Timestamp   time.Time `json:"timestamp"`
-	Service     string    `json:"service"`
-	Environment string    `json:"environment"`
-	Name        string    `json:"name"`
-	Count       int       `json:"count"`
-}
-
-type TaskMetric struct {
-	Timestamp       time.Time `json:"timestamp"`
-	Service         string    `json:"service"`
-	Environment     string    `json:"environment"`
-	Queue           string    `json:"queue"`
-	Success         bool      `json:"success"`
-	ExecutionTimeMs int       `json:"execution_time_ms"`
-}
-
 var Connection driver.Conn
-var EndpointsCol []EndpointMetric
-var EndpointsMutex sync.Mutex
-var IncrementsCol []IncrementMetric
-var IncrementsMutex sync.Mutex
-var TasksCol []TaskMetric
-var TasksMutex sync.Mutex
 
 func Connect() error {
 	conn, err := connect()
@@ -53,23 +17,12 @@ func Connect() error {
 		return err
 	}
 	Connection = *conn
-	EndpointsCol = make([]EndpointMetric, 0)
-	IncrementsCol = make([]IncrementMetric, 0)
-	TasksCol = make([]TaskMetric, 0)
-	EndpointsMutex = sync.Mutex{}
-	IncrementsMutex = sync.Mutex{}
-	TasksMutex = sync.Mutex{}
-	go pushMetrics()
 	return nil
 }
 
-func Close() {
-	Connection.Close()
-}
-
 func connect() (*driver.Conn, error) {
 	var (
-		ctx       = context.Background()
+		ctx = context.Background()
 		conn, err = clickhouse.Open(&clickhouse.Options{
 			Addr: []string{"clickhouse:9000"},
 			Auth: clickhouse.Auth{
@@ -93,169 +46,3 @@ func connect() (*driver.Conn, error) {
 	}
 	return &conn, nil
 }
-
-func pushEndpoints() error {
-	if len(EndpointsCol) == 0 {
-		return nil
-	}
-
-	EndpointsMutex.Lock()
-	newCollection := EndpointsCol
-	EndpointsCol = make([]EndpointMetric, 0)
-	EndpointsMutex.Unlock()
-
-	batch, err := Connection.PrepareBatch(context.Background(), "INSERT INTO endpoints")
-	if err != nil {
-		return err
-	}
-
-	for _, metric := range newCollection {
-		err := batch.Append(
-			metric.Timestamp,
-			metric.Service,
-			metric.Environment,
-			metric.Endpoint,
-			metric.StatusCode,
-			metric.ResponseTime,
-			metric.Method,
-		)
-		if err != nil {
-			return err
-		}
-	}
-
-	return batch.Send()
-}
-
-func pushIncrements() error {
-	if len(IncrementsCol) == 0 {
-		return nil
-	}
-
-	IncrementsMutex.Lock()
-	newCollection := IncrementsCol
-	IncrementsCol = make([]IncrementMetric, 0)
-	IncrementsMutex.Unlock()
-
-	batch, err := Connection.PrepareBatch(context.Background(), "INSERT INTO increments")
-	if err != nil {
-		return err
-	}
-
-	for _, metric := range newCollection {
-		err := batch.Append(
-			metric.Timestamp,
-			metric.Service,
-			metric.Environment,
-			metric.Name,
-			metric.Count,
-		)
-		if err != nil {
-			return err
-		}
-	}
-
-	return batch.Send()
-}
-
-func pushTasks() error {
-	if len(IncrementsCol) == 0 {
-		return nil
-	}
-
-	TasksMutex.Lock()
-	newCollection := TasksCol
-	TasksCol = make([]TaskMetric, 0)
-	TasksMutex.Unlock()
-
-	batch, err := Connection.PrepareBatch(context.Background(), "INSERT INTO tasks")
-	if err != nil {
-		return err
-	}
-
-	for _, metric := range newCollection {
-		err := batch.Append(
-			metric.Timestamp,
-			metric.Service,
-			metric.Environment,
-			metric.Queue,
-			metric.Success,
-			metric.ExecutionTimeMs,
-		)
-		if err != nil {
-			return err
-		}
-	}
-
-	return batch.Send()
-}
-
-func pushMetrics() {
-	for {
-		pushEndpoints()
-		pushIncrements()
-		pushTasks()
-		time.Sleep(time.Second)
-	}
-}
-
-func Migrate() error {
-	err := Connection.Exec(
-		context.TODO(),
-		`CREATE TABLE IF NOT EXISTS endpoints (
-			timestamp        DateTime,
-			service          LowCardinality(String),
-			environment      LowCardinality(String),
-			endpoint         LowCardinality(String),
-			status_code      UInt16,
-			response_time_ms UInt32,
-			method           LowCardinality(String)
-		)
-		ENGINE = MergeTree
-		PARTITION BY toYYYYMM(timestamp)
-		ORDER BY (service, environment, endpoint, method, timestamp);`,
-	)
-	if err != nil {
-		log.Fatal(err)
-		return err
-	}
-
-	err = Connection.Exec(
-		context.TODO(),
-		`CREATE TABLE IF NOT EXISTS tasks (
-			timestamp         DateTime,
-			service           LowCardinality(String),
-			environment       LowCardinality(String),
-			queue             LowCardinality(String),
-			success           Bool,
-			execution_time_ms UInt32
-		)
-		ENGINE = MergeTree
-		PARTITION BY toYYYYMM(timestamp)
-		ORDER BY (service, environment, queue, timestamp);`,
-	)
-	if err != nil {
-		log.Fatal(err)
-		return err
-	}
-
-	err = Connection.Exec(
-		context.TODO(),
-		`CREATE TABLE IF NOT EXISTS increments (
-			timestamp         DateTime,
-			service           LowCardinality(String),
-			environment       LowCardinality(String),
-			name              LowCardinality(String),
-			count             UInt16
-		)
-		ENGINE = MergeTree
-		PARTITION BY toYYYYMM(timestamp)
-		ORDER BY (service, environment, name, timestamp);`,
-	)
-	if err != nil {
-		log.Fatal(err)
-		return err
-	}
-
-	return nil
-}

app/storage/clickhouse/tables/endpoints.go

@@ -1,11 +1,42 @@
 package storage
 
 import (
+	"context"
 	client "monitoring/app/storage/clickhouse"
+
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
 )
 
-func AddEndpointMetric(metric client.EndpointMetric) {
-	client.EndpointsMutex.Lock()
-	defer client.EndpointsMutex.Unlock()
-	client.EndpointsCol = append(client.EndpointsCol, metric)
+type EndpointMetric struct {
+	Timestamp int `json:"timestamp"`
+	Service string `json:"service"`
+	Endpoint string `json:"endpoint"`
+	StatusCode int `json:"status_code"`
+	ResponseTime int `json:"response_time"`
+	Method string `json:"method"`
+}
+
+func AddEndpointMetric(metric EndpointMetric) error {
+	batch, err := connection().PrepareBatch(context.Background(), "INSERT INTO endpoints")
+	if err != nil {
+		return err
+	}
+
+	err = batch.Append(
+		metric.Timestamp,
+		metric.Service,
+		metric.Endpoint,
+		metric.StatusCode,
+		metric.ResponseTime,
+		metric.Method,
+	)
+	if err != nil {
+		return err
+	}
+
+	return batch.Send()
+}
+
+func connection() driver.Conn {
+	return client.Connection
 }

app/storage/clickhouse/tables/increments.go

@@ -1,11 +0,0 @@
-package storage
-
-import (
-	client "monitoring/app/storage/clickhouse"
-)
-
-func AddIncrementMetric(metric client.IncrementMetric) {
-	client.IncrementsMutex.Lock()
-	defer client.IncrementsMutex.Unlock()
-	client.IncrementsCol = append(client.IncrementsCol, metric)
-}

app/storage/clickhouse/tables/tasks.go

@@ -1,11 +0,0 @@
-package storage
-
-import (
-	client "monitoring/app/storage/clickhouse"
-)
-
-func AddTaskMetric(metric client.TaskMetric) {
-	client.TasksMutex.Lock()
-	defer client.TasksMutex.Unlock()
-	client.TasksCol = append(client.TasksCol, metric)
-}

main.go

@@ -4,27 +4,11 @@ import (
 	"encoding/json"
 	"log"
 	endpoint "monitoring/app/routers/metrics"
-	increment "monitoring/app/routers/metrics"
-	task "monitoring/app/routers/metrics"
 	client "monitoring/app/storage/clickhouse"
-	endpoints "monitoring/app/storage/clickhouse/tables"
 	"net/http"
-	"os"
 	"time"
 )
 
-func writeMetric(timestamp time.Time, endpoint string, statusCode int, responseTime int, method string) {
-	endpoints.AddEndpointMetric(client.EndpointMetric{
-		Timestamp:    timestamp.Add(time.Hour * 3),
-		Service:      "monitoring",
-		Environment:  os.Getenv("STAGE"),
-		Endpoint:     endpoint,
-		StatusCode:   statusCode,
-		ResponseTime: responseTime,
-		Method:       method,
-	})
-}
-
 func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		start := time.Now()
@@ -40,7 +24,6 @@ func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.Response
 		} 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))
 	}
 }
@@ -50,16 +33,8 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	defer client.Close()
-
-	err = client.Migrate()
-	if err != nil {
-		panic(err)
-	}
 
 	http.HandleFunc("/api/v1/metrics/endpoint", handlerWrapper(endpoint.AddEndpointMetric))
-	http.HandleFunc("/api/v1/metrics/task", handlerWrapper(task.AddTaskMetric))
-	http.HandleFunc("/api/v1/metrics/increment", handlerWrapper(increment.AddIncrementMetric))
 	log.Printf("Server started")
 	http.ListenAndServe(":1237", nil)
 }