munin/internal/stats/stats.go

// Package stats zählt die Kennzahlen aller Workspaces für die Systemseite
// zusammen (AUSBAUSTUFEN.md 5.16): Bestand, Platz, Stand des Schemas und die
// Aktivität der letzten 30 Tage.
//
// # Warum nicht bei jedem Seitenaufruf?
//
// Für die Zahlen fragt Munin jeden Workspace einzeln ab — bei 40 Workspaces
// rund 300 Abfragen. Das ist kein Problem, wenn es selten geschieht, aber
// unnötig bei jedem Neuladen der Seite. Der Sammler rechnet deshalb höchstens
// alle MinInterval neu und gibt dazwischen das gemerkte Ergebnis zurück.
// Die Seite nennt den Stand ("Stand 14:05").
//
// Gerechnet wird beim Aufruf der Seite, nicht in einer eigenen Goroutine: Wer
// die Seite nie öffnet, kostet nichts, und es gibt keinen Hintergrundlauf,
// der beim Beenden aufgeräumt werden müsste.
package stats

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"

	"munin/internal/store"
	"munin/internal/workspace"
)

// Days ist der Zeitraum der Aktivität. Derselbe wie in der Nutzung eines
// Projekts (store.UsageWindow) — und nicht länger, weil das Suchprotokoll
// nach 30 Tagen aufgeräumt wird. Ältere Tage stünden sonst ohne Suchen da.
const Days = store.UsageWindowDays

// MinInterval ist die Zeit, die ein Ergebnis gilt, bevor neu gerechnet
// wird.
const MinInterval = 5 * time.Minute

// Snapshot ist das Ergebnis einer Erhebung.
type Snapshot struct {
	CollectedAt time.Time
	Duration    time.Duration // wie lange die Erhebung brauchte

	Workspaces []Workspace // nach Name sortiert
	Total      store.Inventory
	DiskUsage  store.DiskUsage // aller Workspaces zusammen

	// SchemaVersion: die neueste Migration, die dieses Programm kennt. Ein
	// Workspace, der davon abweicht, steht mit seinem Stand in Workspace.Schema.
	SchemaVersion string

	Days     []Tag     // genau Tage Einträge, der älteste zuerst, heute zuletzt
	Projects []Project // mit Aktivität, die aktivsten zuerst
	Idle     []Project // ohne jede Aktivität im Zeitraum, nach Workspace und Name

	// WaitCount zählt, wie oft Anfragen seit dem Start auf eine freie
	// Datenbankverbindung warten mussten (alle Pools zusammen), zum Zeitpunkt
	// der Erhebung. PrevWaitCount und PrevCollectedAt sind dieselbe Zahl bei
	// der vorigen Erhebung (0 und Nullzeitpunkt bei der ersten). Die
	// Differenz zeigt, ob gerade gewartet wird — die Summe seit dem Start
	// sagt das nach Tagen Laufzeit nicht mehr.
	WaitCount       int64
	PrevWaitCount   int64
	PrevCollectedAt time.Time

	// Errors sind Workspaces, die sich nicht lesen ließen. Ihre Zahlen fehlen
	// in den Summen; die übrigen sind trotzdem da.
	Errors []string
}

// Workspace sind die Kennzahlen eines Workspace.
type Workspace struct {
	ID, Name  string
	Location  string // Datei oder Schema (store.Backend.Ort)
	Inventory store.Inventory
	DiskUsage store.DiskUsage
	Schema    string // zuletzt angewandte Migration
}

// SchemaMismatch sagt, ob der Workspace auf einem anderen Stand ist als das
// Programm.
func (w Workspace) SchemaMismatch(expected string) bool { return w.Schema != expected }

// Tag ist die Aktivität eines Tages über alle Workspaces.
type Tag struct {
	Date        time.Time // 0 Uhr UTC
	AgentWrites int
	HumanWrites int
	Searches    int
}

// Writes sind alle Schreibvorgänge des Tages.
func (t Tag) Writes() int { return t.AgentWrites + t.HumanWrites }

