Вынос health-check-service в отдельный сервис
Тикер офлайн-детекции (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.
This commit is contained in:
@@ -6,7 +6,6 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
@@ -21,9 +20,6 @@ type Config struct {
|
||||
RedisDB int
|
||||
|
||||
RabbitMQURL string
|
||||
|
||||
HealthCheckTimeout time.Duration
|
||||
HealthCheckInterval time.Duration
|
||||
}
|
||||
|
||||
func Load() (Config, error) {
|
||||
@@ -52,12 +48,6 @@ func Load() (Config, error) {
|
||||
if cfg.RedisDB, err = getEnvInt("REDIS_DB", 0); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.HealthCheckTimeout, err = getEnvDuration("DEVICE_CONTROL_HEALTHCHECK_TIMEOUT", 60*time.Second); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
if cfg.HealthCheckInterval, err = getEnvDuration("DEVICE_CONTROL_HEALTHCHECK_INTERVAL", 15*time.Second); err != nil {
|
||||
return Config{}, err
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
}
|
||||
@@ -80,15 +70,3 @@ func getEnvInt(key string, fallback int) (int, error) {
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func getEnvDuration(key string, fallback time.Duration) (time.Duration, error) {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
return fallback, nil
|
||||
}
|
||||
d, err := time.ParseDuration(v)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
return d, nil
|
||||
}
|
||||
|
||||
@@ -1,93 +0,0 @@
|
||||
// Package healthcheck periodically flips devices to offline when they stop
|
||||
// sending telemetry/acks, per the platform's Device Shadow health-check
|
||||
// design (a goroutine + graceful shutdown, deliberately simple).
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DeviceStore is the subset of shadow.Store the checker needs, kept as an
|
||||
// interface so tests don't require a real Redis.
|
||||
type DeviceStore interface {
|
||||
KnownDevices(ctx context.Context) ([]string, error)
|
||||
MarkOfflineIfStale(ctx context.Context, deviceID string, timeout time.Duration) (bool, error)
|
||||
}
|
||||
|
||||
// OnOfflineFunc is called once per device that just transitioned to offline.
|
||||
type OnOfflineFunc func(deviceID string)
|
||||
|
||||
type Checker struct {
|
||||
store DeviceStore
|
||||
timeout time.Duration
|
||||
interval time.Duration
|
||||
onOffline OnOfflineFunc
|
||||
logger *slog.Logger
|
||||
|
||||
stop chan struct{}
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func New(store DeviceStore, timeout, interval time.Duration, onOffline OnOfflineFunc, logger *slog.Logger) *Checker {
|
||||
return &Checker{
|
||||
store: store,
|
||||
timeout: timeout,
|
||||
interval: interval,
|
||||
onOffline: onOffline,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the periodic scan loop. Call once.
|
||||
func (c *Checker) Start() {
|
||||
c.stop = make(chan struct{})
|
||||
c.done = make(chan struct{})
|
||||
go c.loop()
|
||||
}
|
||||
|
||||
// Stop signals the loop to exit and waits for it to finish.
|
||||
func (c *Checker) Stop() {
|
||||
close(c.stop)
|
||||
<-c.done
|
||||
}
|
||||
|
||||
func (c *Checker) loop() {
|
||||
defer close(c.done)
|
||||
ticker := time.NewTicker(c.interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
c.checkAll()
|
||||
case <-c.stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Checker) checkAll() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), c.interval)
|
||||
defer cancel()
|
||||
|
||||
ids, err := c.store.KnownDevices(ctx)
|
||||
if err != nil {
|
||||
c.logger.Error("healthcheck: list known devices failed", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, id := range ids {
|
||||
changed, err := c.store.MarkOfflineIfStale(ctx, id, c.timeout)
|
||||
if err != nil {
|
||||
c.logger.Error("healthcheck: check device failed", "device_id", id, "error", err)
|
||||
continue
|
||||
}
|
||||
if changed {
|
||||
c.logger.Info("device marked offline", "device_id", id)
|
||||
if c.onOffline != nil {
|
||||
c.onOffline(id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -25,8 +25,14 @@ var (
|
||||
Help: "Total number of telemetry/ack messages received, by event type and outcome.",
|
||||
}, []string{"event_type", "outcome"})
|
||||
|
||||
// Name matches health-check-service's own metric of the same purpose —
|
||||
// this service only ever emits direction="online" (detected via MQTT
|
||||
// telemetry/ack), health-check-service only ever emits "offline".
|
||||
// Prometheus aggregates same-named metrics across scrape targets, so
|
||||
// Grafana's `sum by (direction) (...)` panel works across both without
|
||||
// a query change.
|
||||
HealthTransitionsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
|
||||
Name: "device_control_health_transitions_total",
|
||||
Name: "device_health_transitions_total",
|
||||
Help: "Total number of device online/offline transitions detected.",
|
||||
}, []string{"direction"})
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user