webhook rules for sandbox shutdown

api/v1/helpers.go

@@ -206,6 +206,36 @@ func (v *VerticaDB) GenSubclusterSandboxStatusMap() map[string]string {
 	return scSbMap
 }
 
+// GenStatusSandboxMap() returns a map from status. The key is sandbox name and value is the sandbox pointer
+func (v *VerticaDB) GenStatusSandboxMap() map[string]*SandboxStatus {
+	statusSboxMap := make(map[string]*SandboxStatus)
+	for i := range v.Status.Sandboxes {
+		sBox := &v.Status.Sandboxes[i]
+		statusSboxMap[sBox.Name] = sBox
+	}
+	return statusSboxMap
+}
+
+// GenStatusSandboxMap() returns a map from status. The key is subcluster name and value is the subcluster pointer
+func (v *VerticaDB) GenStatusSubclusterMap() map[string]*SubclusterStatus {
+	statusSclusterMap := make(map[string]*SubclusterStatus)
+	for i := range v.Status.Subclusters {
+		sCluster := &v.Status.Subclusters[i]
+		statusSclusterMap[sCluster.Name] = sCluster
+	}
+	return statusSclusterMap
+}
+
+// GenStatusSClusterIndexMap will organize all of the subclusters into a map so we
+// can quickly find its index in the status.subclusters[] array.
+func (v *VerticaDB) GenStatusSClusterIndexMap() map[string]int {
+	m := make(map[string]int)
+	for i := range v.Status.Subclusters {
+		m[v.Status.Subclusters[i].Name] = i
+	}
+	return m
+}
+
 // GenSandboxSubclusterMapForUnsandbox will compare sandbox status and spec
 // for finding subclusters that need to be unsandboxed, this function returns a map
 // with sandbox name as the key and its subclusters (need to be unsandboxed) as the value

api/v1/verticadb_webhook.go

@@ -54,6 +54,7 @@ const (
 	S3Prefix              = "s3://"
 	GCloudPrefix          = "gs://"
 	AzurePrefix           = "azb://"
+	trueString            = "true"
 )
 
 // hdfsPrefixes are prefixes for an HDFS path.
