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.
77 lines
1.8 KiB
Go
77 lines
1.8 KiB
Go
package healthcheck
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log/slog"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type fakeStore struct {
|
|
mu sync.Mutex
|
|
known []string
|
|
stale map[string]bool // deviceID -> whether MarkOfflineIfStale should report a change
|
|
changes []string // devices actually marked changed, in call order
|
|
}
|
|
|
|
func (f *fakeStore) KnownDevices(context.Context) ([]string, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return append([]string(nil), f.known...), nil
|
|
}
|
|
|
|
func (f *fakeStore) MarkOfflineIfStale(_ context.Context, deviceID string, _ time.Duration) (bool, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if !f.stale[deviceID] {
|
|
return false, nil
|
|
}
|
|
// Only report the transition once, like the real store would.
|
|
f.stale[deviceID] = false
|
|
f.changes = append(f.changes, deviceID)
|
|
return true, nil
|
|
}
|
|
|
|
func discardLogger() *slog.Logger {
|
|
return slog.New(slog.NewTextHandler(io.Discard, nil))
|
|
}
|
|
|
|
func TestChecker_FlipsStaleDevicesAndCallsOnOffline(t *testing.T) {
|
|
store := &fakeStore{
|
|
known: []string{"d1", "d2"},
|
|
stale: map[string]bool{"d1": true, "d2": false},
|
|
}
|
|
|
|
offline := make(chan string, 2)
|
|
c := New(store, time.Minute, 10*time.Millisecond, func(deviceID string) {
|
|
offline <- deviceID
|
|
}, discardLogger())
|
|
c.Start()
|
|
defer c.Stop()
|
|
|
|
select {
|
|
case id := <-offline:
|
|
if id != "d1" {
|
|
t.Fatalf("got offline callback for %q, want d1", id)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("timed out waiting for onOffline callback")
|
|
}
|
|
|
|
select {
|
|
case id := <-offline:
|
|
t.Fatalf("unexpected second onOffline callback for %q (d2 was never stale)", id)
|
|
case <-time.After(50 * time.Millisecond):
|
|
// expected: no further callbacks
|
|
}
|
|
}
|
|
|
|
func TestChecker_StopWaitsForLoopExit(t *testing.T) {
|
|
store := &fakeStore{known: nil}
|
|
c := New(store, time.Minute, time.Hour, nil, discardLogger())
|
|
c.Start()
|
|
c.Stop() // must return without hanging
|
|
}
|