MB-46476: Force fetch cluster info after routine spawns

Backport of MB-46349

Change-Id: I0af895c22d79e68f4942866a85cd8c436b266cbc
Reviewed-on: http://review.couchbase.org/c/eventing/+/153766
Reviewed-by: <abhishek.jindal@couchbase.com>
Reviewed-by: CI Bot
Reviewed-by: Vinayaka Kamath <vinayaka.kamath@couchbase.com>
Tested-by: <ankit.prabhu@couchbase.com>
(cherry picked from commit 00e2bea)
Reviewed-on: http://review.couchbase.org/c/eventing/+/156308
Well-Formed: Restriction Checker

supervisor/util.go

@@ -265,6 +265,8 @@ func (s *SuperSupervisor) undeployFunctionsOnDeletedBkts(deletedBuckets []string
 }
 
 func (s *SuperSupervisor) watchBucketChanges() {
+	logPrefix := "SuperSupervisor::watchBucketChanges"
+
 	config := s.getConfig()
 	s.servicesNotifierRetryTm = 5
 	if tm, ok := config["service_notifier_timeout"]; ok {
@@ -293,6 +295,17 @@ func (s *SuperSupervisor) watchBucketChanges() {
 		go s.watchBucketChanges()
 	}
 
+	deletedBuckets, err := s.bucketRefresh(nil)
+        if err != nil {
+                logging.Errorf("%s Error in bucket Refresh: %v", logPrefix, err)
+                selfRestart()
+                return
+        }
+
+        if len(deletedBuckets) != 0 {
+                delChannel <- deletedBuckets
+        }
+
 	hostPortAddr := net.JoinHostPort(util.Localhost(), s.restPort)
 	clusterAuthURL, err := util.ClusterAuthUrl(hostPortAddr)
 	if err != nil {
@@ -318,6 +331,7 @@ func (s *SuperSupervisor) watchBucketChanges() {
 		select {
 		case notif, ok := <-ch:
 			if !ok {
+				logging.Errorf("%s ServicesChangeNotifier channel closed. Restarting..", logPrefix)
 				selfRestart()
 				return
 			}
@@ -331,7 +345,7 @@ func (s *SuperSupervisor) watchBucketChanges() {
 			np := notif.Msg.(*couchbase.Pool)
 			deletedBuckets, err := s.bucketRefresh(np)
 			if err != nil {
-				logging.Errorf("Refresh bucket Error")
+				logging.Errorf("%s Error in bucket Refresh: %v", logPrefix, err)
 				selfRestart()
 				return
 			}
@@ -343,7 +357,7 @@ func (s *SuperSupervisor) watchBucketChanges() {
 		case <-ticker.C:
 			deletedBuckets, err := s.bucketRefresh(nil)
 			if err != nil {
-				logging.Errorf("Refresh bucket Error")
+				logging.Errorf("%s Error in bucket Refresh: %v", logPrefix, err)
 				selfRestart()
 				return
 			}

util/cluster_info.go

@@ -748,11 +748,19 @@ func (c *ClusterInfoClient) GetClusterInfoCache() *ClusterInfoCache {
 }
 
 func (c *ClusterInfoClient) watchClusterChanges() {
+	logPrefix := "ClusterInfoClient::watchClusterChanges"
+
 	selfRestart := func() {
 		time.Sleep(time.Duration(c.servicesNotifierRetryTm) * time.Millisecond)
 		go c.watchClusterChanges()
 	}
 
+	if err := c.cinfo.FetchWithLock(false); err != nil {
+                logging.Errorf("cic.cinfo.FetchWithLock(): %v\n", err)
+                selfRestart()
+                return
+        }
+
 	clusterAuthURL, err := ClusterAuthUrl(c.clusterURL)
 	if err != nil {
 		logging.Errorf("ClusterInfoClient ClusterAuthUrl(): %v\n", err)
@@ -777,6 +785,7 @@ func (c *ClusterInfoClient) watchClusterChanges() {
 		select {
 		case notif, ok := <-ch:
 			if !ok {
+				logging.Errorf("%s ServicesChangeNotifier channel closed. Restarting..", logPrefix)
 				selfRestart()
 				return
 			}