mesh-controller/cmd/mesh-controller/durations.go

package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"slices"
	"sort"
	"strings"
	"time"

	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
)

// What the core's bounds are set from (novox/hq to-be 45 Phase 0).
//
// **A bound is set from what was measured, not from what seemed reasonable.** Phase 1 puts a watchdog
// on each row of the signals table, and each has a bound: S1 three heartbeat intervals, S2 three times
// a machine's last apply, S3 a tier's build and apply time, S6 a build's timeout. Marked provisional
// in the design until a fortnight of these says what the mesh actually takes. Recorded by the serving
// controller as it hears each — a send's first report, a machine's next word, a plan leaving a tier, a
// build's outcome — and summarised here per machine, repository or module.

// recordBuildDuration measures one build from its ask to its outcome heard.
func recordBuildDuration(ctx context.Context, inv *inventory.Inventory, result link.BuildResult, asked time.Time) {
	if asked.IsZero() || result.ID == "" {
		return
	}
	subject := result.Module
	if subject == "" {
		subject = result.Repository
	}
	detail := "built"
	if result.Failed != "" {
		detail = "failed: " + firstLine(result.Failed)
	}
	if err := inv.RecordDuration(ctx, inventory.Duration{Kind: inventory.DurationBuild, Subject: subject,
		Node: result.On, Ref: result.ID, Started: asked, Took: time.Since(asked), Detail: detail}); err != nil {
		fmt.Fprintf(os.Stderr, "%s: how long it took could not be recorded: %v\n", result.ID, err)
	}
}

// durationSummary is one subject's measurements of one kind.
type durationSummary struct {
	Kind    string `json:"kind"`
	Subject string `json:"subject"`
	Count   int    `json:"count"`
	Median  string `json:"median"`
	P90     string `json:"p90"`
	Max     string `json:"max"`
	// Bound is what to-be 45's rule would make of these, where the rule is a multiple of a measured
	// time: three times the slowest apply (S2), three times the median word interval (S1).
	Suggests string `json:"suggests,omitempty"`
}

func summarise(ds []inventory.Duration) []durationSummary {
	type key struct{ kind, subject string }
	by := map[key][]time.Duration{}
	for _, d := range ds {
		k := key{d.Kind, d.Subject}
		by[k] = append(by[k], d.Took)
	}
	var out []durationSummary
	for k, took := range by {
		slices.Sort(took)
		at := func(q float64) time.Duration { return took[int(q*float64(len(took)-1))] }
		s := durationSummary{Kind: k.kind, Subject: k.subject, Count: len(took),
			Median: round(at(0.5)), P90: round(at(0.9)), Max: round(took[len(took)-1])}
		switch k.kind {
		case inventory.DurationApply:
			s.Suggests = "S2 bound max(2m, 3×last apply) ≈ " + round(max(2*time.Minute, 3*at(0.9))) + " at the p90"
		case inventory.DurationHeartbeatGap:
			s.Suggests = "S1 bound 3×interval ≈ " + round(3*at(0.5))
		}
		out = append(out, s)
	}
	sort.Slice(out, func(i, j int) bool {
		ki, kj := slices.Index(inventory.DurationKinds, out[i].Kind), slices.Index(inventory.DurationKinds, out[j].Kind)
		if ki != kj {
			return ki < kj
		}
		return out[i].Subject < out[j].Subject
	})
	return out
}

func round(d time.Duration) string {
	switch {
	case d < time.Second:
		return d.Round(time.Millisecond).String()
	case d < time.Minute:
		return d.Round(100 * time.Millisecond).String()
	default:
		return d.Round(time.Second).String()
	}
}

// durationsCommand is `durations`: the summary per kind and subject, or every measurement as data.
func durationsCommand(ctx context.Context, args []string) error {
	set := flag.NewFlagSet("durations", flag.ContinueOnError)
	kind := set.String("kind", "", "one kind: "+strings.Join(inventory.DurationKinds, ", "))
	days := set.Int("days", 14, "how many days back")
	asJSON := set.Bool("json", false, "the summary as data")
	all := set.Bool("all", false, "every measurement rather than the summary")
	if _, err := parseAround(set, args); err != nil {
		return err
	}
	if *kind != "" && !slices.Contains(inventory.DurationKinds, *kind) {
		return fmt.Errorf("%q is not a kind of duration: %s", *kind, strings.Join(inventory.DurationKinds, ", "))
	}
	open, err := openStores(ctx)
	if err != nil {
		return err
	}
	defer open.Close()
	ds, err := open.inventory.Durations(ctx, *kind, time.Now().Add(-time.Duration(*days)*24*time.Hour))
	if err != nil {
		return err
	}
	if *all {
		body, err := json.MarshalIndent(ds, "", "  ")
		if err != nil {
			return err
		}
		fmt.Println(string(body))
		return nil
	}
	summary := summarise(ds)
	if *asJSON {
		body, err := json.MarshalIndent(map[string]any{"days": *days, "durations": summary}, "", "  ")
		if err != nil {
			return err
		}
		fmt.Println(string(body))
		return nil
	}
	if len(summary) == 0 {
		fmt.Printf("nothing measured in the last %d day(s): the serving controller records apply, heartbeat-gap, "+
			"plan-tier and build durations as it hears them\n", *days)
		return nil
	}
	fmt.Printf("durations over the last %d day(s) — what the core's bounds are set from (to-be 45 Phase 0)\n\n", *days)
	fmt.Printf("  %-14s %-28s %6s %10s %10s %10s\n", "kind", "of", "count", "median", "p90", "max")
	for _, s := range summary {
		fmt.Printf("  %-14s %-28s %6d %10s %10s %10s\n", s.Kind, s.Subject, s.Count, s.Median, s.P90, s.Max)
		if s.Suggests != "" {
			fmt.Printf("  %-14s %-28s %s\n", "", "", s.Suggests)
		}
	}
	return nil
}

// forgettingOldDurations removes what is older than a month, at start and daily after.
func forgettingOldDurations(ctx context.Context, inv *inventory.Inventory) {
	for {
		if n, err := inv.ForgetOldDurations(ctx); err != nil {
			fmt.Fprintf(os.Stderr, "durations older than %s could not be removed: %v\n", inventory.DurationsKeptFor, err)
		} else if n > 0 {
			fmt.Printf("removed %d duration(s) older than %s\n", n, inventory.DurationsKeptFor)
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(24 * time.Hour):
		}
	}
}