cacto
17af76c159
Тикер офлайн-детекции (internal/healthcheck) переехал из device-control-service в новый самостоятельный Go-сервис — как и планировалось с самого начала (см. изначальный README-заглушку). Детекция online остаётся в device-control-service: она происходит как побочный эффект уже имеющейся там MQTT-подписки на devices/+/telemetry и devices/+/ack, заводить для этого отдельный сервис с дублирующей MQTT-подпиской избыточно. health-check-service — свой Go-модуль (без зависимости от proto/, сборка из собственного контекста), читает только узкое read+offline-подмножество Device Shadow keyspace в Redis (devices:known, status, last_seen) — полный Shadow API с записью desired/reported state остаётся только в device-control-service. Метрика device_control_health_transitions_total переименована в device_health_transitions_total и теперь публикуется с этим именем из ДВУХ сервисов (device-control-service — direction=online, health-check-service — direction=offline): Prometheus агрегирует одноимённые метрики с разных таргетов прозрачно, поэтому Grafana- дашборд адаптирован простой сменой имени метрики в запросе, без переделки панели. Проверено вживую через docker compose: реальный online→offline переход (публикация тестовой телеметрии + ожидание таймаута) корректно долетает до Redis и RabbitMQ (device.status_changed), Prometheus видит новый scrape-таргет как up.
79 lines
1.9 KiB
Go
79 lines
1.9 KiB
Go
// Package rabbitmq publishes device online/offline transitions for
|
|
// notification-service to consume. Delivery reliability matters more than
|
|
// latency here, so messages are persistent and the queue is durable.
|
|
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
const StatusChangedQueue = "device.status_changed"
|
|
|
|
type Publisher struct {
|
|
conn *amqp.Connection
|
|
ch *amqp.Channel
|
|
}
|
|
|
|
func Connect(url string) (*Publisher, error) {
|
|
conn, err := amqp.Dial(url)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("dial rabbitmq: %w", err)
|
|
}
|
|
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
conn.Close()
|
|
return nil, fmt.Errorf("open channel: %w", err)
|
|
}
|
|
|
|
if _, err := ch.QueueDeclare(StatusChangedQueue, true, false, false, false, nil); err != nil {
|
|
ch.Close()
|
|
conn.Close()
|
|
return nil, fmt.Errorf("declare queue %s: %w", StatusChangedQueue, err)
|
|
}
|
|
|
|
return &Publisher{conn: conn, ch: ch}, nil
|
|
}
|
|
|
|
func (p *Publisher) Close() error {
|
|
if err := p.ch.Close(); err != nil {
|
|
p.conn.Close()
|
|
return err
|
|
}
|
|
return p.conn.Close()
|
|
}
|
|
|
|
type statusChangedEvent struct {
|
|
DeviceID string `json:"device_id"`
|
|
Status string `json:"status"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// PublishStatusChanged emits one event when a device transitions online or offline.
|
|
func (p *Publisher) PublishStatusChanged(ctx context.Context, deviceID, status string) error {
|
|
body, err := json.Marshal(statusChangedEvent{
|
|
DeviceID: deviceID,
|
|
Status: status,
|
|
At: time.Now(),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("marshal event: %w", err)
|
|
}
|
|
|
|
err = p.ch.PublishWithContext(ctx, "", StatusChangedQueue, false, false, amqp.Publishing{
|
|
ContentType: "application/json",
|
|
DeliveryMode: amqp.Persistent,
|
|
Timestamp: time.Now(),
|
|
Body: body,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("publish status change for device %s: %w", deviceID, err)
|
|
}
|
|
return nil
|
|
}
|