munin/last_test.go

//go:build last

// Lasttest: viele Agenten greifen gleichzeitig über echtes HTTP und MCP zu.
//
// Er läuft NICHT bei jedem "go test ./...", weil er Zeit braucht und die
// Maschine auslastet. Das Build-Tag oben schaltet ihn erst ein, wenn man
// ausdrücklich danach fragt:
//
//	go test -tags last -run TestLast -v .
//
// Mit eigenen Werten (alle Schalter siehe die flag.-Zeilen unten):
//
//	go test -tags last -run TestLast -v . -last.agenten=64 -last.dauer=30s
//
// Drei Aufstellungen mit derselben Zahl Agenten, damit die Zahlen vergleichbar
// sind:
//
//   - ein Projekt: alle Agenten im selben Projekt
//   - ein Workspace, mehrere Projekte: teilen sich eine Datei und damit die
//     eine Schreibverbindung (siehe openPool in internal/store/store.go)
//   - mehrere Workspaces: je eine eigene Datei, schreiben echt parallel
//
// Am Ende prüft jede Aufstellung, ob nichts verloren ging und ob der
// Suchindex zum Stand der Datenbank passt.
package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"log"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/modelcontextprotocol/go-sdk/mcp"

	"munin/internal/auth"
	"munin/internal/search"
	"munin/internal/store"
	"munin/internal/workspace"
)

var (
	lastAgenten   = flag.Int("last.agenten", 32, "Zahl der gleichzeitigen Agenten je Aufstellung")
	lastDauer     = flag.Duration("last.dauer", 10*time.Second, "Laufzeit je Aufstellung")
	lastProjekte  = flag.Int("last.projekte", 4, "Projekte bzw. Workspaces in den Aufstellungen mit mehreren")
	lastEintraege = flag.Int("last.eintraege", 500, "Einträge, die jedes Projekt vorab bekommt")
	lastSlugs     = flag.Int("last.slugs", 20, "Bezeichner, auf die die Agenten je Projekt schreiben — klein heißt viel Gedränge auf denselben Einträgen")
	lastSchreiben = flag.Int("last.schreibanteil", 30, "Anteil schreibender Aufrufe in Prozent, der Rest ist recall und get")
)

func TestLast(t *testing.T) {
	// Jede Anfrage würde sonst eine Logzeile schreiben — bei zehntausenden
	// Aufrufen geht die Auswertung darin unter.
	log.SetOutput(io.Discard)
	slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))

	n := *lastProjekte
	aufstellungen := []struct {
		name                 string
		workspaces, projekte int // projekte = je Workspace
	}{
		{"ein Projekt", 1, 1},
		{fmt.Sprintf("ein Workspace, %d Projekte", n), 1, n},
		{fmt.Sprintf("%d Workspaces, je 1 Projekt", n), n, 1},
	}

	for _, a := range aufstellungen {
		t.Run(a.name, func(t *testing.T) {
			lauf(t, a.workspaces, a.projekte)
		})
	}
}

// ziel ist ein Projekt samt den Keys, mit denen Agenten es ansprechen.
type ziel struct {
	ws      *workspace.Workspace
	projekt store.Project
	keys    []string
}

