Files
deploy-felhom-compose/controller/internal/report/pusher.go
T
admin 8aebbb8902 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>
2026-02-20 18:53:21 +01:00

196 lines
4.4 KiB
Go

package report
import (
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"strings"
"sync"
"time"
"gitea.dooplex.hu/admin/felhom-controller/internal/config"
)
// PushStatus tracks the last hub push attempt and result.
type PushStatus struct {
LastAttempt time.Time
LastSuccess time.Time
LastError string
Consecutive int // consecutive failures
}
// Pusher sends reports to the central hub.
type Pusher struct {
hubURL string
apiKey string
httpClient *http.Client
logger *log.Logger
enabled bool
statusMu sync.RWMutex
status PushStatus
}
// NewPusher creates a new report pusher from hub configuration.
func NewPusher(cfg *config.HubConfig, logger *log.Logger) *Pusher {
return &Pusher{
hubURL: strings.TrimRight(cfg.URL, "/"),
apiKey: cfg.APIKey,
httpClient: &http.Client{
Timeout: 30 * time.Second,
},
logger: logger,
enabled: cfg.Enabled,
}
}
// Push sends a report to the hub. Retries 3 times with 5s backoff.
func (p *Pusher) Push(report *Report) error {
if !p.enabled {
return nil
}
data, err := json.Marshal(report)
if err != nil {
return fmt.Errorf("marshal report: %w", err)
}
url := p.hubURL + "/api/v1/report"
p.statusMu.Lock()
p.status.LastAttempt = time.Now()
p.statusMu.Unlock()
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
if attempt > 0 {
time.Sleep(5 * time.Second)
}
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data))
if err != nil {
lastErr = err
continue
}
req.Header.Set("Content-Type", "application/json")
if p.apiKey != "" {
req.Header.Set("Authorization", "Bearer "+p.apiKey)
}
resp, err := p.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 {
p.logger.Printf("[INFO] Hub report pushed successfully (%d bytes)", len(data))
p.statusMu.Lock()
p.status.LastSuccess = time.Now()
p.status.LastError = ""
p.status.Consecutive = 0
p.statusMu.Unlock()
return nil
}
lastErr = fmt.Errorf("HTTP %d", resp.StatusCode)
}
p.statusMu.Lock()
p.status.LastError = lastErr.Error()
p.status.Consecutive++
p.statusMu.Unlock()
return fmt.Errorf("hub push failed after 3 attempts: %w", lastErr)
}
// GetStatus returns a snapshot of the current push status.
func (p *Pusher) GetStatus() PushStatus {
p.statusMu.RLock()
defer p.statusMu.RUnlock()
return p.status
}
// PushInfraBackup sends the infrastructure backup payload to the Hub.
// Uses the same retry logic as Push.
func (p *Pusher) PushInfraBackup(data []byte) error {
if !p.enabled {
return nil
}
url := p.hubURL + "/api/v1/infra-backup"
var lastErr error
for attempt := 0; attempt < 3; attempt++ {
if attempt > 0 {
time.Sleep(5 * time.Second)
}
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data))
if err != nil {
lastErr = err
continue
}
req.Header.Set("Content-Type", "application/json")
if p.apiKey != "" {
req.Header.Set("Authorization", "Bearer "+p.apiKey)
}
resp, err := p.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 {
p.logger.Printf("[INFO] Infra backup pushed to Hub (%d bytes)", len(data))
return nil
}
lastErr = fmt.Errorf("HTTP %d", resp.StatusCode)
}
return fmt.Errorf("infra backup push failed after 3 attempts: %w", lastErr)
}
// PushOnce sends a single report regardless of the enabled flag.
// Used for one-time notifications (e.g., reporting-disabled on startup).
func (p *Pusher) PushOnce(report *Report) error {
if p.hubURL == "" || p.apiKey == "" {
return nil
}
data, err := json.Marshal(report)
if err != nil {
p.logger.Printf("[WARN] Hub report marshal failed: %v", err)
return nil
}
url := p.hubURL + "/api/v1/report"
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(data))
if err != nil {
return nil
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+p.apiKey)
resp, err := p.httpClient.Do(req)
if err != nil {
p.logger.Printf("[WARN] Hub disabled-notification failed: %v", err)
return nil
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
p.logger.Printf("[INFO] Hub disabled-notification sent (%d bytes)", len(data))
}
return nil
}