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.
92 lines
2.8 KiB
Go
92 lines
2.8 KiB
Go
// Package shadow gives health-check-service read/offline-detection access
|
|
// to the Device Shadow keyspace in Redis. device-control-service owns the
|
|
// full Shadow API (desired/reported state patches, online-touch on
|
|
// telemetry/ack) — this is deliberately the narrow subset a periodic
|
|
// offline-scanner needs, kept as its own small copy rather than a shared
|
|
// module: the two services only overlap on reading/writing `status` and
|
|
// `last_seen`, and duplicating ~2 key-naming functions is cheaper than a
|
|
// fourth shared Go module for that.
|
|
package shadow
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
const knownDevicesKey = "devices:known"
|
|
|
|
func statusKey(deviceID string) string { return "device:" + deviceID + ":status" }
|
|
func lastSeenKey(deviceID string) string { return "device:" + deviceID + ":last_seen" }
|
|
|
|
const (
|
|
StatusOnline = "online"
|
|
StatusOffline = "offline"
|
|
)
|
|
|
|
type Store struct {
|
|
rdb *redis.Client
|
|
}
|
|
|
|
func New(rdb *redis.Client) *Store {
|
|
return &Store{rdb: rdb}
|
|
}
|
|
|
|
func Connect(ctx context.Context, addr, password string, db int) (*Store, error) {
|
|
rdb := redis.NewClient(&redis.Options{Addr: addr, Password: password, DB: db})
|
|
if err := rdb.Ping(ctx).Err(); err != nil {
|
|
return nil, fmt.Errorf("ping redis: %w", err)
|
|
}
|
|
return New(rdb), nil
|
|
}
|
|
|
|
func (s *Store) Close() error {
|
|
return s.rdb.Close()
|
|
}
|
|
|
|
// KnownDevices lists every device ID device-control-service has heard from.
|
|
func (s *Store) KnownDevices(ctx context.Context) ([]string, error) {
|
|
ids, err := s.rdb.SMembers(ctx, knownDevicesKey).Result()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list known devices: %w", err)
|
|
}
|
|
return ids, nil
|
|
}
|
|
|
|
// MarkOfflineIfStale flips deviceID to offline if it is currently online and
|
|
// its last_seen is older than timeout. It reports whether a change was made.
|
|
func (s *Store) MarkOfflineIfStale(ctx context.Context, deviceID string, timeout time.Duration) (changed bool, err error) {
|
|
status, err := s.rdb.Get(ctx, statusKey(deviceID)).Result()
|
|
if err == redis.Nil || status != StatusOnline {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, fmt.Errorf("get status for %s: %w", deviceID, err)
|
|
}
|
|
|
|
lastSeenRaw, err := s.rdb.Get(ctx, lastSeenKey(deviceID)).Result()
|
|
if err != nil && err != redis.Nil {
|
|
return false, fmt.Errorf("get last_seen for %s: %w", deviceID, err)
|
|
}
|
|
|
|
var lastSeen time.Time
|
|
if lastSeenRaw != "" {
|
|
sec, err := strconv.ParseInt(lastSeenRaw, 10, 64)
|
|
if err != nil {
|
|
return false, fmt.Errorf("parse last_seen for %s: %w", deviceID, err)
|
|
}
|
|
lastSeen = time.Unix(sec, 0)
|
|
}
|
|
|
|
if time.Since(lastSeen) <= timeout {
|
|
return false, nil
|
|
}
|
|
if err := s.rdb.Set(ctx, statusKey(deviceID), StatusOffline, 0).Err(); err != nil {
|
|
return false, fmt.Errorf("set status offline for %s: %w", deviceID, err)
|
|
}
|
|
return true, nil
|
|
}
|