// lauf baut eine Aufstellung auf, lässt die Agenten laufen und wertet aus.
func lauf(t *testing.T, anzahlWorkspaces, projekteJeWorkspace int) {
	ctx := context.Background()
	dir := t.TempDir()

	control, err := store.OpenControl(filepath.Join(dir, "control.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { control.Close() })
	workspaces := workspace.NewManager(dir)
	t.Cleanup(func() { workspaces.Close() })

	// Dieselbe Verdrahtung wie im Betrieb. Schreibbudget 0 = unbegrenzt — es
	// soll eine Schleife stoppen und würde hier die Messung abwürgen.
	handler, err := baueHandler(control, workspaces, nil, config{
		sessionSecret:    strings.Repeat("x", 32),
		adminToken:       strings.Repeat("y", 32),
		maxAPIBody:       defaultMaxAPIBodyBytes,
		writeLimitHourly: 0,
	})
	if err != nil {
		t.Fatal(err)
	}
	server := httptest.NewServer(handler)
	t.Cleanup(server.Close)

	// ── Daten anlegen ────────────────────────────────────────────────────────
	agenten := *lastAgenten
	anzahlZiele := anzahlWorkspaces * projekteJeWorkspace
	var ziele []*ziel

	for w := 0; w < anzahlWorkspaces; w++ {
		wsEintrag, err := control.CreateWorkspace(ctx, fmt.Sprintf("Workspace %d", w))
		if err != nil {
			t.Fatal(err)
		}
		ws, err := workspaces.Get(wsEintrag.ID)
		if err != nil {
			t.Fatal(err)
		}
		for p := 0; p < projekteJeWorkspace; p++ {
			projekt, err := ws.Store.CreateProject(ctx, wsEintrag.ID, fmt.Sprintf("projekt-%d", p), "Projekt", "")
			if err != nil {
				t.Fatal(err)
			}
			z := &ziel{ws: ws, projekt: projekt}
			ziele = append(ziele, z)

			for i := 0; i < *lastEintraege; i++ {
				if _, err := ws.Store.PutEntry(ctx, store.EntryWrite{
					ProjectID: projekt.ID, Slug: fmt.Sprintf("vorab-%04d", i), Title: fmt.Sprintf("Vorab %d", i),
					Content: fuellText(i), Source: store.SourceHuman,
				}); err != nil {
					t.Fatal(err)
				}
			}
		}
	}

	// Jeder Agent bekommt einen eigenen Key, wie ein echter Nutzer.
	for a := 0; a < agenten; a++ {
		z := ziele[a%anzahlZiele]
		klartext, err := auth.GenerateKey()
		if err != nil {
			t.Fatal(err)
		}
		hash := auth.HashKey(klartext)
		if _, err := z.ws.Store.CreateAPIKey(ctx, z.projekt.ID, fmt.Sprintf("agent-%d", a), hash, string(auth.RoleWrite)); err != nil {
			t.Fatal(err)
		}
		if err := control.IndexKey(ctx, hash, z.ws.ID); err != nil {
			t.Fatal(err)
		}
		z.keys = append(z.keys, klartext)
	}

	// ── Agenten laufen lassen ────────────────────────────────────────────────
	// Ein gemeinsamer Transport, der je Agent eine Verbindung offen hält. Mit
	// den Vorgaben von Go (2 offene Verbindungen je Ziel) öffnen die Agenten
	// sonst laufend neue, bis dem Testrechner die lokalen Ports ausgehen
	// ("can't assign requested address"). Echte Agenten sitzen auf
	// verschiedenen Rechnern und kennen das Problem nicht — es wäre also ein
	// Fehler der Messung, nicht des Servers.
	httpTransport := http.DefaultTransport.(*http.Transport).Clone()
	httpTransport.MaxIdleConnsPerHost = agenten
	t.Cleanup(httpTransport.CloseIdleConnections)

	messung := newMessung()
	ende := time.Now().Add(*lastDauer)
	var wg sync.WaitGroup

	for a := 0; a < agenten; a++ {
		z := ziele[a%anzahlZiele]
		key := z.keys[a/anzahlZiele]
		wg.Add(1)
		go func(a int) {
			defer wg.Done()
			agent(ctx, httpTransport, server.URL, key, a, ende, messung)
		}(a)
	}
	wg.Wait()

	// ── Auswerten ────────────────────────────────────────────────────────────
	messung.bericht(t, *lastDauer)

	if anzahl, erste := messung.fehler(); anzahl > 0 {
		t.Errorf("%d fehlgeschlagene Aufrufe, erste: %v", anzahl, erste)
	}

	pruefeBestand(t, ziele, messung.erfolgreicheSchreibvorgaenge())
}

// agent verhält sich wie ein Agent in einer Schleife: suchen, lesen,
// gelegentlich schreiben — bis die Zeit um ist.
func agent(ctx context.Context, httpTransport http.RoundTripper, url, key string, nr int, ende time.Time, m *messung) {
	httpClient := &http.Client{
		Transport: &mitKey{key: key, next: httpTransport},
		Timeout:   30 * time.Second,
	}
	transport := &mcp.StreamableClientTransport{
		Endpoint:   url + "/mcp",
		HTTPClient: httpClient,
		// Der zusätzliche Rückkanal für Nachrichten vom Server wird hier nicht
		// gebraucht und würde nur eine weitere offene Verbindung je Agent kosten.
		DisableStandaloneSSE: true,
	}
	client := mcp.NewClient(&mcp.Implementation{Name: fmt.Sprintf("agent-%d", nr), Version: "1"}, nil)

	start := time.Now()
	sitzung, err := client.Connect(ctx, transport, nil)
	m.erfasse("verbinden", time.Since(start), err)
	if err != nil {
		return
	}
	defer sitzung.Close()

	zufall := rand.New(rand.NewPCG(uint64(nr), 42))
	for i := 0; time.Now().Before(ende); i++ {
		var werkzeug string
		var args map[string]any

		switch w := zufall.IntN(100); {
		case w < *lastSchreiben:
			werkzeug = "remember"
			// Ein eindeutiges Wort je Schreibvorgang — daran erkennt
			// pruefeBestand, ob der Suchindex den letzten Stand kennt.
			args = map[string]any{
				"slug":    fmt.Sprintf("gemeinsam-%02d", zufall.IntN(*lastSlugs)),
				"title":   "Gemeinsamer Eintrag",
				"content": "Stand " + eindeutigesWort(nr, i) + "\n\n" + fuellText(i),
			}
		case w < *lastSchreiben+(100-*lastSchreiben)*2/3:
			werkzeug = "recall"
			args = map[string]any{"query": suchbegriffe[zufall.IntN(len(suchbegriffe))]}
		default:
			werkzeug = "get"
			args = map[string]any{"slug": fmt.Sprintf("vorab-%04d", zufall.IntN(max(1, *lastEintraege)))}
		}

		start := time.Now()
		erg, err := sitzung.CallTool(ctx, &mcp.CallToolParams{Name: werkzeug, Arguments: args})
		if err == nil && erg.IsError {
			err = fmt.Errorf("%s: %s", werkzeug, ergebnisText(erg))
		}
		m.erfasse(werkzeug, time.Since(start), err)
	}
}

// pruefeBestand vergleicht am Ende Datenbank und Suchindex.
//
// Zwei Dinge müssen stimmen:
//   - Kein Schreibvorgang ging verloren: Jeder erfolgreiche remember-Aufruf
//     hat eine Fassung erzeugt. Die Summe der Fassungen aller umkämpften
//     Einträge muss also genau die Zahl der erfolgreichen Aufrufe sein.
//   - Der Suchindex kennt den letzten Stand: Das eindeutige Wort der letzten
//     Fassung muss den Eintrag finden.
func pruefeBestand(t *testing.T, ziele []*ziel, erfolgreich int) {
	t.Helper()
	ctx := context.Background()

	summeFassungen, veraltet := 0, 0
	for _, z := range ziele {
		idx, err := z.ws.Search.For(ctx, z.projekt.ID)
		if err != nil {
			t.Fatal(err)
		}
		for s := 0; s < *lastSlugs; s++ {
			slug := fmt.Sprintf("gemeinsam-%02d", s)
			e, err := z.ws.Store.EntryBySlug(ctx, z.projekt.ID, slug)
			if err != nil {
				continue // nie beschrieben
			}
			summeFassungen += e.Version

			wort := strings.Fields(e.Content)[1]
			gefunden := slices.ContainsFunc(idx.Search(wort, 5), func(r search.Result) bool { return r.DocID == slug })
			if !gefunden {
				veraltet++
			}
		}
	}

	if summeFassungen != erfolgreich {
		t.Errorf("%d erfolgreiche remember-Aufrufe, aber %d Fassungen gespeichert", erfolgreich, summeFassungen)
	}
	if veraltet > 0 {
		t.Errorf("%d Einträge: Der Suchindex kennt die letzte Fassung nicht", veraltet)
	}
	t.Logf("Bestand geprüft: %d Fassungen, Suchindex %s", summeFassungen,
		map[bool]string{true: "aktuell", false: "VERALTET"}[veraltet == 0])
}

// ── Messung ──────────────────────────────────────────────────────────────────

// messung sammelt Dauer und Fehler aller Aufrufe, getrennt nach Werkzeug.
type messung struct {
	mu     sync.Mutex
	dauern map[string][]time.Duration
	fehl   map[string]int
	erste  []error
}

func newMessung() *messung {
	return &messung{dauern: map[string][]time.Duration{}, fehl: map[string]int{}}
}

func (m *messung) erfasse(werkzeug string, d time.Duration, err error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if err != nil {
		m.fehl[werkzeug]++
		if len(m.erste) < 10 {
			m.erste = append(m.erste, err)
		}
		return
	}
	m.dauern[werkzeug] = append(m.dauern[werkzeug], d)
}

// fehler gibt die Zahl aller fehlgeschlagenen Aufrufe und die ersten davon
// zurück.
func (m *messung) fehler() (anzahl int, erste []error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, n := range m.fehl {
		anzahl += n
	}
	return anzahl, m.erste
}

func (m *messung) erfolgreicheSchreibvorgaenge() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.dauern["remember"])
}