// Project ist die Aktivität eines Projekts im Zeitraum.
type Project struct {
	Workspace string // Name
	ID        string
	Slug      string
	Name      string

	AgentWrites int
	HumanWrites int
	Searches    int
}

// Writes sind alle Schreibvorgänge des Projekts im Zeitraum.
func (p Project) Writes() int { return p.AgentWrites + p.HumanWrites }

// Total ist alles zusammen — danach wird sortiert.
func (p Project) Total() int { return p.Writes() + p.Searches }

// Collector erhebt den Stand und merkt ihn sich für MinInterval.
type Collector struct {
	control    *store.ControlStore
	workspaces *workspace.Manager

	// mu schützt stand und sorgt zugleich dafür, dass bei zwei
	// gleichzeitigen Aufrufen nur einer rechnet; der zweite wartet und
	// bekommt dasselbe Ergebnis.
	mu       sync.Mutex
	snapshot Snapshot
}

// NewCollector legt einen Sammler an.
func NewCollector(control *store.ControlStore, workspaces *workspace.Manager) *Collector {
	return &Collector{control: control, workspaces: workspaces}
}

// Current liefert einen Stand, der höchstens MinInterval alt ist, und
// erhebt dafür nötigenfalls neu.
func (s *Collector) Current(ctx context.Context, now time.Time) Snapshot {
	s.mu.Lock()
	defer s.mu.Unlock()

	if !s.snapshot.CollectedAt.IsZero() && now.Sub(s.snapshot.CollectedAt) < MinInterval {
		return s.snapshot
	}
	newer := s.collect(ctx, now)
	newer.PrevWaitCount, newer.PrevCollectedAt = s.snapshot.WaitCount, s.snapshot.CollectedAt
	s.snapshot = newer
	return s.snapshot
}

// WaitCount zählt die Wartevorgänge aller gerade offenen Pools zusammen: der
// Kontroll-Datenbank und jedes geöffneten Workspace.
func (s *Collector) WaitCount() int64 {
	total := s.control.PoolStats().WaitCount()
	for _, w := range s.workspaces.OpenWorkspaces() {
		total += w.Store.PoolStats().WaitCount()
	}
	return total
}

// ForWorkspace erhebt den Stand eines einzelnen Workspace — für den
// Überblick des Workspace-Admins (AUSBAUSTUFEN.md 5.17). Ohne Zwischenspeicher:
// Ein Workspace sind nur eine Handvoll Abfragen, und der Admin soll nach
// einer Änderung (Key angelegt) sofort den neuen Stand sehen.
func (s *Collector) ForWorkspace(ctx context.Context, w store.Workspace, now time.Time) Snapshot {
	return s.collectAll(ctx, now, []store.Workspace{w})
}

// collect fragt jeden Workspace ab und zählt zusammen.
//
// Dafür werden alle Workspaces geöffnet, auch lange nicht benutzte. Das tun
// das tägliche Aufräumen und der Export (main.go) ohnehin schon beim Start;
// zusätzliche Verbindungen kostet die Erhebung also nicht.
func (s *Collector) collect(ctx context.Context, now time.Time) Snapshot {
	all, err := s.control.ListWorkspaces(ctx)
	if err != nil {
		return Snapshot{CollectedAt: now, Errors: []string{fmt.Sprintf("Workspaces konnten nicht gelesen werden: %v", err)}}
	}
	return s.collectAll(ctx, now, all)
}

