feat: allow v2 volume to attach on the an old and running instance-ma…

…nager pod

Longhorn 9383

Signed-off-by: Derek Su <[email protected]>

datastore/longhorn.go

@@ -3752,30 +3752,39 @@ func (s *DataStore) GetInstanceManagerByInstanceRO(obj interface{}) (*longhorn.I
 		name       string // name of the object
 		nodeID     string
 		dataEngine longhorn.DataEngineType
+		image      string
 	)
 
 	switch obj := obj.(type) {
 	case *longhorn.Engine:
 		name = obj.Name
 		nodeID = obj.Spec.NodeID
 		dataEngine = obj.Spec.DataEngine
+		image = obj.Spec.Image
 	case *longhorn.Replica:
 		name = obj.Name
 		nodeID = obj.Spec.NodeID
 		dataEngine = obj.Spec.DataEngine
+		image = obj.Spec.Image
 	default:
 		return nil, fmt.Errorf("unknown type for GetInstanceManagerByInstance, %+v", obj)
 	}
 	if nodeID == "" {
 		return nil, fmt.Errorf("invalid request for GetInstanceManagerByInstance: no NodeID specified for instance %v", name)
 	}
 
-	image, err := s.GetSettingValueExisted(types.SettingNameDefaultInstanceManagerImage)
+	instanceManagerImage, err := s.GetSettingValueExisted(types.SettingNameDefaultInstanceManagerImage)
 	if err != nil {
 		return nil, err
 	}
 
-	imMap, err := s.ListInstanceManagersBySelectorRO(nodeID, image, longhorn.InstanceManagerTypeAllInOne, dataEngine)
+	// Because there is only one active instance manager image for v2 data engine,
+	// use spec.image as the instance manager image instead.
+	if dataEngine == longhorn.DataEngineTypeV2 {
+		instanceManagerImage = image
+	}
+
+	imMap, err := s.ListInstanceManagersBySelectorRO(nodeID, instanceManagerImage, longhorn.InstanceManagerTypeAllInOne, dataEngine)
 	if err != nil {
 		return nil, err
 	}
@@ -3785,7 +3794,8 @@ func (s *DataStore) GetInstanceManagerByInstanceRO(obj interface{}) (*longhorn.I
 		}
 
 	}
-	return nil, fmt.Errorf("cannot find the only available instance manager for instance %v, node %v, instance manager image %v, type %v", name, nodeID, image, longhorn.InstanceManagerTypeAllInOne)
+	return nil, fmt.Errorf("cannot find the only available instance manager for instance %v, node %v, instance manager image %v, type %v",
+		name, nodeID, instanceManagerImage, longhorn.InstanceManagerTypeAllInOne)
 }
 
 func (s *DataStore) ListInstanceManagersByNodeRO(node string, imType longhorn.InstanceManagerType, dataEngine longhorn.DataEngineType) (map[string]*longhorn.InstanceManager, error) {

webhook/resources/volume/mutator.go

@@ -46,6 +46,45 @@ func (v *volumeMutator) Resource() admission.Resource {
 	}
 }
 
+func (v *volumeMutator) areAllDefaultInstanceManagersStopped(defaultInstanceManagerImage string) (bool, error) {
+	ims, err := v.ds.ListInstanceManagersBySelectorRO("", defaultInstanceManagerImage, longhorn.InstanceManagerTypeAllInOne, longhorn.DataEngineTypeV2)
+	if err != nil {
+		return false, errors.Wrap(err, "failed to list instance managers")
+	}
+
+	for _, im := range ims {
+		if im.Status.CurrentState != longhorn.InstanceManagerStateStopped {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+func (v *volumeMutator) getActiveInstanceManagerImage(defaultInstanceManagerImage string) (string, error) {
+	// Check whether all default instance managers are stopped.
+	// If all default instance managers are stopped, we can use the current active instance manager image.
+	// If not, we use the default instance manager image because system is in the middle of upgrade.
+	allStopped, err := v.areAllDefaultInstanceManagersStopped(defaultInstanceManagerImage)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to check whether all default instance managers are stopped")
+	}
+
+	if allStopped {
+		// Get the active instance manager image from the non-default instance manager.
+		ims, err := v.ds.ListInstanceManagersBySelectorRO("", "", longhorn.InstanceManagerTypeAllInOne, longhorn.DataEngineTypeV2)
+		if err != nil {
+			return "", errors.Wrap(err, "failed to list instance managers")
+		}
+		for _, im := range ims {
+			if im.Spec.Image != defaultInstanceManagerImage {
+				return im.Spec.Image, nil
+			}
+		}
+	}
+
+	return defaultInstanceManagerImage, nil
+}
+
 func (v *volumeMutator) Create(request *admission.Request, newObj runtime.Object) (admission.PatchOps, error) {
 	volume, ok := newObj.(*longhorn.Volume)
 	if !ok {
@@ -169,6 +208,14 @@ func (v *volumeMutator) Create(request *admission.Request, newObj runtime.Object
 	if defaultImage == "" {
 		return nil, werror.NewInvalidError(fmt.Sprintf("invalid empty setting %s", defaultImageSetting), "")
 	}
+	if types.IsDataEngineV2(volume.Spec.DataEngine) {
+		activeInstanceManagerImage, err := v.getActiveInstanceManagerImage(defaultImage)
+		if err != nil {
+			return nil, werror.NewInvalidError(fmt.Sprintf("failed to get active instance manager image for volume %v: %v", name, err), "")
+		}
+		defaultImage = activeInstanceManagerImage
+	}
+
 	patchOps = append(patchOps, fmt.Sprintf(`{"op": "replace", "path": "/spec/image", "value": "%s"}`, defaultImage))
 
 	// Mutate the backup compression method to the default one