Move first ES cluster state observation out of go routine

[jiuker]

There is a process to execute in Start, but the execution point is uncertain. As a result, the lastHealth in the observer structure is unknown, and pkg/controller/elasticsearch/pdb/reconcile.go:196 returns 0. The probability of restarting the operator will cause *v1beta1.PodDisruptionBudgetStatus MinAvailable to be updated from 1 (it is the correct 1 before operator restart) to 2. After the observe is executed, the lastHealth is green, and the MinAvailable is updated from 2 to 1. Therefore, restart the operator *v1beta1.PodDisruptionBudgetStatus is updated twice.

Repro steps:

1. Add fmt.Println("update!") line 182 of pkg/controller/common/reconciler/reconciler.go`:
2. start operator
3. kubectl apply -f config/samples/elasticsearch/elasticsearch.yaml。
4. wait all pod ready
5. restart operator
6. output:

update!
...
update!

Resolves #5812.

[elasticmachine]

Since this is a community submitted pull request, a Jenkins build has not been kicked off automatically. Can an Elastic organization member please verify the contents of this patch and then kick off a build manually?

[thbkrkr]

Nice catch!

If you look at runPeriodically(), there was an intention to trigger a first observation before doing it periodically but it has been done before the go routine starts. So I would rather we fix runPeriodically() / Start(). I think we can merge the two because the indirection provided by runPeriodically() makes reading the flow a bit harder and let us make this bug exist, and then move the first observation out of the go routine, like this:

diff:

diff --git a/pkg/controller/elasticsearch/observer/observer.go b/pkg/controller/elasticsearch/observer/observer.go
index 1fb8bc8aa..31fd5d415 100644
--- a/pkg/controller/elasticsearch/observer/observer.go
+++ b/pkg/controller/elasticsearch/observer/observer.go
@@ -65,9 +65,26 @@ func NewObserver(cluster types.NamespacedName, esClient client.Client, settings
        return &observer
 }
 
-// Start the observer in a separate goroutine
+// Start starts the Observer in a separate goroutine.
+// The cluster state is observed periodically until the observer stop channel is closed.
 func (o *Observer) Start() {
-       go o.runPeriodically()
+       // initial synchronous observation
+       o.observe()
+       // periodic asynchronous observations
+       go func() {
+               ticker := time.NewTicker(o.settings.ObservationInterval)
+               defer ticker.Stop()
+
+               for {
+                       select {
+                       case <-ticker.C:
+                               o.observe()
+                       case <-o.stopChan:
+                               log.Info("Stopping observer for cluster", "namespace", o.cluster.Namespace, "es_name", o.cluster.Name)
+                               return
+                       }
+               }
+       }()
 }
 
 // Stop the observer loop
@@ -85,25 +102,6 @@ func (o *Observer) LastHealth() esv1.ElasticsearchHealth {
        return o.lastHealth
 }
 
-// runPeriodically triggers a state retrieval every tick,
-// until the given context is cancelled
-func (o *Observer) runPeriodically() {
-       o.observe()
-
-       ticker := time.NewTicker(o.settings.ObservationInterval)
-       defer ticker.Stop()
-
-       for {
-               select {
-               case <-ticker.C:
-                       o.observe()
-               case <-o.stopChan:
-                       log.Info("Stopping observer for cluster", "namespace", o.cluster.Namespace, "es_name", o.cluster.Name)
-                       return
-               }
-       }
-}
-

[cla-checker-service]

💚 CLA has been signed

[thbkrkr]

Jenkins test this please

This breaks the unit tests.

[jiuker]

I have make event chan asynchronous。But I don't know if it's consistent with the original logic。 @thbkrkr
In particular, I want to make it clear that this submission has modified the non test code. Please confirm carefully.

pkg/controller/elasticsearch/observer/watch.go
@@ -17,8 +17,9 @@ import (
 // whose health has changed between 2 observations.
 // Aimed to be used for triggering a reconciliation.
 func WatchClusterHealthChange(m *Manager) *source.Channel {
- 	evtChan := make(chan event.GenericEvent)
+	var evtChan chan event.GenericEvent
	m.AddObservationListener(healthChangeListener(evtChan))
+ 	evtChan = make(chan event.GenericEvent, len(m.listeners))
 	return &source.Channel{
 		// Each event in Source will be consumed and turned into
 		// a reconciliation request.

[thbkrkr]

I opened #5812 to discuss it because it is more involved than I thought.

[jiuker]

I modified the E2E. Becase our usage scenarios do not conform to the E2E. So there is a problem with the E2E test, and the program runs without problem. @thbkrkr
The receiver is asynchronous. Therefore, the E2E test should also be asynchronous reception. You can see it in detail
$GOPATH/go/pkg/mod/sigs. k8s. io/controller- runtime@v0.12.1/pkg/source/source. go:273

code

	// Start implements Source and should only be called by the Controller.
func (cs *Channel) Start(
	ctx context.Context,
	handler handler.EventHandler,
	queue workqueue.RateLimitingInterface,
	prct ...predicate.Predicate) error {
	// Source should have been specified by the user.
	if cs.Source == nil {
		return fmt.Errorf("must specify Channel.Source")
	}

	// stop should have been injected before Start was called
	if cs.stop == nil {
		return fmt.Errorf("must call InjectStop on Channel before calling Start")
	}

	// use default value if DestBufferSize not specified
	if cs.DestBufferSize == 0 {
		cs.DestBufferSize = defaultBufferSize
	}

	dst := make(chan event.GenericEvent, cs.DestBufferSize)

	cs.destLock.Lock()
	cs.dest = append(cs.dest, dst)
	cs.destLock.Unlock()

	cs.once.Do(func() {
		// Distribute GenericEvents to all EventHandler / Queue pairs Watching this source
		go cs.syncLoop(ctx)
	})

	go func() {
		for evt := range dst {
			shouldHandle := true
			for _, p := range prct {
				if !p.Generic(evt) {
					shouldHandle = false
					break
				}
			}

			if shouldHandle {
				handler.Generic(evt, queue)
			}
		}
	}()

	return nil
}

E2E diff

			eventsCluster2 <- cluster
 		}
 	})

+ 	doneCh := make(chan struct{})
+ 	go func() {
+ 		// events should be propagated to both listeners
+ 		<-eventsCluster1
+ 		<-eventsCluster2
+ 		<-eventsCluster1
+ 		<-eventsCluster2
+ 		doneCh <- struct{}{}
+ 	}()
 	// observe 2 clusters
 	obs1 := m.Observe(cluster1, fakeEsClient200(client.BasicAuth{}))
 	defer obs1.Stop()
 	obs2 := m.Observe(cluster2, fakeEsClient200(client.BasicAuth{}))
 	defer obs2.Stop()

- 	// events should be propagated to both listeners
- 	<-eventsCluster1
- 	<-eventsCluster2
- 	<-eventsCluster1
- 	<-eventsCluster2
+	<-doneCh
 }

 func esObject(n types.NamespacedName) esv1.Elasticsearch {

[jiuker]

It is possible that other test cases are affected. I should do more tests.

[jiuker]

I have tested package observer all test Fun。Make jenkins test again。 @thbkrkr

[jiuker]

review again @barkbay

[jiuker]

review again @thbkrkr

[thbkrkr]

Jenkins test this please

[barkbay]

LGTM

[barkbay]

I think this PR has actually introduced some performance issues described here: #6078