227 lines
8.3 KiB
Go
227 lines
8.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"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
|
|
}
|
|
|
|
// probeEngineHealth is H-engine: every machine heard from has reported its current declaration — the
|
|
// last one it was sent — under a node-engine build the mesh delivered, or is inside the bound of a send.
|
|
func probeEngineHealth(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
|
|
return machinesHealth(ctx, d, hostModule)
|
|
}
|
|
|
|
// probeToolsHealth is H-tools: every machine heard from that is assigned the node tools has them
|
|
// announced, answering the bus's discovery.
|
|
func probeToolsHealth(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
|
|
return machinesHealth(ctx, d, broker.RuntimeModule)
|
|
}
|
|
|
|
// machinesHealth judges a component on every machine running it and heard from, by its health
|
|
// definition, as the gate does — with no condition taken as the build's doing.
|
|
func machinesHealth(ctx context.Context, d *doctor, component string) ([]conditions.Observation, error) {
|
|
inv := d.open.inventory
|
|
running, err := inv.Running(ctx, component)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
shelf, err := inv.Catalogue(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
f, err := gatherGateFacts(ctx, d.open, coreComponent(component))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
f.judged = false
|
|
heard := heardMachines(d)
|
|
var out []conditions.Observation
|
|
for _, node := range running {
|
|
if !heard[node] {
|
|
continue // a machine not heard from is S1's
|
|
}
|
|
if r, said := f.reports[node]; said && !r.Current && r.Sent != nil && time.Since(*r.Sent) < gateBound {
|
|
continue // inside the bound of a send: S2's, and the gate's
|
|
}
|
|
// What the machine did with what it was sent is not the component's health: the node-engine's is
|
|
// that it reported its current declaration at all, the node tools' that they answer.
|
|
if r := f.reports[node]; r.Current || component == broker.RuntimeModule {
|
|
now := time.Now()
|
|
r.Current, r.Outcome = true, inventory.OutcomeApplied
|
|
if r.At == nil {
|
|
r.At = &now
|
|
}
|
|
f.reports[node] = r
|
|
}
|
|
if component == hostModule {
|
|
// A node-engine reporting a build the mesh delivered, current or previous, is the version
|
|
// split's (D10), not ill health; a build the mesh did not deliver is.
|
|
f.engines[node] = deliveredOr(shelf[component], f.engines[node])
|
|
}
|
|
h, why := judgeHealth(component, coreComponent(component), shelf[component], node, time.Time{}, f)
|
|
if h == healthGood {
|
|
continue
|
|
}
|
|
out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: coreComponent(component) + "." + node,
|
|
Token: "unhealthy", Kind: kindCoreUnhealthy, Machine: node, Severity: conditions.Warning,
|
|
Summary: fmt.Sprintf("the %s on %s is not healthy: %s", componentWords(component), node, why)})
|
|
}
|
|
return sortedFound(out), nil
|
|
}
|
|
|
|
// deliveredOr is the reported engine build, or the current delivered one when the report names any
|
|
// build at all: what H-engine judges is that it reports, not which.
|
|
func deliveredOr(m catalogue.Manifest, reported string) string {
|
|
if reported == "" {
|
|
return reported
|
|
}
|
|
if v := deliveredVersions(m); len(v) > 0 {
|
|
return v[0]
|
|
}
|
|
return reported
|
|
}
|
|
|
|
// componentWords is a core component as a sentence names it.
|
|
func componentWords(component string) string {
|
|
switch component {
|
|
case hostModule:
|
|
return "node-engine"
|
|
case broker.RuntimeModule:
|
|
return "node tools"
|
|
case catalogue.ControllerSeatName:
|
|
return "controller"
|
|
}
|
|
return component
|
|
}
|
|
|
|
// 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
|
|
}
|