426 lines
12 KiB
Go
426 lines
12 KiB
Go
// Package agent ist `sultan agent`: Er registriert den Host beim Server
|
||
// (Enrollment mit Einladung und Pinning), hält den SSE-Stream offen,
|
||
// beobachtet Docker und meldet seinen vollständigen Zustand
|
||
// (service_state) – nach jedem (Wieder-)Verbinden und bei jeder Änderung.
|
||
//
|
||
// Im Datenverzeichnis (Default ./suldata) liegen node.key (0600) und
|
||
// node.json (inkl. server_fingerprint, dem Pin des Servers).
|
||
package agent
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log/slog"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"sultan/pkg/auth"
|
||
"sultan/pkg/crypto"
|
||
"sultan/pkg/docker"
|
||
"sultan/pkg/policy"
|
||
"sultan/pkg/registry"
|
||
"sultan/pkg/stream"
|
||
)
|
||
|
||
// NoDocker als DockerSocket bzw. ConfigDir schaltet die jeweilige Quelle ab.
|
||
const NoDocker = "none"
|
||
|
||
// Config ist die Konfiguration des Agenten; Nullwerte bekommen Defaults.
|
||
type Config struct {
|
||
DataDir string // Default ./suldata
|
||
ServerURL string // https://ingress:8443 – Override; sonst aus der Einladung bzw. node.json
|
||
Token string // Einladung, nur beim ersten Start
|
||
NodeID string // Default aus dem Hostnamen
|
||
Hostname string // Default os.Hostname()
|
||
|
||
DockerSocket string // Default /var/run/docker.sock; "none" schaltet Docker ab
|
||
BackendHost string // Adresse, unter der der Ingress die Dienste erreicht
|
||
BackendInterface string // alternativ: IP dieses Interfaces (z. B. wt0)
|
||
|
||
ConfigDir string // Default /etc/sultan.d (*.yaml, Spec 4.1b); "none" schaltet ab
|
||
ConfigPoll time.Duration // Default 5s: Prüfintervall für Änderungen in ConfigDir
|
||
|
||
Logger *slog.Logger
|
||
Debounce time.Duration // Default 500ms: Änderungen sammeln, bevor gesendet wird
|
||
RetryInterval time.Duration // Default 5s: erneuter Sendeversuch nach Fehler
|
||
ReconnectMin time.Duration // Default 1s: kleinste Wartezeit vor Neuverbindung
|
||
}
|
||
|
||
// Agent ist ein startbereiter Agent.
|
||
type Agent struct {
|
||
cfg Config
|
||
log *slog.Logger
|
||
nodeIDErr error
|
||
|
||
signer *auth.Signer
|
||
id Identity
|
||
serverURL string // tatsächlich genutzte Adresse (Override oder node.json)
|
||
|
||
mu sync.Mutex
|
||
fileSpecs []ServiceSpec
|
||
health healthTracker
|
||
|
||
keyMu sync.RWMutex // schützt signer (rotate_key tauscht ihn im Betrieb)
|
||
kick func() // Zustand neu senden (in Run gesetzt)
|
||
runCtx context.Context
|
||
hc *http.Client
|
||
}
|
||
|
||
// httpClient liefert den gepinnten Client für den Server.
|
||
func (a *Agent) httpClient() *http.Client {
|
||
if a.hc == nil {
|
||
a.hc = &http.Client{Transport: &http.Transport{TLSClientConfig: crypto.PinnedTLSConfig(a.id.ServerFingerprint)}}
|
||
}
|
||
return a.hc
|
||
}
|
||
|
||
// New prüft die Konfiguration und setzt Defaults; Dateien werden erst von
|
||
// Enroll/Run angefasst.
|
||
func New(cfg Config) (*Agent, error) {
|
||
if cfg.DataDir == "" {
|
||
cfg.DataDir = "./suldata"
|
||
}
|
||
if cfg.DockerSocket == "" {
|
||
cfg.DockerSocket = "/var/run/docker.sock"
|
||
}
|
||
if cfg.ConfigDir == "" {
|
||
cfg.ConfigDir = "/etc/sultan.d"
|
||
}
|
||
if cfg.ConfigPoll <= 0 {
|
||
cfg.ConfigPoll = 5 * time.Second
|
||
}
|
||
if cfg.Logger == nil {
|
||
cfg.Logger = slog.New(slog.NewTextHandler(os.Stderr, nil))
|
||
}
|
||
if cfg.Debounce <= 0 {
|
||
cfg.Debounce = 500 * time.Millisecond
|
||
}
|
||
if cfg.RetryInterval <= 0 {
|
||
cfg.RetryInterval = 5 * time.Second
|
||
}
|
||
if cfg.ReconnectMin <= 0 {
|
||
cfg.ReconnectMin = time.Second
|
||
}
|
||
if cfg.ServerURL != "" {
|
||
u, err := crypto.ValidateServerURL(cfg.ServerURL)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("agent: --server: %w", err)
|
||
}
|
||
cfg.ServerURL = u
|
||
}
|
||
if cfg.Hostname == "" {
|
||
cfg.Hostname, _ = os.Hostname()
|
||
}
|
||
a := &Agent{cfg: cfg, log: cfg.Logger}
|
||
if cfg.NodeID != "" {
|
||
if !policy.ValidID(cfg.NodeID) {
|
||
return nil, fmt.Errorf("agent: Knoten-ID %q ungültig (a–z, 0–9, - und _, kein \"__\")", cfg.NodeID)
|
||
}
|
||
} else {
|
||
a.cfg.NodeID, a.nodeIDErr = DefaultNodeID(cfg.Hostname)
|
||
}
|
||
return a, nil
|
||
}
|
||
|
||
// errNotConnected: Der Server nimmt den Zustand erst bei offenem Stream an.
|
||
var errNotConnected = errors.New("Stream noch nicht verbunden")
|
||
|
||
// Run registriert den Agenten (falls nötig) und arbeitet, bis ctx endet.
|
||
func (a *Agent) Run(ctx context.Context) error {
|
||
if err := a.Enroll(ctx); err != nil {
|
||
return err
|
||
}
|
||
host, err := backendHost(a.cfg.BackendHost, a.cfg.BackendInterface, a.serverURL)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
hc := a.httpClient()
|
||
a.runCtx = ctx
|
||
// Abgebrochener Schlüsselwechsel? Vor dem Verbinden abschließen.
|
||
if _, err := os.Stat(a.path("node.key.new")); err == nil {
|
||
if err := a.completeRotation(ctx); err != nil {
|
||
a.log.Error("Begonnenen Schlüsselwechsel abschließen", "err", err)
|
||
}
|
||
}
|
||
a.log.Info("Agent gestartet", "node", a.id.NodeID, "server", a.serverURL, "backend", host,
|
||
"docker", a.cfg.DockerSocket, "config_dir", a.cfg.ConfigDir)
|
||
|
||
trigger := make(chan struct{}, 1)
|
||
kick := func() {
|
||
select {
|
||
case trigger <- struct{}{}:
|
||
default:
|
||
}
|
||
}
|
||
a.kick = kick
|
||
var sc *stream.Client
|
||
sc = &stream.Client{
|
||
URL: a.currentServerURL() + "/api/v1/stream",
|
||
HTTPClient: hc,
|
||
Prepare: func(r *http.Request) error { return a.currentSigner().Sign(r) },
|
||
MinBackoff: a.cfg.ReconnectMin,
|
||
OnConnect: func() {
|
||
a.log.Info("Mit Server verbunden")
|
||
a.rememberServerURL()
|
||
kick() // Server hat die Routen beim Trennen entfernt: Zustand neu senden
|
||
},
|
||
OnError: func(err error, retryIn time.Duration) {
|
||
// Ausweichweg: nächster Versuch über die nächste bekannte Adresse.
|
||
next := a.nextServerURL()
|
||
sc.URL = next + "/api/v1/stream"
|
||
a.log.Warn("Verbindung zum Server unterbrochen", "err", err, "retry_in", retryIn, "next", next)
|
||
},
|
||
}
|
||
|
||
var wg sync.WaitGroup
|
||
wg.Go(func() { sc.Run(ctx, a.handleEvent) })
|
||
if a.cfg.DockerSocket != NoDocker {
|
||
wg.Go(func() { a.watchDocker(ctx, kick) })
|
||
}
|
||
if a.cfg.ConfigDir != NoDocker {
|
||
wg.Go(func() { a.watchServiceDir(ctx, kick) })
|
||
}
|
||
wg.Go(func() { a.sendLoop(ctx, hc, host, trigger, kick) })
|
||
<-ctx.Done()
|
||
wg.Wait()
|
||
a.log.Info("Agent beendet")
|
||
return nil
|
||
}
|
||
|
||
// sendLoop sendet den Zustand, sobald ausgelöst – gesammelt über Debounce.
|
||
func (a *Agent) sendLoop(ctx context.Context, hc *http.Client, host string, trigger <-chan struct{}, kick func()) {
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-trigger:
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-time.After(a.cfg.Debounce):
|
||
}
|
||
|
||
state, err := a.collect(ctx, host)
|
||
if err == nil {
|
||
err = a.postState(ctx, hc, state)
|
||
}
|
||
switch {
|
||
case err == nil:
|
||
a.log.Debug("Zustand gesendet", "workloads", len(state.Workloads))
|
||
case ctx.Err() != nil:
|
||
return
|
||
case errors.Is(err, errNotConnected):
|
||
// Kein erneuter Versuch nötig: OnConnect löst ohnehin aus.
|
||
default:
|
||
a.log.Warn("Zustand senden fehlgeschlagen", "err", err, "retry_in", a.cfg.RetryInterval)
|
||
time.AfterFunc(a.cfg.RetryInterval, kick)
|
||
}
|
||
}
|
||
}
|
||
|
||
// collect ermittelt den aktuellen Zustand: Docker-Container und Dienste aus
|
||
// ConfigDir samt Ausfällen (gestoppte Container, Healthchecks). Ohne Docker-Socket (Bare-Metal)
|
||
// entfallen die Container; schlägt die Docker-API fehl, wird nichts
|
||
// gesendet (sonst würden alle Container-Routen entfernt).
|
||
func (a *Agent) collect(ctx context.Context, host string) (registry.NodeState, error) {
|
||
state := registry.NodeState{Workloads: []registry.Workload{}}
|
||
if a.dockerAvailable() {
|
||
client := &docker.Client{Socket: a.cfg.DockerSocket}
|
||
containers, err := client.List(ctx)
|
||
if err != nil {
|
||
return state, err
|
||
}
|
||
// Gestoppte Container nennen in der Liste keine Ports.
|
||
for i, c := range containers {
|
||
if c.State == "running" || !enabled(c.Labels) {
|
||
continue
|
||
}
|
||
ports, err := client.PortBindings(ctx, c.ID)
|
||
if err != nil {
|
||
a.log.Debug("Portzuordnung nicht lesbar", "container", c.Name(), "err", err)
|
||
continue // inzwischen entfernt o. Ä. → Workloads warnt
|
||
}
|
||
containers[i].Ports = ports
|
||
}
|
||
ws, down, warnings := Workloads(containers, host)
|
||
for _, w := range warnings {
|
||
a.log.Warn("Container nicht übernommen", "grund", w)
|
||
}
|
||
state.Workloads = append(state.Workloads, ws...)
|
||
state.DownServices = append(state.DownServices, down...)
|
||
}
|
||
a.mu.Lock()
|
||
specs := a.fileSpecs
|
||
a.mu.Unlock()
|
||
state.Workloads = append(state.Workloads, FileWorkloads(specs, host)...)
|
||
state.DownServices = append(state.DownServices, a.health.down()...)
|
||
return state, nil
|
||
}
|
||
|
||
func (a *Agent) dockerAvailable() bool {
|
||
if a.cfg.DockerSocket == NoDocker {
|
||
return false
|
||
}
|
||
_, err := os.Stat(a.cfg.DockerSocket)
|
||
return !errors.Is(err, os.ErrNotExist)
|
||
}
|
||
|
||
// watchServiceDir prüft ConfigDir regelmäßig auf Änderungen (Polling statt
|
||
// inotify: zuverlässig auch bei Bind-Mounts und atomar ersetzten Dateien).
|
||
// Bei einer Änderung werden die Dienste neu geladen, die Healthchecks neu
|
||
// gestartet und der Zustand gesendet.
|
||
func (a *Agent) watchServiceDir(ctx context.Context, kick func()) {
|
||
var last string
|
||
var stopHealth context.CancelFunc = func() {}
|
||
var healthWG sync.WaitGroup
|
||
defer func() { stopHealth(); healthWG.Wait() }()
|
||
|
||
ticker := time.NewTicker(a.cfg.ConfigPoll)
|
||
defer ticker.Stop()
|
||
for {
|
||
if fp := dirFingerprint(a.cfg.ConfigDir); fp != last {
|
||
last = fp
|
||
specs, warnings := LoadServiceDir(a.cfg.ConfigDir)
|
||
for _, w := range warnings {
|
||
a.log.Warn("Dienstdatei fehlerhaft", "grund", w)
|
||
}
|
||
a.log.Info("Dienste aus Konfigurationsverzeichnis geladen", "dir", a.cfg.ConfigDir, "anzahl", len(specs))
|
||
a.mu.Lock()
|
||
a.fileSpecs = specs
|
||
a.mu.Unlock()
|
||
|
||
stopHealth()
|
||
healthWG.Wait()
|
||
hctx, cancel := context.WithCancel(ctx)
|
||
stopHealth = cancel
|
||
keep := map[string]bool{}
|
||
for _, s := range specs {
|
||
if s.Healthcheck != nil {
|
||
keep[s.Name] = true
|
||
healthWG.Go(func() { a.healthLoop(hctx, s.Name, *s.Healthcheck, kick) })
|
||
}
|
||
}
|
||
a.health.retain(keep)
|
||
kick()
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}
|
||
|
||
func (a *Agent) healthLoop(ctx context.Context, name string, spec HealthSpec, kick func()) {
|
||
client := &http.Client{}
|
||
ticker := time.NewTicker(spec.Interval.Duration)
|
||
defer ticker.Stop()
|
||
for {
|
||
err := checkHealth(ctx, client, spec)
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
if a.health.report(name, spec.Failures, err == nil) {
|
||
if err != nil {
|
||
a.log.Warn("Dienst ausgefallen – Wartungsseite aktiv", "service", name, "grund", err)
|
||
} else {
|
||
a.log.Info("Dienst wieder erreichbar", "service", name)
|
||
}
|
||
kick()
|
||
}
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}
|
||
|
||
// dirFingerprint fasst Namen und Inhalte aller YAML-Dateien zusammen.
|
||
func dirFingerprint(dir string) string {
|
||
entries, err := os.ReadDir(dir)
|
||
if err != nil {
|
||
return "fehlt"
|
||
}
|
||
h := sha256.New()
|
||
for _, e := range entries {
|
||
if ext := filepath.Ext(e.Name()); e.IsDir() || (ext != ".yaml" && ext != ".yml") {
|
||
continue
|
||
}
|
||
data, _ := os.ReadFile(filepath.Join(dir, e.Name()))
|
||
fmt.Fprintf(h, "%s\x00%d\x00", e.Name(), len(data))
|
||
h.Write(data)
|
||
}
|
||
return hex.EncodeToString(h.Sum(nil))
|
||
}
|
||
|
||
func (a *Agent) postState(ctx context.Context, hc *http.Client, state registry.NodeState) error {
|
||
body, err := json.Marshal(state)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.currentServerURL()+"/api/v1/state", bytes.NewReader(body))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
if err := a.currentSigner().Sign(req); err != nil {
|
||
return err
|
||
}
|
||
resp, err := hc.Do(req)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer resp.Body.Close()
|
||
switch resp.StatusCode {
|
||
case http.StatusNoContent:
|
||
return nil
|
||
case http.StatusConflict:
|
||
return errNotConnected
|
||
default:
|
||
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||
return fmt.Errorf("Server antwortet %d: %s", resp.StatusCode, strings.TrimSpace(string(msg)))
|
||
}
|
||
}
|
||
|
||
// watchDocker löst bei jedem Container-Event und nach jedem (Wieder-)Aufbau
|
||
// des Event-Streams ein erneutes Senden aus.
|
||
func (a *Agent) watchDocker(ctx context.Context, kick func()) {
|
||
client := &docker.Client{Socket: a.cfg.DockerSocket}
|
||
backoff := a.cfg.ReconnectMin
|
||
for ctx.Err() == nil {
|
||
events, errc := client.Events(ctx)
|
||
kick() // verpasste Änderungen nachholen
|
||
received := false
|
||
for range events {
|
||
received = true
|
||
kick()
|
||
}
|
||
err := <-errc
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
if received {
|
||
backoff = a.cfg.ReconnectMin
|
||
}
|
||
a.log.Debug("Docker-Events unterbrochen", "err", err, "retry_in", backoff)
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-time.After(backoff):
|
||
}
|
||
backoff = min(backoff*2, 30*time.Second)
|
||
}
|
||
}
|