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

Reviewed-on: #19

.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

@@ -5,10 +5,8 @@ services:
     image: mathwave/sprint-repo:monitoring
     networks:
       - clickhouse
-      - monitoring
     environment:
       CLICKHOUSE_PASSWORD: $CLICKHOUSE_PASSWORD_PROD
-      STAGE: "production"
     deploy:
       mode: replicated
       restart_policy:
@@ -20,5 +18,3 @@ services:
 networks:
   clickhouse:
     external: true
-  monitoring:
-    external: true

app/routers/metrics/endpoint.go

@@ -3,20 +3,19 @@ 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())
 		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

@@ -3,20 +3,19 @@ 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{}
+	body := tasks.TaskMetric{}
 	err := d.Decode(&body)
 	if err != nil {
 		return nil, http.StatusBadRequest
 	}
 
-	tasks.AddTaskMetric(body)
+	err = tasks.AddTaskMetric(body)
 	if err != nil {
 		log.Print(err.Error())
 		return nil, http.StatusInternalServerError

app/storage/clickhouse/client.go

@@ -5,47 +5,12 @@ import (
 	"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,13 +18,6 @@ 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
 }
 
@@ -94,111 +52,6 @@ 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(),
@@ -238,24 +91,5 @@ func Migrate() error {
 		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,45 @@
 package storage
 
 import (
+	"context"
 	client "monitoring/app/storage/clickhouse"
+	"time"
+
+	"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    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"`
+}
+
+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.Environment,
+		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 +1,36 @@
 package storage
 
 import (
-	client "monitoring/app/storage/clickhouse"
+	"context"
+	"time"
 )
 
-func AddTaskMetric(metric client.TaskMetric) {
-	client.TasksMutex.Lock()
-	defer client.TasksMutex.Unlock()
-	client.TasksCol = append(client.TasksCol, metric)
+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"`
+}
+
+func AddTaskMetric(metric TaskMetric) error {
+	batch, err := connection().PrepareBatch(context.Background(), "INSERT INTO tasks")
+	if err != nil {
+		return err
+	}
+
+	err = batch.Append(
+		metric.Timestamp,
+		metric.Service,
+		metric.Environment,
+		metric.Queue,
+		metric.Success,
+		metric.ExecutionTimeMs,
+	)
+	if err != nil {
+		return err
+	}
+
+	return batch.Send()
 }

main.go

@@ -4,27 +4,12 @@ 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 +25,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))
 	}
 }
@@ -59,7 +43,6 @@ func main() {
 
 	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)
 }