mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/health.go

package main

// The module's health once its records are on the ledger (novox/hq to-be 55 §1.1, ADR 0592 §33 changed by ADR 0597 §12):
// unhealthy, naming the cause, when a kept thing does not answer, when the search index is behind the ledger, when a head lags
// behind its history, and when the stream of the verdicts is at its cap. It names where the records are served from and the
// result of the move's last check, and counts the drafts idle more than a week, the records the form-only check refused and the
// parts waiting more than 30 minutes; such a part is asked of the operator once, and the ask clears when it is judged. A record
// waiting for its review is not the module failing, so those counts leave it healthy.

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/buscalls"
)

// PartWaits is how long a part may wait on the judge before the operator is asked once.
const PartWaits = 30 * time.Minute

// ledgerHealth adds what the ledger form owes to the verdict of the watch.
func (k *KB) ledgerHealth(base HealthAnswer) HealthAnswer {
	ms, _ := k.Store.(*movableStore)
	l, err := k.ledger()
	if err != nil {
		if ms != nil {
			base.Why += "; the records are served from the " + SourceDatabase
		}
		return base
	}
	var broken, notes []string
	notes = append(notes, "the records are served from the ledger")
	// A kept thing that does not answer, and a head behind its history.
	var lag buscalls.Lagged
	if err := l.pipe.Call(buscalls.LedgerLag, buscalls.LedgerRequest{Ledger: LedgerName}, &lag); err != nil {
		broken = append(broken, "the ledger does not answer: "+err.Error())
	} else if len(lag.Instances) > 0 {
		names := make([]string, 0, 3)
		for i, in := range lag.Instances {
			if i == 3 {
				break
			}
			names = append(names, fmt.Sprintf("%s (state at %d of %d)", in.Instance, in.State, in.History))
		}
		broken = append(broken, fmt.Sprintf("%d head(s) are behind their history: %s", len(lag.Instances), strings.Join(names, ", ")))
	}
	// The index is behind the ledger when it holds fewer records than the ledger does.
	st := l.IndexState()
	l.mu.Lock()
	heads, indexed := len(l.heads), 0
	if l.index != nil {
		indexed = len(l.index.docsOf)
	}
	l.mu.Unlock()
	switch {
	case !st.Built:
		notes = append(notes, "the search index is being built")
	case heads > 0 && indexed < heads-reservedCount(l):
		broken = append(broken, fmt.Sprintf("the search index holds %d of %d records: it is behind the ledger", indexed, heads))
	default:
		notes = append(notes, fmt.Sprintf("the search index holds %d documents", st.Docs))
	}
	// The stream of the verdicts at its cap.
	var fill buscalls.StreamFilled
	if err := l.pipe.Call(buscalls.StreamFill, buscalls.StreamRequest{Stream: "reviews"}, &fill); err != nil {
		broken = append(broken, "the stream of the verdicts does not answer: "+err.Error())
	} else if fill.MaxBytes > 0 && fill.Bytes >= fill.MaxBytes {
		broken = append(broken, fmt.Sprintf("the stream of the verdicts is at its cap of %d bytes", fill.MaxBytes))
	}
	// The move's last check.
	if kv := k.States; kv != nil {
		if e, err := kv.State(findingsState).Get(checkKey); err == nil && e != nil {
			var c struct {
				Differences []string  `json:"differences"`
				At          time.Time `json:"at"`
			}
			if json.Unmarshal(e.Value, &c) == nil {
				if len(c.Differences) == 0 {
					notes = append(notes, "the last check of the database against the ledger was clean ("+c.At.Local().Format("2006-01-02 15:04")+")")
				} else {
					notes = append(notes, fmt.Sprintf("the last check of the database against the ledger found %d difference(s)", len(c.Differences)))
				}
			}
		}
	}
	// What waits on a person, counted and left healthy.
	idle, waiting := k.waitingCounts()
	if idle > 0 {
		notes = append(notes, fmt.Sprintf("%d draft(s) untouched for more than a week", idle))
	}
	if waiting > 0 {
		notes = append(notes, fmt.Sprintf("%d part(s) waiting on the judge for more than 30 minutes", waiting))
	}
	if refused := k.formRefused(); refused > 0 {
		notes = append(notes, fmt.Sprintf("%d record(s) the form-only check refused", refused))
	}
	if len(broken) > 0 {
		base.State, base.Healthy = "unhealthy", false
		base.Why = strings.Join(broken, "; ") + "; " + strings.Join(notes, "; ")
		return base
	}
	base.Why += "; " + strings.Join(notes, "; ")
	return base
}