@@ -105,7 +106,7 @@ func (v *VerticaDB) ValidateCreate() error {
 	verticadblog.Info("validate create", "name", v.Name, "GroupVersion", GroupVersion)
 
 	allErrs := v.validateVerticaDBSpec()
-	if allErrs == nil {
+	if len(allErrs) == 0 {
 		return nil
 	}
 	return apierrors.NewInvalid(schema.GroupKind{Group: Group, Kind: VerticaDBKind}, v.Name, allErrs)
@@ -116,7 +117,8 @@ func (v *VerticaDB) ValidateUpdate(old runtime.Object) error {
 	verticadblog.Info("validate update", "name", v.Name, "GroupVersion", GroupVersion)
 
 	allErrs := append(v.validateImmutableFields(old), v.validateVerticaDBSpec()...)
-	if allErrs == nil {
+
+	if len(allErrs) == 0 {
 		return nil
 	}
 	return apierrors.NewInvalid(schema.GroupKind{Group: Group, Kind: VerticaDBKind}, v.Name, allErrs)
@@ -148,6 +150,11 @@ func (v *VerticaDB) validateImmutableFields(old runtime.Object) field.ErrorList
 	allErrs = v.checkImmutableStsName(oldObj, allErrs)
 	allErrs = v.checkValidSubclusterTypeTransition(oldObj, allErrs)
 	allErrs = v.checkSandboxesDuringUpgrade(oldObj, allErrs)
+	allErrs = v.validateShutdownSandboxImage(old, allErrs)
+	allErrs = v.validateTerminatingSandboxes(old, allErrs)
+	allErrs = v.validateUnsandboxShutdownConditions(old, allErrs)
+	allErrs = v.validateAnnotatedSubclustersInShutdownSandbox(old, allErrs)
+	allErrs = v.validateNewSBoxOrSClusterShutdownUnset(allErrs)
 	return allErrs
 }
 
@@ -1591,6 +1598,23 @@ func (v *VerticaDB) findPersistScsInSandbox(oldObj *VerticaDB) map[string]int {
 	return persistScsWithSbIndex
 }
 
+// findPersistSandboxes returns a slice of sandbox names that are in both new and old vdb.
+func (v *VerticaDB) findPersistSandboxes(oldObj *VerticaDB) []string {
+	oldSandboxes := oldObj.Spec.Sandboxes
+	newSandboxes := v.Spec.Sandboxes
+	oldSandboxMap := make(map[string]int)
+	for _, oldSandbox := range oldSandboxes {
+		oldSandboxMap[oldSandbox.Name] = 1
+	}
+	persistSandboxes := make([]string, 0)
+	for _, newSandbox := range newSandboxes {
+		if _, ok := oldSandboxMap[newSandbox.Name]; ok {
+			persistSandboxes = append(persistSandboxes, newSandbox.Name)
+		}
+	}
+	return persistSandboxes
+}
+
 // checkImmutableSubclusterInSandbox ensures we do not scale and remove any subcluster that is in a sandbox
 func (v *VerticaDB) checkImmutableSubclusterInSandbox(oldObj *VerticaDB, allErrs field.ErrorList) field.ErrorList {
 	// if either old vdb or new vdb does not have any sandboxes, skip this check
@@ -1668,13 +1692,15 @@ func (v *VerticaDB) checkSandboxPrimary(allErrs field.ErrorList, oldObj *Vertica
 	oldSbMap := oldObj.GenSandboxMap()
 	newSbMap := v.GenSandboxMap()
 	for oldScName, oldSbName := range oldScInSandbox {
-		newSbName, newFound := newScInSandbox[oldScName]
 		sc := oldScMap[oldScName]
 
 		// sandbox is removed
 		if _, sandboxExist := newSbMap[oldSbName]; !sandboxExist {
 			continue
 		}
+
+		newSbName, newFound := newScInSandbox[oldScName]
+		// old sandbox still exits
 		if !newFound && sc.Type == SandboxPrimarySubcluster {
 			i := oldScIndexMap[oldScName]
 			err := field.Invalid(path.Index(i),
@@ -1683,11 +1709,12 @@ func (v *VerticaDB) checkSandboxPrimary(allErrs field.ErrorList, oldObj *Vertica
 			allErrs = append(allErrs, err)
 			continue
 		}
-
+		// old sandbox exists and subcluster either exists (primary or nonprimary) or removed (nonprimary)
 		// Remaining check is concerned with subclusters moving between sandboxes.
 		if oldSbName == newSbName {
 			continue
 		}
+		// old sandbox name does not match new sandbox name
 		if sc.Type == SandboxPrimarySubcluster {
 			i := oldSbIndexMap[oldSbName]
 			p := field.NewPath("spec").Child("sandboxes")
@@ -1701,6 +1728,194 @@ func (v *VerticaDB) checkSandboxPrimary(allErrs field.ErrorList, oldObj *Vertica
 	return allErrs
 }
 
+// ensures a new subcluster or sandbox to be added to a vdb has Shutdown field set to false
+func (v *VerticaDB) validateNewSBoxOrSClusterShutdownUnset(allErrs field.ErrorList) field.ErrorList {
+	statusSClusterMap := v.GenStatusSubclusterMap()
+	statusSBoxMap := v.GenStatusSandboxMap()
+	for i := range v.Spec.Sandboxes {
+		newSBox := v.Spec.Sandboxes[i]
+		if _, foundInStatus := statusSBoxMap[newSBox.Name]; !foundInStatus && newSBox.Shutdown {
+			p := field.NewPath("spec").Child("sandboxes").Index(i)
+			err := field.Invalid(p,
+				newSBox.Shutdown,
+				fmt.Sprintf("shutdown must be false when adding sandbox %q",
+					newSBox.Name))
+			allErrs = append(allErrs, err)
+		}
+	}
+	for i := range v.Spec.Subclusters {
+		newSCluster := &v.Spec.Subclusters[i]
+		if _, foundInStatus := statusSClusterMap[newSCluster.Name]; !foundInStatus && newSCluster.Shutdown {
+			p := field.NewPath("spec").Child("subclusters").Index(i)
+			err := field.Invalid(p,
+				newSCluster.Shutdown,
+				fmt.Sprintf("shutdown must be false when adding subcluster %q",
+					newSCluster.Name))
+			allErrs = append(allErrs, err)
+		}
+	}
+	return allErrs
+}
+
+// return a map whose key is the pointer of a subcluster that is to be unsandboxed. The value is the pointer to a sandbox
+// where the subcluster lives in
+func (v *VerticaDB) findSclustersToUnsandbox(old runtime.Object) map[*Subcluster]*Sandbox {
+	oldObj := old.(*VerticaDB)
+	oldSubclusterInSandbox := oldObj.GenSubclusterSandboxMap()
+	newSuclusterInSandbox := v.GenSubclusterSandboxMap()
+	oldSubclusterMap := oldObj.GenSubclusterMap()
+	oldSandboxMap := oldObj.GenSandboxMap()
+	newSubclusterMap := v.GenSubclusterMap()
+	sclusterSboxMap := make(map[*Subcluster]*Sandbox)
+	for oldSubclusterName, oldSandboxName := range oldSubclusterInSandbox {
+		_, oldSubclusterInNewSandboxes := newSuclusterInSandbox[oldSubclusterName]
+		_, oldSubclusterInPersist := newSubclusterMap[oldSubclusterName]
+		// for unsandboxing, check shutdown field of the subcluster and sandbox
+		if !oldSubclusterInNewSandboxes && oldSubclusterInPersist {
+			oldSubcluster := oldSubclusterMap[oldSubclusterName]
+			oldSandbox := oldSandboxMap[oldSandboxName]
+			sclusterSboxMap[oldSubcluster] = oldSandbox
+		}
+	}
+	return sclusterSboxMap
+}
+
+// validateUnsandboxShutdownConditions ensures we will not unsandbox a subcluster that has shutdown set to true, or a subcluster
+// in a sandbox where the shutdown field for the sandbox is set to true
+func (v *VerticaDB) validateUnsandboxShutdownConditions(old runtime.Object, allErrs field.ErrorList) field.ErrorList {
+	statusSClusterIndexMap := v.GenStatusSClusterIndexMap()
+	sclustersToUnsandbox := v.findSclustersToUnsandbox(old)
+	for oldSubcluster, oldSandbox := range sclustersToUnsandbox {
+		oldSubclusterIndex := statusSClusterIndexMap[oldSubcluster.Name]
+		if oldSubcluster.Shutdown || v.Status.Subclusters[oldSubclusterIndex].Shutdown {
+			p := field.NewPath("spec").Child("subclusters")
+			err := field.Invalid(p,
+				oldSubcluster.Name,
+				fmt.Sprintf("cannot unsandbox subcluster %q that has Shutdown field set to true",
+					oldSubcluster.Name))
+			allErrs = append(allErrs, err)
+			continue
+		}
+		if oldSandbox.Shutdown {
+			p := field.NewPath("spec").Child("sandboxes")
+			err := field.Invalid(p,
+				oldSubcluster.Name,
+				fmt.Sprintf("cannot unsandbox subcluster %q in sandbox %q that has Shutdown field set to true",
+					oldSubcluster.Name, oldSandbox.Name))
+			allErrs = append(allErrs, err)
+			continue
+		}
+	}
+	return allErrs
+}
+
+// ensure: when a sandbox is to be shutdown, if a subcluster in it has annotation vertica.com/shutdown-driven-by-sandbox set to true, the
+// subcluster's shutdown field should not be updated
+func (v *VerticaDB) validateAnnotatedSubclustersInShutdownSandbox(old runtime.Object, allErrs field.ErrorList) field.ErrorList {
+	oldObj := old.(*VerticaDB)
+	persistSandboxes := v.findPersistSandboxes(oldObj)
+	newSandboxMap := v.GenSandboxMap()
+	oldSandboxMap := oldObj.GenSandboxMap()
+	newSubclusterIndexMap := v.GenSubclusterIndexMap()
+	newSubclusterMap := v.GenSubclusterMap()
+	oldSubclusterMap := oldObj.GenSubclusterMap()
+	for _, sandboxName := range persistSandboxes {
+		newSandbox := newSandboxMap[sandboxName]
+		oldSandbox := oldSandboxMap[sandboxName]
+		if newSandbox.Shutdown {
+			for _, subclusterName := range oldSandbox.Subclusters {
+				oldSubcluster := oldSubclusterMap[subclusterName.Name]
+				if vmeta.GetShutdownDrivenBySandbox(oldSubcluster.Annotations) {
+					newSubcluster, oldSclusterPersist := newSubclusterMap[subclusterName.Name]
+					if oldSclusterPersist && oldSubcluster.Shutdown != newSubcluster.Shutdown {
+						index := newSubclusterIndexMap[subclusterName.Name]
+						p := field.NewPath("spec").Child("subclusters")
+						err := field.Invalid(p.Index(index),
+							subclusterName.Name,
+							fmt.Sprintf("cannot change Shutdown field for subcluster %q that has annotation \"vertica.com/shutdown-driven-by-sandbox\" set to true",
+								subclusterName.Name))
+						allErrs = append(allErrs, err)
+					}
+				}
+			}
+		}
+	}
+	return allErrs
+}
+
+// ensures a sandbox's image is not updated after the sandbox or any of its subclusters are shut down
+func (v *VerticaDB) validateShutdownSandboxImage(old runtime.Object, allErrs field.ErrorList) field.ErrorList {
+	oldObj := old.(*VerticaDB)
+	persistSandboxes := v.findPersistSandboxes(oldObj)
+	oldSandboxMap := oldObj.GenSandboxMap()
+	newSandboxMap := v.GenSandboxMap()
+	newSandboxIndexMap := v.GenSandboxIndexMap()
+	newSubclusterMap := v.GenSubclusterMap()
+	statusSClusterIndexMap := v.GenStatusSClusterIndexMap()
+	for _, sandboxName := range persistSandboxes {
+		oldSandbox := oldSandboxMap[sandboxName]
+		newSandbox := newSandboxMap[sandboxName]
+		if oldSandbox.Image != newSandbox.Image {
+			newSandboxIndex := newSandboxIndexMap[sandboxName]
+			errMsg := v.checkSboxForShutdown(newSandbox, newSubclusterMap, statusSClusterIndexMap)
+			if errMsg != "" {
+				p := field.NewPath("spec").Child("sandboxes").Index(newSandboxIndex)
+				err := field.Invalid(p.Child("image"),
+					sandboxName,
+					fmt.Sprintf("cannot change the image for sandbox %q because %q",
+						sandboxName, errMsg))
+				allErrs = append(allErrs, err)
+			}
+		}
+	}
+	return allErrs
+}
+
+// ensure the sandbox to terminate is not shut down
+func (v *VerticaDB) validateTerminatingSandboxes(old runtime.Object, allErrs field.ErrorList) field.ErrorList {
+	oldObj := old.(*VerticaDB)
+	oldSandboxMap := oldObj.GenSandboxMap()
+	newSandboxMap := v.GenSandboxMap()
+	sandboxesToBeRemoved := vutil.MapKeyDiff(oldSandboxMap, newSandboxMap)
+	newSubclusterMap := v.GenSubclusterMap()
+	for _, sandboxToBeRemoved := range sandboxesToBeRemoved {
+		numOfSclustersInSbox := len(oldSandboxMap[sandboxToBeRemoved].Subclusters)
+		sclustersToTerminate := numOfSclustersInSbox
+		for _, subcluster := range oldSandboxMap[sandboxToBeRemoved].Subclusters {
+			if _, ok := newSubclusterMap[subcluster.Name]; ok {
+				sclustersToTerminate--
+			}
+		}
+		// sandboxToBeRemoved and all its subclusters are not found in new spec
+		if sclustersToTerminate == numOfSclustersInSbox {
+			if oldSandboxMap[sandboxToBeRemoved].Shutdown {
+				p := field.NewPath("spec").Child("sandboxes")
+				err := field.Invalid(p,
+					sandboxToBeRemoved,
+					fmt.Sprintf("cannot terminate sandbox %q that has Shutdown field set to true",
+						sandboxToBeRemoved))
+				allErrs = append(allErrs, err)
+			} // else looks good
+		} // else is expected
+	}
+	return allErrs
+}
+
+// checkSboxForShutdown checks if sbox or its scluster has shutdown field set to true in spec/status
+func (v *VerticaDB) checkSboxForShutdown(newSandbox *Sandbox, newSClusterMap map[string]*Subcluster, statusSClusterIndexMap map[string]int) string {
+	if newSandbox.Shutdown {
+		return fmt.Sprintf("Shutdown field of sandbox %q is set to true", newSandbox.Name)
+	}
+
+	for _, subcluster := range newSandbox.Subclusters {
+		i, foundInStatus := statusSClusterIndexMap[subcluster.Name]
+		if foundInStatus && (newSClusterMap[subcluster.Name].Shutdown || v.Status.Subclusters[i].Shutdown) {
+			return fmt.Sprintf("Subcluster %q is marked for shut down or has already been shut down", subcluster.Name)
+		}
+	}
+	return ""
+}
+
 // checkImmutableStsName ensures the statefulset name of the subcluster stays constant
 func (v *VerticaDB) checkImmutableStsName(oldObj *VerticaDB, allErrs field.ErrorList) field.ErrorList {
 	// We have an annotation to control the sts name. We are not going to allow