// bericht schreibt eine Tabelle: Aufrufe je Sekunde und wie lange die
// typische (p50), die langsamsten 5 % (p95) und 1 % (p99) Aufrufe dauerten.
func (m *messung) bericht(t *testing.T, dauer time.Duration) {
	m.mu.Lock()
	defer m.mu.Unlock()

	var b strings.Builder
	fmt.Fprintf(&b, "\n%-10s %8s %8s %9s %9s %9s %9s %7s\n", "Werkzeug", "Aufrufe", "je s", "p50", "p95", "p99", "max", "Fehler")
	for _, w := range []string{"verbinden", "recall", "get", "remember"} {
		d := m.dauern[w]
		slices.Sort(d)
		je := float64(len(d)) / dauer.Seconds()
		if w == "verbinden" {
			je = 0 // einmal je Agent, keine Rate
		}
		fmt.Fprintf(&b, "%-10s %8d %8.0f %9s %9s %9s %9s %7d\n", w, len(d), je,
			perzentil(d, 50), perzentil(d, 95), perzentil(d, 99), perzentil(d, 100), m.fehl[w])
	}
	t.Log(b.String())
}

func perzentil(sortiert []time.Duration, p int) string {
	if len(sortiert) == 0 {
		return "-"
	}
	i := (len(sortiert) - 1) * p / 100
	return sortiert[i].Round(10 * time.Microsecond).String()
}

