11diff --git a/opensearch-operator/controllers/opensearchController.go b/opensearch-operator/controllers/opensearchController.go
2- index 6bf9f9c..ed543da 100644
2+ index 6bf9f9c..554a51f 100644
33--- a/opensearch-operator/controllers/opensearchController.go
44+++ b/opensearch-operator/controllers/opensearchController.go
5- @@ -234,6 +234,15 @@ func (r *OpenSearchClusterReconciler) reconcilePhasePending(ctx context.Context)
5+ @@ -201,6 +201,10 @@ func (r *OpenSearchClusterReconciler) deleteExternalResources(ctx context.Contex
6+ r.Instance,
7+ )
8+
9+ + // Not wrapped: every DeleteResources at this version returns immediately
10+ + // (tls.go:578, securityconfig.go:254, configuration.go:182, cluster.go:272,
11+ + // dashboards.go:188), so a watchdog here could only ever spawn goroutines
12+ + // for a call that cannot block.
13+ componentReconcilers := []reconcilers.ComponentReconciler{
14+ tls.DeleteResources,
15+ securityconfig.DeleteResources,
16+ @@ -234,6 +238,15 @@ func (r *OpenSearchClusterReconciler) reconcilePhasePending(ctx context.Context)
617 return ctrl.Result{Requeue: true}, nil
718 }
819
920+ // 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- + )
21+ + // controller requeues itself every 30s, and on the e2e runs that prompted this
22+ + // the reconciles of a settled cluster came about 10s apart, driven by watch
23+ + // events on top of that requeue. A minute is past anything normal and still
24+ + // early enough to name a stall while whoever is watching is still watching,
25+ + // and it is both how long a call may run before it is reported and how often
26+ + // the report then repeats.
27+ + const componentStallReportEvery = time.Minute
1728+
1829 func (r *OpenSearchClusterReconciler) reconcilePhaseRunning(ctx context.Context) (ctrl.Result, error) {
1930 // Update initialized status first
2031 if !r.Instance.Status.Initialized {
21- @@ -313,16 +322 ,19 @@ func (r *OpenSearchClusterReconciler) reconcilePhaseRunning(ctx context.Context)
32+ @@ -313,16 +326 ,19 @@ func (r *OpenSearchClusterReconciler) reconcilePhaseRunning(ctx context.Context)
2233 r.Instance,
2334 )
2435
@@ -35,24 +46,24 @@ index 6bf9f9c..ed543da 100644
3546- upgrade.Reconcile,
3647- restart.Reconcile,
3748- 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),
49+ + reconcilers.WithStallWatchdog(r.Logger, "tls", componentStallReportEvery, componentStallReportEvery , tls.Reconcile),
50+ + reconcilers.WithStallWatchdog(r.Logger, "securityconfig", componentStallReportEvery, componentStallReportEvery , securityconfig.Reconcile),
51+ + reconcilers.WithStallWatchdog(r.Logger, "config", componentStallReportEvery, componentStallReportEvery , config.Reconcile),
52+ + reconcilers.WithStallWatchdog(r.Logger, "cluster", componentStallReportEvery, componentStallReportEvery , cluster.Reconcile),
53+ + reconcilers.WithStallWatchdog(r.Logger, "scaler", componentStallReportEvery, componentStallReportEvery , scaler.Reconcile),
54+ + reconcilers.WithStallWatchdog(r.Logger, "dashboards", componentStallReportEvery, componentStallReportEvery , dashboards.Reconcile),
55+ + reconcilers.WithStallWatchdog(r.Logger, "upgrade", componentStallReportEvery, componentStallReportEvery , upgrade.Reconcile),
56+ + reconcilers.WithStallWatchdog(r.Logger, "restart", componentStallReportEvery, componentStallReportEvery , restart.Reconcile),
57+ + reconcilers.WithStallWatchdog(r.Logger, "snapshotrepository", componentStallReportEvery, componentStallReportEvery , snapshotrepository.Reconcile),
4758 }
4859 for _, rec := range componentReconcilers {
4960 result, err := rec()
5061diff --git a/opensearch-operator/pkg/reconcilers/stallwatchdog.go b/opensearch-operator/pkg/reconcilers/stallwatchdog.go
5162new file mode 100644
52- index 0000000..f01b989
63+ index 0000000..9822729
5364--- /dev/null
5465+++ b/opensearch-operator/pkg/reconcilers/stallwatchdog.go
55- @@ -0,0 +1,56 @@
66+ @@ -0,0 +1,73 @@
5667+ package reconcilers
5768+
5869+ import (
@@ -65,25 +76,32 @@ index 0000000..f01b989
6576+ // WithStallWatchdog wraps a component reconciler so a call that stops returning
6677+ // keeps saying so, under the name of the component it is stuck in.
6778+ //
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
79+ + // None of the components the cluster controller runs in sequence logs
80+ + // unconditionally when it is entered. The TLS one has a line above its work,
81+ + // but it sits on the early return for a cluster with no security section, which
82+ + // the chart in this repository never produces, and the rolling restart one
83+ + // opens at V(1), off at the verbosity the operator ships with. So a call that
84+ + // never comes back leaves nothing behind at all: the last line in the operator
85+ + // log stays whatever was written before the stall, the object stops being
7286+ // reconciled, and the two look identical from the outside. Logging every entry
7387+ // instead would put nine lines per cluster into the log every reconcile, so
7488+ // nothing is written while a component behaves.
7589+ //
7690+ // 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.
91+ + // during a stall says which component is in flight and for how long. The
92+ + // collector that reads this log in CI tails it at 100 lines, so a single line
93+ + // written when the stall began is exactly what gets cut.
7994+ func WithStallWatchdog(logger logr.Logger, name string, warnAfter, repeatEvery time.Duration, rec ComponentReconciler) ComponentReconciler {
8095+ return func() (reconcile.Result, error) {
8196+ done := make(chan struct{})
8297+ defer close(done)
8398+
84- + go func() {
85- + started := time.Now()
99+ + // Taken here rather than inside the goroutine: on a host loaded enough to
100+ + // be worth a stall report, the two differ by however long the scheduler
101+ + // took, which is measured into the one number the report carries.
102+ + started := time.Now()
86103+
104+ + go func() {
87105+ timer := time.NewTimer(warnAfter)
88106+ defer timer.Stop()
89107+ select {
@@ -95,9 +113,19 @@ index 0000000..f01b989
95113+ ticker := time.NewTicker(repeatEvery)
96114+ defer ticker.Stop()
97115+ for {
116+ + // A call that has returned is not reported as stuck: without this
117+ + // a tick that fired at the same moment as the return still writes
118+ + // its line, because a select with both cases ready picks either.
119+ + select {
120+ + case <-done:
121+ + return
122+ + default:
123+ + }
124+ +
98125+ logger.Info("component reconciler has not returned",
99126+ "component", name,
100127+ "elapsed", time.Since(started).String())
128+ +
101129+ select {
102130+ case <-done:
103131+ return
@@ -111,14 +139,15 @@ index 0000000..f01b989
111139+ }
112140diff --git a/opensearch-operator/pkg/reconcilers/stallwatchdog_test.go b/opensearch-operator/pkg/reconcilers/stallwatchdog_test.go
113141new file mode 100644
114- index 0000000..4314196
142+ index 0000000..a927f55
115143--- /dev/null
116144+++ b/opensearch-operator/pkg/reconcilers/stallwatchdog_test.go
117- @@ -0,0 +1,90 @@
145+ @@ -0,0 +1,155 @@
118146+ package reconcilers
119147+
120148+ import (
121149+ "errors"
150+ + "regexp"
122151+ "sync"
123152+ "time"
124153+
@@ -149,6 +178,8 @@ index 0000000..4314196
149178+
150179+ var errStallWatchdogTest = errors.New("component failed")
151180+
181+ + var elapsedPattern = regexp.MustCompile(`"elapsed"="([^"]+)"`)
182+ +
152183+ var _ = Describe("component reconciler stall watchdog", func() {
153184+ var sink *recordingSink
154185+
@@ -169,7 +200,12 @@ index 0000000..4314196
169200+ })
170201+
171202+ 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) {
203+ + // The call takes long enough for a watchdog that ignored warnAfter to
204+ + // get its report out, and warnAfter is far enough away that a loaded CI
205+ + // host cannot reach it: what is pinned is the threshold, not the speed
206+ + // of the scheduler.
207+ + wrapped := WithStallWatchdog(funcr.New(sink.write, funcr.Options{}), "cluster", 10*time.Second, 10*time.Second, func() (reconcile.Result, error) {
208+ + time.Sleep(150 * time.Millisecond)
173209+ return reconcile.Result{}, nil
174210+ })
175211+
@@ -179,6 +215,60 @@ index 0000000..4314196
179215+ Consistently(sink.snapshot, 200*time.Millisecond, 20*time.Millisecond).Should(BeEmpty())
180216+ })
181217+
218+ + It("drops the watchdog when the call returns before the threshold", func() {
219+ + wrapped := WithStallWatchdog(funcr.New(sink.write, funcr.Options{}), "cluster", 120*time.Millisecond, 120*time.Millisecond, func() (reconcile.Result, error) {
220+ + return reconcile.Result{}, nil
221+ + })
222+ +
223+ + _, err := wrapped()
224+ + Expect(err).ToNot(HaveOccurred())
225+ +
226+ + // Well past the threshold: a watchdog that does not notice the call
227+ + // returned would report here, once per wrapped call on every reconcile
228+ + // of every cluster, which is the cost this design claims not to have.
229+ + Consistently(sink.snapshot, 400*time.Millisecond, 40*time.Millisecond).Should(BeEmpty())
230+ + })
231+ +
232+ + It("keeps reporting while the call stays blocked", func() {
233+ + release := make(chan struct{})
234+ + returned := make(chan struct{})
235+ + wrapped := WithStallWatchdog(funcr.New(sink.write, funcr.Options{}), "restart", 10*time.Millisecond, 10*time.Millisecond, func() (reconcile.Result, error) {
236+ + <-release
237+ + return reconcile.Result{}, nil
238+ + })
239+ +
240+ + go func() {
241+ + defer GinkgoRecover()
242+ + defer close(returned)
243+ + _, err := wrapped()
244+ + Expect(err).ToNot(HaveOccurred())
245+ + }()
246+ +
247+ + // A single report would satisfy the spec above and still be lost: the
248+ + // collector that reads this log in CI tails it, so the line that
249+ + // matters is one written while somebody is looking, not the first one.
250+ + Eventually(func() int { return len(sink.snapshot()) }).Should(BeNumerically(">=", 3))
251+ +
252+ + // The report carries how long the call has been in flight, and that
253+ + // number has to move: a frozen or zero elapsed says nothing about
254+ + // whether the stall started a second or an hour ago.
255+ + elapsed := func() []string {
256+ + var out []string
257+ + for _, line := range sink.snapshot() {
258+ + if m := elapsedPattern.FindStringSubmatch(line); m != nil {
259+ + out = append(out, m[1])
260+ + }
261+ + }
262+ + return out
263+ + }
264+ + Expect(elapsed()).To(HaveLen(len(sink.snapshot())))
265+ + Expect(elapsed()[0]).ToNot(Equal("0s"))
266+ + Expect(elapsed()[1]).ToNot(Equal(elapsed()[0]))
267+ +
268+ + close(release)
269+ + Eventually(returned).Should(BeClosed())
270+ + })
271+ +
182272+ It("names the component while its call is still in flight", func() {
183273+ release := make(chan struct{})
184274+ returned := make(chan struct{})
@@ -200,7 +290,10 @@ index 0000000..4314196
200290+ Eventually(returned).Should(BeClosed())
201291+
202292+ // The goroutine has to stop with the call, or every stalled reconcile
203- + // leaves one behind writing for the life of the process.
293+ + // leaves one behind writing for the life of the process. A report that
294+ + // was already being written when the call returned is allowed to land,
295+ + // so the count is taken after a few intervals rather than immediately.
296+ + time.Sleep(50 * time.Millisecond)
204297+ settled := len(sink.snapshot())
205298+ Consistently(func() int { return len(sink.snapshot()) }, 100*time.Millisecond, 20*time.Millisecond).Should(Equal(settled))
206299+ })
0 commit comments