MB-47946 : Repair metadata handles, dcp streams, restart goroutines if

encryption level changes during lifecycle operation Change-Id: I17f75aa04ea52910f7af1016c9f631ac91181def Reviewed-on: http://review.couchbase.org/c/eventing/+/159915 Tested-by: <abhishek.jindal@couchbase.com> Reviewed-by: <abhishek.jindal@couchbase.com>
couchbase · Sep 2, 2021 · a6911ee · a6911ee
1 parent 9876db0
commit a6911ee
Show file tree

Hide file tree

Showing 17 changed files with 413 additions and 69 deletions.
diff --git a/common/common.go b/common/common.go
@@ -147,7 +147,10 @@ type SecuritySetting struct {
 	RootCAs            *x509.CertPool
 }
 
-var ErrRetryTimeout = errors.New("retry timeout")
+var (
+	ErrRetryTimeout           = errors.New("retry timeout")
+	ErrEncryptionLevelChanged = errors.New("Encryption Level changed during boostrap")
+)
 
 // EventingProducer interface to export functions from eventing_producer
 type EventingProducer interface {

diff --git a/consumer/bucket_ops.go b/consumer/bucket_ops.go
diff --git a/consumer/checkpoint.go b/consumer/checkpoint.go
@@ -55,28 +55,35 @@ func (c *Consumer) doLastSeqNoCheckpoint() {
 						continue
 					}
 					// Metadata blob doesn't exist probably the app is deployed for the first time.
+					var operr error
 					err = util.Retry(util.NewFixedBackoff(bucketOpRetryInterval), c.retryCount, getOpCallback,
-						c, c.producer.AddMetadataPrefix(vbKey), &vbBlob, &cas, true, &isNoEnt)
+						c, c.producer.AddMetadataPrefix(vbKey), &vbBlob, &cas, &operr, true, &isNoEnt)
 					if err == common.ErrRetryTimeout {
 						logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 						return
+					} else if operr == common.ErrEncryptionLevelChanged {
+						continue
 					}
 
 					if isNoEnt {
 						logging.Infof("%s [%s:%s:%d] vb: %d Creating the initial metadata blob entry",
 							logPrefix, c.workerName, c.tcpPort, c.Pid(), vb)
 
 						err = util.Retry(util.NewFixedBackoff(bucketOpRetryInterval), c.retryCount, recreateCheckpointBlobsFromVbStatsCallback, c,
-							c.producer.AddMetadataPrefix(vbKey), &vbBlob)
+							c.producer.AddMetadataPrefix(vbKey), &vbBlob, &operr)
 						if err == common.ErrRetryTimeout {
 							logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 							return
+						} else if operr == common.ErrEncryptionLevelChanged {
+							continue
 						}
 
 						err = c.updateCheckpointInfo(vbKey, vb, &vbBlob)
 						if err == common.ErrRetryTimeout {
 							logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 							return
+						} else if err == common.ErrEncryptionLevelChanged {
+							continue
 						}
 
 						continue
@@ -88,6 +95,8 @@ func (c *Consumer) doLastSeqNoCheckpoint() {
 						if err == common.ErrRetryTimeout {
 							logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 							return
+						} else if err == common.ErrEncryptionLevelChanged {
+							continue
 						}
 
 						continue
@@ -101,6 +110,8 @@ func (c *Consumer) doLastSeqNoCheckpoint() {
 						if err == common.ErrRetryTimeout {
 							logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 							return
+						} else if err == common.ErrEncryptionLevelChanged {
+							continue
 						}
 
 						continue
@@ -137,11 +148,14 @@ func (c *Consumer) updateCheckpointInfo(vbKey string, vb uint16, vbBlob *vbucket
 	vbBlob.VBuuid = c.vbProcessingStats.getVbStat(vb, "vb_uuid").(uint64)
 	vbBlob.ManifestUID = c.vbProcessingStats.getVbStat(vb, "manifest_id").(string)
 
+	var operr error
 	err := util.Retry(util.NewFixedBackoff(bucketOpRetryInterval), c.retryCount, periodicCheckpointCallback,
-		c, c.producer.AddMetadataPrefix(vbKey), vbBlob)
+		c, c.producer.AddMetadataPrefix(vbKey), vbBlob, &operr)
 	if err == common.ErrRetryTimeout {
 		logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 		return err
+	} else if operr == common.ErrEncryptionLevelChanged {
+		return operr
 	}
 
 	return nil

diff --git a/consumer/control_routine.go b/consumer/control_routine.go
@@ -138,6 +138,8 @@ func (c *Consumer) controlRoutine() error {
 					if err == common.ErrRetryTimeout {
 						logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 						return err
+					} else if err == common.ErrEncryptionLevelChanged {
+						continue
 					}
 
 					if err == nil {
@@ -198,6 +200,8 @@ func (c *Consumer) controlRoutine() error {
 				if err == common.ErrRetryTimeout {
 					logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 					return err
+				} else if err == common.ErrEncryptionLevelChanged {
+					continue
 				}
 			}
 
@@ -228,11 +232,14 @@ func (c *Consumer) controlRoutine() error {
 
 				logging.Infof("%s [%s:%s:%d] vb: %v, reclaiming it back by restarting dcp stream",
 					logPrefix, c.workerName, c.tcpPort, c.Pid(), vb)
+				var operr error
 				err := util.Retry(util.NewFixedBackoff(bucketOpRetryInterval), c.retryCount, getOpCallback,
-					c, c.producer.AddMetadataPrefix(vbKey), &vbBlob, &cas, true, &isNoEnt, true)
+					c, c.producer.AddMetadataPrefix(vbKey), &vbBlob, &cas, &operr, true, &isNoEnt, true)
 				if err == common.ErrRetryTimeout {
 					logging.Errorf("%s [%s:%s:%d] Exiting due to timeout", logPrefix, c.workerName, c.tcpPort, c.Pid())
 					return err
+				} else if operr == common.ErrEncryptionLevelChanged {
+					continue
 				}
 
 				err = c.updateVbOwnerAndStartDCPStream(vbKey, vb, &vbBlob)

diff --git a/consumer/defs.go b/consumer/defs.go
@@ -194,6 +194,7 @@ type Consumer struct {
 	inflightDcpStreamsRWMutex     *sync.RWMutex
 	ipcType                       string // ipc mechanism used to communicate with cpp workers - af_inet/af_unix
 	isBootstrapping               bool
+	initEncryptData               bool
 	isRebalanceOngoing            bool
 	isTerminateRunning            uint32                        // To signify if Consumer::Stop is running
 	kvHostDcpFeedMap              map[string]*couchbase.DcpFeed // Access controlled by hostDcpFeedRWMutex