self/monitoring
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: self:6a1b0c7b7f40da1a250ff455a00ffd22222eed41
Branches Tags
self:master
self:prod
self:dev
..
compare: self:028db059fe2cca50edb1308c623293ddb3130f87
Branches Tags
self:prod
self:dev
self:master

No commits in common. "6a1b0c7b7f40da1a250ff455a00ffd22222eed41" and "028db059fe2cca50edb1308c623293ddb3130f87" have entirely different histories.
6a1b0c7b7f ... 028db059fe

8 changed files with 108 additions and 173 deletions
Show Stats Expand all files Collapse all files

5
app/routers/metrics/endpoint.go
View File

@ -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) {
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

5
app/routers/metrics/increment.go
View File

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

5
app/routers/metrics/task.go
View File

@ -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

147
app/storage/clickhouse/client.go
View File

@ -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(),

42
app/storage/clickhouse/tables/endpoints.go
View File

@ -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
}

33
app/storage/clickhouse/tables/increments.go
View File

@ -1,11 +1,34 @@
package storage
import (
client "monitoring/app/storage/clickhouse"
"context"
"time"
)
func AddIncrementMetric(metric client.IncrementMetric) {
client.IncrementsMutex.Lock()
defer client.IncrementsMutex.Unlock()
client.IncrementsCol = append(client.IncrementsCol, metric)
type IncrementMetric struct {
Timestamp time.Time `json:"timestamp"`
Service string `json:"service"`
Environment string `json:"environment"`
Name string `json:"name"`
Count int `json:"count"`
}
func AddIncrementMetric(metric IncrementMetric) error {
batch, err := connection().PrepareBatch(context.Background(), "INSERT INTO increments")
if err != nil {
return err
}
err = batch.Append(
metric.Timestamp,
metric.Service,
metric.Environment,
metric.Name,
metric.Count,
)
if err != nil {
return err
}
return batch.Send()
}

35
app/storage/clickhouse/tables/tasks.go
View File

@ -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()
}

5
main.go
View File

@ -14,7 +14,7 @@ import (
)
func writeMetric(timestamp time.Time, endpoint string, statusCode int, responseTime int, method string) {
endpoints.AddEndpointMetric(client.EndpointMetric{
err := endpoints.AddEndpointMetric(endpoints.EndpointMetric{
Timestamp: timestamp.Add(time.Hour * 3),
Service: "monitoring",
Environment: os.Getenv("STAGE"),
@ -23,6 +23,9 @@ func writeMetric(timestamp time.Time, endpoint string, statusCode int, responseT
ResponseTime: responseTime,
Method: method,
})
if err != nil {
log.Printf("Error sending metrics %s", err.Error())
}
}
func handlerWrapper(f func(*http.Request) (interface{}, int)) func(http.ResponseWriter, *http.Request) {