From 81040eec621cdfd36dab29d8ba582c856bcfa548 Mon Sep 17 00:00:00 2001 From: Dmitry Gammel Date: Tue, 11 Aug 2026 21:58:59 +0500 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D1=8B=20=D0=BC=D0=B5=D1=82=D1=80=D0=B8=D0=BA=D0=B8=20Pro?= =?UTF-8?q?metheus=20=D0=B8=20=D0=B4=D0=B0=D1=88=D0=B1=D0=BE=D1=80=D0=B4?= =?UTF-8?q?=D1=8B=20Grafana=20(=D1=8D=D1=82=D0=B0=D0=BF=202.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Все три Go-сервиса (ingest, device-control, rule-engine) отдают /metrics в формате Prometheus: счётчики MQTT/RabbitMQ/ClickHouse операций, гистограммы длительности батч-флашей и диспатча правил, переходы устройств online/offline. Добавлены сервисы prometheus и grafana в docker-compose с провижининг конфигом (datasource + два готовых дашборда: "Ingest & Telemetry" и "Automation & Devices"). --- .env.example | 8 ++ docker-compose.yml | 36 ++++++++ monitoring/grafana/.gitkeep | 0 .../dashboards/automation-devices.json | 83 +++++++++++++++++ .../provisioning/dashboards/dashboards.yml | 11 +++ .../dashboards/ingest-telemetry.json | 88 +++++++++++++++++++ .../provisioning/datasources/datasource.yml | 9 ++ monitoring/prometheus/.gitkeep | 0 monitoring/prometheus/prometheus.yml | 15 ++++ services/device-control-service/README.md | 7 ++ .../cmd/device-control-service/main.go | 19 +++- services/device-control-service/go.mod | 15 +++- services/device-control-service/go.sum | 40 ++++++--- .../internal/metrics/metrics.go | 32 +++++++ .../internal/server/server.go | 40 ++++++--- .../internal/server/server_test.go | 27 ++++++ services/ingest-service/README.md | 7 ++ .../ingest-service/cmd/ingest-service/main.go | 50 ++++++++++- services/ingest-service/go.mod | 15 +++- services/ingest-service/go.sum | 34 +++++-- .../ingest-service/internal/config/config.go | 5 ++ .../internal/metrics/metrics.go | 48 ++++++++++ services/rule-engine-service/README.md | 6 ++ .../cmd/rule-engine-service/main.go | 22 +++++ services/rule-engine-service/go.mod | 16 +++- services/rule-engine-service/go.sum | 34 +++++-- .../internal/config/config.go | 19 ++++ .../internal/engine/engine.go | 9 ++ .../internal/engine/engine_test.go | 30 +++++++ .../internal/metrics/metrics.go | 37 ++++++++ .../internal/rulecache/cache.go | 3 + 31 files changed, 712 insertions(+), 53 deletions(-) delete mode 100644 monitoring/grafana/.gitkeep create mode 100644 monitoring/grafana/provisioning/dashboards/automation-devices.json create mode 100644 monitoring/grafana/provisioning/dashboards/dashboards.yml create mode 100644 monitoring/grafana/provisioning/dashboards/ingest-telemetry.json create mode 100644 monitoring/grafana/provisioning/datasources/datasource.yml delete mode 100644 monitoring/prometheus/.gitkeep create mode 100644 monitoring/prometheus/prometheus.yml create mode 100644 services/device-control-service/internal/metrics/metrics.go create mode 100644 services/ingest-service/internal/metrics/metrics.go create mode 100644 services/rule-engine-service/internal/metrics/metrics.go diff --git a/.env.example b/.env.example index 9bd111f..32262de 100644 --- a/.env.example +++ b/.env.example @@ -43,10 +43,12 @@ INGEST_MQTT_TOPIC=devices/+/telemetry INGEST_BATCH_MAX_SIZE=500 INGEST_BATCH_FLUSH_INTERVAL=5s INGEST_BATCH_FLUSH_TIMEOUT=10s +INGEST_METRICS_PORT=9101 # --- rule-engine-service --- DEVICE_CONTROL_HOST=device-control-service RULE_ENGINE_CACHE_REFRESH_INTERVAL=15s +RULE_ENGINE_METRICS_PORT=9102 # --- Laravel app (stage 2) --- # Regenerate for anything beyond local dev: php artisan key:generate --show @@ -55,3 +57,9 @@ APP_ENV=local APP_DEBUG=true APP_URL=http://localhost:8000 LARAVEL_HTTP_PORT=8000 + +# --- Prometheus + Grafana (stage 2.5) --- +PROMETHEUS_PORT=9090 +GRAFANA_PORT=3000 +GRAFANA_ADMIN_USER=admin +GRAFANA_ADMIN_PASSWORD=change_me diff --git a/docker-compose.yml b/docker-compose.yml index c50e603..e9afd92 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,6 +8,8 @@ volumes: redis-data: rabbitmq-data: mosquitto-data: + prometheus-data: + grafana-data: services: mosquitto: @@ -92,6 +94,8 @@ services: ingest-service: build: ./services/ingest-service + ports: + - "${INGEST_METRICS_PORT}:9101" environment: MQTT_HOST: mosquitto MQTT_PORT: 1883 @@ -104,6 +108,7 @@ services: RABBITMQ_PORT: 5672 RABBITMQ_USER: ${RABBITMQ_USER} RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD} + INGEST_METRICS_PORT: 9101 depends_on: mosquitto: condition: service_started @@ -152,6 +157,8 @@ services: build: context: . dockerfile: services/rule-engine-service/Dockerfile + ports: + - "${RULE_ENGINE_METRICS_PORT}:9102" environment: POSTGRES_HOST: postgres POSTGRES_PORT: 5432 @@ -165,6 +172,7 @@ services: RABBITMQ_USER: ${RABBITMQ_USER} RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD} RULE_ENGINE_CACHE_REFRESH_INTERVAL: ${RULE_ENGINE_CACHE_REFRESH_INTERVAL} + RULE_ENGINE_METRICS_PORT: 9102 depends_on: postgres: condition: service_healthy @@ -222,5 +230,33 @@ services: - home-automation restart: unless-stopped + prometheus: + image: prom/prometheus:v3.1.0 + ports: + - "${PROMETHEUS_PORT}:9090" + volumes: + - ./monitoring/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro + - prometheus-data:/prometheus + networks: + - home-automation + restart: unless-stopped + + grafana: + image: grafana/grafana:11.4.0 + ports: + - "${GRAFANA_PORT}:3000" + environment: + GF_SECURITY_ADMIN_USER: ${GRAFANA_ADMIN_USER} + GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_ADMIN_PASSWORD} + GF_USERS_ALLOW_SIGN_UP: "false" + volumes: + - grafana-data:/var/lib/grafana + - ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro + depends_on: + - prometheus + networks: + - home-automation + restart: unless-stopped + # Remaining app services (health-check-service, esp32-emulator) are added # here as they get implemented — see services/*/README.md for the plan. diff --git a/monitoring/grafana/.gitkeep b/monitoring/grafana/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/monitoring/grafana/provisioning/dashboards/automation-devices.json b/monitoring/grafana/provisioning/dashboards/automation-devices.json new file mode 100644 index 0000000..b43f736 --- /dev/null +++ b/monitoring/grafana/provisioning/dashboards/automation-devices.json @@ -0,0 +1,83 @@ +{ + "uid": "automation-devices", + "title": "Automation & Devices", + "tags": ["home-automation"], + "timezone": "browser", + "schemaVersion": 39, + "version": 1, + "refresh": "10s", + "time": { "from": "now-30m", "to": "now" }, + "panels": [ + { + "id": 1, + "title": "Rules triggered (rate, by outcome)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "targets": [ + { + "expr": "sum by (outcome) (rate(rule_engine_rules_triggered_total[1m]))", + "legendFormat": "{{outcome}}", + "refId": "A" + } + ] + }, + { + "id": 2, + "title": "device-control-service dispatch duration (p95, by action)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "targets": [ + { + "expr": "histogram_quantile(0.95, sum(rate(rule_engine_dispatch_duration_seconds_bucket[5m])) by (le, action_type))", + "legendFormat": "{{action_type}}", + "refId": "A" + } + ] + }, + { + "id": 3, + "title": "Device commands (rate, by action + outcome)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 }, + "targets": [ + { + "expr": "sum by (action, outcome) (rate(device_control_commands_total[1m]))", + "legendFormat": "{{action}} / {{outcome}}", + "refId": "A" + } + ] + }, + { + "id": 4, + "title": "Device online/offline transitions (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "targets": [ + { + "expr": "sum by (direction) (rate(device_control_health_transitions_total[5m]))", + "legendFormat": "{{direction}}", + "refId": "A" + } + ] + }, + { + "id": 5, + "title": "Automation rule cache refreshes (rate, by outcome)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "targets": [ + { + "expr": "sum by (outcome) (rate(rule_engine_cache_refresh_total[1m]))", + "legendFormat": "{{outcome}}", + "refId": "A" + } + ] + } + ] +} diff --git a/monitoring/grafana/provisioning/dashboards/dashboards.yml b/monitoring/grafana/provisioning/dashboards/dashboards.yml new file mode 100644 index 0000000..03847a8 --- /dev/null +++ b/monitoring/grafana/provisioning/dashboards/dashboards.yml @@ -0,0 +1,11 @@ +apiVersion: 1 + +providers: + - name: default + orgId: 1 + folder: "" + type: file + disableDeletion: false + updateIntervalSeconds: 30 + options: + path: /etc/grafana/provisioning/dashboards diff --git a/monitoring/grafana/provisioning/dashboards/ingest-telemetry.json b/monitoring/grafana/provisioning/dashboards/ingest-telemetry.json new file mode 100644 index 0000000..58d4914 --- /dev/null +++ b/monitoring/grafana/provisioning/dashboards/ingest-telemetry.json @@ -0,0 +1,88 @@ +{ + "uid": "ingest-telemetry", + "title": "Ingest & Telemetry", + "tags": ["home-automation"], + "timezone": "browser", + "schemaVersion": 39, + "version": 1, + "refresh": "10s", + "time": { "from": "now-30m", "to": "now" }, + "panels": [ + { + "id": 1, + "title": "MQTT messages received (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 }, + "targets": [ + { + "expr": "sum(rate(ingest_mqtt_messages_received_total[1m]))", + "legendFormat": "received/s", + "refId": "A" + } + ] + }, + { + "id": 2, + "title": "MQTT messages invalid / dropped (rate)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 }, + "targets": [ + { + "expr": "sum(rate(ingest_mqtt_messages_invalid_total[1m]))", + "legendFormat": "invalid/s", + "refId": "A" + }, + { + "expr": "sum(rate(ingest_mqtt_messages_dropped_total[1m]))", + "legendFormat": "dropped (backlog full)/s", + "refId": "B" + } + ] + }, + { + "id": 3, + "title": "ClickHouse batch flushes (rate, by outcome)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 8 }, + "targets": [ + { + "expr": "sum by (outcome) (rate(ingest_batch_flush_total[1m]))", + "legendFormat": "{{outcome}}", + "refId": "A" + } + ] + }, + { + "id": 4, + "title": "ClickHouse batch flush duration (p95)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 8 }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "targets": [ + { + "expr": "histogram_quantile(0.95, sum(rate(ingest_batch_flush_duration_seconds_bucket[5m])) by (le))", + "legendFormat": "p95", + "refId": "A" + } + ] + }, + { + "id": 5, + "title": "RabbitMQ publishes (rate, by outcome)", + "type": "timeseries", + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 16 }, + "targets": [ + { + "expr": "sum by (outcome) (rate(ingest_rabbitmq_publish_total[1m]))", + "legendFormat": "{{outcome}}", + "refId": "A" + } + ] + } + ] +} diff --git a/monitoring/grafana/provisioning/datasources/datasource.yml b/monitoring/grafana/provisioning/datasources/datasource.yml new file mode 100644 index 0000000..bb009bb --- /dev/null +++ b/monitoring/grafana/provisioning/datasources/datasource.yml @@ -0,0 +1,9 @@ +apiVersion: 1 + +datasources: + - name: Prometheus + type: prometheus + access: proxy + url: http://prometheus:9090 + isDefault: true + editable: false diff --git a/monitoring/prometheus/.gitkeep b/monitoring/prometheus/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/monitoring/prometheus/prometheus.yml b/monitoring/prometheus/prometheus.yml new file mode 100644 index 0000000..54912a3 --- /dev/null +++ b/monitoring/prometheus/prometheus.yml @@ -0,0 +1,15 @@ +global: + scrape_interval: 15s + +scrape_configs: + - job_name: ingest-service + static_configs: + - targets: ["ingest-service:9101"] + + - job_name: device-control-service + static_configs: + - targets: ["device-control-service:8090"] + + - job_name: rule-engine-service + static_configs: + - targets: ["rule-engine-service:9102"] diff --git a/services/device-control-service/README.md b/services/device-control-service/README.md index a88e806..e693472 100644 --- a/services/device-control-service/README.md +++ b/services/device-control-service/README.md @@ -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 diff --git a/services/device-control-service/cmd/device-control-service/main.go b/services/device-control-service/cmd/device-control-service/main.go index c0090c6..38b7535 100644 --- a/services/device-control-service/cmd/device-control-service/main.go +++ b/services/device-control-service/cmd/device-control-service/main.go @@ -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 { diff --git a/services/device-control-service/go.mod b/services/device-control-service/go.mod index b102eca..63a2d40 100644 --- a/services/device-control-service/go.mod +++ b/services/device-control-service/go.mod @@ -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 ) diff --git a/services/device-control-service/go.sum b/services/device-control-service/go.sum index 1ec4e21..caf6949 100644 --- a/services/device-control-service/go.sum +++ b/services/device-control-service/go.sum @@ -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= diff --git a/services/device-control-service/internal/metrics/metrics.go b/services/device-control-service/internal/metrics/metrics.go new file mode 100644 index 0000000..1aae263 --- /dev/null +++ b/services/device-control-service/internal/metrics/metrics.go @@ -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"}) +) diff --git a/services/device-control-service/internal/server/server.go b/services/device-control-service/internal/server/server.go index 4c7e3be..b9fa29c 100644 --- a/services/device-control-service/internal/server/server.go +++ b/services/device-control-service/internal/server/server.go @@ -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} } diff --git a/services/device-control-service/internal/server/server_test.go b/services/device-control-service/internal/server/server_test.go index ae5f424..1a3f956 100644 --- a/services/device-control-service/internal/server/server_test.go +++ b/services/device-control-service/internal/server/server_test.go @@ -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) + } +} diff --git a/services/ingest-service/README.md b/services/ingest-service/README.md index 4848eae..f7a579a 100644 --- a/services/ingest-service/README.md +++ b/services/ingest-service/README.md @@ -20,6 +20,13 @@ MQTT/ClickHouse/RabbitMQ, как и на диаграмме архитектур понадобится lookup через реестр устройств (с кэшированием, как rule-engine кэширует правила) — это естественное развитие, пока не реализовано. +## Метрики + +`GET /metrics` (порт `INGEST_METRICS_PORT`, по умолчанию 9101) — формат +Prometheus. Счётчики принятых/невалидных/отброшенных (backlog переполнен) +MQTT-сообщений, флашей батча в ClickHouse (с latency и размером батча), +публикаций в RabbitMQ. + ## Запуск ```bash diff --git a/services/ingest-service/cmd/ingest-service/main.go b/services/ingest-service/cmd/ingest-service/main.go index ac198cb..dc5a526 100644 --- a/services/ingest-service/cmd/ingest-service/main.go +++ b/services/ingest-service/cmd/ingest-service/main.go @@ -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 + } +} diff --git a/services/ingest-service/go.mod b/services/ingest-service/go.mod index 15f9002..9593abd 100644 --- a/services/ingest-service/go.mod +++ b/services/ingest-service/go.mod @@ -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 ) diff --git a/services/ingest-service/go.sum b/services/ingest-service/go.sum index ad9d49d..fd21f92 100644 --- a/services/ingest-service/go.sum +++ b/services/ingest-service/go.sum @@ -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= diff --git a/services/ingest-service/internal/config/config.go b/services/ingest-service/internal/config/config.go index 1b4e913..d07ac37 100644 --- a/services/ingest-service/internal/config/config.go +++ b/services/ingest-service/internal/config/config.go @@ -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 } diff --git a/services/ingest-service/internal/metrics/metrics.go b/services/ingest-service/internal/metrics/metrics.go new file mode 100644 index 0000000..f898461 --- /dev/null +++ b/services/ingest-service/internal/metrics/metrics.go @@ -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"}) +) diff --git a/services/rule-engine-service/README.md b/services/rule-engine-service/README.md index e51d80b..a209df0 100644 --- a/services/rule-engine-service/README.md +++ b/services/rule-engine-service/README.md @@ -39,6 +39,12 @@ - **Кэш правил — только для чтения активных правил**, никакой записи назад в Postgres. CRUD правил — это зона ответственности Laravel (этап 2). +## Метрики + +`GET /metrics` (порт `RULE_ENGINE_METRICS_PORT`, по умолчанию 9102) — +счётчики обработанных показаний, сработавших правил (по action_type и +исходу), latency вызова device-control-service, обновлений кэша правил. + ## Запуск ```bash diff --git a/services/rule-engine-service/cmd/rule-engine-service/main.go b/services/rule-engine-service/cmd/rule-engine-service/main.go index 5989360..8c60ae1 100644 --- a/services/rule-engine-service/cmd/rule-engine-service/main.go +++ b/services/rule-engine-service/cmd/rule-engine-service/main.go @@ -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 } diff --git a/services/rule-engine-service/go.mod b/services/rule-engine-service/go.mod index a12021e..f9feabd 100644 --- a/services/rule-engine-service/go.mod +++ b/services/rule-engine-service/go.mod @@ -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 ) diff --git a/services/rule-engine-service/go.sum b/services/rule-engine-service/go.sum index a5ecd8b..89b896d 100644 --- a/services/rule-engine-service/go.sum +++ b/services/rule-engine-service/go.sum @@ -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= diff --git a/services/rule-engine-service/internal/config/config.go b/services/rule-engine-service/internal/config/config.go index 72cd856..fe9958e 100644 --- a/services/rule-engine-service/internal/config/config.go +++ b/services/rule-engine-service/internal/config/config.go @@ -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 == "" { diff --git a/services/rule-engine-service/internal/engine/engine.go b/services/rule-engine-service/internal/engine/engine.go index 1c72acd..894f65e 100644 --- a/services/rule-engine-service/internal/engine/engine.go +++ b/services/rule-engine-service/internal/engine/engine.go @@ -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 { diff --git a/services/rule-engine-service/internal/engine/engine_test.go b/services/rule-engine-service/internal/engine/engine_test.go index c91b22b..8c3d21a 100644 --- a/services/rule-engine-service/internal/engine/engine_test.go +++ b/services/rule-engine-service/internal/engine/engine_test.go @@ -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) + } +} diff --git a/services/rule-engine-service/internal/metrics/metrics.go b/services/rule-engine-service/internal/metrics/metrics.go new file mode 100644 index 0000000..4fa2cd3 --- /dev/null +++ b/services/rule-engine-service/internal/metrics/metrics.go @@ -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"}) +) diff --git a/services/rule-engine-service/internal/rulecache/cache.go b/services/rule-engine-service/internal/rulecache/cache.go index e084a7f..749182f 100644 --- a/services/rule-engine-service/internal/rulecache/cache.go +++ b/services/rule-engine-service/internal/rulecache/cache.go @@ -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 }