mesh-controller/cmd/mesh-controller/stuck_probe.go

package main

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"time"

	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

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

// What stopped moving (novox/hq ADR 0433 §2, §3, issue 1156). The watchman writes `stuck.<bucket>.<key>` in its own
// bucket for each key whose pulse expired, and again as moved when the key moves; it never deletes one. The controller
// reads that bucket in two places and holds no timer for either: a probe of the self-check raises each key not moved as
// a condition of the delivery scope, and the check that a machine is silent takes a node's presence key as its verdict.
const (
	kindStuck    = "stuck"
	probeStuckID = "DM"
)

// presenceVerdict is the watchman's word on whether a machine is online, from its stuck presence key.
type presenceVerdict int

const (
	// presenceUnsaid: the watchman has written nothing for the machine; the heartbeat rule judges it, as before.
	presenceUnsaid presenceVerdict = iota
	// presenceHeard: the key says moved, the machine is back online.
	presenceHeard
	// presenceSilent: the key says not moved, the machine was offline past its bound.
	presenceSilent
)

// presenceEntry is presence.<node> as the watchman wrote it: whether the machine is online and when the entry was made.
// The watchman writes it only when it changes, so for an online machine the time is when it connected.
type presenceEntry struct {
	Online bool
	At     time.Time
}

// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none).
// Last heard is lastHeard's: now when online, the entry's time when offline, the store's last report only without an entry.
func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) {
	switch {
	case stuck == nil:
		m.presence = presenceUnsaid
	case stuck.Moved:
		m.presence = presenceHeard
	default:
		m.presence = presenceSilent
	}
	m.lastHeard = lastHeard(m.lastHeard, entry, now)
}

// silent says the machine is silent: the watchman's verdict where it gave one, else the heartbeat rule. A machine never
// heard from is not silent, it has not joined.
func (m machineFacts) silent(now time.Time) bool {
	switch m.presence {
	case presenceSilent:
		return true
	case presenceHeard:
		return false
	}
	return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > silentBound
}

// stuckObservations are the conditions of what stopped moving: one for each key not moved, in the delivery scope, kind
// stuck, naming the key and what it waited for. A node's presence key is left to the silent check; a moved key raises
// nothing, so its condition clears.
func stuckObservations(entries map[string]sdkdelivery.Stuck) []conditions.Observation {
	keys := make([]string, 0, len(entries))
	for k := range entries {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	var out []conditions.Observation
	for _, k := range keys {
		s := entries[k]
		if s.Moved {
			continue
		}
		if s.Bucket == sdkdelivery.PresenceBucket {
			if node, _, _, err := sdkdelivery.ParsePresenceKey(s.Key); err == nil && node != "" {
				continue
			}
		}
		waited := s.Waited
		if waited == "" {
			waited = "a move"
		}
		out = append(out, conditions.Observation{Scope: conditions.ScopeDelivery, ID: k, Kind: kindStuck,
			Severity: conditions.Warning,
			Summary: fmt.Sprintf("%s has not moved since %s: it waited for %s", k, s.Since.UTC().Format("2006-01-02 15:04 MST"),
				waited),
			Said: fmt.Sprintf("%s: waiting for %s since %s", k, waited, s.Since.UTC().Format(time.RFC3339))})
	}
	return out
}

// readStuck is every key of the stuck bucket and its value; none when the bucket is not there yet (the watchman not
// delivered). A value of another version is left out, said.
func readStuck(ctx context.Context, conn *nats.Conn) (map[string]sdkdelivery.Stuck, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	kv, err := api.KeyValue(ctx, sdkdelivery.StuckBucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("the watchman's stuck bucket: %w", err)
	}
	keys, err := boundedCall(ctx, "listing the stuck keys", func(ctx context.Context) ([]string, error) { return kv.Keys(ctx) })
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	out := map[string]sdkdelivery.Stuck{}
	for _, key := range keys {
		e, err := kv.Get(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			continue
		}
		if err != nil {
			return nil, err
		}
		s, err := sdkdelivery.DecodeStuck(e.Value())
		if err != nil {
			fmt.Printf("the stuck key %s is not read: %v\n", key, err)
			continue
		}
		out[key] = s
	}
	return out, nil
}

// readPresence is, by node, presence.<node>'s entry; a node without an entry is left out.
func readPresence(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]presenceEntry, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	kv, err := api.KeyValue(ctx, sdkdelivery.PresenceBucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("the watchman's presence bucket: %w", err)
	}
	out := map[string]presenceEntry{}
	for _, node := range nodes {
		key, err := sdkdelivery.PresenceNodeKey(node)
		if err != nil {
			continue
		}
		e, err := kv.Get(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			continue
		}
		if err != nil {
			return nil, err
		}
		p, err := sdkdelivery.DecodePresence(e.Value())
		if err != nil {
			fmt.Printf("the presence key %s is not read: %v\n", key, err)
			continue
		}
		out[node] = presenceEntry{Online: p.Online, At: e.Created()}
	}
	return out, nil
}

// withPresence says of each machine what the watchman says: its stuck presence key and its presence entry's time.
func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts, now time.Time) error {
	stuck, err := readStuck(ctx, conn)
	if err != nil {
		return err
	}
	names := make([]string, len(machines))
	for i, m := range machines {
		names[i] = m.name
	}
	entries, err := readPresence(ctx, conn, names)
	if err != nil {
		return err
	}
	for i := range machines {
		var said *sdkdelivery.Stuck
		if key, err := sdkdelivery.StuckKey(sdkdelivery.PresenceBucket, machines[i].name); err == nil {
			if s, ok := stuck[key]; ok {
				said = &s
			}
		}
		var entry *presenceEntry
		if e, ok := entries[machines[i].name]; ok {
			entry = &e
		}
		applyPresence(&machines[i], said, entry, now)
	}
	return nil
}

// probeStuck is DM: every key the watchman says stopped moving is raised, but a node's presence, which the silent check
// takes.
func probeStuck(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
	if d.js == nil {
		return nil, nil
	}
	entries, err := readStuck(ctx, d.js.Conn())
	if err != nil {
		return nil, err
	}
	return stuckObservations(entries), nil
}