Refuse to create a duplicate engine

Longhorn 6682

Signed-off-by: Eric Weber <eric.weber@suse.com>
Signed-off-by: Phan Le <phan.le@suse.com>

datastore/uncached.go

@@ -3,17 +3,18 @@ package datastore
 import (
 	"context"
 
-	"k8s.io/apimachinery/pkg/runtime"
-
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	"github.com/longhorn/longhorn-manager/types"
 	"github.com/longhorn/longhorn-manager/util"
 
 	longhornapis "github.com/longhorn/longhorn-manager/k8s/pkg/apis/longhorn"
+	longhorn "github.com/longhorn/longhorn-manager/k8s/pkg/apis/longhorn/v1beta2"
 )
 
 // GetLonghornEventList returns an uncached list of longhorn events for the
@@ -344,6 +345,18 @@ func (s *DataStore) GetAllLonghornVolumes() (runtime.Object, error) {
 	return s.lhClient.LonghornV1beta2().Volumes(s.namespace).List(context.TODO(), metav1.ListOptions{})
 }
 
+// ListVolumeEnginesUncached returns an uncached list of Volume's engines in
+// Longhorn namespace directly from the API server.
+func (s *DataStore) ListVolumeEnginesUncached(volumeName string) ([]longhorn.Engine, error) {
+	engineList, err := s.lhClient.LonghornV1beta2().Engines(s.namespace).List(context.TODO(), metav1.ListOptions{
+		LabelSelector: labels.Set(types.GetVolumeLabels(volumeName)).String(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return engineList.Items, nil
+}
+
 // GetAllLonghornRecurringJobs returns an uncached list of RecurringJobs in
 // Longhorn namespace directly from the API server.
 // Using cached informers should be preferred but current lister doesn't have a

webhook/resources/engine/validator.go

@@ -43,6 +43,15 @@ func (e *engineValidator) Resource() admission.Resource {
 func (e *engineValidator) Create(request *admission.Request, newObj runtime.Object) error {
 	engine := newObj.(*longhorn.Engine)
 
+	volume, err := e.ds.GetVolume(engine.Spec.VolumeName)
+	if err != nil {
+		if datastore.ErrorIsNotFound(err) {
+			return werror.NewInvalidError("volume does not exist for engine", "spec.volumeName")
+		}
+		err = errors.Wrap(err, "failed to get volume for engine")
+		return werror.NewInternalError(err.Error())
+	}
+
 	if engine.Spec.BackendStoreDriver == longhorn.BackendStoreDriverTypeV2 {
 		v2DataEngineEnabled, err := e.ds.GetSettingAsBool(types.SettingNameV2DataEngine)
 		if err != nil {
@@ -54,7 +63,7 @@ func (e *engineValidator) Create(request *admission.Request, newObj runtime.Obje
 		}
 	}
 
-	return nil
+	return e.validateNumberOfEngines(engine, volume)
 }
 
 func (e *engineValidator) Update(request *admission.Request, oldObj runtime.Object, newObj runtime.Object) error {
@@ -70,3 +79,23 @@ func (e *engineValidator) Update(request *admission.Request, oldObj runtime.Obje
 
 	return nil
 }
+
+func (e *engineValidator) validateNumberOfEngines(newEngine *longhorn.Engine, volume *longhorn.Volume) error {
+	volumeEngines, err := e.ds.ListVolumeEnginesUncached(newEngine.Spec.VolumeName)
+	if err != nil {
+		err = errors.Wrap(err, "failed to list engines for volume")
+		return werror.NewInternalError(err.Error())
+	}
+
+	newNumVolumeEngines := len(volumeEngines) + 1
+	if volume.Spec.Migratable && newNumVolumeEngines > 2 {
+		message := fmt.Sprintf("engine creation would result in %d engines for migratable volume", newNumVolumeEngines)
+		return werror.NewInvalidError(message, "")
+	}
+	if !volume.Spec.Migratable && newNumVolumeEngines > 1 {
+		message := fmt.Sprintf("engine creation would result in %d engines for non-migratable volume", newNumVolumeEngines)
+		return werror.NewInvalidError(message, "")
+	}
+
+	return nil
+}