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

Merge pull request 'fix' (#28) from master into prod

Browse Source 
Reviewed-on: https://gitea.chocomarsh.com/self/monitoring/pulls/28
This commit is contained in:
emmatveev 2025-06-21 12:09:01 +03:00
commit 6a1b0c7b7f
8 changed files with 173 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,23 +3,24 @@ 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
}
return nil, http.StatusAccepted
}
}

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

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

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

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

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

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

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

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

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

@ -1,34 +1,11 @@
package storage
import (
"context"
"time"
client "monitoring/app/storage/clickhouse"
)
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()
func AddIncrementMetric(metric client.IncrementMetric) {
client.IncrementsMutex.Lock()
defer client.IncrementsMutex.Unlock()
client.IncrementsCol = append(client.IncrementsCol, metric)
}

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

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

5
main.go
View File

@ -14,7 +14,7 @@ import (
)
func writeMetric(timestamp time.Time, endpoint string, statusCode int, responseTime int, method string) {
err := endpoints.AddEndpointMetric(endpoints.EndpointMetric{
endpoints.AddEndpointMetric(client.EndpointMetric{
Timestamp: timestamp.Add(time.Hour * 3),
Service: "monitoring",
Environment: os.Getenv("STAGE"),
@ -23,9 +23,6 @@ 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) {