feat: v2 data engine live upgrade

app/daemon.go

@@ -169,7 +169,7 @@ func startManager(c *cli.Context) error {
 	if err != nil {
 		return err
 	}
-	if err := webhook.StartWebhook(ctx, types.WebhookTypeConversion, clientsWithoutDatastore); err != nil {
+	if err := webhook.StartWebhook(ctx, currentNodeID, types.WebhookTypeConversion, clientsWithoutDatastore); err != nil {
 		return err
 	}
 
@@ -195,7 +195,7 @@ func startManager(c *cli.Context) error {
 		return err
 	}
 
-	if err := webhook.StartWebhook(ctx, types.WebhookTypeAdmission, clients); err != nil {
+	if err := webhook.StartWebhook(ctx, currentNodeID, types.WebhookTypeAdmission, clients); err != nil {
 		return err
 	}
 	if err := clients.Datastore.AddLabelToManagerPod(currentNodeID, types.GetAdmissionWebhookLabel()); err != nil {

controller/backup_controller.go

@@ -596,6 +596,14 @@ func (bc *BackupController) isResponsibleFor(b *longhorn.Backup, defaultEngineIm
 		err = errors.Wrap(err, "error while checking isResponsibleFor")
 	}()
 
+	upgradeRequested, err := bc.ds.IsNodeDataEngineUpgradeRequested(bc.controllerID)
+	if err != nil {
+		return false, err
+	}
+	if upgradeRequested {
+		return false, nil
+	}
+
 	isResponsible := isControllerResponsibleFor(bc.controllerID, bc.ds, b.Name, "", b.Status.OwnerID)
 
 	currentOwnerEngineAvailable, err := bc.ds.CheckEngineImageReadiness(defaultEngineImage, b.Status.OwnerID)

controller/controller_manager.go

@@ -41,7 +41,7 @@ func StartControllers(logger logrus.FieldLogger, clients *client.Clients,
 	stopCh := clients.StopCh
 
 	// Longhorn controllers
-	replicaController, err := NewReplicaController(logger, ds, scheme, kubeClient, namespace, controllerID)
+	replicaController, err := NewReplicaController(logger, ds, scheme, kubeClient, namespace, controllerID, proxyConnCounter)
 	if err != nil {
 		return nil, err
 	}
@@ -154,6 +154,16 @@ func StartControllers(logger logrus.FieldLogger, clients *client.Clients,
 		return nil, err
 	}
 
+	dataEngineUpgradeManagerController, err := NewDataEngineUpgradeManagerController(logger, ds, scheme, kubeClient, controllerID, namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	nodeDataEngineUpgradeController, err := NewNodeDataEngineUpgradeController(logger, ds, scheme, kubeClient, controllerID, namespace)
+	if err != nil {
+		return nil, err
+	}
+
 	// Kubernetes controllers
 	kubernetesPVController, err := NewKubernetesPVController(logger, ds, scheme, kubeClient, controllerID)
 	if err != nil {
@@ -213,6 +223,9 @@ func StartControllers(logger logrus.FieldLogger, clients *client.Clients,
 	go volumeEvictionController.Run(Workers, stopCh)
 	go volumeCloneController.Run(Workers, stopCh)
 	go volumeExpansionController.Run(Workers, stopCh)
+	// The two controllers do not support upgrade in parallel
+	go dataEngineUpgradeManagerController.Run(1, stopCh)
+	go nodeDataEngineUpgradeController.Run(1, stopCh)
 
 	// Start goroutines for Kubernetes controllers
 	go kubernetesPVController.Run(Workers, stopCh)