cacto/home_automatization
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Добавлены метрики Prometheus и дашборды Grafana (этап 2.5)

Browse Source 
Все три Go-сервиса (ingest, device-control, rule-engine) отдают
/metrics в формате Prometheus: счётчики MQTT/RabbitMQ/ClickHouse
операций, гистограммы длительности батч-флашей и диспатча правил,
переходы устройств online/offline. Добавлены сервисы prometheus и
grafana в docker-compose с провижининг конфигом (datasource +
два готовых дашборда: "Ingest & Telemetry" и "Automation & Devices").
This commit is contained in:
cacto
2026-08-11 21:58:59 +05:00
parent 3ade5c2512
commit 81040eec62
31 changed files with 712 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/device-control-service/README.md
+7
View File
@@ -49,6 +49,13 @@
Laravel→device-control. gRPC-контракт между Go-сервисами (rule-engine)
не тронут.
## Метрики
`GET /metrics` на том же порту, что и команды (`DEVICE_CONTROL_HTTP_PORT`)
— не открывали отдельный порт ради этого. Счётчики/latency команд
(TurnOn/TurnOff/SetLevel по action+outcome), MQTT-событий (telemetry/ack),
переходов online/offline health-check.
## Запуск
```bash
services/device-control-service/cmd/device-control-service/main.go
+17 -2
View File
@@ -16,12 +16,14 @@ import (
"syscall"
"time"
"github.com/prometheus/client_golang/prometheus/promhttp"
"google.golang.org/grpc"
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/config"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/healthcheck"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/httpapi"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/mqttclient"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/rabbitmq"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/server"
@@ -74,6 +76,8 @@ func run(logger *slog.Logger) error {
}
checker := healthcheck.New(shadowStore, cfg.HealthCheckTimeout, cfg.HealthCheckInterval, func(deviceID string) {
metrics.HealthTransitionsTotal.WithLabelValues("offline").Inc()
pubCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := publisher.PublishStatusChanged(pubCtx, deviceID, shadow.StatusOffline); err != nil {
@@ -93,10 +97,14 @@ func run(logger *slog.Logger) error {
devicecontrol.RegisterDeviceControlServer(grpcServer, srv)
// Same server instance, second transport — for callers where a full
// gRPC client is impractical (Laravel/PHP; see README).
// gRPC client is impractical (Laravel/PHP; see README). /metrics rides
// along on this same port rather than opening a third one.
mux := http.NewServeMux()
mux.Handle("/", httpapi.NewRouter(srv, logger))
mux.Handle("/metrics", promhttp.Handler())
httpServer := &http.Server{
Addr: fmt.Sprintf(":%d", cfg.HTTPPort),
Handler: httpapi.NewRouter(srv, logger),
Handler: mux,
}
go func() {
@@ -145,8 +153,10 @@ func telemetryHandler(store *shadow.Store, publisher *rabbitmq.Publisher, logger
deviceID, ok := deviceIDFromTopic(topic)
if !ok {
logger.Warn("dropping telemetry from unparseable topic", "topic", topic)
metrics.MQTTEventsTotal.WithLabelValues("telemetry", "invalid").Inc()
return
}
metrics.MQTTEventsTotal.WithLabelValues("telemetry", "ok").Inc()
touch(context.Background(), store, publisher, deviceID, logger)
}
}
@@ -160,14 +170,17 @@ func ackHandler(store *shadow.Store, publisher *rabbitmq.Publisher, logger *slog
deviceID, ok := deviceIDFromTopic(topic)
if !ok {
logger.Warn("dropping ack from unparseable topic", "topic", topic)
metrics.MQTTEventsTotal.WithLabelValues("ack", "invalid").Inc()
return
}
var ack ackPayload
if err := json.Unmarshal(payload, &ack); err != nil {
logger.Warn("dropping invalid ack payload", "device_id", deviceID, "error", err)
metrics.MQTTEventsTotal.WithLabelValues("ack", "invalid").Inc()
return
}
metrics.MQTTEventsTotal.WithLabelValues("ack", "ok").Inc()
ctx := context.Background()
if len(ack.State) > 0 {
@@ -186,6 +199,8 @@ func touch(ctx context.Context, store *shadow.Store, publisher *rabbitmq.Publish
return
}
if becameOnline {
metrics.HealthTransitionsTotal.WithLabelValues("online").Inc()
pubCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
if err := publisher.PublishStatusChanged(pubCtx, deviceID, shadow.StatusOnline); err != nil {
services/device-control-service/go.mod
+11 -4
View File
@@ -6,20 +6,27 @@ require (
git.cactoz.su/cacto/home_automatization/proto v0.0.0
github.com/alicebob/miniredis/v2 v2.38.0
github.com/eclipse/paho.mqtt.golang v1.5.1
github.com/prometheus/client_golang v1.24.1
github.com/rabbitmq/amqp091-go v1.13.0
github.com/redis/go-redis/v9 v9.21.0
google.golang.org/grpc v1.82.1
)
require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/kylelemons/godebug v1.1.0 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.70.1 // indirect
github.com/prometheus/procfs v0.21.1 // indirect
github.com/yuin/gopher-lua v1.1.1 // indirect
go.uber.org/atomic v1.11.0 // indirect
golang.org/x/net v0.53.0 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.43.0 // indirect
golang.org/x/text v0.36.0 // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
google.golang.org/protobuf v1.36.11 // indirect
)
services/device-control-service/go.sum
+30 -10
View File
@@ -1,5 +1,7 @@
github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw=
github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
@@ -22,16 +24,30 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E=
github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
@@ -52,14 +68,16 @@ go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw=
@@ -68,3 +86,5 @@ google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
services/device-control-service/internal/metrics/metrics.go
+32
View File
@@ -0,0 +1,32 @@
// Package metrics defines device-control-service's Prometheus metrics.
// Registered automatically (via promauto) into the default registry on
// import; served alongside the command HTTP API's /metrics route.
package metrics
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
CommandsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "device_control_commands_total",
Help: "Total number of TurnOn/TurnOff/SetLevel commands dispatched, by action and outcome.",
}, []string{"action", "outcome"})
CommandDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "device_control_command_duration_seconds",
Help: "Duration of a command dispatch (Redis desired_state patch + MQTT publish), by action.",
Buckets: prometheus.DefBuckets,
}, []string{"action"})
MQTTEventsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "device_control_mqtt_events_total",
Help: "Total number of telemetry/ack messages received, by event type and outcome.",
}, []string{"event_type", "outcome"})
HealthTransitionsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "device_control_health_transitions_total",
Help: "Total number of device online/offline transitions detected.",
}, []string{"direction"})
)
services/device-control-service/internal/server/server.go
+28 -12
View File
@@ -9,8 +9,10 @@ import (
"encoding/json"
"fmt"
"log/slog"
"time"
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics"
)
// ShadowPatcher is the subset of shadow.Store the server needs.
@@ -36,21 +38,21 @@ func New(shadow ShadowPatcher, mqtt CommandPublisher, logger *slog.Logger) *Serv
}
func (s *Server) TurnOn(ctx context.Context, req *devicecontrol.TurnOnRequest) (*devicecontrol.CommandResult, error) {
return s.dispatch(ctx, req.GetDeviceId(),
return s.dispatch(ctx, "turn_on", req.GetDeviceId(),
map[string]any{"power": "on"},
map[string]any{"action": "turn_on"},
)
}
func (s *Server) TurnOff(ctx context.Context, req *devicecontrol.TurnOffRequest) (*devicecontrol.CommandResult, error) {
return s.dispatch(ctx, req.GetDeviceId(),
return s.dispatch(ctx, "turn_off", req.GetDeviceId(),
map[string]any{"power": "off"},
map[string]any{"action": "turn_off"},
)
}
func (s *Server) SetLevel(ctx context.Context, req *devicecontrol.SetLevelRequest) (*devicecontrol.CommandResult, error) {
return s.dispatch(ctx, req.GetDeviceId(),
return s.dispatch(ctx, "set_level", req.GetDeviceId(),
map[string]any{"level": req.GetLevel()},
map[string]any{"action": "set_level", "level": req.GetLevel()},
)
@@ -60,27 +62,41 @@ func (s *Server) SetLevel(ctx context.Context, req *devicecontrol.SetLevelReques
// gets an instant, optimistic view even if the device is offline), then hand
// the command to MQTT. Business failures come back as CommandResult.Error,
// not a gRPC error — the wire contract is success/error in one message.
func (s *Server) dispatch(ctx context.Context, deviceID string, desiredPatch, commandPayload map[string]any) (*devicecontrol.CommandResult, error) {
func (s *Server) dispatch(ctx context.Context, action, deviceID string, desiredPatch, commandPayload map[string]any) (*devicecontrol.CommandResult, error) {
start := time.Now()
result := s.doDispatch(ctx, action, deviceID, desiredPatch, commandPayload)
metrics.CommandDuration.WithLabelValues(action).Observe(time.Since(start).Seconds())
outcome := "success"
if !result.Success {
outcome = "error"
}
metrics.CommandsTotal.WithLabelValues(action, outcome).Inc()
return result, nil
}
func (s *Server) doDispatch(ctx context.Context, action, deviceID string, desiredPatch, commandPayload map[string]any) *devicecontrol.CommandResult {
if deviceID == "" {
return &devicecontrol.CommandResult{Success: false, Error: "device_id is required"}, nil
return &devicecontrol.CommandResult{Success: false, Error: "device_id is required"}
}
if err := s.shadow.PatchDesiredState(ctx, deviceID, desiredPatch); err != nil {
s.logger.Error("patch desired state failed", "device_id", deviceID, "error", err)
return &devicecontrol.CommandResult{Success: false, Error: "failed to update desired state"}, nil
s.logger.Error("patch desired state failed", "action", action, "device_id", deviceID, "error", err)
return &devicecontrol.CommandResult{Success: false, Error: "failed to update desired state"}
}
body, err := json.Marshal(commandPayload)
if err != nil {
s.logger.Error("encode command failed", "device_id", deviceID, "error", err)
return &devicecontrol.CommandResult{Success: false, Error: "failed to encode command"}, nil
s.logger.Error("encode command failed", "action", action, "device_id", deviceID, "error", err)
return &devicecontrol.CommandResult{Success: false, Error: "failed to encode command"}
}
topic := fmt.Sprintf("devices/%s/commands", deviceID)
if err := s.mqtt.Publish(topic, 1, false, body); err != nil {
s.logger.Error("publish command failed", "device_id", deviceID, "topic", topic, "error", err)
return &devicecontrol.CommandResult{Success: false, Error: "failed to dispatch command"}, nil
s.logger.Error("publish command failed", "action", action, "device_id", deviceID, "topic", topic, "error", err)
return &devicecontrol.CommandResult{Success: false, Error: "failed to dispatch command"}
}
return &devicecontrol.CommandResult{Success: true}, nil
return &devicecontrol.CommandResult{Success: true}
}
services/device-control-service/internal/server/server_test.go
+27
View File
@@ -8,7 +8,10 @@ import (
"log/slog"
"testing"
"github.com/prometheus/client_golang/prometheus/testutil"
devicecontrol "git.cactoz.su/cacto/home_automatization/proto/device_control"
"git.cactoz.su/cacto/home_automatization/services/device-control-service/internal/metrics"
)
type fakeShadow struct {
@@ -122,3 +125,27 @@ func TestDispatch_PublishFailure(t *testing.T) {
t.Fatal("expected failure when mqtt publish errors")
}
}
func TestDispatch_RecordsMetrics(t *testing.T) {
successCounter := metrics.CommandsTotal.WithLabelValues("turn_on", "success")
errorCounter := metrics.CommandsTotal.WithLabelValues("turn_on", "error")
before := testutil.ToFloat64(successCounter)
s := New(&fakeShadow{}, &fakeMQTT{}, discardLogger())
if _, err := s.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"}); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got := testutil.ToFloat64(successCounter); got != before+1 {
t.Fatalf("got turn_on/success counter %v, want %v", got, before+1)
}
beforeErr := testutil.ToFloat64(errorCounter)
s2 := New(&fakeShadow{err: errors.New("redis down")}, &fakeMQTT{}, discardLogger())
if _, err := s2.TurnOn(context.Background(), &devicecontrol.TurnOnRequest{DeviceId: "d1"}); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got := testutil.ToFloat64(errorCounter); got != beforeErr+1 {
t.Fatalf("got turn_on/error counter %v, want %v", got, beforeErr+1)
}
}
services/ingest-service/README.md
+7
View File
@@ -20,6 +20,13 @@ MQTT/ClickHouse/RabbitMQ, как и на диаграмме архитектур
понадобится lookup через реестр устройств (с кэшированием, как rule-engine
кэширует правила) — это естественное развитие, пока не реализовано.
## Метрики
`GET /metrics` (порт `INGEST_METRICS_PORT`, по умолчанию 9101) — формат
Prometheus. Счётчики принятых/невалидных/отброшенных (backlog переполнен)
MQTT-сообщений, флашей батча в ClickHouse (с latency и размером батча),
публикаций в RabbitMQ.
## Запуск
```bash
services/ingest-service/cmd/ingest-service/main.go
+49 -1
View File
@@ -5,16 +5,21 @@ package main
import (
"context"
"fmt"
"log/slog"
"net/http"
"os"
"os/signal"
"sync"
"syscall"
"time"
"github.com/prometheus/client_golang/prometheus/promhttp"
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/batch"
chstore "git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/clickhouse"
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/config"
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/metrics"
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/mqttclient"
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/rabbitmq"
"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
@@ -59,7 +64,7 @@ func run(logger *slog.Logger) error {
}
defer publisher.Close()
batcher := batch.New(cfg.BatchMaxSize, cfg.BatchFlushInterval, cfg.BatchFlushTimeout, store.InsertBatch, logger)
batcher := batch.New(cfg.BatchMaxSize, cfg.BatchFlushInterval, cfg.BatchFlushTimeout, instrumentedInsert(store), logger)
batcher.Start()
defer batcher.Stop()
@@ -77,6 +82,9 @@ func run(logger *slog.Logger) error {
cancel()
if err != nil {
logger.Error("publish reading event failed", "device_id", reading.DeviceID, "error", err)
metrics.RabbitMQPublishes.WithLabelValues("error").Inc()
} else {
metrics.RabbitMQPublishes.WithLabelValues("success").Inc()
}
}
}()
@@ -87,9 +95,12 @@ func run(logger *slog.Logger) error {
Topic: cfg.MQTTTopic,
QoS: 1,
}, func(payload []byte, receivedAt time.Time) {
metrics.MQTTMessagesReceived.Inc()
reading, err := telemetry.ParseReading(payload, receivedAt)
if err != nil {
logger.Warn("dropping invalid telemetry payload", "error", err)
metrics.MQTTMessagesInvalid.Inc()
return
}
@@ -97,12 +108,24 @@ func run(logger *slog.Logger) error {
case incoming <- reading:
default:
logger.Error("dropping reading: worker backlog full", "device_id", reading.DeviceID)
metrics.MQTTMessagesDropped.Inc()
}
}, logger)
if err != nil {
return err
}
metricsServer := &http.Server{
Addr: fmt.Sprintf(":%d", cfg.MetricsPort),
Handler: promhttp.Handler(),
}
go func() {
logger.Info("metrics server started", "metrics_port", cfg.MetricsPort)
if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logger.Error("metrics server stopped", "error", err)
}
}()
logger.Info("ingest-service started", "mqtt_topic", cfg.MQTTTopic)
stop := make(chan os.Signal, 1)
@@ -114,5 +137,30 @@ func run(logger *slog.Logger) error {
close(incoming)
workerWG.Wait()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := metricsServer.Shutdown(shutdownCtx); err != nil {
logger.Error("metrics server shutdown failed", "error", err)
}
return nil
}
// instrumentedInsert wraps store.InsertBatch with duration/size/outcome
// metrics, without teaching the batch or clickhouse packages about Prometheus.
func instrumentedInsert(store *chstore.Store) batch.FlushFunc {
return func(ctx context.Context, readings []telemetry.Reading) error {
start := time.Now()
err := store.InsertBatch(ctx, readings)
metrics.BatchFlushDuration.Observe(time.Since(start).Seconds())
metrics.BatchSize.Observe(float64(len(readings)))
if err != nil {
metrics.BatchFlushes.WithLabelValues("error").Inc()
} else {
metrics.BatchFlushes.WithLabelValues("success").Inc()
}
return err
}
}
services/ingest-service/go.mod
+11 -4
View File
@@ -5,26 +5,33 @@ go 1.25.0
require (
github.com/ClickHouse/clickhouse-go/v2 v2.47.0
github.com/eclipse/paho.mqtt.golang v1.5.1
github.com/prometheus/client_golang v1.24.1
github.com/rabbitmq/amqp091-go v1.13.0
)
require (
github.com/ClickHouse/ch-go v0.73.0 // indirect
github.com/andybalholm/brotli v1.2.1 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/go-faster/city v1.0.1 // indirect
github.com/go-faster/errors v0.7.1 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/klauspost/compress v1.18.6 // indirect
github.com/klauspost/compress v1.19.1 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/paulmach/orb v0.13.0 // indirect
github.com/pierrec/lz4/v4 v4.1.27 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.70.1 // indirect
github.com/prometheus/procfs v0.21.1 // indirect
github.com/segmentio/asm v1.2.1 // indirect
github.com/shopspring/decimal v1.4.0 // indirect
go.opentelemetry.io/otel v1.44.0 // indirect
go.opentelemetry.io/otel/trace v1.44.0 // indirect
go.yaml.in/yaml/v3 v3.0.4 // indirect
golang.org/x/net v0.56.0 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.46.0 // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/sync v0.21.0 // indirect
golang.org/x/sys v0.47.0 // indirect
google.golang.org/protobuf v1.36.11 // indirect
)
services/ingest-service/go.sum
+26 -8
View File
@@ -4,6 +4,8 @@ github.com/ClickHouse/clickhouse-go/v2 v2.47.0 h1:ZDAzrnKSOPTIsm4tdUNfrii2yc8dk4
github.com/ClickHouse/clickhouse-go/v2 v2.47.0/go.mod h1:sPj7C7UYQ2MWHcfX+4eGN6nwnCqwUKfgO6PcwKpd6K8=
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -20,14 +22,26 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/paulmach/orb v0.13.0 h1:r7n7mQGGF+cj/CbcivEj9J3HGK+XR+yXnvzRdq9saIw=
github.com/paulmach/orb v0.13.0/go.mod h1:6scRWINywA2Jf05dcjOfLfxrUIMECvTSG2MVbRLxu/k=
github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk=
github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0=
@@ -44,14 +58,18 @@ go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/
go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
services/ingest-service/internal/config/config.go
+5
View File
@@ -24,6 +24,8 @@ type Config struct {
BatchMaxSize int
BatchFlushInterval time.Duration
BatchFlushTimeout time.Duration
MetricsPort int
}
func Load() (Config, error) {
@@ -55,6 +57,9 @@ func Load() (Config, error) {
if cfg.BatchFlushTimeout, err = getEnvDuration("INGEST_BATCH_FLUSH_TIMEOUT", 10*time.Second); err != nil {
return Config{}, err
}
if cfg.MetricsPort, err = getEnvInt("INGEST_METRICS_PORT", 9101); err != nil {
return Config{}, err
}
return cfg, nil
}
services/ingest-service/internal/metrics/metrics.go
+48
View File
@@ -0,0 +1,48 @@
// Package metrics defines ingest-service's Prometheus metrics. Registered
// automatically (via promauto) into the default registry on import; served
// by main.go's dedicated metrics HTTP server.
package metrics
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
MQTTMessagesReceived = promauto.NewCounter(prometheus.CounterOpts{
Name: "ingest_mqtt_messages_received_total",
Help: "Total number of telemetry messages received from MQTT.",
})
MQTTMessagesInvalid = promauto.NewCounter(prometheus.CounterOpts{
Name: "ingest_mqtt_messages_invalid_total",
Help: "Total number of telemetry messages dropped for failing validation.",
})
MQTTMessagesDropped = promauto.NewCounter(prometheus.CounterOpts{
Name: "ingest_mqtt_messages_dropped_total",
Help: "Total number of readings dropped because the worker backlog was full.",
})
BatchFlushes = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "ingest_batch_flush_total",
Help: "Total number of batch flushes to ClickHouse, by outcome.",
}, []string{"outcome"})
BatchFlushDuration = promauto.NewHistogram(prometheus.HistogramOpts{
Name: "ingest_batch_flush_duration_seconds",
Help: "Duration of ClickHouse batch insert calls.",
Buckets: prometheus.DefBuckets,
})
BatchSize = promauto.NewHistogram(prometheus.HistogramOpts{
Name: "ingest_batch_size",
Help: "Number of readings per flushed batch.",
Buckets: []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000},
})
RabbitMQPublishes = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "ingest_rabbitmq_publish_total",
Help: "Total number of new-reading events published to RabbitMQ, by outcome.",
}, []string{"outcome"})
)
services/rule-engine-service/README.md
+6
View File
@@ -39,6 +39,12 @@
- **Кэш правил — только для чтения активных правил**, никакой записи назад в
Postgres. CRUD правил — это зона ответственности Laravel (этап 2).
## Метрики
`GET /metrics` (порт `RULE_ENGINE_METRICS_PORT`, по умолчанию 9102) —
счётчики обработанных показаний, сработавших правил (по action_type и
исходу), latency вызова device-control-service, обновлений кэша правил.
## Запуск
```bash
services/rule-engine-service/cmd/rule-engine-service/main.go
+22
View File
@@ -5,12 +5,16 @@ package main
import (
"context"
"fmt"
"log/slog"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/prometheus/client_golang/prometheus/promhttp"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/config"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
@@ -63,6 +67,17 @@ func run(logger *slog.Logger) error {
eng := engine.New(cache, dcClient, mq, logger)
metricsServer := &http.Server{
Addr: fmt.Sprintf(":%d", cfg.MetricsPort),
Handler: promhttp.Handler(),
}
go func() {
logger.Info("metrics server started", "metrics_port", cfg.MetricsPort)
if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
logger.Error("metrics server stopped", "error", err)
}
}()
logger.Info("rule-engine-service started")
err = mq.ConsumeReadings(ctx, func(ctx context.Context, event rabbitmq.ReadingEvent) rabbitmq.HandleResult {
@@ -84,5 +99,12 @@ func run(logger *slog.Logger) error {
}
logger.Info("shutting down")
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := metricsServer.Shutdown(shutdownCtx); err != nil {
logger.Error("metrics server shutdown failed", "error", err)
}
return nil
}
services/rule-engine-service/go.mod
+12 -4
View File
@@ -5,18 +5,26 @@ go 1.25.0
require (
git.cactoz.su/cacto/home_automatization/proto v0.0.0-00010101000000-000000000000
github.com/jackc/pgx/v5 v5.10.0
github.com/prometheus/client_golang v1.24.1
github.com/rabbitmq/amqp091-go v1.13.0
google.golang.org/grpc v1.82.1
)
require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
golang.org/x/net v0.53.0 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.43.0 // indirect
golang.org/x/text v0.36.0 // indirect
github.com/kylelemons/godebug v1.1.0 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.70.1 // indirect
github.com/prometheus/procfs v0.21.1 // indirect
golang.org/x/net v0.57.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
google.golang.org/protobuf v1.36.11 // indirect
)
services/rule-engine-service/go.sum
+26 -8
View File
@@ -1,3 +1,5 @@
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -21,8 +23,22 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk=
github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -44,14 +60,16 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg=
golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164=
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw=
services/rule-engine-service/internal/config/config.go
+19
View File
@@ -5,6 +5,7 @@ package config
import (
"fmt"
"os"
"strconv"
"time"
)
@@ -16,6 +17,8 @@ type Config struct {
RabbitMQURL string
RuleCacheRefreshInterval time.Duration
MetricsPort int
}
func Load() (Config, error) {
@@ -47,6 +50,10 @@ func Load() (Config, error) {
}
cfg.RuleCacheRefreshInterval = interval
if cfg.MetricsPort, err = getEnvInt("RULE_ENGINE_METRICS_PORT", 9102); err != nil {
return Config{}, err
}
return cfg, nil
}
@@ -57,6 +64,18 @@ func getEnv(key, fallback string) string {
return fallback
}
func getEnvInt(key string, fallback int) (int, error) {
v := os.Getenv(key)
if v == "" {
return fallback, nil
}
n, err := strconv.Atoi(v)
if err != nil {
return 0, fmt.Errorf("%s: %w", key, err)
}
return n, nil
}
func getEnvDuration(key string, fallback time.Duration) (time.Duration, error) {
v := os.Getenv(key)
if v == "" {
services/rule-engine-service/internal/engine/engine.go
+9
View File
@@ -11,6 +11,7 @@ import (
"time"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/metrics"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
)
@@ -57,6 +58,8 @@ func New(ruleSource RuleSource, dispatcher Dispatcher, publisher TriggerPublishe
// the whole reading for. A rule the device itself rejected (bad device_id,
// unsupported action) is terminal and does not cause a retry.
func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
metrics.ReadingsConsumed.Inc()
matched := e.rules.MatchingRules(r.DeviceID, r.SensorType)
var firstTransportErr error
@@ -69,8 +72,11 @@ func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
if !matchesCondition {
continue
}
metrics.RulesMatched.Inc()
start := time.Now()
result, dispatchErr := e.dispatcher.Dispatch(ctx, rule.TargetDeviceID, rule.ActionType, rule.ActionParams)
metrics.DispatchDuration.WithLabelValues(rule.ActionType).Observe(time.Since(start).Seconds())
success := dispatchErr == nil && result.Success
errMsg := result.Error
@@ -78,13 +84,16 @@ func (e *Engine) HandleReading(ctx context.Context, r Reading) error {
case dispatchErr != nil:
errMsg = dispatchErr.Error()
e.logger.Error("dispatch action failed", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "error", dispatchErr)
metrics.RulesTriggered.WithLabelValues(rule.ActionType, "transport_error").Inc()
if firstTransportErr == nil {
firstTransportErr = dispatchErr
}
case !result.Success:
e.logger.Warn("device rejected command", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "error", result.Error)
metrics.RulesTriggered.WithLabelValues(rule.ActionType, "rejected").Inc()
default:
e.logger.Info("rule triggered", "rule_id", rule.ID, "target_device_id", rule.TargetDeviceID, "action_type", rule.ActionType)
metrics.RulesTriggered.WithLabelValues(rule.ActionType, "success").Inc()
}
if pubErr := e.publisher.PublishRuleTriggered(ctx, rule.ID, rule.ZoneID, rule.TargetDeviceID, rule.ActionType, success, errMsg); pubErr != nil {
services/rule-engine-service/internal/engine/engine_test.go
+30
View File
@@ -7,7 +7,10 @@ import (
"log/slog"
"testing"
"github.com/prometheus/client_golang/prometheus/testutil"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/devicecontrolclient"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/metrics"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
)
@@ -154,3 +157,30 @@ func TestHandleReading_InvalidOperatorSkipsRuleButContinues(t *testing.T) {
t.Fatalf("got dispatch calls %+v, want only the valid rule dispatched", dispatcher.calls)
}
}
func TestHandleReading_RecordsMetrics(t *testing.T) {
source := &fakeRuleSource{rules: []rules.Rule{
{ID: 1, ConditionOperator: ">", ConditionValue: 28, TargetDeviceID: "fan-1", ActionType: "turn_on"},
}}
dispatcher := &fakeDispatcher{result: devicecontrolclient.Result{Success: true}}
e := New(source, dispatcher, &fakePublisher{}, discardLogger())
triggeredCounter := metrics.RulesTriggered.WithLabelValues("turn_on", "success")
beforeConsumed := testutil.ToFloat64(metrics.ReadingsConsumed)
beforeMatched := testutil.ToFloat64(metrics.RulesMatched)
beforeTriggered := testutil.ToFloat64(triggeredCounter)
if err := e.HandleReading(context.Background(), Reading{Value: 30}); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got := testutil.ToFloat64(metrics.ReadingsConsumed); got != beforeConsumed+1 {
t.Fatalf("got readings_consumed %v, want %v", got, beforeConsumed+1)
}
if got := testutil.ToFloat64(metrics.RulesMatched); got != beforeMatched+1 {
t.Fatalf("got rules_matched %v, want %v", got, beforeMatched+1)
}
if got := testutil.ToFloat64(triggeredCounter); got != beforeTriggered+1 {
t.Fatalf("got rules_triggered{turn_on,success} %v, want %v", got, beforeTriggered+1)
}
}
services/rule-engine-service/internal/metrics/metrics.go
+37
View File
@@ -0,0 +1,37 @@
// Package metrics defines rule-engine-service's Prometheus metrics.
// Registered automatically (via promauto) into the default registry on
// import; served by main.go's dedicated metrics HTTP server.
package metrics
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
ReadingsConsumed = promauto.NewCounter(prometheus.CounterOpts{
Name: "rule_engine_readings_consumed_total",
Help: "Total number of telemetry.new_reading events consumed from RabbitMQ.",
})
RulesMatched = promauto.NewCounter(prometheus.CounterOpts{
Name: "rule_engine_rules_matched_total",
Help: "Total number of rule conditions that evaluated true and were dispatched.",
})
RulesTriggered = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "rule_engine_rules_triggered_total",
Help: "Total number of rule dispatch attempts, by action type and outcome.",
}, []string{"action_type", "outcome"})
DispatchDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "rule_engine_dispatch_duration_seconds",
Help: "Duration of the gRPC/HTTP call to device-control-service, by action type.",
Buckets: prometheus.DefBuckets,
}, []string{"action_type"})
CacheRefreshes = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "rule_engine_cache_refresh_total",
Help: "Total number of automation_rules cache refreshes from PostgreSQL, by outcome.",
}, []string{"outcome"})
)
services/rule-engine-service/internal/rulecache/cache.go
+3
View File
@@ -10,6 +10,7 @@ import (
"sync"
"time"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/metrics"
"git.cactoz.su/cacto/home_automatization/services/rule-engine-service/internal/rules"
)
@@ -88,12 +89,14 @@ func (c *Cache) loop() {
func (c *Cache) refresh(ctx context.Context) error {
dbRows, err := c.fetcher.FetchActiveRules(ctx)
if err != nil {
metrics.CacheRefreshes.WithLabelValues("error").Inc()
return err
}
index := buildIndex(dbRows, c.logger)
c.mu.Lock()
c.index = index
c.mu.Unlock()
metrics.CacheRefreshes.WithLabelValues("success").Inc()
return nil
}