// ── Hilfen ───────────────────────────────────────────────────────────────────

// mitKey hängt den Bearer-Key an jede Anfrage — so wie es ein MCP-Client mit
// "headers" in seiner Konfiguration tut.
type mitKey struct {
	key  string
	next http.RoundTripper
}

func (m *mitKey) RoundTrip(r *http.Request) (*http.Response, error) {
	r = r.Clone(r.Context())
	r.Header.Set("Authorization", "Bearer "+m.key)
	return m.next.RoundTrip(r)
}

func ergebnisText(r *mcp.CallToolResult) string {
	var b strings.Builder
	for _, c := range r.Content {
		if t, ok := c.(*mcp.TextContent); ok {
			b.WriteString(t.Text)
		}
	}
	return b.String()
}

// eindeutigesWort bildet aus Agent und Durchlauf ein Wort nur aus Buchstaben,
// damit der Tokenizer es nicht an Ziffern zerlegt.
func eindeutigesWort(agent, durchlauf int) string {
	s := []byte(fmt.Sprintf("%04d%07d", agent, durchlauf))
	for i := range s {
		s[i] = 'a' + (s[i] - '0')
	}
	return "marke" + string(s)
}

var suchbegriffe = []string{"deployment", "server", "datenbank", "zugang", "backup", "monitoring", "release", "fehler"}

// fuellText erzeugt einen Absatz mit wechselnden Fachbegriffen, damit der
// Suchindex realistisch viele Begriffe und Treffer hat.
func fuellText(i int) string {
	var b strings.Builder
	for j := 0; j < 40; j++ {
		b.WriteString(suchbegriffe[(i+j*3)%len(suchbegriffe)])
		b.WriteString(" wird hier beschrieben, mit Details zu Schritt ")
		b.WriteString(fmt.Sprint(j))
		b.WriteString(". ")
	}
	return b.String()
}