func reservedCount(l *ledgerStore) int {
	n := 0
	for _, h := range l.heads {
		if st := h.Entry.Current.Status; st == StatusReserved || st == StatusWithdrawn || st == StatusRetracted {
			n++
		}
	}
	return n
}

// waitingCounts are the drafts idle more than DraftIdle and the parts waiting more than PartWaits; a part that waits is
// asked of the operator once, and the ask is cleared when it is judged (the mark leaves `asked`).
func (k *KB) waitingCounts() (idle, waiting int) {
	if k.States == nil {
		return 0, 0
	}
	now := k.now()
	if keys, err := k.States.State(draftsState).Keys(); err == nil {
		for _, key := range keys {
			if e, err := k.States.State(draftsState).Get(key); err == nil && e != nil {
				var d Draft
				if json.Unmarshal(e.Value, &d) == nil && now.Sub(d.Touched) > DraftIdle {
					idle++
				}
			}
		}
	}
	live := map[string]bool{}
	if keys, err := k.States.State(askedState).Keys(); err == nil {
		for _, key := range keys {
			e, err := k.States.State(askedState).Get(key)
			if err != nil || e == nil {
				continue
			}
			var a struct {
				Part   string    `json:"part"`
				Record string    `json:"record"`
				At     time.Time `json:"at"`
			}
			if json.Unmarshal(e.Value, &a) != nil || now.Sub(a.At) <= PartWaits {
				continue
			}
			waiting++
			live[key] = true
			k.askPartWaits(key, a.Record, a.Part)
		}
	}
	k.clearPartWaits(live)
	return idle, waiting
}

const partWaitPrefix = "part-wait."

// askPartWaits asks the operator once, through the operator channel, about a part waiting more than 30 minutes.
func (k *KB) askPartWaits(key, record, part string) {
	if k.Notify == nil || k.Store == nil {
		return
	}
	ctx := context.Background()
	if first, err := k.Store.Told(ctx, partWaitPrefix+key); err != nil || !first {
		return
	}
	if err := k.Notify("knowledge."+partWaitPrefix+key, fmt.Sprintf("The review of %s waits on the judge: its part %q has waited more than 30 minutes.", record, part),
		"A review is waiting on the judge: "+record, "The knowledge base asked the judge for a part of this record and no answer has come. It asks again by itself; the tool knowledge_asked lists what is in flight.",
		record+" is judged again"); err != nil {
		_ = k.Store.Untold(ctx, partWaitPrefix+key)
	}
}

// clearPartWaits clears the ask of every part that is no longer waiting.
func (k *KB) clearPartWaits(live map[string]bool) {
	if k.Clear == nil || k.Store == nil {
		return
	}
	ctx := context.Background()
	told, err := k.Store.ToldOf(ctx, partWaitPrefix)
	if err != nil {
		return
	}
	for _, t := range told {
		if live[strings.TrimPrefix(t, partWaitPrefix)] {
			continue
		}
		if err := k.Clear("knowledge."+t, "its part was judged"); err == nil {
			_ = k.Store.Untold(ctx, t)
		}
	}
}

// formRefused counts the records a form-only version would be refused for.
func (k *KB) formRefused() int {
	done, err := k.FormOnly("", nil, false)
	if err != nil {
		return 0
	}
	n := 0
	for _, d := range done {
		if strings.HasPrefix(d.Skipped, "the check refused") {
			n++
		}
	}
	return n
}

// pageHealth adds the overview page to a verdict: the last rebuild and each source's last answer, and the module is said degraded
// while any source is unanswered. The module stays healthy: a source that does not answer is that module's fault, not this one's.
func (k *KB) pageHealth(base HealthAnswer) HealthAnswer {
	if k.Overview == nil {
		return base
	}
	k.Overview.mu.Lock()
	rebuilt := k.Overview.rebuilt
	k.Overview.mu.Unlock()
	if bad := k.Overview.Unanswered(); len(bad) > 0 {
		var says []string
		for _, s := range bad {
			why := "no answer yet"
			if s.Mark != nil {
				why = s.Mark.Cause + ": " + s.Mark.Why
			}
			says = append(says, fmt.Sprintf("%s (%s; last answer %s)", s.Name, why, s.Last.Local().Format("15:04")))
		}
		base.Why += "; the overview page is degraded, it does not answer: " + strings.Join(says, ", ")
		return base
	}
	base.Why += "; the overview page was last rebuilt " + rebuilt.Local().Format("2006-01-02 15:04")
	return base
}