software steht so weit.
jetzt build + docker build und testen
This commit is contained in:
br committed
1 parent
42f0c0accb
commit
ca6cdcca63
90 files changed
+10298
-480
No files matched your search
@@ -0,0 +1,411 @@
|
||||
// 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 Healthcheck-Ausfällen. 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() {
|
||||
containers, err := (&docker.Client{Socket: a.cfg.DockerSocket}).List(ctx)
|
||||
if err != nil {
|
||||
return state, err
|
||||
}
|
||||
ws, warnings := Workloads(containers, host)
|
||||
for _, w := range warnings {
|
||||
a.log.Warn("Container nicht übernommen", "grund", w)
|
||||
}
|
||||
state.Workloads = append(state.Workloads, ws...)
|
||||
}
|
||||
a.mu.Lock()
|
||||
specs := a.fileSpecs
|
||||
a.mu.Unlock()
|
||||
state.Workloads = append(state.Workloads, FileWorkloads(specs, host)...)
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user