// collectAll zählt die gegebenen Workspaces zusammen.
func (s *Collector) collectAll(ctx context.Context, now time.Time, all []store.Workspace) Snapshot {
	begin := time.Now()
	snapshot := Snapshot{CollectedAt: now}

	// Die Tage stehen von vornherein fest, auch die ohne jede Aktivität —
	// sonst hätte das Diagramm Lücken, die man nicht sieht.
	today := time.Date(now.UTC().Year(), now.UTC().Month(), now.UTC().Day(), 0, 0, 0, 0, time.UTC)
	first := today.AddDate(0, 0, -(Days - 1))
	tagNum := map[string]int{}
	for i := 0; i < Days; i++ {
		d := first.AddDate(0, 0, i)
		snapshot.Days = append(snapshot.Days, Tag{Date: d})
		tagNum[d.Format("2006-01-02")] = i
	}

	for _, w := range all {
		ws, projects, err := s.collectWorkspace(ctx, w, now, first, &snapshot, tagNum)
		if err != nil {
			snapshot.Errors = append(snapshot.Errors, fmt.Sprintf("%s: %v", w.Name, err))
			continue
		}
		snapshot.Workspaces = append(snapshot.Workspaces, ws)
		snapshot.Total = snapshot.Total.Plus(ws.Inventory)
		snapshot.DiskUsage.Data += ws.DiskUsage.Data
		snapshot.DiskUsage.WAL += ws.DiskUsage.WAL

		for _, p := range projects {
			if p.Total() > 0 {
				snapshot.Projects = append(snapshot.Projects, p)
			} else {
				snapshot.Idle = append(snapshot.Idle, p)
			}
		}
	}

	sort.Slice(snapshot.Workspaces, func(i, j int) bool { return snapshot.Workspaces[i].Name < snapshot.Workspaces[j].Name })
	// Die aktivsten zuerst; bei Gleichstand alphabetisch, damit die
	// Reihenfolge beim Neuladen nicht springt.
	sort.Slice(snapshot.Projects, func(i, j int) bool {
		a, b := snapshot.Projects[i], snapshot.Projects[j]
		if a.Total() != b.Total() {
			return a.Total() > b.Total()
		}
		return a.Workspace+a.Name < b.Workspace+b.Name
	})
	sort.Slice(snapshot.Idle, func(i, j int) bool {
		a, b := snapshot.Idle[i], snapshot.Idle[j]
		return a.Workspace+a.Name < b.Workspace+b.Name
	})

	snapshot.WaitCount = s.WaitCount()
	snapshot.Duration = time.Since(begin)
	return snapshot
}

// collectWorkspace sammelt die Zahlen eines Workspace. Die Aktivität je Tag
// zählt sie direkt in snapshot.Days ein; die je Projekt gibt sie zurück.
func (s *Collector) collectWorkspace(ctx context.Context, w store.Workspace, now, since time.Time,
	snapshot *Snapshot, tagNum map[string]int) (Workspace, []Project, error) {

	result := Workspace{ID: w.ID, Name: w.Name, Location: s.workspaces.Location(w.ID)}

	opened, err := s.workspaces.Get(w.ID)
	if err != nil {
		return result, nil, err
	}
	st := opened.Store

	if result.Inventory, err = st.Inventory(ctx, now); err != nil {
		return result, nil, err
	}
	if result.DiskUsage, err = st.DiskUsage(ctx); err != nil {
		return result, nil, err
	}
	actual, expected, err := st.SchemaVersion(ctx)
	if err != nil {
		return result, nil, err
	}
	result.Schema = actual
	snapshot.SchemaVersion = expected // für alle Workspaces derselbe: er kommt aus dem Programm

	// Jedes Projekt bekommt einen Eintrag, auch ohne Aktivität — gerade die
	// ruhigen sind interessant.
	list, err := st.ListProjects(ctx, w.ID)
	if err != nil {
		return result, nil, err
	}
	projects := make([]Project, len(list))
	toID := map[string]*Project{}
	for i, p := range list {
		projects[i] = Project{Workspace: w.Name, ID: p.ID, Slug: p.Slug, Name: p.Name}
		toID[p.ID] = &projects[i]
	}

	activity, err := st.Activity(ctx, since)
	if err != nil {
		return result, nil, err
	}
	for _, a := range activity {
		if i, ok := tagNum[a.Tag]; ok {
			snapshot.Days[i].AgentWrites += a.AgentWrites
			snapshot.Days[i].HumanWrites += a.HumanWrites
			snapshot.Days[i].Searches += a.Searches
		}
		// Zeilen eines inzwischen gelöschten Projekts zählen für den Tag,
		// haben aber kein Projekt mehr, dem sie gehören.
		if p := toID[a.ProjectID]; p != nil {
			p.AgentWrites += a.AgentWrites
			p.HumanWrites += a.HumanWrites
			p.Searches += a.Searches
		}
	}

	return result, projects, nil
}