mesh-controller/cmd/mesh-controller/health_probes.go

package main

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strings"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/lease"
)

// The core components' health definitions, as probes in the self-check's registry (novox/hq to-be 45
// §8, §4 `H-*`). The same definitions judge a core component's new build on its first machine (the
// gate, gate.go); here they are run against every machine on the self-check's schedule, so a core
// component that stops being healthy between upgrades is said too.
//
//	controller   holds the lease, and says it is ready: its self-check ran, `status` answered in bound
//	node-engine  has reported its current declaration, under a build the mesh delivered
//	node tools   announced, and answer the bus's discovery
//	bus          every stream and durable consumer the mesh defines is there, and a request crosses the
//	             bus to every machine's node tools and back
const kindCoreUnhealthy = "core-unhealthy"

// probeControllerHealth is H-controller: the lease is held, fresh, by a controller that says it is
// ready — or one that started less than the witness's bound ago.
func probeControllerHealth(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
	h, found, err := theLease.holder(ctx)
	if err != nil {
		return nil, err
	}
	unhealthy := func(why string) []conditions.Observation {
		node := d.host
		if found && h.Host != "" {
			node = h.Host
		}
		return []conditions.Observation{{Scope: conditions.ScopeCore, ID: lease.ComponentController + "." + node,
			Token: "unhealthy", Kind: kindCoreUnhealthy, Machine: node, Severity: conditions.Urgent,
			Summary: "the controller is not healthy: " + why}}
	}
	switch {
	case !found:
		return unhealthy("nobody holds the controller lease"), nil
	case time.Since(h.Renewed) > lease.FreshWithin:
		return unhealthy(fmt.Sprintf("the lease was last renewed %s ago", time.Since(h.Renewed).Round(time.Second))), nil
	case (h.Health == nil || !h.Health.Ready) && time.Since(h.Taken) > lease.ReadyWithin:
		why := "the controller holding the lease does not say it is ready"
		if h.Health != nil && h.Health.Why != "" {
			why += ": " + h.Health.Why
		}
		return unhealthy(why), nil
	}
	return nil, nil
}

// probeBusHealth is H-bus: every stream and durable consumer the mesh defines is on the bus, and a
// request crosses it to every machine's node tools and back.
func probeBusHealth(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
	problems, err := busHealth(ctx, d)
	if err != nil {
		return nil, err
	}
	if len(problems) == 0 {
		return nil, nil
	}
	return []conditions.Observation{{Scope: conditions.ScopeBus, ID: "mesh", Token: "unhealthy",
		Kind: kindCoreUnhealthy, Severity: conditions.Urgent,
		Summary: fmt.Sprintf("the bus is not healthy: %s", strings.Join(problems, "; ")),
		Said:    strings.Join(problems, "; ")}}, nil
}

// busHealth is the bus's health definition, as what is wanting: nothing when healthy. What the bus's
// planned step checks after the bus is replaced, too.
var busHealth = func(ctx context.Context, d *doctor) ([]string, error) {
	if d.js == nil {
		return nil, errors.New("this controller has no bus to ask")
	}
	streams, consumers, err := expectedBusObjects(ctx, d.open.inventory)
	if err != nil {
		return nil, err
	}
	js := d.js.Context()
	var problems []string
	for _, s := range streams {
		if _, err := js.StreamInfo(s.Name, nats.Context(ctx)); errors.Is(err, nats.ErrStreamNotFound) {
			problems = append(problems, "the stream "+s.Name+" is missing")
		} else if err != nil {
			return nil, fmt.Errorf("the stream %s cannot be read: %w", s.Name, err)
		}
	}
	for _, c := range consumers {
		_, err := js.ConsumerInfo(c.Stream, c.Name, nats.Context(ctx))
		if errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound) {
			problems = append(problems, consumerWords(c)+" is missing")
		} else if err != nil {
			return nil, fmt.Errorf("%s cannot be read: %w", consumerWords(c), err)
		}
	}
	// The round trip: every machine heard from that runs the node tools answers across the bus.
	running, err := d.open.inventory.Running(ctx, broker.RuntimeModule)
	if err != nil {
		return nil, err
	}
	heard := heardMachines(d)
	var expected []string
	for _, n := range running {
		if heard[n] {
			expected = append(expected, n)
		}
	}
	if len(expected) > 0 {
		answered, err := servedOnTheBus(ctx, d.js.Conn())
		if err != nil {
			return nil, err
		}
		var silent []string
		for _, n := range expected {
			if !answered[n].runtime {
				silent = append(silent, n)
			}
		}
		// One machine's tools silent is that machine's (H-tools); none answering is the bus.
		if len(silent) == len(expected) {
			slices.Sort(silent)
			problems = append(problems, "no request crossed the bus and came back: no machine's node tools answered ("+
				strings.Join(silent, ", ")+")")
		}
	}
	return problems, nil
}