diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index ba26c8a..a798bf4 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -22,6 +22,22 @@ rules: - get - list - watch +- apiGroups: + - crawl.example.com + resources: + - providerhealths + verbs: + - create + - delete + - get + - list +- apiGroups: + - crawl.example.com + resources: + - providerhealths/status + verbs: + - get + - patch - apiGroups: - crawl.example.com resources: diff --git a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md index 8566d90..861ce6a 100644 --- a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md +++ b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md @@ -5,7 +5,7 @@ Plan: [docs/plans/2026-09-11-1526-provider-health-checks.md](../plans/2026-09-11 - [x] Step 1 — ProviderHealth API and CRD - [x] Step 2 — Contract plus the kubernetes and GCP checks - [x] Step 3 — KVM check -- [ ] Step 4 — Monitor +- [x] Step 4 — Monitor - [ ] Step 5 — Metrics and wiring - [ ] Step 6 — Docs - [ ] Step 7 — Validate, record, push, open the PR @@ -138,3 +138,53 @@ Worth noting: - `TestDialError_keepsClassification`. - **Not run yet:** `TestKVMCheck_realHost` in the integration file only compiles here. It needs beelink, so it goes on the Step 7 checklist for the user to run. + +## Step 4 — Monitor + +`internal/providerhealth` has the `Monitor` runnable plus the pure `buildStatus`. + +- **Scheduling:** the Monitor is leader-only. It cleans up once at start, checks + immediately, then checks on a ticker. +- **Each round:** providers are checked concurrently under one timeout, then logged and + published one after another. A failed write is logged, and the next round writes + again. +- **Reads:** objects are read through the reader it is given. `main.go` passes the + manager's API reader (Step 5). + +The first `make manifests` left `config/rbac/role.yaml` unchanged, with no error. +I had put the `+kubebuilder:rbac` markers inside `Monitor`'s doc comment, and +controller-gen only reads RBAC markers from package-level comment groups. So they were +silently attached to the type and ignored. An empty `git diff config/rbac/role.yaml` was +the only sign. Moving them into their own comment block fixed it: + +```bash +make manifests && git diff config/rbac/role.yaml +# + providerhealths: create, delete, get, list +# + providerhealths/status: get, patch +``` + +Vet, race tests and the envtest round-trip (using the `bin/k8s/1.36.2-linux-amd64` +binaries `make setup-envtest` had already fetched): + +```bash +go vet ./internal/providerhealth/ && go test -race -count=1 -v ./internal/providerhealth/ +# --- PASS: TestMonitor_envtestRoundTrip (6.45s) ... ok +``` + +Worth noting: +- **Fake client scope:** the fake client needs a RESTMapper that registers + ProviderHealth as `RESTScopeRoot`. Its default mapper treats every kind as namespaced. +- **Why the envtest calls `publish` directly:** `checkOnce` only logs write errors, so a + schema rejection would look like a missing object. Calling `publish` makes the API + server's answer the test failure. + - Its first round writes a 2000-character message; the stored message is truncated to + 1024. + - Its second round proves that only the target that flipped gets a new + `lastTransitionTime`. +- **No guard against a hung checker:** the Monitor trusts `Checker`'s contract to return + once the context ends, and doesn't abandon a checker that ignores it. All three + checkers stop when the context ends: KVM through `awaitOrCancel`, and kubernetes and + GCP through their clients. +- **Test renamed:** the planned `TestMonitor_timeoutBoundsBlockingChecker` is + `TestMonitor_checkOnce_timeoutBoundsEachRound`. It also asserts that a fast provider + still publishes next to a hung one. diff --git a/internal/providerhealth/envtest_test.go b/internal/providerhealth/envtest_test.go new file mode 100644 index 0000000..0fc5f10 --- /dev/null +++ b/internal/providerhealth/envtest_test.go @@ -0,0 +1,100 @@ +package providerhealth + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" + "unicode/utf8" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/envtest" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// TestMonitor_envtestRoundTrip writes through a real API server, which is +// the only thing that validates the generated schema: cluster scope, the +// status subresource, required fields, list-map keys and the message limit. +func TestMonitor_envtestRoundTrip(t *testing.T) { + if testing.Short() { + t.Skip("envtest skipped under -short") + } + env := &envtest.Environment{ + CRDDirectoryPaths: []string{filepath.Join("..", "..", "config", "crd", "bases")}, + ErrorIfCRDPathMissing: true, + BinaryAssetsDirectory: envtestBinaryDir(), + } + cfg, err := env.Start() + if err != nil { + t.Fatalf("starting envtest (run `make setup-envtest` first): %v", err) + } + t.Cleanup(func() { _ = env.Stop() }) + scheme := runtime.NewScheme() + if err := crawlv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + c, err := client.New(cfg, client.Options{Scheme: scheme}) + if err != nil { + t.Fatal(err) + } + ctx := context.Background() + m := &Monitor{Client: c, Reader: c} + + first := Result{Provider: "kvm-home", Type: "kvm", Supported: true, CheckedAt: t0, Targets: []provider.TargetStatus{ + {Name: "nas1", Address: "nas1:22", Ready: true, Reason: provider.ReasonReachable, + Latency: 42 * time.Millisecond, Details: map[string]string{"libvirtVersion": "10.2.0"}}, + {Name: "nas2", Address: "nas2:22", Reason: provider.ReasonUnreachable, Message: strings.Repeat("x", 2000)}, + }} + if err := m.publish(ctx, first); err != nil { + t.Fatalf("first publish: %v", err) + } + got := getHealth(t, c, "kvm-home") + if got.Status.TargetsReady != "1/2" || got.Status.Targets[0].Details["libvirtVersion"] != "10.2.0" || + got.Status.Targets[0].LatencyMillis != 42 || utf8.RuneCountInString(got.Status.Targets[1].Message) != maxMessageLen { + t.Errorf("status after the first round = %+v", got.Status) + } + + second := Result{Provider: "kvm-home", Type: "kvm", Supported: true, CheckedAt: t1, + Targets: []provider.TargetStatus{up("nas1"), up("nas2")}} + if err := m.publish(ctx, second); err != nil { + t.Fatalf("second publish: %v", err) + } + got = getHealth(t, c, "kvm-home") + if cond := readyCondition(got); cond == nil || cond.Reason != reasonAllTargetsReady { + t.Errorf("Ready after recovery = %+v, want %s", cond, reasonAllTargetsReady) + } + if lt := got.Status.Targets[0].LastTransitionTime.Time; !lt.Equal(t0) { + t.Errorf("nas1 lastTransitionTime = %s, want %s", lt, t0) + } + if lt := got.Status.Targets[1].LastTransitionTime.Time; !lt.Equal(t1) { + t.Errorf("nas2 lastTransitionTime = %s, want %s", lt, t1) + } + + m.cleanup(ctx, nil) + var gone crawlv1alpha1.ProviderHealth + if err := c.Get(ctx, client.ObjectKey{Name: "kvm-home"}, &gone); !apierrors.IsNotFound(err) { + t.Errorf("Get after cleanup = %v, want NotFound", err) + } +} + +// envtestBinaryDir finds the binaries `make setup-envtest` downloads, so +// the test also runs outside make (which sets KUBEBUILDER_ASSETS instead). +func envtestBinaryDir() string { + base := filepath.Join("..", "..", "bin", "k8s") + entries, err := os.ReadDir(base) + if err != nil { + return "" + } + for _, e := range entries { + if e.IsDir() { + return filepath.Join(base, e.Name()) + } + } + return "" +} diff --git a/internal/providerhealth/monitor.go b/internal/providerhealth/monitor.go new file mode 100644 index 0000000..e001ea7 --- /dev/null +++ b/internal/providerhealth/monitor.go @@ -0,0 +1,290 @@ +// Package providerhealth checks every configured provider on an interval +// and publishes the results: one ProviderHealth object per provider, a log +// line whenever a target goes down or comes back, and a snapshot the +// metrics collector reads at scrape time. +package providerhealth + +import ( + "context" + "fmt" + "slices" + "sync" + "time" + + "github.com/go-logr/logr" + "go.opentelemetry.io/otel/attribute" + "k8s.io/apimachinery/pkg/api/equality" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/validation" + "sigs.k8s.io/controller-runtime/pkg/client" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +const ( + // labelManagedBy marks the ProviderHealth objects this operator owns; + // cleanup never deletes an object without it. + labelManagedBy = "app.kubernetes.io/managed-by" + managedByValue = "egress-proxies-operator" + + minInterval = 10 * time.Second + defaultTimeout = 15 * time.Second +) + +// Target is one configured provider. Checker is nil for a provider that +// cannot check itself. +type Target struct { + Name string + Type string + Checker provider.Checker +} + +// Result is the outcome of one provider's latest check. +type Result struct { + Provider string + Type string + // Supported is false when the provider has no Checker. + Supported bool + Targets []provider.TargetStatus + CheckedAt time.Time +} + +// controller-gen reads RBAC markers only at package level, so they stay +// out of Monitor's doc comment. +// +kubebuilder:rbac:groups=crawl.example.com,resources=providerhealths,verbs=get;list;create;delete +// +kubebuilder:rbac:groups=crawl.example.com,resources=providerhealths/status,verbs=get;patch + +// Monitor is the manager Runnable that checks the providers. +type Monitor struct { + // Client writes the ProviderHealth objects. + Client client.Client + // Reader reads them; pass the manager's API reader. The objects are + // cluster-scoped, and reading them through the cache would start a + // cluster-wide informer needing watch rights, whatever --proxy-namespace + // says, for a handful of reads a minute. + Reader client.Reader + // Targets are the configured providers, in config order. + Targets []Target + // Interval between rounds of checks; 0 turns checking off. The first + // round runs at start: nothing races it, and early visibility is the + // point. + Interval time.Duration + // Timeout bounds each round (default 15s). + Timeout time.Duration + + now func() time.Time + + mu sync.Mutex + latest []Result + + // prev holds each target's previous verdict for transition logging. + // Only the loop goroutine touches it. + prev map[string]provider.TargetStatus +} + +// NeedLeaderElection is true: only the leader dials the hypervisors and +// writes the objects, and only its gauges describe the checks. +func (m *Monitor) NeedLeaderElection() bool { return true } + +// ValidateSchedule checks --provider-check-interval and +// --provider-check-timeout. The interval is 0 (off) or at least 10s, which +// is what keeps one status write per provider per round cheap; the timeout +// is positive and no longer than the interval. +func ValidateSchedule(interval, timeout time.Duration) error { + switch { + case interval == 0: + return nil + case interval < minInterval: + return fmt.Errorf("--provider-check-interval must be 0 (off) or at least %s, got %s", minInterval, interval) + case timeout <= 0 || timeout > interval: + return fmt.Errorf("--provider-check-timeout must be positive and at most the interval (%s), got %s", interval, timeout) + } + return nil +} + +// Start deletes the objects of providers that are no longer configured, +// then checks every Interval until ctx ends. The providers config is read +// once per process, so cleaning up once is enough. With checking off, +// every object goes, so none is left claiming a stale Ready. +func (m *Monitor) Start(ctx context.Context) error { + if m.Timeout == 0 { + m.Timeout = defaultTimeout + } + if m.now == nil { + m.now = time.Now + } + log := logf.FromContext(ctx).WithName("provider-health") + ctx = logf.IntoContext(ctx, log) + + keep := map[string]bool{} + if m.Interval > 0 { + for _, t := range m.Targets { + keep[t.Name] = true + } + } + m.cleanup(ctx, keep) + if m.Interval == 0 { + log.Info("Provider checks are off (--provider-check-interval=0)") + return nil + } + for _, t := range m.Targets { + if errs := validation.IsDNS1123Subdomain(t.Name); len(errs) > 0 { + log.Info("WARNING: provider name is not a valid object name; its ProviderHealth object is not written, its metrics are", + "provider", t.Name, "problems", errs) + } + } + + m.checkOnce(ctx) + ticker := time.NewTicker(m.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + m.checkOnce(ctx) + } + } +} + +// Snapshot returns the latest results in config order. +func (m *Monitor) Snapshot() []Result { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]Result, len(m.latest)) + for i, r := range m.latest { + r.Targets = slices.Clone(r.Targets) + out[i] = r + } + return out +} + +// checkOnce runs every provider's check concurrently under one timeout, +// then records, logs and publishes the results provider by provider. A +// failed write is logged and retried by the next round. +func (m *Monitor) checkOnce(ctx context.Context) { + ctx, span := tracing.Start(ctx, "providerhealth.check") + defer span.End() + log := logf.FromContext(ctx) + + checkCtx, cancel := context.WithTimeout(ctx, m.Timeout) + defer cancel() + results := make([]Result, len(m.Targets)) + var wg sync.WaitGroup + for i, t := range m.Targets { + wg.Go(func() { results[i] = m.run(checkCtx, t) }) + } + wg.Wait() + + m.mu.Lock() + m.latest = results + m.mu.Unlock() + + notReady := 0 + for _, r := range results { + m.logTransitions(log, r) + for _, ts := range r.Targets { + if !ts.Ready { + notReady++ + } + } + if err := m.publish(ctx, r); err != nil { + log.Error(err, "Writing provider health", "provider", r.Provider) + } + } + span.SetAttributes( + attribute.Int("providerhealth.providers", len(results)), + attribute.Int("providerhealth.targets_not_ready", notReady), + ) +} + +func (m *Monitor) run(ctx context.Context, t Target) Result { + r := Result{Provider: t.Name, Type: t.Type, Supported: t.Checker != nil} + if t.Checker != nil { + r.Targets = t.Checker.Check(ctx) + } + r.CheckedAt = m.now() + return r +} + +// logTransitions logs a target going down, staying down for a different +// reason, or coming back. Steady states stay quiet; a target that is ready +// the first time it is seen only logs at V(1). +func (m *Monitor) logTransitions(log logr.Logger, r Result) { + if m.prev == nil { + m.prev = map[string]provider.TargetStatus{} + } + for _, ts := range r.Targets { + key := r.Provider + "/" + ts.Name + prev, seen := m.prev[key] + m.prev[key] = ts + kv := []any{"provider", r.Provider, "target", ts.Name, "address", ts.Address, "reason", ts.Reason} + switch { + case !ts.Ready && (!seen || prev.Ready || prev.Reason != ts.Reason): + log.Info("WARNING: provider target not ready", append(kv, "message", ts.Message)...) + case ts.Ready && seen && !prev.Ready: + log.Info("Provider target ready", append(kv, "latency", ts.Latency.Round(time.Millisecond))...) + case ts.Ready && !seen: + log.V(1).Info("Provider target ready", append(kv, "latency", ts.Latency.Round(time.Millisecond))...) + } + } +} + +// publish creates the provider's ProviderHealth object when missing and +// patches its status. The status changes on every round, since +// lastCheckTime moves; the comparison only skips the patch when a clock +// did not advance. +func (m *Monitor) publish(ctx context.Context, r Result) error { + if len(validation.IsDNS1123Subdomain(r.Provider)) > 0 { + return nil + } + var ph crawlv1alpha1.ProviderHealth + err := m.Reader.Get(ctx, client.ObjectKey{Name: r.Provider}, &ph) + switch { + case apierrors.IsNotFound(err): + ph = crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{ + Name: r.Provider, + Labels: map[string]string{labelManagedBy: managedByValue}, + }} + if err := m.Client.Create(ctx, &ph); err != nil { + return fmt.Errorf("creating ProviderHealth %s: %w", r.Provider, err) + } + case err != nil: + return fmt.Errorf("reading ProviderHealth %s: %w", r.Provider, err) + } + base := ph.DeepCopy() + ph.Status = buildStatus(ph.Status, r) + if equality.Semantic.DeepEqual(base.Status, ph.Status) { + return nil + } + if err := m.Client.Status().Patch(ctx, &ph, client.MergeFrom(base)); err != nil { + return fmt.Errorf("patching ProviderHealth %s status: %w", r.Provider, err) + } + return nil +} + +// cleanup deletes this operator's ProviderHealth objects whose name is not +// in keep. +func (m *Monitor) cleanup(ctx context.Context, keep map[string]bool) { + log := logf.FromContext(ctx) + var list crawlv1alpha1.ProviderHealthList + if err := m.Reader.List(ctx, &list, client.MatchingLabels{labelManagedBy: managedByValue}); err != nil { + log.Error(err, "Listing provider health objects; stale ones stay until the next start") + return + } + for i := range list.Items { + ph := &list.Items[i] + if keep[ph.Name] { + continue + } + if err := m.Client.Delete(ctx, ph); client.IgnoreNotFound(err) != nil { + log.Error(err, "Deleting provider health of an unconfigured provider", "name", ph.Name) + continue + } + log.Info("Deleted provider health of an unconfigured provider", "name", ph.Name) + } +} diff --git a/internal/providerhealth/monitor_test.go b/internal/providerhealth/monitor_test.go new file mode 100644 index 0000000..b2616d3 --- /dev/null +++ b/internal/providerhealth/monitor_test.go @@ -0,0 +1,354 @@ +package providerhealth + +import ( + "context" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/go-logr/logr/funcr" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +var ( + t0 = time.Date(2026, 9, 11, 15, 0, 0, 0, time.UTC) + t1 = t0.Add(time.Minute) +) + +type checkFunc func(ctx context.Context) []provider.TargetStatus + +func (f checkFunc) Check(ctx context.Context) []provider.TargetStatus { return f(ctx) } + +func verdicts(ts ...provider.TargetStatus) checkFunc { + return func(context.Context) []provider.TargetStatus { return ts } +} + +func up(name string) provider.TargetStatus { + return provider.TargetStatus{Name: name, Address: name + ":22", Ready: true, Reason: provider.ReasonReachable} +} + +func down(name, reason string) provider.TargetStatus { + return provider.TargetStatus{Name: name, Address: name + ":22", Reason: reason, Message: name + " is down"} +} + +func managed(name string) *crawlv1alpha1.ProviderHealth { + return &crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{labelManagedBy: managedByValue}, + }} +} + +// newFakeClient registers ProviderHealth as cluster-scoped; the fake's +// default mapper would treat it as namespaced. +func newFakeClient(t *testing.T, objs ...client.Object) client.WithWatch { + t.Helper() + scheme := runtime.NewScheme() + if err := crawlv1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + mapper := apimeta.NewDefaultRESTMapper([]schema.GroupVersion{crawlv1alpha1.GroupVersion}) + mapper.Add(crawlv1alpha1.GroupVersion.WithKind("ProviderHealth"), apimeta.RESTScopeRoot) + return fake.NewClientBuilder(). + WithScheme(scheme). + WithRESTMapper(mapper). + WithStatusSubresource(&crawlv1alpha1.ProviderHealth{}). + WithObjects(objs...). + Build() +} + +func newTestMonitor(c client.Client, targets ...Target) *Monitor { + return &Monitor{ + Client: c, + Reader: c, + Targets: targets, + Interval: time.Minute, + Timeout: time.Second, + now: func() time.Time { return t1 }, + } +} + +func getHealth(t *testing.T, c client.Reader, name string) *crawlv1alpha1.ProviderHealth { + t.Helper() + var ph crawlv1alpha1.ProviderHealth + if err := c.Get(context.Background(), client.ObjectKey{Name: name}, &ph); err != nil { + t.Fatalf("Get ProviderHealth %s: %v", name, err) + } + return &ph +} + +func readyCondition(ph *crawlv1alpha1.ProviderHealth) *metav1.Condition { + return apimeta.FindStatusCondition(ph.Status.Conditions, crawlv1alpha1.ConditionReady) +} + +func TestMonitor_checkOnce_createsLabelledObjects(t *testing.T) { + t.Parallel() + c := newFakeClient(t) + m := newTestMonitor(c, + Target{Name: "kvm-home", Type: "kvm", Checker: verdicts(up("nas1"), down("nas2", provider.ReasonUnreachable))}, + Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))}, + ) + + m.checkOnce(context.Background()) + + kvm := getHealth(t, c, "kvm-home") + if kvm.Labels[labelManagedBy] != managedByValue { + t.Errorf("labels = %v, want %s=%s", kvm.Labels, labelManagedBy, managedByValue) + } + if cond := readyCondition(kvm); cond == nil || cond.Status != metav1.ConditionFalse || cond.Reason != reasonTargetsNotReady { + t.Errorf("kvm-home Ready = %+v, want False/%s", cond, reasonTargetsNotReady) + } + if kvm.Status.Type != "kvm" || kvm.Status.TargetsReady != "1/2" || kvm.Status.LastCheckTime == nil || !kvm.Status.LastCheckTime.Time.Equal(t1) { + t.Errorf("kvm-home status = %+v, want type kvm, 1/2 ready, checked at %s", kvm.Status, t1) + } + local := getHealth(t, c, "local") + if cond := readyCondition(local); cond == nil || cond.Status != metav1.ConditionTrue { + t.Errorf("local Ready = %+v, want True", cond) + } +} + +func TestMonitor_checkOnce_patchesExisting(t *testing.T) { + t.Parallel() + existing := managed("local") + existing.Status = crawlv1alpha1.ProviderHealthStatus{ + Type: "kubernetes", + Targets: []crawlv1alpha1.ProviderTarget{readyAt("apiserver", t0)}, + } + c := newFakeClient(t, existing) + m := newTestMonitor(c, Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))}) + + m.checkOnce(context.Background()) + + got := getHealth(t, c, "local") + if got.Status.LastCheckTime == nil || !got.Status.LastCheckTime.Time.Equal(t1) { + t.Errorf("lastCheckTime = %v, want %s", got.Status.LastCheckTime, t1) + } + if lt := got.Status.Targets[0].LastTransitionTime.Time; !lt.Equal(t0) { + t.Errorf("apiserver lastTransitionTime = %s, want %s kept: it was ready before too", lt, t0) + } +} + +func TestMonitor_checkOnce_timeoutBoundsEachRound(t *testing.T) { + t.Parallel() + c := newFakeClient(t) + var hadDeadline atomic.Bool + hanging := checkFunc(func(ctx context.Context) []provider.TargetStatus { + _, ok := ctx.Deadline() + hadDeadline.Store(ok) + <-ctx.Done() + return []provider.TargetStatus{down("nas1", provider.ReasonTimeout)} + }) + m := newTestMonitor(c, + Target{Name: "kvm-home", Type: "kvm", Checker: hanging}, + Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))}, + ) + m.Timeout = 50 * time.Millisecond + + done := make(chan struct{}) + go func() { + m.checkOnce(context.Background()) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("checkOnce still running 5s after a 50ms timeout") + } + + if !hadDeadline.Load() { + t.Error("the checker got a context without a deadline") + } + if cond := readyCondition(getHealth(t, c, "kvm-home")); cond == nil || cond.Status != metav1.ConditionFalse { + t.Errorf("kvm-home Ready = %+v, want False", cond) + } + if cond := readyCondition(getHealth(t, c, "local")); cond == nil || cond.Status != metav1.ConditionTrue { + t.Errorf("local Ready = %+v, want True despite the slow neighbour", cond) + } +} + +func TestMonitor_checkOnce_noCheckerIsUnknown(t *testing.T) { + t.Parallel() + c := newFakeClient(t) + m := newTestMonitor(c, Target{Name: "gcp-eu", Type: "gcp"}) + + m.checkOnce(context.Background()) + + got := getHealth(t, c, "gcp-eu") + if cond := readyCondition(got); cond == nil || cond.Status != metav1.ConditionUnknown || cond.Reason != reasonChecksNotImplemented { + t.Errorf("Ready = %+v, want Unknown/%s", cond, reasonChecksNotImplemented) + } +} + +func TestMonitor_checkOnce_invalidNameSkipsTheObject(t *testing.T) { + t.Parallel() + c := newFakeClient(t) + m := newTestMonitor(c, Target{Name: "Home_KVM", Type: "kvm", Checker: verdicts(up("nas1"))}) + + m.checkOnce(context.Background()) + + var list crawlv1alpha1.ProviderHealthList + if err := c.List(context.Background(), &list); err != nil { + t.Fatal(err) + } + if len(list.Items) != 0 { + t.Errorf("wrote %d objects, want none for an invalid name", len(list.Items)) + } + snap := m.Snapshot() + if len(snap) != 1 || snap[0].Provider != "Home_KVM" || !snap[0].Targets[0].Ready { + t.Errorf("Snapshot() = %+v, want the result kept for metrics", snap) + } +} + +func TestMonitor_logsOnlyTransitions(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var lines []string + log := funcr.New(func(_, args string) { + mu.Lock() + defer mu.Unlock() + lines = append(lines, args) + }, funcr.Options{}) + ctx := logr.NewContext(context.Background(), log) + rounds := [][]provider.TargetStatus{ + {up("nas1")}, + {up("nas1")}, + {down("nas1", provider.ReasonUnreachable)}, + {down("nas1", provider.ReasonUnreachable)}, + {down("nas1", provider.ReasonTimeout)}, + {up("nas1")}, + } + round := 0 + m := newTestMonitor(newFakeClient(t), Target{Name: "kvm-home", Type: "kvm", Checker: checkFunc( + func(context.Context) []provider.TargetStatus { return rounds[round] })}) + + for round = range rounds { + m.checkOnce(ctx) + } + + mu.Lock() + defer mu.Unlock() + want := [][]string{ + {`"msg"="WARNING: provider target not ready"`, `"reason"="Unreachable"`}, + {`"msg"="WARNING: provider target not ready"`, `"reason"="Timeout"`}, + {`"msg"="Provider target ready"`}, + } + if len(lines) != len(want) { + t.Fatalf("logged %d lines, want %d:\n%s", len(lines), len(want), strings.Join(lines, "\n")) + } + for i, subs := range want { + for _, sub := range subs { + if !strings.Contains(lines[i], sub) { + t.Errorf("line %d missing %s:\n%s", i, sub, lines[i]) + } + } + } +} + +func TestMonitor_cleanup_deletesOnlyLabelledUnconfigured(t *testing.T) { + t.Parallel() + foreign := &crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{Name: "foreign"}} + c := newFakeClient(t, managed("kvm-home"), managed("retired"), foreign) + m := newTestMonitor(c) + + m.cleanup(context.Background(), map[string]bool{"kvm-home": true}) + + getHealth(t, c, "kvm-home") + getHealth(t, c, "foreign") + var gone crawlv1alpha1.ProviderHealth + if err := c.Get(context.Background(), client.ObjectKey{Name: "retired"}, &gone); !apierrors.IsNotFound(err) { + t.Errorf("Get retired = %v, want NotFound", err) + } +} + +func TestMonitor_Start_offCleansUpAndReturns(t *testing.T) { + t.Parallel() + c := newFakeClient(t, managed("kvm-home")) + var checked atomic.Bool + m := newTestMonitor(c, Target{Name: "kvm-home", Type: "kvm", Checker: checkFunc( + func(context.Context) []provider.TargetStatus { checked.Store(true); return nil })}) + m.Interval = 0 + + errc := make(chan error, 1) + go func() { errc <- m.Start(context.Background()) }() + select { + case err := <-errc: + if err != nil { + t.Fatalf("Start() = %v, want nil", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Start did not return with checking off") + } + + if checked.Load() { + t.Error("a check ran with checking off") + } + var gone crawlv1alpha1.ProviderHealth + if err := c.Get(context.Background(), client.ObjectKey{Name: "kvm-home"}, &gone); !apierrors.IsNotFound(err) { + t.Errorf("Get kvm-home = %v, want NotFound so no stale Ready is left behind", err) + } +} + +func TestMonitor_Start_checksAtOnceAndStops(t *testing.T) { + t.Parallel() + c := newFakeClient(t) + m := newTestMonitor(c, Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))}) + m.Interval = time.Hour + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + errc := make(chan error, 1) + go func() { errc <- m.Start(ctx) }() + deadline := time.Now().Add(5 * time.Second) + for len(m.Snapshot()) == 0 { + if time.Now().After(deadline) { + t.Fatal("no check within 5s of Start, want one right away") + } + time.Sleep(10 * time.Millisecond) + } + cancel() + + select { + case err := <-errc: + if err != nil { + t.Errorf("Start() = %v, want nil", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Start did not return after its context ended") + } +} + +func TestValidateSchedule(t *testing.T) { + t.Parallel() + tests := []struct { + name string + interval, timeout time.Duration + wantErr bool + }{ + {"off", 0, 0, false}, + {"defaults", time.Minute, 15 * time.Second, false}, + {"shortest interval", 10 * time.Second, 10 * time.Second, false}, + {"too frequent", 5 * time.Second, time.Second, true}, + {"negative interval", -time.Minute, time.Second, true}, + {"no timeout", time.Minute, 0, true}, + {"timeout longer than interval", time.Minute, 2 * time.Minute, true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if err := ValidateSchedule(tc.interval, tc.timeout); (err != nil) != tc.wantErr { + t.Errorf("ValidateSchedule(%s, %s) = %v, want error %v", tc.interval, tc.timeout, err, tc.wantErr) + } + }) + } +} diff --git a/internal/providerhealth/status.go b/internal/providerhealth/status.go new file mode 100644 index 0000000..6dcaeec --- /dev/null +++ b/internal/providerhealth/status.go @@ -0,0 +1,98 @@ +package providerhealth + +import ( + "fmt" + "slices" + "strings" + "unicode/utf8" + + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" +) + +// Reasons on the Ready condition. +const ( + reasonAllTargetsReady = "AllTargetsReady" + reasonTargetsNotReady = "TargetsNotReady" + reasonChecksNotImplemented = "ChecksNotImplemented" + reasonNoTargets = "NoTargets" +) + +// maxMessageLen matches the CRD's limit on a target's message. +const maxMessageLen = 1024 + +// buildStatus derives a ProviderHealth status from the previous one and a +// fresh result. Every timestamp comes from r.CheckedAt, and prev is left +// untouched. A target keeps its lastTransitionTime while ready does not +// flip, which is how several hosts each show when they went down. +func buildStatus(prev crawlv1alpha1.ProviderHealthStatus, r Result) crawlv1alpha1.ProviderHealthStatus { + now := metav1.NewTime(r.CheckedAt) + out := crawlv1alpha1.ProviderHealthStatus{ + Type: r.Type, + Conditions: slices.Clone(prev.Conditions), + LastCheckTime: &now, + } + before := make(map[string]crawlv1alpha1.ProviderTarget, len(prev.Targets)) + for _, t := range prev.Targets { + before[t.Name] = t + } + ready := 0 + var failing []string + for _, ts := range r.Targets { + t := crawlv1alpha1.ProviderTarget{ + Name: ts.Name, + Address: ts.Address, + Ready: ts.Ready, + Reason: ts.Reason, + Message: truncate(ts.Message, maxMessageLen), + LatencyMillis: ts.Latency.Milliseconds(), + LastTransitionTime: now, + Details: ts.Details, + } + if b, ok := before[ts.Name]; ok && b.Ready == ts.Ready { + t.LastTransitionTime = b.LastTransitionTime + } + if ts.Ready { + ready++ + } else { + failing = append(failing, ts.Name+": "+ts.Reason) + } + out.Targets = append(out.Targets, t) + } + if r.Supported { + out.TargetsReady = fmt.Sprintf("%d/%d", ready, len(r.Targets)) + } + + cond := metav1.Condition{Type: crawlv1alpha1.ConditionReady, LastTransitionTime: now} + switch { + case !r.Supported: + cond.Status = metav1.ConditionUnknown + cond.Reason = reasonChecksNotImplemented + cond.Message = fmt.Sprintf("%s providers have no health check", r.Type) + case len(r.Targets) == 0: + cond.Status = metav1.ConditionUnknown + cond.Reason = reasonNoTargets + cond.Message = "the provider has nothing to check" + case len(failing) == 0: + cond.Status = metav1.ConditionTrue + cond.Reason = reasonAllTargetsReady + cond.Message = fmt.Sprintf("%d of %d targets ready", ready, len(r.Targets)) + default: + cond.Status = metav1.ConditionFalse + cond.Reason = reasonTargetsNotReady + cond.Message = truncate(strings.Join(failing, "; "), maxMessageLen) + } + apimeta.SetStatusCondition(&out.Conditions, cond) + return out +} + +// truncate cuts s to at most n characters, which is how the API server +// counts maxLength. +func truncate(s string, n int) string { + if utf8.RuneCountInString(s) <= n { + return s + } + return string([]rune(s)[:n-1]) + "…" +} diff --git a/internal/providerhealth/status_test.go b/internal/providerhealth/status_test.go new file mode 100644 index 0000000..6e3ac0e --- /dev/null +++ b/internal/providerhealth/status_test.go @@ -0,0 +1,135 @@ +package providerhealth + +import ( + "strings" + "testing" + "time" + "unicode/utf8" + + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +func readyAt(name string, at time.Time) crawlv1alpha1.ProviderTarget { + return crawlv1alpha1.ProviderTarget{Name: name, Ready: true, Reason: provider.ReasonReachable, LastTransitionTime: metav1.NewTime(at)} +} + +func TestBuildStatus(t *testing.T) { + t.Parallel() + tests := []struct { + name string + prev crawlv1alpha1.ProviderHealthStatus + result Result + wantStatus metav1.ConditionStatus + wantReason string + wantMessage string + wantTargetsReady string + // wantTransitions is each target's lastTransitionTime, in order. + wantTransitions []time.Time + }{ + { + name: "all ready", + result: Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1"), up("nas2")}, CheckedAt: t1}, + wantStatus: metav1.ConditionTrue, + wantReason: reasonAllTargetsReady, + wantMessage: "2 of 2 targets ready", + wantTargetsReady: "2/2", + wantTransitions: []time.Time{t1, t1}, + }, + { + name: "one target down", + result: Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1"), down("nas2", provider.ReasonUnreachable)}, CheckedAt: t1}, + wantStatus: metav1.ConditionFalse, + wantReason: reasonTargetsNotReady, + wantMessage: "nas2: Unreachable", + wantTargetsReady: "1/2", + wantTransitions: []time.Time{t1, t1}, + }, + { + name: "only a flip moves a target's transition time", + prev: crawlv1alpha1.ProviderHealthStatus{Targets: []crawlv1alpha1.ProviderTarget{readyAt("nas1", t0), readyAt("nas2", t0)}}, + result: Result{Type: "kvm", Supported: true, CheckedAt: t1, + Targets: []provider.TargetStatus{up("nas1"), down("nas2", provider.ReasonTimeout)}}, + wantStatus: metav1.ConditionFalse, + wantReason: reasonTargetsNotReady, + wantMessage: "nas2: Timeout", + wantTargetsReady: "1/2", + wantTransitions: []time.Time{t0, t1}, + }, + { + name: "no checker", + result: Result{Type: "gcp", CheckedAt: t1}, + wantStatus: metav1.ConditionUnknown, + wantReason: reasonChecksNotImplemented, + }, + { + name: "nothing to check", + result: Result{Type: "kvm", Supported: true, CheckedAt: t1}, + wantStatus: metav1.ConditionUnknown, + wantReason: reasonNoTargets, + wantTargetsReady: "0/0", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := buildStatus(tc.prev, tc.result) + + cond := apimeta.FindStatusCondition(got.Conditions, crawlv1alpha1.ConditionReady) + if cond == nil || cond.Status != tc.wantStatus || cond.Reason != tc.wantReason { + t.Fatalf("Ready = %+v, want %s/%s", cond, tc.wantStatus, tc.wantReason) + } + if tc.wantMessage != "" && cond.Message != tc.wantMessage { + t.Errorf("Ready message = %q, want %q", cond.Message, tc.wantMessage) + } + if got.TargetsReady != tc.wantTargetsReady { + t.Errorf("targetsReady = %q, want %q", got.TargetsReady, tc.wantTargetsReady) + } + if got.Type != tc.result.Type || got.LastCheckTime == nil || !got.LastCheckTime.Time.Equal(t1) { + t.Errorf("type/lastCheckTime = %q/%v, want %q/%s", got.Type, got.LastCheckTime, tc.result.Type, t1) + } + if len(got.Targets) != len(tc.wantTransitions) { + t.Fatalf("got %d targets, want %d", len(got.Targets), len(tc.wantTransitions)) + } + for i, want := range tc.wantTransitions { + if lt := got.Targets[i].LastTransitionTime.Time; !lt.Equal(want) { + t.Errorf("target %s lastTransitionTime = %s, want %s", got.Targets[i].Name, lt, want) + } + } + }) + } +} + +func TestBuildStatus_conditionKeepsTransitionWhileUnchanged(t *testing.T) { + t.Parallel() + first := buildStatus(crawlv1alpha1.ProviderHealthStatus{}, + Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1")}, CheckedAt: t0}) + firstCond := apimeta.FindStatusCondition(first.Conditions, crawlv1alpha1.ConditionReady) + + second := buildStatus(first, Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1")}, CheckedAt: t1}) + + cond := apimeta.FindStatusCondition(second.Conditions, crawlv1alpha1.ConditionReady) + if !cond.LastTransitionTime.Time.Equal(t0) { + t.Errorf("Ready lastTransitionTime = %s, want %s from the first round", cond.LastTransitionTime, t0) + } + if !firstCond.LastTransitionTime.Time.Equal(t0) || len(first.Conditions) != 1 { + t.Errorf("buildStatus modified the previous status: %+v", first.Conditions) + } +} + +func TestBuildStatus_truncatesLongMessages(t *testing.T) { + t.Parallel() + long := strings.Repeat("é", 3000) + got := buildStatus(crawlv1alpha1.ProviderHealthStatus{}, Result{Type: "kvm", Supported: true, CheckedAt: t1, + Targets: []provider.TargetStatus{{Name: "nas1", Reason: provider.ReasonUnreachable, Message: long}}}) + + if n := utf8.RuneCountInString(got.Targets[0].Message); n != maxMessageLen { + t.Errorf("message has %d characters, want %d", n, maxMessageLen) + } + if !strings.HasSuffix(got.Targets[0].Message, "…") { + t.Error("truncated message does not end with an ellipsis") + } +}