Merge pull request 'master' (#27) from master into dev

Reviewed-on: #27

.deploy/deploy-dev.yaml

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

.deploy/deploy-prod.yaml

@@ -8,6 +8,7 @@ services:
       - monitoring
     environment:
       CLICKHOUSE_PASSWORD: $CLICKHOUSE_PASSWORD_PROD
+      STAGE: "production"
     deploy:
       mode: replicated
       restart_policy:

app/routers/metrics/endpoint.go

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

app/routers/metrics/increment.go

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

app/storage/clickhouse/client.go

@@ -5,12 +5,47 @@ 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()
@@ -18,6 +53,13 @@ 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
 }
 
@@ -52,6 +94,111 @@ 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(),
@@ -91,5 +238,24 @@ 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,45 +1,11 @@
 package storage
 
 import (
-	"context"
 	client "monitoring/app/storage/clickhouse"
-	"time"
-
-	"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"`
-}
-
-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
+func AddEndpointMetric(metric client.EndpointMetric) {
+	client.EndpointsMutex.Lock()
+	defer client.EndpointsMutex.Unlock()
+	client.EndpointsCol = append(client.EndpointsCol, metric)
 }

app/storage/clickhouse/tables/increments.go

@@ -0,0 +1,11 @@
+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,36 +1,11 @@
 package storage
 
 import (
-	"context"
-	"time"
+	client "monitoring/app/storage/clickhouse"
 )
 
-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()
+func AddTaskMetric(metric client.TaskMetric) {
+	client.TasksMutex.Lock()
+	defer client.TasksMutex.Unlock()
+	client.TasksCol = append(client.TasksCol, metric)
 }

main.go

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