feat: scaffold agentmon services and k8s deploy

Adds Go microservices (ingest-gateway, event-processor, query-api, web-ui), NATS+Postgres wiring, initial schema/init job, ingress manifests for LAN+tailnet, and a multi-arch image build script.
2026-01-17 01:06:57 -08:00
parent a584d7e274
commit 256b841cbf
28 changed files with 1554 additions and 0 deletions
@@ -0,0 +1,133 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	qnats "agentmon/internal/queue/nats"
+	"agentmon/internal/store/postgres"
+)
+
+type envelope struct {
+	Schema struct {
+		Name    string `json:"name"`
+		Version int    `json:"version"`
+	} `json:"schema"`
+	Event struct {
+		ID     string `json:"id"`
+		Type   string `json:"type"`
+		TS     any    `json:"ts"`
+		Source struct {
+			Framework string `json:"framework"`
+			ClientID  string `json:"client_id"`
+		} `json:"source"`
+	} `json:"event"`
+	Correlation *struct {
+		SessionID    string `json:"session_id,omitempty"`
+		RunID        string `json:"run_id,omitempty"`
+		TraceID      string `json:"trace_id,omitempty"`
+		SpanID       string `json:"span_id,omitempty"`
+		ParentSpanID string `json:"parent_span_id,omitempty"`
+	} `json:"correlation,omitempty"`
+}
+
+func main() {
+	log.Printf("event-processor starting")
+
+	dsn := os.Getenv("DATABASE_URL")
+	if dsn == "" {
+		log.Fatalf("DATABASE_URL is required")
+	}
+
+	natsURL := envDefault("NATS_URL", "nats://nats:4222")
+	natsTopic := envDefault("NATS_TOPIC", "agentmon.events.v1")
+
+	db, err := postgres.Open(dsn)
+	if err != nil {
+		log.Fatalf("failed to open DB: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+
+	if err := db.Ping(ctx); err != nil {
+		log.Fatalf("failed to ping DB: %v", err)
+	}
+
+	sub, err := qnats.NewSubscriber(natsURL, natsTopic)
+	if err != nil {
+		log.Fatalf("failed to connect to NATS: %v", err)
+	}
+	defer sub.Close()
+
+	log.Printf("subscribed to %s (%s)", natsTopic, natsURL)
+
+	err = sub.Subscribe(ctx, func(msg []byte) error {
+		var env envelope
+		if err := json.Unmarshal(msg, &env); err != nil {
+			return err
+		}
+
+		ts, err := parseTS(env.Event.TS)
+		if err != nil {
+			return err
+		}
+
+		e := postgres.InsertEvent{
+			EventID:         env.Event.ID,
+			TS:              ts,
+			Type:            env.Event.Type,
+			Payload:         json.RawMessage(msg),
+			SourceFramework: sqlNull(env.Event.Source.Framework),
+			ClientID:        sqlNull(env.Event.Source.ClientID),
+		}
+
+		if env.Correlation != nil {
+			e.SessionID = sqlNull(env.Correlation.SessionID)
+			e.RunID = sqlNull(env.Correlation.RunID)
+			e.TraceID = sqlNull(env.Correlation.TraceID)
+			e.SpanID = sqlNull(env.Correlation.SpanID)
+			e.ParentSpanID = sqlNull(env.Correlation.ParentSpanID)
+		}
+
+		return db.InsertEvent(ctx, e)
+	})
+
+	if err != nil && err != context.Canceled {
+		log.Printf("processor stopped: %v", err)
+	}
+}
+
+func envDefault(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+func sqlNull(s string) sql.NullString {
+	if s == "" {
+		return sql.NullString{}
+	}
+	return sql.NullString{String: s, Valid: true}
+}
+
+func parseTS(v any) (time.Time, error) {
+	// Accept RFC3339 string or unix-ms number.
+	switch t := v.(type) {
+	case string:
+		return time.Parse(time.RFC3339Nano, t)
+	case float64:
+		ms := int64(t)
+		return time.Unix(0, ms*int64(time.Millisecond)), nil
+	default:
+		return time.Time{}, postgres.ErrMissingField
+	}
+}
@@ -0,0 +1,109 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"net/http"
+	"os"
+
+	"agentmon/internal/httpx"
+	qnats "agentmon/internal/queue/nats"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/go-chi/chi/v5/middleware"
+	"github.com/gorilla/websocket"
+)
+
+func main() {
+	addr := envDefault("AGENTMON_ADDR", ":8080")
+	natsURL := envDefault("NATS_URL", "nats://nats:4222")
+	natsTopic := envDefault("NATS_TOPIC", "agentmon.events.v1")
+
+	pub, err := qnats.NewPublisher(natsURL, natsTopic)
+	if err != nil {
+		log.Fatalf("failed to connect to NATS: %v", err)
+	}
+	defer pub.Close()
+
+	r := chi.NewRouter()
+	r.Use(middleware.RequestID)
+	r.Use(middleware.RealIP)
+	r.Use(middleware.Logger)
+	r.Use(middleware.Recoverer)
+
+	r.Get("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	})
+
+	r.Post("/v1/events", func(w http.ResponseWriter, r *http.Request) {
+		var events []json.RawMessage
+		if err := json.NewDecoder(r.Body).Decode(&events); err != nil {
+			httpx.WriteJSON(w, http.StatusBadRequest, map[string]any{"error": "invalid_json"})
+			return
+		}
+
+		accepted := 0
+		rejected := 0
+		for _, raw := range events {
+			if len(raw) == 0 {
+				rejected++
+				continue
+			}
+			if err := pub.Publish(r.Context(), raw); err != nil {
+				rejected++
+				continue
+			}
+			accepted++
+		}
+
+		httpx.WriteJSON(w, http.StatusAccepted, map[string]any{"accepted": accepted, "rejected": rejected})
+	})
+
+	r.Get("/v1/ws", wsHandler(pub))
+
+	log.Printf("ingest-gateway listening on %s", addr)
+	log.Fatal(http.ListenAndServe(addr, r))
+}
+
+func wsHandler(pub *qnats.Publisher) http.HandlerFunc {
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+
+		for {
+			_, msg, err := conn.ReadMessage()
+			if err != nil {
+				return
+			}
+
+			var v map[string]any
+			if err := json.Unmarshal(msg, &v); err != nil {
+				_ = conn.WriteJSON(map[string]any{"error": "invalid_json"})
+				continue
+			}
+
+			if err := pub.Publish(r.Context(), msg); err != nil {
+				_ = conn.WriteJSON(map[string]any{"error": "publish_failed"})
+				continue
+			}
+
+			_ = v
+			_ = conn.WriteJSON(map[string]any{"ack": map[string]any{"up_to_seq": nil}})
+		}
+	}
+}
+
+func envDefault(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
@@ -0,0 +1,59 @@
+package main
+
+import (
+	"log"
+	"net/http"
+	"os"
+	"strconv"
+
+	"agentmon/internal/httpx"
+	"agentmon/internal/store/postgres"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/go-chi/chi/v5/middleware"
+)
+
+func main() {
+	addr := envDefault("AGENTMON_QUERY_ADDR", ":8081")
+	dsn := os.Getenv("DATABASE_URL")
+	if dsn == "" {
+		log.Fatalf("DATABASE_URL is required")
+	}
+
+	db, err := postgres.Open(dsn)
+	if err != nil {
+		log.Fatalf("failed to open DB: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+
+	r := chi.NewRouter()
+	r.Use(middleware.RequestID)
+	r.Use(middleware.RealIP)
+	r.Use(middleware.Logger)
+	r.Use(middleware.Recoverer)
+
+	r.Get("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	})
+
+	r.Get("/v1/events", func(w http.ResponseWriter, r *http.Request) {
+		limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
+		events, err := db.ListRecentEvents(r.Context(), limit)
+		if err != nil {
+			httpx.WriteJSON(w, http.StatusInternalServerError, map[string]any{"error": "db_error"})
+			return
+		}
+		httpx.WriteJSON(w, http.StatusOK, map[string]any{"events": events})
+	})
+
+	log.Printf("query-api listening on %s", addr)
+	log.Fatal(http.ListenAndServe(addr, r))
+}
+
+func envDefault(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
@@ -0,0 +1,56 @@
+package main
+
+import (
+	"encoding/json"
+	"io"
+	"log"
+	"net/http"
+	"os"
+)
+
+func main() {
+	addr := envDefault("AGENTMON_UI_ADDR", ":8082")
+
+	queryAPIBase := envDefault("AGENTMON_QUERY_BASE", "http://query-api")
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	})
+
+	mux.HandleFunc("/api/events", func(w http.ResponseWriter, r *http.Request) {
+		resp, err := http.Get(queryAPIBase + "/v1/events?limit=100")
+		if err != nil {
+			w.WriteHeader(http.StatusBadGateway)
+			_, _ = w.Write([]byte("query-api unreachable"))
+			return
+		}
+		defer resp.Body.Close()
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(resp.StatusCode)
+		_, _ = io.Copy(w, resp.Body)
+	})
+
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/html; charset=utf-8")
+
+		payload, _ := json.Marshal(map[string]any{"query_api": queryAPIBase})
+		_, _ = w.Write([]byte("<html><body><h1>agentmon</h1><p>Recent events:</p><pre id='out'>loading...</pre>"))
+		_, _ = w.Write([]byte("<script>\n"))
+		_, _ = w.Write([]byte("const cfg=" + string(payload) + ";\n"))
+		_, _ = w.Write([]byte("fetch('/api/events').then(r=>r.json()).then(j=>{document.getElementById('out').textContent=JSON.stringify(j,null,2);}).catch(e=>{document.getElementById('out').textContent=String(e);});\n"))
+		_, _ = w.Write([]byte("</script></body></html>"))
+	})
+
+	log.Printf("web-ui listening on %s", addr)
+	log.Fatal(http.ListenAndServe(addr, mux))
+}
+
+func envDefault(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}