|
| 1 | +diff --git a/opensearch-operator/controllers/opensearchController.go b/opensearch-operator/controllers/opensearchController.go |
| 2 | +index 6bf9f9c..ed543da 100644 |
| 3 | +--- a/opensearch-operator/controllers/opensearchController.go |
| 4 | ++++ b/opensearch-operator/controllers/opensearchController.go |
| 5 | +@@ -234,6 +234,15 @@ func (r *OpenSearchClusterReconciler) reconcilePhasePending(ctx context.Context) |
| 6 | + return ctrl.Result{Requeue: true}, nil |
| 7 | + } |
| 8 | + |
| 9 | ++// A pass through the components takes seconds on a healthy cluster: the |
| 10 | ++// controller requeues itself every 30s and successive reconciles of an idle |
| 11 | ++// cluster land about 10s apart. A minute is past anything normal and still |
| 12 | ++// early enough to name a stall while whoever is watching is still watching. |
| 13 | ++const ( |
| 14 | ++ componentStallWarnAfter = time.Minute |
| 15 | ++ componentStallRepeatEvery = time.Minute |
| 16 | ++) |
| 17 | ++ |
| 18 | + func (r *OpenSearchClusterReconciler) reconcilePhaseRunning(ctx context.Context) (ctrl.Result, error) { |
| 19 | + // Update initialized status first |
| 20 | + if !r.Instance.Status.Initialized { |
| 21 | +@@ -313,16 +322,19 @@ func (r *OpenSearchClusterReconciler) reconcilePhaseRunning(ctx context.Context) |
| 22 | + r.Instance, |
| 23 | + ) |
| 24 | + |
| 25 | ++ // Wrapped so a component that stops returning is named in the log. Without |
| 26 | ++ // it the loop goes quiet mid-pass and the object is left unreconciled with |
| 27 | ++ // nothing written about it. |
| 28 | + componentReconcilers := []reconcilers.ComponentReconciler{ |
| 29 | +- tls.Reconcile, |
| 30 | +- securityconfig.Reconcile, |
| 31 | +- config.Reconcile, |
| 32 | +- cluster.Reconcile, |
| 33 | +- scaler.Reconcile, |
| 34 | +- dashboards.Reconcile, |
| 35 | +- upgrade.Reconcile, |
| 36 | +- restart.Reconcile, |
| 37 | +- snapshotrepository.Reconcile, |
| 38 | ++ reconcilers.WithStallWatchdog(r.Logger, "tls", componentStallWarnAfter, componentStallRepeatEvery, tls.Reconcile), |
| 39 | ++ reconcilers.WithStallWatchdog(r.Logger, "securityconfig", componentStallWarnAfter, componentStallRepeatEvery, securityconfig.Reconcile), |
| 40 | ++ reconcilers.WithStallWatchdog(r.Logger, "config", componentStallWarnAfter, componentStallRepeatEvery, config.Reconcile), |
| 41 | ++ reconcilers.WithStallWatchdog(r.Logger, "cluster", componentStallWarnAfter, componentStallRepeatEvery, cluster.Reconcile), |
| 42 | ++ reconcilers.WithStallWatchdog(r.Logger, "scaler", componentStallWarnAfter, componentStallRepeatEvery, scaler.Reconcile), |
| 43 | ++ reconcilers.WithStallWatchdog(r.Logger, "dashboards", componentStallWarnAfter, componentStallRepeatEvery, dashboards.Reconcile), |
| 44 | ++ reconcilers.WithStallWatchdog(r.Logger, "upgrade", componentStallWarnAfter, componentStallRepeatEvery, upgrade.Reconcile), |
| 45 | ++ reconcilers.WithStallWatchdog(r.Logger, "restart", componentStallWarnAfter, componentStallRepeatEvery, restart.Reconcile), |
| 46 | ++ reconcilers.WithStallWatchdog(r.Logger, "snapshotrepository", componentStallWarnAfter, componentStallRepeatEvery, snapshotrepository.Reconcile), |
| 47 | + } |
| 48 | + for _, rec := range componentReconcilers { |
| 49 | + result, err := rec() |
| 50 | +diff --git a/opensearch-operator/pkg/reconcilers/stallwatchdog.go b/opensearch-operator/pkg/reconcilers/stallwatchdog.go |
| 51 | +new file mode 100644 |
| 52 | +index 0000000..f01b989 |
| 53 | +--- /dev/null |
| 54 | ++++ b/opensearch-operator/pkg/reconcilers/stallwatchdog.go |
| 55 | +@@ -0,0 +1,56 @@ |
| 56 | ++package reconcilers |
| 57 | ++ |
| 58 | ++import ( |
| 59 | ++ "time" |
| 60 | ++ |
| 61 | ++ "github.com/go-logr/logr" |
| 62 | ++ "sigs.k8s.io/controller-runtime/pkg/reconcile" |
| 63 | ++) |
| 64 | ++ |
| 65 | ++// WithStallWatchdog wraps a component reconciler so a call that stops returning |
| 66 | ++// keeps saying so, under the name of the component it is stuck in. |
| 67 | ++// |
| 68 | ++// Of the components the cluster controller runs in sequence, only the TLS one |
| 69 | ++// logs when it is entered, so a reconcile that never comes back out of any of |
| 70 | ++// the others leaves nothing behind at all: the last line in the operator log |
| 71 | ++// stays whatever was written before the stall, the object stops being |
| 72 | ++// reconciled, and the two look identical from the outside. Logging every entry |
| 73 | ++// instead would put nine lines per cluster into the log every reconcile, so |
| 74 | ++// nothing is written while a component behaves. |
| 75 | ++// |
| 76 | ++// The report repeats rather than firing once, so a log collected at any point |
| 77 | ++// during a stall says which component is in flight and for how long, instead of |
| 78 | ++// only whoever read it at the right second knowing. |
| 79 | ++func WithStallWatchdog(logger logr.Logger, name string, warnAfter, repeatEvery time.Duration, rec ComponentReconciler) ComponentReconciler { |
| 80 | ++ return func() (reconcile.Result, error) { |
| 81 | ++ done := make(chan struct{}) |
| 82 | ++ defer close(done) |
| 83 | ++ |
| 84 | ++ go func() { |
| 85 | ++ started := time.Now() |
| 86 | ++ |
| 87 | ++ timer := time.NewTimer(warnAfter) |
| 88 | ++ defer timer.Stop() |
| 89 | ++ select { |
| 90 | ++ case <-done: |
| 91 | ++ return |
| 92 | ++ case <-timer.C: |
| 93 | ++ } |
| 94 | ++ |
| 95 | ++ ticker := time.NewTicker(repeatEvery) |
| 96 | ++ defer ticker.Stop() |
| 97 | ++ for { |
| 98 | ++ logger.Info("component reconciler has not returned", |
| 99 | ++ "component", name, |
| 100 | ++ "elapsed", time.Since(started).String()) |
| 101 | ++ select { |
| 102 | ++ case <-done: |
| 103 | ++ return |
| 104 | ++ case <-ticker.C: |
| 105 | ++ } |
| 106 | ++ } |
| 107 | ++ }() |
| 108 | ++ |
| 109 | ++ return rec() |
| 110 | ++ } |
| 111 | ++} |
| 112 | +diff --git a/opensearch-operator/pkg/reconcilers/stallwatchdog_test.go b/opensearch-operator/pkg/reconcilers/stallwatchdog_test.go |
| 113 | +new file mode 100644 |
| 114 | +index 0000000..4314196 |
| 115 | +--- /dev/null |
| 116 | ++++ b/opensearch-operator/pkg/reconcilers/stallwatchdog_test.go |
| 117 | +@@ -0,0 +1,90 @@ |
| 118 | ++package reconcilers |
| 119 | ++ |
| 120 | ++import ( |
| 121 | ++ "errors" |
| 122 | ++ "sync" |
| 123 | ++ "time" |
| 124 | ++ |
| 125 | ++ "github.com/go-logr/logr/funcr" |
| 126 | ++ . "github.com/onsi/ginkgo/v2" |
| 127 | ++ . "github.com/onsi/gomega" |
| 128 | ++ "sigs.k8s.io/controller-runtime/pkg/reconcile" |
| 129 | ++) |
| 130 | ++ |
| 131 | ++type recordingSink struct { |
| 132 | ++ mu sync.Mutex |
| 133 | ++ lines []string |
| 134 | ++} |
| 135 | ++ |
| 136 | ++func (r *recordingSink) write(prefix, args string) { |
| 137 | ++ r.mu.Lock() |
| 138 | ++ defer r.mu.Unlock() |
| 139 | ++ r.lines = append(r.lines, prefix+args) |
| 140 | ++} |
| 141 | ++ |
| 142 | ++func (r *recordingSink) snapshot() []string { |
| 143 | ++ r.mu.Lock() |
| 144 | ++ defer r.mu.Unlock() |
| 145 | ++ out := make([]string, len(r.lines)) |
| 146 | ++ copy(out, r.lines) |
| 147 | ++ return out |
| 148 | ++} |
| 149 | ++ |
| 150 | ++var errStallWatchdogTest = errors.New("component failed") |
| 151 | ++ |
| 152 | ++var _ = Describe("component reconciler stall watchdog", func() { |
| 153 | ++ var sink *recordingSink |
| 154 | ++ |
| 155 | ++ BeforeEach(func() { |
| 156 | ++ sink = &recordingSink{} |
| 157 | ++ }) |
| 158 | ++ |
| 159 | ++ It("passes the wrapped result and error through untouched", func() { |
| 160 | ++ want := reconcile.Result{Requeue: true, RequeueAfter: 7 * time.Second} |
| 161 | ++ wrapped := WithStallWatchdog(funcr.New(sink.write, funcr.Options{}), "cluster", time.Hour, time.Hour, func() (reconcile.Result, error) { |
| 162 | ++ return want, errStallWatchdogTest |
| 163 | ++ }) |
| 164 | ++ |
| 165 | ++ got, err := wrapped() |
| 166 | ++ |
| 167 | ++ Expect(got).To(Equal(want)) |
| 168 | ++ Expect(err).To(MatchError(errStallWatchdogTest)) |
| 169 | ++ }) |
| 170 | ++ |
| 171 | ++ It("stays silent while the component returns promptly", func() { |
| 172 | ++ wrapped := WithStallWatchdog(funcr.New(sink.write, funcr.Options{}), "cluster", 50*time.Millisecond, 50*time.Millisecond, func() (reconcile.Result, error) { |
| 173 | ++ return reconcile.Result{}, nil |
| 174 | ++ }) |
| 175 | ++ |
| 176 | ++ _, err := wrapped() |
| 177 | ++ Expect(err).ToNot(HaveOccurred()) |
| 178 | ++ |
| 179 | ++ Consistently(sink.snapshot, 200*time.Millisecond, 20*time.Millisecond).Should(BeEmpty()) |
| 180 | ++ }) |
| 181 | ++ |
| 182 | ++ It("names the component while its call is still in flight", func() { |
| 183 | ++ release := make(chan struct{}) |
| 184 | ++ returned := make(chan struct{}) |
| 185 | ++ wrapped := WithStallWatchdog(funcr.New(sink.write, funcr.Options{}), "securityconfig", 10*time.Millisecond, 10*time.Millisecond, func() (reconcile.Result, error) { |
| 186 | ++ <-release |
| 187 | ++ return reconcile.Result{}, nil |
| 188 | ++ }) |
| 189 | ++ |
| 190 | ++ go func() { |
| 191 | ++ defer GinkgoRecover() |
| 192 | ++ defer close(returned) |
| 193 | ++ _, err := wrapped() |
| 194 | ++ Expect(err).ToNot(HaveOccurred()) |
| 195 | ++ }() |
| 196 | ++ |
| 197 | ++ Eventually(sink.snapshot).Should(ContainElement(ContainSubstring("securityconfig"))) |
| 198 | ++ |
| 199 | ++ close(release) |
| 200 | ++ Eventually(returned).Should(BeClosed()) |
| 201 | ++ |
| 202 | ++ // The goroutine has to stop with the call, or every stalled reconcile |
| 203 | ++ // leaves one behind writing for the life of the process. |
| 204 | ++ settled := len(sink.snapshot()) |
| 205 | ++ Consistently(func() int { return len(sink.snapshot()) }, 100*time.Millisecond, 20*time.Millisecond).Should(Equal(settled)) |
| 206 | ++ }) |
| 207 | ++}) |
0 commit comments