feat: Hub monitoring takeover — event push system + config cleanup (v0.21.0)

Replace external Healthchecks.io with Hub-native event system. Controller
now pushes structured events via POST /api/v1/event with typed detail
structs. Hub handles dead man's switch, notification dispatch, and cooldowns.

Phase 5: PushEvent() core method, 21 event types, expanded notification
settings (11 toggles), Hub connection monitoring on dashboard, alerts.
Phase 6: Deprecation log for ping UUIDs, pinger kept for transition.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-20 18:53:21 +01:00
parent 55abe401ee
commit 8aebbb8902
13 changed files with 722 additions and 318 deletions
+358 -207
View File
@@ -7,15 +7,15 @@ import (
"io"
"log"
"net/http"
"strings"
"sync"
"time"
"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
)
// Notifier sends notification events to the hub relay service.
// Notifier sends structured events to the hub via /api/v1/event.
// Non-blocking: fires requests in goroutines, logs errors but doesn't retry aggressively.
// Cooldown logic is handled by the Hub — the controller sends all events unconditionally.
type Notifier struct {
hubURL string
apiKey string
@@ -26,11 +26,7 @@ type Notifier struct {
settings *settings.Settings
mu sync.Mutex
cooldowns map[string]time.Time // event_type -> last notification time
perEventCooldown map[string]time.Duration // per-event override cooldown durations
// prevHealthStatus tracks the previous health check status for change detection
prevHealthStatus string
prevHealthStatus string // tracks previous health check status for change detection
}
// New creates a new Notifier. Returns a no-op notifier if hub is not enabled.
@@ -50,11 +46,6 @@ func New(hubURL, apiKey, customerID string, sett *settings.Settings, logger *log
logger: logger,
enabled: enabled,
settings: sett,
cooldowns: make(map[string]time.Time),
perEventCooldown: map[string]time.Duration{
"storage_disconnected": 1 * time.Hour,
"storage_reconnected": 1 * time.Hour,
},
}
}
@@ -63,17 +54,310 @@ func (n *Notifier) IsEnabled() bool {
return n.enabled
}
// preferencesRequest is the JSON payload sent to the hub preferences endpoint.
// ── Detail structs ───────────────────────────────────────────────────
// BackupDetails holds structured data for backup events.
type BackupDetails struct {
DriveCount int `json:"drive_count,omitempty"`
SnapshotID string `json:"snapshot_id,omitempty"`
DurationSec int `json:"duration_sec,omitempty"`
DataAdded string `json:"data_added,omitempty"`
Error string `json:"error,omitempty"`
}
// DBDumpDetails holds structured data for DB dump events.
type DBDumpDetails struct {
DatabaseCount int `json:"database_count,omitempty"`
TotalSize string `json:"total_size,omitempty"`
DurationSec int `json:"duration_sec,omitempty"`
Error string `json:"error,omitempty"`
}
// DiskDetails holds structured data for disk warning/critical events.
type DiskDetails struct {
Mount string `json:"mount,omitempty"`
UsagePercent float64 `json:"usage_percent,omitempty"`
Label string `json:"label,omitempty"`
}
// HealthDetails holds structured data for health events.
type HealthDetails struct {
PreviousStatus string `json:"previous_status,omitempty"`
CurrentStatus string `json:"current_status,omitempty"`
Issues []string `json:"issues,omitempty"`
Warnings []string `json:"warnings,omitempty"`
}
// StorageDetails holds structured data for storage events.
type StorageDetails struct {
DrivePath string `json:"drive_path,omitempty"`
Label string `json:"label,omitempty"`
StoppedApps []string `json:"stopped_apps,omitempty"`
}
// UpdateDetails holds structured data for controller update events.
type UpdateDetails struct {
FromVersion string `json:"from_version,omitempty"`
ToVersion string `json:"to_version,omitempty"`
Error string `json:"error,omitempty"`
}
// AppDetails holds structured data for app lifecycle events.
type AppDetails struct {
StackName string `json:"stack_name,omitempty"`
DisplayName string `json:"display_name,omitempty"`
}
// CrossDriveDetails holds structured data for cross-drive backup events.
type CrossDriveDetails struct {
StackName string `json:"stack_name,omitempty"`
Method string `json:"method,omitempty"`
DestPath string `json:"dest_path,omitempty"`
Duration string `json:"duration,omitempty"`
Error string `json:"error,omitempty"`
}
// ── Core event push ──────────────────────────────────────────────────
// eventRequest is the JSON payload sent to /api/v1/event.
type eventRequest struct {
CustomerID string `json:"customer_id"`
EventType string `json:"event_type"`
Severity string `json:"severity"`
Message string `json:"message"`
Details json.RawMessage `json:"details,omitempty"`
}
// PushEvent sends a structured event to the hub's /api/v1/event endpoint.
// Non-blocking (goroutine). Retries twice with 3s backoff.
// details may be nil (omitted from JSON) or a struct that marshals to JSON.
func (n *Notifier) PushEvent(eventType, severity, message string, details interface{}) {
if !n.enabled {
return
}
var detailsJSON json.RawMessage
if details != nil {
b, err := json.Marshal(details)
if err != nil {
n.logger.Printf("[WARN] PushEvent: failed to marshal details for %s: %v", eventType, err)
} else {
detailsJSON = b
}
}
payload := eventRequest{
CustomerID: n.customerID,
EventType: eventType,
Severity: severity,
Message: message,
Details: detailsJSON,
}
jsonData, err := json.Marshal(payload)
if err != nil {
n.logger.Printf("[ERROR] PushEvent: marshal failed for %s: %v", eventType, err)
return
}
go func() {
url := n.hubURL + "/api/v1/event"
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
if attempt > 0 {
time.Sleep(3 * time.Second)
}
req, err := http.NewRequest("POST", url, bytes.NewReader(jsonData))
if err != nil {
lastErr = err
continue
}
req.Header.Set("Authorization", "Bearer "+n.apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := n.httpClient.Do(req)
if err != nil {
lastErr = err
continue
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
n.logger.Printf("[INFO] Event pushed: %s (%s) — %s", eventType, severity, message)
return
}
lastErr = fmt.Errorf("HTTP %d", resp.StatusCode)
}
n.logger.Printf("[WARN] Event push failed after 3 attempts (%s/%s): %v", eventType, severity, lastErr)
}()
}
// ── Convenience methods ──────────────────────────────────────────────
// NotifyHealthChange checks if health status changed and sends appropriate events.
// Detects both degradation (ok→warn, ok→fail, warn→fail) and recovery (fail→ok, warn→ok, fail→warn).
func (n *Notifier) NotifyHealthChange(status string, issues, warnings []string) {
if !n.enabled {
return
}
n.mu.Lock()
prev := n.prevHealthStatus
n.prevHealthStatus = status
n.mu.Unlock()
if prev == "" {
return // First run, just record status
}
if status == prev {
return
}
details := HealthDetails{
PreviousStatus: prev,
CurrentStatus: status,
Issues: issues,
Warnings: warnings,
}
prevRank := statusRank(prev)
newRank := statusRank(status)
if newRank > prevRank {
// Degradation
if status == "fail" {
n.PushEvent("health_critical", "error",
fmt.Sprintf("Rendszer állapot kritikus (volt: %s)", prev), details)
} else if status == "warn" {
n.PushEvent("health_degraded", "warning",
fmt.Sprintf("Rendszer állapot romlott (volt: %s)", prev), details)
}
} else {
// Recovery
n.PushEvent("health_recovered", "info",
fmt.Sprintf("Rendszer állapot helyreállt: %s (volt: %s)", status, prev), details)
}
}
// NotifyBackupFailed sends a backup failure event.
func (n *Notifier) NotifyBackupFailed(message, errMsg string) {
n.PushEvent("backup_failed", "error", message, BackupDetails{Error: errMsg})
}
// NotifyBackupCompleted sends a backup success event.
func (n *Notifier) NotifyBackupCompleted(details BackupDetails) {
n.PushEvent("backup_completed", "info", "Biztonsági mentés elkészült", details)
}
// NotifyDBDumpFailed sends a DB dump failure event.
func (n *Notifier) NotifyDBDumpFailed(message, errMsg string) {
n.PushEvent("db_dump_failed", "error", message, DBDumpDetails{Error: errMsg})
}
// NotifyDBDumpCompleted sends a DB dump success event.
func (n *Notifier) NotifyDBDumpCompleted(details DBDumpDetails) {
n.PushEvent("db_dump_completed", "info", "Adatbázis mentés elkészült", details)
}
// NotifyIntegrityFailed sends a backup integrity check failure event.
func (n *Notifier) NotifyIntegrityFailed(message, errMsg string) {
n.PushEvent("backup_integrity_failed", "error", message, &BackupDetails{Error: errMsg})
}
// NotifyIntegrityOK sends a backup integrity check success event.
func (n *Notifier) NotifyIntegrityOK(message string) {
n.PushEvent("backup_integrity_ok", "info", message, nil)
}
// NotifyControllerUpdated sends a controller update event.
func (n *Notifier) NotifyControllerUpdated(fromVer, toVer string, success bool) {
severity := "info"
msg := fmt.Sprintf("Controller frissítve: %s → %s", fromVer, toVer)
details := UpdateDetails{FromVersion: fromVer, ToVersion: toVer}
if !success {
severity = "error"
msg = fmt.Sprintf("Controller frissítés sikertelen: %s → %s", fromVer, toVer)
}
n.PushEvent("controller_updated", severity, msg, details)
}
// NotifyControllerStarted sends a controller startup event.
func (n *Notifier) NotifyControllerStarted(version string) {
n.PushEvent("controller_started", "info",
fmt.Sprintf("Controller elindult (v%s)", version), nil)
}
// NotifyStorageDisconnected sends a drive disconnection event.
func (n *Notifier) NotifyStorageDisconnected(label string, stoppedApps []string) {
msg := fmt.Sprintf("Meghajtó váratlanul leválasztva: %s", label)
n.PushEvent("storage_disconnected", "error", msg, StorageDetails{
Label: label,
StoppedApps: stoppedApps,
})
}
// NotifyStorageReconnected sends a drive reconnection event.
func (n *Notifier) NotifyStorageReconnected(label string) {
n.PushEvent("storage_reconnected", "info",
fmt.Sprintf("Meghajtó újra csatlakoztatva: %s", label), StorageDetails{Label: label})
}
// NotifyAppDeployed sends an app deployment event.
func (n *Notifier) NotifyAppDeployed(stackName, displayName string) {
n.PushEvent("app_deployed", "info",
fmt.Sprintf("Alkalmazás telepítve: %s", displayName),
AppDetails{StackName: stackName, DisplayName: displayName})
}
// NotifyAppRemoved sends an app removal event.
func (n *Notifier) NotifyAppRemoved(stackName, displayName string) {
n.PushEvent("app_removed", "info",
fmt.Sprintf("Alkalmazás eltávolítva: %s", displayName),
AppDetails{StackName: stackName, DisplayName: displayName})
}
// NotifyCrossDriveCompleted sends a cross-drive backup success event.
func (n *Notifier) NotifyCrossDriveCompleted(details CrossDriveDetails) {
n.PushEvent("crossdrive_completed", "info",
fmt.Sprintf("Másodlagos mentés elkészült: %s", details.StackName), details)
}
// NotifyCrossDriveFailed sends a cross-drive backup failure event.
func (n *Notifier) NotifyCrossDriveFailed(details CrossDriveDetails) {
n.PushEvent("crossdrive_failed", "error",
fmt.Sprintf("Másodlagos mentés sikertelen: %s", details.StackName), details)
}
// NotifyDRStarted sends a disaster recovery start event.
func (n *Notifier) NotifyDRStarted(appCount int) {
n.PushEvent("disaster_recovery_started", "warning",
fmt.Sprintf("Katasztrófa helyreállítás elindítva (%d alkalmazás)", appCount), nil)
}
// NotifyDRCompleted sends a disaster recovery completion event.
func (n *Notifier) NotifyDRCompleted(successCount, failCount int) {
severity := "info"
if failCount > 0 {
severity = "warning"
}
n.PushEvent("disaster_recovery_completed", severity,
fmt.Sprintf("Katasztrófa helyreállítás befejezve (%d sikeres, %d sikertelen)", successCount, failCount), nil)
}
// ── Preferences sync ─────────────────────────────────────────────────
type preferencesRequest struct {
CustomerID string `json:"customer_id"`
Email string `json:"email"`
EnabledEvents []string `json:"enabled_events"`
CooldownHours int `json:"cooldown_hours,omitempty"`
}
// SyncPreferences pushes the current notification preferences to the hub.
// Called after the user saves notification settings on the settings page.
// Synchronous — returns error for the handler to display to the user.
func (n *Notifier) SyncPreferences(email string, enabledEvents []string) error {
func (n *Notifier) SyncPreferences(email string, enabledEvents []string, cooldownHours int) error {
if !n.enabled {
return fmt.Errorf("hub nem konfigurált")
}
@@ -82,6 +366,7 @@ func (n *Notifier) SyncPreferences(email string, enabledEvents []string) error {
CustomerID: n.customerID,
Email: email,
EnabledEvents: enabledEvents,
CooldownHours: cooldownHours,
}
jsonData, err := json.Marshal(payload)
@@ -108,172 +393,23 @@ func (n *Notifier) SyncPreferences(email string, enabledEvents []string) error {
return fmt.Errorf("hub hiba (%d): %s", resp.StatusCode, string(body))
}
n.logger.Printf("[INFO] Notification preferences synced to hub: email=%s, events=%v", email, enabledEvents)
n.logger.Printf("[INFO] Notification preferences synced to hub: email=%s, events=%v, cooldown=%dh", email, enabledEvents, cooldownHours)
return nil
}
// notifyRequest is the JSON payload sent to the hub.
type notifyRequest struct {
CustomerID string `json:"customer_id"`
EventType string `json:"event_type"`
Severity string `json:"severity"`
Message string `json:"message"`
Details string `json:"details,omitempty"`
}
// ── Test notification ────────────────────────────────────────────────
// Notify sends a notification event to the hub relay.
// Checks local cooldown and event preferences before sending.
// Non-blocking: fires the HTTP request in a goroutine.
func (n *Notifier) Notify(eventType, severity, message, details string) {
if !n.enabled {
return
}
prefs := n.settings.GetNotificationPrefs()
if prefs.Email == "" {
return // No email configured, skip
}
// Check if event is enabled in preferences
eventEnabled := false
for _, e := range prefs.EnabledEvents {
if e == eventType {
eventEnabled = true
break
}
}
if !eventEnabled {
return
}
// Check cooldown — per-event override takes priority over global
cooldownDuration := time.Duration(prefs.CooldownHours) * time.Hour
if cooldownDuration == 0 {
cooldownDuration = 6 * time.Hour
}
if override, ok := n.perEventCooldown[eventType]; ok {
cooldownDuration = override
}
n.mu.Lock()
lastSent, exists := n.cooldowns[eventType]
if exists && time.Since(lastSent) < cooldownDuration {
n.mu.Unlock()
n.logger.Printf("[DEBUG] Notification cooldown active for %s (sent %s ago)", eventType, time.Since(lastSent).Round(time.Minute))
return
}
n.cooldowns[eventType] = time.Now()
n.mu.Unlock()
// Fire the notification in a goroutine (non-blocking)
go func() {
payload := notifyRequest{
CustomerID: n.customerID,
EventType: eventType,
Severity: severity,
Message: message,
Details: details,
}
jsonData, err := json.Marshal(payload)
if err != nil {
n.logger.Printf("[ERROR] Failed to marshal notification: %v", err)
return
}
url := n.hubURL + "/api/v1/notify"
req, err := http.NewRequest("POST", url, bytes.NewReader(jsonData))
if err != nil {
n.logger.Printf("[ERROR] Failed to create notification request: %v", err)
return
}
req.Header.Set("Authorization", "Bearer "+n.apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := n.httpClient.Do(req)
if err != nil {
n.logger.Printf("[WARN] Failed to send notification to hub: %v", err)
return
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
n.logger.Printf("[WARN] Hub notification returned %d for %s/%s", resp.StatusCode, eventType, severity)
return
}
n.logger.Printf("[INFO] Notification sent: %s (%s) — %s", eventType, severity, message)
}()
}
// NotifyHealthChange checks if health status changed and sends appropriate notifications.
// Call this after each health check with the new report status/issues/warnings.
func (n *Notifier) NotifyHealthChange(status string, issues, warnings []string) {
if !n.enabled {
return
}
prev := n.prevHealthStatus
n.prevHealthStatus = status
// Only notify on status degradation (ok→warn, ok→fail, warn→fail)
if prev == "" {
return // First run, just record status
}
if statusRank(status) <= statusRank(prev) {
return // Status improved or stayed the same
}
// Notify about each issue/warning
for _, issue := range issues {
n.Notify("container_unhealthy", "critical", issue, "")
}
for _, w := range warnings {
// Determine specific event type from warning message
eventType := classifyWarning(w)
n.Notify(eventType, "warning", w, "")
}
}
// NotifyBackupFailed sends a notification about a backup failure.
func (n *Notifier) NotifyBackupFailed(message, details string) {
n.Notify("backup_failed", "critical", message, details)
}
// NotifyDBDumpFailed sends a notification about a database dump failure.
func (n *Notifier) NotifyDBDumpFailed(message, details string) {
n.Notify("db_dump_failed", "critical", message, details)
}
// NotifyIntegrityFailed sends a notification about a backup integrity check failure.
func (n *Notifier) NotifyIntegrityFailed(message, details string) {
n.Notify("integrity_failed", "warning", message, details)
}
// NotifyUpdateSuccess sends a notification about a successful controller update.
func (n *Notifier) NotifyUpdateSuccess(fromVer, toVer string) {
n.Notify("update_success", "info",
fmt.Sprintf("Controller frissítve: %s → %s", fromVer, toVer), "")
}
// NotifyUpdateFailed sends a notification about a failed controller update.
func (n *Notifier) NotifyUpdateFailed(targetVer, errMsg string) {
n.Notify("update_failed", "warning",
fmt.Sprintf("Controller frissítés sikertelen: %s — %s", targetVer, errMsg), "")
}
// SendTest sends a test notification for verifying the notification flow.
// SendTest sends a test event for verifying the notification flow (synchronous).
func (n *Notifier) SendTest() error {
if !n.enabled {
return fmt.Errorf("notifications not enabled (hub not configured)")
}
payload := notifyRequest{
payload := eventRequest{
CustomerID: n.customerID,
EventType: "test",
Severity: "info",
Message: "Teszt értesítés a Felhom rendszerből",
Details: "Ha ezt az emailt megkapta, az értesítések megfelelően működnek.",
}
jsonData, err := json.Marshal(payload)
@@ -281,7 +417,7 @@ func (n *Notifier) SendTest() error {
return fmt.Errorf("marshal: %w", err)
}
url := n.hubURL + "/api/v1/notify"
url := n.hubURL + "/api/v1/event"
req, err := http.NewRequest("POST", url, bytes.NewReader(jsonData))
if err != nil {
return fmt.Errorf("request: %w", err)
@@ -302,6 +438,59 @@ func (n *Notifier) SendTest() error {
return nil
}
// ── Backward compatibility ───────────────────────────────────────────
// notifyRequest is the JSON payload for the legacy /api/v1/notify endpoint.
type notifyRequest struct {
CustomerID string `json:"customer_id"`
EventType string `json:"event_type"`
Severity string `json:"severity"`
Message string `json:"message"`
Details string `json:"details,omitempty"`
}
// Notify sends a legacy notification to /api/v1/notify (backward compat).
// Kept for old Hub instances that don't support /api/v1/event yet.
// No local cooldown — Hub handles cooldowns.
func (n *Notifier) Notify(eventType, severity, message, details string) {
if !n.enabled {
return
}
go func() {
payload := notifyRequest{
CustomerID: n.customerID,
EventType: eventType,
Severity: severity,
Message: message,
Details: details,
}
jsonData, err := json.Marshal(payload)
if err != nil {
n.logger.Printf("[ERROR] Failed to marshal notification: %v", err)
return
}
url := n.hubURL + "/api/v1/notify"
req, err := http.NewRequest("POST", url, bytes.NewReader(jsonData))
if err != nil {
return
}
req.Header.Set("Authorization", "Bearer "+n.apiKey)
req.Header.Set("Content-Type", "application/json")
resp, err := n.httpClient.Do(req)
if err != nil {
return
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
}()
}
// ── Helpers ──────────────────────────────────────────────────────────
func statusRank(status string) int {
switch status {
case "ok":
@@ -315,41 +504,3 @@ func statusRank(status string) int {
}
}
func classifyWarning(message string) string {
// Try to classify the warning message into a specific event type
switch {
case contains(message, "disk") || contains(message, "Disk") || contains(message, "SSD") || contains(message, "HDD"):
if contains(message, "critical") || contains(message, "Critical") {
return "disk_critical"
}
return "disk_warning"
case contains(message, "Memory") || contains(message, "memory"):
return "disk_warning" // group memory under system warnings
case contains(message, "Temperature") || contains(message, "temperature"):
return "disk_warning" // group temp under system warnings
case contains(message, "container") || contains(message, "Container"):
return "container_unhealthy"
default:
return "disk_warning" // fallback to generic system warning
}
}
func contains(s, substr string) bool {
return strings.Contains(s, substr)
}
// NotifyStorageDisconnected sends a notification about a drive disconnection.
func (n *Notifier) NotifyStorageDisconnected(label string, stoppedApps []string) {
msg := fmt.Sprintf("Meghajtó váratlanul leválasztva: %s", label)
details := ""
if len(stoppedApps) > 0 {
details = fmt.Sprintf("Leállított alkalmazások: %s", strings.Join(stoppedApps, ", "))
}
n.Notify("storage_disconnected", "critical", msg, details)
}
// NotifyStorageReconnected sends a notification about a drive reconnection.
func (n *Notifier) NotifyStorageReconnected(label string) {
n.Notify("storage_reconnected", "info",
fmt.Sprintf("Meghajtó újra csatlakoztatva: %s. Az alkalmazások manuálisan indíthatók.", label), "")
}