249 lines
8.0 KiB
Go
249 lines
8.0 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
"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/inventory"
|
|
"github.com/novox/mesh-controller/internal/testbus"
|
|
)
|
|
|
|
func TestNoHealerActsWithoutTheLease(t *testing.T) {
|
|
open := aMesh(t)
|
|
ctx := t.Context()
|
|
h, told, _ := healingOn(t, open)
|
|
sentNotReported(t, "laptop")
|
|
h.epoch = func(context.Context) (uint64, error) { return 0, nil }
|
|
h.tick(ctx)
|
|
h.epoch = func(context.Context) (uint64, error) { return 0, errors.New("the lease is not held") }
|
|
h.tick(ctx)
|
|
h.epoch = func(context.Context) (uint64, error) { return 57, nil }
|
|
h.acting = func() bool { return false }
|
|
h.tick(ctx)
|
|
if len(told.said()) != 0 {
|
|
t.Fatalf("a healer acted without the lease: %+v", told.said())
|
|
}
|
|
}
|
|
|
|
func TestH4ResetsOnlyWhatTheTableMarksResettable(t *testing.T) {
|
|
open := aMesh(t)
|
|
ctx := t.Context()
|
|
h, _, _ := healingOn(t, open)
|
|
marked := 0
|
|
for _, c := range broker.MeshConsumers() {
|
|
if c.Resettable != "" {
|
|
marked++
|
|
if c.Stream != broker.EventsStream || c.Name != broker.ControllerName {
|
|
t.Errorf("%s on %s is marked resettable: only the controller's own events consumer is", c.Name, c.Stream)
|
|
}
|
|
}
|
|
}
|
|
if marked != 1 {
|
|
t.Fatalf("%d consumers are marked resettable, want the controller's events consumer alone", marked)
|
|
}
|
|
for _, who := range []string{"EVENTS.anchor_shop", "CONTROL.controller"} {
|
|
if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopeBus, ID: who, Token: "behind",
|
|
Kind: kindConsumerBehind, Severity: conditions.Warning, Summary: who + " is far behind", Source: "D6"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
h.tick(ctx) // the fake reset fails the test if it is called
|
|
reset := ""
|
|
h.resetConsumer = func(stream, name string) (string, error) {
|
|
reset = stream + "." + name
|
|
return "it was 1500 behind", nil
|
|
}
|
|
if _, err := conditionsFrom.Observe(ctx, conditions.Observation{Scope: conditions.ScopeBus, ID: "EVENTS.controller",
|
|
Token: "behind", Kind: kindConsumerBehind, Severity: conditions.Warning, Summary: "far behind", Source: "D6"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
h.tick(ctx)
|
|
if reset != "EVENTS.controller" {
|
|
t.Fatalf("reset %q", reset)
|
|
}
|
|
}
|
|
|
|
func TestNatsH3AssertsAMissingConsumerAgainAndBrakesAfterItsBudget(t *testing.T) {
|
|
url := testbus.URL(t)
|
|
open := aMesh(t)
|
|
ctx := t.Context()
|
|
js, err := broker.Dial(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(js.Close)
|
|
for _, s := range []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS"} {
|
|
_ = js.Context().DeleteStream(s)
|
|
}
|
|
if _, err := assertBusObjects(ctx, open.inventory, js); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
h, told, clock := healingOn(t, open)
|
|
h.js = js
|
|
h.assertObjects = func(ctx context.Context) error { return assertOnSend(ctx, open.inventory, js, " ") }
|
|
d := &doctor{open: open, js: js}
|
|
probe := func() {
|
|
t.Helper()
|
|
found, err := probeConsumers(ctx, d)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := conditionsFrom.Reconcile(ctx, "D6", kindedAs(found, "consumer-wrong")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
const key = "bus.NODES.laptop.missing"
|
|
|
|
if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
probe()
|
|
if c, raised, _ := conditionsFrom.Get(ctx, key); !raised || c.Kind != "consumer-lost" {
|
|
t.Fatalf("the deleted consumer was not raised: %+v", c)
|
|
}
|
|
h.tick(ctx)
|
|
if _, err := js.Context().ConsumerInfo("NODES", "laptop"); err != nil {
|
|
t.Fatalf("H3 did not make the consumer again: %v", err)
|
|
}
|
|
c, still, _ := conditionsFrom.Get(ctx, key)
|
|
if !still || len(c.Tried) != 1 || c.Tried[0].By != "healer H3" || c.Resolver != "healer:H3" {
|
|
t.Fatalf("the act is not in the condition, or the healer cleared it: %+v", c)
|
|
}
|
|
probe()
|
|
if _, still, _ := conditionsFrom.Get(ctx, key); still {
|
|
t.Fatal("the probe's next run did not clear what H3 repaired")
|
|
}
|
|
// Kept in the history as the keeper says it, a moment later.
|
|
var cleared *conditions.Event
|
|
for wait := time.Now().Add(5 * time.Second); cleared == nil && time.Now().Before(wait); time.Sleep(20 * time.Millisecond) {
|
|
history, err := conditionsFrom.HistorySince(ctx, time.Now().Add(-time.Minute))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for i := range history {
|
|
if history[i].Key == key && history[i].Change == conditions.ChangeCleared {
|
|
cleared = &history[i]
|
|
}
|
|
}
|
|
}
|
|
if cleared == nil || !strings.Contains(cleared.Why, "D6 no longer observes it") || len(cleared.Tried) != 1 {
|
|
t.Fatalf("the clearing is not the probe's, or forgets what was tried: %+v", cleared)
|
|
}
|
|
|
|
// Again within the hour: the budget is one, so after its settle the operator is told, and H3 stops.
|
|
clock.pass(10 * time.Minute)
|
|
if err := js.Context().DeleteConsumer("NODES", "laptop"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
probe()
|
|
h.assertObjects = func(context.Context) error { t.Error("H3 acted past its budget"); return nil }
|
|
h.tick(ctx)
|
|
c, _, _ = conditionsFrom.Get(ctx, key)
|
|
if !c.Escalated() || c.Severity != conditions.Urgent || c.Count != 2 {
|
|
t.Fatalf("the spent budget was not handed to the operator: %+v", c)
|
|
}
|
|
acts := told.said()
|
|
if len(acts) != 2 || acts[0].Outcome != inventory.HealActed || acts[1].Outcome != inventory.HealEscalated {
|
|
t.Fatalf("said %+v", acts)
|
|
}
|
|
clock.pass(2 * time.Hour)
|
|
h.tick(ctx)
|
|
if len(told.said()) != 2 {
|
|
t.Fatal("a healer acted on what the operator holds")
|
|
}
|
|
}
|
|
|
|
func TestNatsH4ResetsTheControllersEventsConsumerAndItStillDelivers(t *testing.T) {
|
|
url := testbus.URL(t)
|
|
open := aMesh(t)
|
|
ctx := t.Context()
|
|
js, err := broker.Dial(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(js.Close)
|
|
for _, s := range []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS"} {
|
|
_ = js.Context().DeleteStream(s)
|
|
}
|
|
if _, err := assertBusObjects(ctx, open.inventory, js); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Bound as the controller binds it (link/receive_nats.go), taking one and acknowledging none.
|
|
events := make(chan *nats.Msg, 64)
|
|
sub, err := js.Context().ChanSubscribe("", events, nats.Bind(broker.EventsStream, broker.ControllerName))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
|
followed := broker.ControllerFollows[0]
|
|
for i := 0; i < consumerFarBehind+200; i++ {
|
|
if _, err := js.Context().Publish(followed, []byte(`{"n":`+strconv.Itoa(i)+`}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
d := &doctor{open: open, js: js}
|
|
probe := func() []conditions.Observation {
|
|
t.Helper()
|
|
found, err := probeConsumers(ctx, d)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := conditionsFrom.Reconcile(ctx, "D6", kindedAs(found, "consumer-wrong")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return found
|
|
}
|
|
probe()
|
|
const key = "bus.EVENTS.controller.behind"
|
|
if c, raised, _ := conditionsFrom.Get(ctx, key); !raised || c.Kind != kindConsumerBehind {
|
|
t.Fatalf("a consumer %d behind was not raised: %+v", consumerFarBehind+200, c)
|
|
}
|
|
h, told, _ := healingOn(t, open)
|
|
h.resetConsumer = func(stream, name string) (string, error) {
|
|
before, after, err := js.ResetConsumer(stream, name)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return "it was " + strconv.FormatUint(before.Pending, 10) + " behind; " + strconv.FormatUint(after.Pending, 10) +
|
|
" pending now", nil
|
|
}
|
|
h.tick(ctx)
|
|
if acts := told.said(); len(acts) != 1 || acts[0].Healer != "H4" || acts[0].Outcome != inventory.HealActed {
|
|
t.Fatalf("said %+v", acts)
|
|
}
|
|
if found := probe(); len(found) != 0 {
|
|
t.Fatalf("after the reset the probe still finds %+v", found)
|
|
}
|
|
if _, still, _ := conditionsFrom.Get(ctx, key); still {
|
|
t.Fatal("the probe's next run did not clear what H4 repaired")
|
|
}
|
|
// What comes next still reaches the controller.
|
|
for len(events) > 0 {
|
|
<-events
|
|
}
|
|
if _, err := js.Context().Publish(followed, []byte(`{"after":"the reset"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
deadline := time.After(10 * time.Second)
|
|
for {
|
|
select {
|
|
case m := <-events:
|
|
if strings.Contains(string(m.Data), "after") {
|
|
return
|
|
}
|
|
_ = m.Ack()
|
|
case <-deadline:
|
|
t.Fatal("the controller's subscription heard nothing after its consumer was reset: it would be deaf")
|
|
}
|
|
}
|
|
}
|