jbergner
440423c5b6
All checks were successful
release-tag / release-image (push) Successful in 2m43s
135 lines
7.8 KiB
Go
135 lines
7.8 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/local/glpi-neural-brain/internal/activity"
|
|
"github.com/local/glpi-neural-brain/internal/config"
|
|
"github.com/local/glpi-neural-brain/internal/engine"
|
|
"github.com/local/glpi-neural-brain/internal/graph"
|
|
"github.com/local/glpi-neural-brain/internal/ingest"
|
|
"github.com/local/glpi-neural-brain/internal/sourceagent"
|
|
webui "github.com/local/glpi-neural-brain/internal/web"
|
|
)
|
|
|
|
const buildVersion = "production-readiness-v1.2"
|
|
|
|
const agentStatusHTML = `<!doctype html>
|
|
<html lang="de"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>Neural Brain Source Agent</title><style>
|
|
body{margin:0;background:#0b1018;color:#e8edf5;font:14px/1.45 system-ui,sans-serif}main{max-width:980px;margin:40px auto;padding:0 20px}.card{background:#121a26;border:1px solid #26364a;border-radius:12px;padding:22px}h1{margin:0 0 4px;font-size:22px}.muted{color:#90a0b5}.state{display:inline-block;padding:5px 9px;border-radius:999px;background:#253246;margin:12px 0}.ok{color:#6ee7a8}.bad{color:#ff9b9b}dl{display:grid;grid-template-columns:220px 1fr;gap:8px 16px;margin:18px 0}dt{color:#90a0b5}dd{margin:0;word-break:break-word}pre{white-space:pre-wrap;background:#0a0f16;padding:12px;border-radius:8px;border:1px solid #26364a}.warn{background:#352b15;border:1px solid #6c5721;padding:12px;border-radius:8px;color:#ffd98a}a{color:#87bfff}</style></head>
|
|
<body><main><div class="card"><h1>Neural Brain · Source Agent</h1><div class="muted">Leichtgewichtiger Poller-Modus · Status aktualisiert sich automatisch.</div><div id="state" class="state">lade…</div><div id="warning"></div><dl id="details"></dl><pre id="error" style="display:none"></pre><div class="muted">JSON: <a href="/api/status">/api/status</a> · Health: <a href="/healthz">/healthz</a></div></div></main>
|
|
<script>
|
|
const fields=[['agent_id','Agent ID'],['brain_url','Brain URL'],['version','Version'],['configured_tasks','Tasks'],['config_source','Config-Quelle'],['config_issued_at','Config ausgestellt'],['last_config_success_at','Letzter Config-Erfolg'],['last_heartbeat_success_at','Letzter Heartbeat-Erfolg'],['last_task_run_at','Letzter Task-Lauf']];
|
|
function fmt(v){if(v===null||v===undefined||v===''||v==='0001-01-01T00:00:00Z')return '—';return String(v)}
|
|
async function refresh(){try{const r=await fetch('/api/status',{cache:'no-store'}),s=await r.json();const st=document.getElementById('state');st.textContent=s.connection_state==='degraded'?'● Brain erreichbar, Config/Heartbeat fehlerhaft':(s.brain_connected?'● mit Brain verbunden':'● keine aktuelle Brain-Verbindung');st.className='state '+(s.brain_connected&&s.connection_state!=='degraded'?'ok':'bad');document.getElementById('details').innerHTML=fields.map(([k,l])=>'<dt>'+l+'</dt><dd>'+fmt(s[k])+'</dd>').join('');const w=document.getElementById('warning');const msg=s.brain_url_warning||s.connection_hint||'';w.innerHTML=msg?'<div class="warn">'+msg.replace(/[&<>]/g,c=>({'&':'&','<':'<','>':'>'}[c]))+'</div>':'';const e=document.getElementById('error'),err=s.last_connection_error||s.last_config_error||s.last_heartbeat_error||s.last_task_error||'';e.style.display=err?'block':'none';e.textContent=err?'Letzter Fehler: '+err:''}catch(e){document.getElementById('state').textContent='Status nicht lesbar: '+e}}
|
|
refresh();setInterval(refresh,5000);
|
|
</script></body></html>`
|
|
|
|
func main() {
|
|
cfg, err := config.Load()
|
|
if err != nil {
|
|
slog.Error("configuration invalid", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})))
|
|
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer cancel()
|
|
if cfg.Mode == "agent" {
|
|
runAgent(ctx, cfg)
|
|
return
|
|
}
|
|
runBrain(ctx, cancel, cfg)
|
|
}
|
|
|
|
func runAgent(ctx context.Context, cfg config.Config) {
|
|
runner, err := sourceagent.NewRunner(sourceagent.RunnerConfig{
|
|
BrainURL: cfg.AgentBrainURL, AgentID: cfg.AgentID, Token: cfg.AgentToken, DataDir: cfg.DataDir,
|
|
ConfigFile: cfg.AgentConfigFile, ConfigRefresh: cfg.AgentConfigRefresh, HTTPTimeout: cfg.AgentHTTPTimeout,
|
|
Concurrency: cfg.AgentConcurrency, BatchSize: cfg.AgentBatchSize, AllowPrivate: cfg.AgentAllowPrivate, Version: buildVersion,
|
|
ComputeEnabled: cfg.AgentComputeEnabled, ComputePollInterval: cfg.AgentComputePollInterval, ComputeMaxBytes: cfg.AgentComputeMaxBytes,
|
|
DockerControllerEnabled: cfg.AgentDockerControllerEnabled, DockerSocket: cfg.AgentDockerSocket, DockerComposeBinary: cfg.AgentDockerComposeBinary,
|
|
ControllerPollInterval: cfg.AgentControllerPollInterval, ControllerMaxDuration: cfg.AgentControllerMaxDuration,
|
|
})
|
|
if err != nil {
|
|
slog.Error("source agent initialization failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
defer runner.Close()
|
|
runner.Start(ctx)
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
_, _ = w.Write([]byte(agentStatusHTML))
|
|
})
|
|
mux.HandleFunc("GET /api/status", func(w http.ResponseWriter, r *http.Request) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(runner.Status())
|
|
})
|
|
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) })
|
|
srv := &http.Server{Addr: cfg.ListenAddr, Handler: mux, ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 30 * time.Second}
|
|
go func() {
|
|
slog.Info("neural brain source-agent listening", "addr", cfg.ListenAddr, "agent_id", cfg.AgentID, "brain_url", cfg.AgentBrainURL)
|
|
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
slog.Error("agent status server failed", "error", err)
|
|
}
|
|
}()
|
|
<-ctx.Done()
|
|
shutdown, c := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer c()
|
|
_ = srv.Shutdown(shutdown)
|
|
}
|
|
|
|
func runBrain(ctx context.Context, cancel context.CancelFunc, cfg config.Config) {
|
|
g, err := graph.Open(cfg.DataDir)
|
|
if err != nil {
|
|
slog.Error("graph open failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
broker := activity.New(120)
|
|
sourceStore, err := sourceagent.OpenStore(cfg.DataDir)
|
|
if err != nil {
|
|
slog.Error("source agent store open failed", "error", err)
|
|
_ = g.Close()
|
|
os.Exit(1)
|
|
}
|
|
defer sourceStore.Close()
|
|
if requeued, err := sourceStore.EnsureInboxClassifierVersion(ctx, engine.SourceInboxClassifierVersion); err != nil {
|
|
slog.Warn("source inbox classifier migration failed", "error", err)
|
|
} else if requeued > 0 {
|
|
slog.Info("source inbox documents queued for classifier upgrade", "documents", requeued, "classifier_version", engine.SourceInboxClassifierVersion)
|
|
}
|
|
eng := engine.New(cfg, g, broker)
|
|
eng.SetSourceInbox(sourceStore)
|
|
eng.Start(ctx)
|
|
watcher := ingest.NewAgentWatcher(cfg.AgentRunsFiles, g, broker)
|
|
watcher.Start(ctx)
|
|
ui := &webui.Server{Engine: eng, Graph: g, Broker: broker, APIKey: cfg.APIKey, PublicURL: cfg.BrainPublicURL, SourceAgents: sourceStore}
|
|
srv := &http.Server{Addr: cfg.ListenAddr, Handler: ui.Handler(), ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 30 * time.Second, WriteTimeout: 10 * time.Minute, IdleTimeout: 90 * time.Second}
|
|
go func() {
|
|
storage := g.StorageStatus()
|
|
slog.Info("neural brain listening", "mode", "brain", "addr", cfg.ListenAddr, "knowledge_dirs", cfg.KnowledgeDirs, "staging_dirs", cfg.StagingDirs, "graph_backend", storage.Backend, "graph_db", storage.Path)
|
|
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
slog.Error("server failed", "error", err)
|
|
cancel()
|
|
}
|
|
}()
|
|
<-ctx.Done()
|
|
shutdown, c := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer c()
|
|
_ = srv.Shutdown(shutdown)
|
|
_ = eng.Flush(shutdown)
|
|
if err := g.Close(); err != nil {
|
|
slog.Error("graph close failed", "error", err)
|
|
}
|
|
}
|