feat: support snapshot max count and size

Signed-off-by: PoAn Yang <poan.yang@suse.com>

api/model.go

@@ -330,6 +330,14 @@ type UpdateReplicaDiskSoftAntiAffinityInput struct {
 	ReplicaDiskSoftAntiAffinity string `json:"replicaDiskSoftAntiAffinity"`
 }
 
+type UpdateSnapshotMaxCount struct {
+	SnapshotMaxCount int `json:"snapshotMaxCount"`
+}
+
+type UpdateSnapshotMaxSize struct {
+	SnapshotMaxSize int64 `json:"snapshotMaxSize"`
+}
+
 type PVCreateInput struct {
 	PVName string `json:"pvName"`
 	FSType string `json:"fsType"`

api/router.go

@@ -75,6 +75,8 @@ func NewRouter(s *Server) *mux.Router {
 		"updateDataLocality":                s.VolumeUpdateDataLocality,
 		"updateAccessMode":                  s.VolumeUpdateAccessMode,
 		"updateUnmapMarkSnapChainRemoved":   s.VolumeUpdateUnmapMarkSnapChainRemoved,
+		"updateSnapshotMaxCount":            s.VolumeUpdateSnapshotMaxCount,
+		"updateSnapshotMaxSize":             s.VolumeUpdateSnapshotMaxSize,
 		"updateReplicaSoftAntiAffinity":     s.VolumeUpdateReplicaSoftAntiAffinity,
 		"updateReplicaZoneSoftAntiAffinity": s.VolumeUpdateReplicaZoneSoftAntiAffinity,
 		"updateReplicaDiskSoftAntiAffinity": s.VolumeUpdateReplicaDiskSoftAntiAffinity,

api/volume.go

@@ -779,3 +779,47 @@ func (s *Server) EngineUpgrade(rw http.ResponseWriter, req *http.Request) error
 
 	return s.responseWithVolume(rw, req, id, v)
 }
+
+func (s *Server) VolumeUpdateSnapshotMaxCount(rw http.ResponseWriter, req *http.Request) error {
+	var input UpdateSnapshotMaxCount
+	id := mux.Vars(req)["name"]
+
+	apiContext := api.GetApiContext(req)
+	if err := apiContext.Read(&input); err != nil {
+		return errors.Wrap(err, "failed to read SnapshotMaxCount input")
+	}
+
+	obj, err := util.RetryOnConflictCause(func() (interface{}, error) {
+		return s.m.UpdateSnapshotMaxCount(id, input.SnapshotMaxCount)
+	})
+	if err != nil {
+		return err
+	}
+	v, ok := obj.(*longhorn.Volume)
+	if !ok {
+		return fmt.Errorf("failed to convert to volume %v object", id)
+	}
+	return s.responseWithVolume(rw, req, "", v)
+}
+
+func (s *Server) VolumeUpdateSnapshotMaxSize(rw http.ResponseWriter, req *http.Request) error {
+	var input UpdateSnapshotMaxSize
+	id := mux.Vars(req)["name"]
+
+	apiContext := api.GetApiContext(req)
+	if err := apiContext.Read(&input); err != nil {
+		return errors.Wrap(err, "failed to read SnapshotMaxSize input")
+	}
+
+	obj, err := util.RetryOnConflictCause(func() (interface{}, error) {
+		return s.m.UpdateSnapshotMaxSize(id, input.SnapshotMaxSize)
+	})
+	if err != nil {
+		return err
+	}
+	v, ok := obj.(*longhorn.Volume)
+	if !ok {
+		return fmt.Errorf("failed to convert to volume %v object", id)
+	}
+	return s.responseWithVolume(rw, req, "", v)
+}

controller/engine_controller.go

@@ -972,6 +972,24 @@ func (m *EngineMonitor) refresh(engine *longhorn.Engine) error {
 						volumeInfo.UnmapMarkSnapChainRemoved, engine.Spec.UnmapMarkSnapChainRemovedEnabled)
 				}
 			}
+
+			engine.Status.SnapshotMaxCount = volumeInfo.MaximumSnapshotCount
+			if engine.Spec.SnapshotMaxCount != volumeInfo.MaximumSnapshotCount {
+				logrus.Infof("Correcting flag SnapshotMaxCount from %d to %d", volumeInfo.MaximumSnapshotCount, engine.Spec.SnapshotMaxCount)
+				if err := engineClientProxy.VolumeMaximumSnapshotCountSet(engine); err != nil {
+					return errors.Wrapf(err, "failed to correct flag SnapshotMaxCount from %d to %d",
+						volumeInfo.MaximumSnapshotCount, engine.Spec.SnapshotMaxCount)
+				}
+			}
+
+			engine.Status.SnapshotMaxSize = volumeInfo.MaximumTotalSnapshotSize
+			if engine.Spec.SnapshotMaxSize != volumeInfo.MaximumTotalSnapshotSize {
+				logrus.Infof("Correcting flag SnapshotMaxSize from %d to %d", volumeInfo.MaximumTotalSnapshotSize, engine.Spec.SnapshotMaxSize)
+				if err := engineClientProxy.VolumeMaximumTotalSnapshotSizeSet(engine); err != nil {
+					return errors.Wrapf(err, "failed to correct flag SnapshotMaxSize from %d to %d",
+						volumeInfo.MaximumTotalSnapshotSize, engine.Spec.SnapshotMaxSize)
+				}
+			}
 		}
 	} else {
 		// For incompatible running engine, the current size is always `engine.Spec.VolumeSize`.

controller/volume_controller.go

@@ -455,6 +455,10 @@ func (c *VolumeController) syncVolume(key string) (err error) {
 		return err
 	}
 
+	if err := c.syncVolumeSnapshotSetting(volume, engines, replicas); err != nil {
+		return err
+	}
+
 	if err := c.updateRecurringJobs(volume); err != nil {
 		return err
 	}
@@ -1164,6 +1168,23 @@ func (c *VolumeController) syncVolumeUnmapMarkSnapChainRemovedSetting(v *longhor
 	return nil
 }
 
+func (c *VolumeController) syncVolumeSnapshotSetting(v *longhorn.Volume, es map[string]*longhorn.Engine, rs map[string]*longhorn.Replica) error {
+	if es == nil && rs == nil {
+		return nil
+	}
+
+	for _, e := range es {
+		e.Spec.SnapshotMaxCount = v.Spec.SnapshotMaxCount
+		e.Spec.SnapshotMaxSize = v.Spec.SnapshotMaxSize
+	}
+	for _, r := range rs {
+		r.Spec.SnapshotMaxCount = v.Spec.SnapshotMaxCount
+		r.Spec.SnapshotMaxSize = v.Spec.SnapshotMaxSize
+	}
+
+	return nil
+}
+
 // ReconcileVolumeState handles the attaching and detaching of volume
 func (c *VolumeController) ReconcileVolumeState(v *longhorn.Volume, es map[string]*longhorn.Engine, rs map[string]*longhorn.Replica) (err error) {
 	defer func() {
@@ -3314,6 +3335,8 @@ func (c *VolumeController) createEngine(v *longhorn.Volume, isNewEngine bool) (*
 			ReplicaAddressMap:         map[string]string{},
 			UpgradedReplicaAddressMap: map[string]string{},
 			RevisionCounterDisabled:   v.Spec.RevisionCounterDisabled,
+			SnapshotMaxCount:          v.Spec.SnapshotMaxCount,
+			SnapshotMaxSize:           v.Spec.Size,
 		},
 	}
 
@@ -3365,6 +3388,8 @@ func (c *VolumeController) createReplica(v *longhorn.Volume, e *longhorn.Engine,
 			HardNodeAffinity:                 hardNodeAffinity,
 			RevisionCounterDisabled:          v.Spec.RevisionCounterDisabled,
 			UnmapMarkDiskChainRemovedEnabled: e.Spec.UnmapMarkSnapChainRemovedEnabled,
+			SnapshotMaxCount:                 v.Spec.SnapshotMaxCount,
+			SnapshotMaxSize:                  v.Spec.SnapshotMaxSize,
 		},
 	}
 	if isRebuildingReplica {

datastore/longhorn.go

@@ -373,6 +373,14 @@ func (s *DataStore) ValidateSetting(name, value string) (err error) {
 				return err
 			}
 		}
+	case types.SettingNameSnapshotMaxCount:
+		v, err := strconv.Atoi(value)
+		if err != nil {
+			return err
+		}
+		if v < 2 || v > 250 {
+			return fmt.Errorf("%s should be between 2 and 250", name)
+		}
 	}
 	return nil
 }

engineapi/engine.go

@@ -289,6 +289,26 @@ func (e *EngineBinary) VolumeUnmapMarkSnapChainRemovedSet(engine *longhorn.Engin
 	return nil
 }
 
+// VolumeMaximumSnapshotCountSet calls engine binary
+// TODO: Deprecated, replaced by gRPC proxy
+func (e *EngineBinary) VolumeMaximumSnapshotCountSet(engine *longhorn.Engine) error {
+	cmdline := []string{"snapshot-max-count", strconv.Itoa(engine.Spec.SnapshotMaxCount)}
+	if _, err := e.ExecuteEngineBinary(cmdline...); err != nil {
+		return errors.Wrapf(err, "error setting volume flag SnapshotMaxCount to %d", engine.Spec.SnapshotMaxCount)
+	}
+	return nil
+}
+
+// VolumeMaximumTotalSnapshotSizeSet calls engine binary
+// TODO: Deprecated, replaced by gRPC proxy
+func (e *EngineBinary) VolumeMaximumTotalSnapshotSizeSet(engine *longhorn.Engine) error {
+	cmdline := []string{"snapshot-max-size", strconv.FormatInt(engine.Spec.SnapshotMaxSize, 10)}
+	if _, err := e.ExecuteEngineBinary(cmdline...); err != nil {
+		return errors.Wrapf(err, "error setting volume flag SnapshotMaxSize to %d", engine.Spec.SnapshotMaxSize)
+	}
+	return nil
+}
+
 // ReplicaRebuildVerify calls engine binary
 // TODO: Deprecated, replaced by gRPC proxy
 func (e *EngineBinary) ReplicaRebuildVerify(engine *longhorn.Engine, replicaName, url string) error {

engineapi/enginesim.go

@@ -245,6 +245,14 @@ func (e *EngineSimulator) VolumeUnmapMarkSnapChainRemovedSet(*longhorn.Engine) e
 	return fmt.Errorf(ErrNotImplement)
 }
 
+func (e *EngineSimulator) VolumeMaximumSnapshotCountSet(*longhorn.Engine) error {
+	return fmt.Errorf(ErrNotImplement)
+}
+
+func (e *EngineSimulator) VolumeMaximumTotalSnapshotSizeSet(*longhorn.Engine) error {
+	return fmt.Errorf(ErrNotImplement)
+}
+
 func (e *EngineSimulator) ReplicaRebuildVerify(engine *longhorn.Engine, replicaName, url string) error {
 	return fmt.Errorf(ErrNotImplement)
 }

engineapi/instance_manager.go

@@ -338,6 +338,8 @@ func getBinaryAndArgsForEngineProcessCreation(e *longhorn.Engine,
 
 	if engineCLIAPIVersion >= 9 {
 		args = append([]string{"--engine-instance-name", e.Name}, args...)
+		args = append(args, "--snapshot-max-count", strconv.Itoa(e.Spec.SnapshotMaxCount))
+		args = append(args, "--snapshot-max-size", strconv.FormatInt(e.Spec.SnapshotMaxSize, 10))
 	}
 
 	for _, addr := range e.Status.CurrentReplicaAddressMap {
@@ -379,6 +381,8 @@ func getBinaryAndArgsForReplicaProcessCreation(r *longhorn.Replica,
 	if engineCLIAPIVersion >= 9 {
 		args = append(args, "--replica-instance-name", r.Name)
 		args = append([]string{"--volume-name", r.Spec.VolumeName}, args...)
+		args = append(args, "--snapshot-max-count", strconv.Itoa(r.Spec.SnapshotMaxCount))
+		args = append(args, "--snapshot-max-size", strconv.FormatInt(r.Spec.SnapshotMaxSize, 10))
 	}
 
 	// 3 ports are already used by replica server, data server and syncagent server

engineapi/proxy_volume.go

@@ -43,3 +43,13 @@ func (p *Proxy) VolumeUnmapMarkSnapChainRemovedSet(e *longhorn.Engine) error {
 	return p.grpcClient.VolumeUnmapMarkSnapChainRemovedSet(string(e.Spec.BackendStoreDriver), e.Name, e.Spec.VolumeName,
 		p.DirectToURL(e), e.Spec.UnmapMarkSnapChainRemovedEnabled)
 }
+
+func (p *Proxy) VolumeMaximumSnapshotCountSet(e *longhorn.Engine) error {
+	return p.grpcClient.VolumeMaximumSnapshotCountSet(string(e.Spec.BackendStoreDriver), e.Name, e.Spec.VolumeName,
+		p.DirectToURL(e), e.Spec.SnapshotMaxCount)
+}
+
+func (p *Proxy) VolumeMaximumTotalSnapshotSizeSet(e *longhorn.Engine) error {
+	return p.grpcClient.VolumeMaximumTotalSnapshotSizeSet(string(e.Spec.BackendStoreDriver), e.Name, e.Spec.VolumeName,
+		p.DirectToURL(e), e.Spec.SnapshotMaxSize)
+}

engineapi/types.go

@@ -78,6 +78,8 @@ type EngineClient interface {
 	VolumeFrontendStart(*longhorn.Engine) error
 	VolumeFrontendShutdown(*longhorn.Engine) error
 	VolumeUnmapMarkSnapChainRemovedSet(engine *longhorn.Engine) error
+	VolumeMaximumSnapshotCountSet(engine *longhorn.Engine) error
+	VolumeMaximumTotalSnapshotSizeSet(engine *longhorn.Engine) error
 
 	ReplicaList(*longhorn.Engine) (map[string]*Replica, error)
 	ReplicaAdd(engine *longhorn.Engine, replicaName, url string, isRestoreVolume, fastSync bool, replicaFileSyncHTTPClientTimeout int64) error
@@ -129,6 +131,8 @@ type Volume struct {
 	LastExpansionError        string `json:"lastExpansionError"`
 	LastExpansionFailedAt     string `json:"lastExpansionFailedAt"`
 	UnmapMarkSnapChainRemoved bool   `json:"unmapMarkSnapChainRemoved"`
+	MaximumSnapshotCount      int    `json:"maximumSnapshotCount"`
+	MaximumTotalSnapshotSize  int64  `json:"maximumTotalSnapshotSize"`
 }
 
 type BackupTarget struct {

manager/volume.go

@@ -1130,3 +1130,55 @@ func (m *VolumeManager) verifyDataSourceForVolumeCreation(dataSource longhorn.Vo
 	}
 	return nil
 }
+
+func (m *VolumeManager) UpdateSnapshotMaxCount(name string, snapshotMaxCount int) (v *longhorn.Volume, err error) {
+	defer func() {
+		err = errors.Wrapf(err, "unable to update field SnapshotMaxCount for volume %s", name)
+	}()
+
+	v, err = m.ds.GetVolume(name)
+	if err != nil {
+		return nil, err
+	}
+
+	if v.Spec.SnapshotMaxCount == snapshotMaxCount {
+		logrus.Debugf("Volume %s already set field SnapshotMaxCount to %d", v.Name, snapshotMaxCount)
+		return v, nil
+	}
+
+	oldSnapshotMaxCount := v.Spec.SnapshotMaxCount
+	v.Spec.SnapshotMaxCount = snapshotMaxCount
+	v, err = m.ds.UpdateVolume(v)
+	if err != nil {
+		return nil, err
+	}
+
+	logrus.Infof("Updated volume %s field SnapshotMaxCount from %d to %d", v.Name, oldSnapshotMaxCount, snapshotMaxCount)
+	return v, nil
+}
+
+func (m *VolumeManager) UpdateSnapshotMaxSize(name string, snapshotMaxSize int64) (v *longhorn.Volume, err error) {
+	defer func() {
+		err = errors.Wrapf(err, "unable to update field SnapshotMaxSize for volume %s", name)
+	}()
+
+	v, err = m.ds.GetVolume(name)
+	if err != nil {
+		return nil, err
+	}
+
+	if v.Spec.SnapshotMaxSize == snapshotMaxSize {
+		logrus.Debugf("Volume %s already set field SnapshotMaxSize to %d", v.Name, snapshotMaxSize)
+		return v, nil
+	}
+
+	oldSnapshotMaxSize := v.Spec.SnapshotMaxSize
+	v.Spec.SnapshotMaxSize = snapshotMaxSize
+	v, err = m.ds.UpdateVolume(v)
+	if err != nil {
+		return nil, err
+	}
+
+	logrus.Infof("Updated volume %s field SnapshotMaxSize from %d to %d", v.Name, oldSnapshotMaxSize, snapshotMaxSize)
+	return v, nil
+}

types/setting.go

@@ -98,6 +98,7 @@ const (
 	SettingNameSnapshotDataIntegrity                                    = SettingName("snapshot-data-integrity")
 	SettingNameSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation = SettingName("snapshot-data-integrity-immediate-check-after-snapshot-creation")
 	SettingNameSnapshotDataIntegrityCronJob                             = SettingName("snapshot-data-integrity-cronjob")
+	SettingNameSnapshotMaxCount                                         = SettingName("snapshot-max-count")
 	SettingNameRestoreVolumeRecurringJobs                               = SettingName("restore-volume-recurring-jobs")
 	SettingNameRemoveSnapshotsDuringFilesystemTrim                      = SettingName("remove-snapshots-during-filesystem-trim")
 	SettingNameFastReplicaRebuildEnabled                                = SettingName("fast-replica-rebuild-enabled")
@@ -173,6 +174,7 @@ var (
 		SettingNameSnapshotDataIntegrity,
 		SettingNameSnapshotDataIntegrityCronJob,
 		SettingNameSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation,
+		SettingNameSnapshotMaxCount,
 		SettingNameRestoreVolumeRecurringJobs,
 		SettingNameRemoveSnapshotsDuringFilesystemTrim,
 		SettingNameFastReplicaRebuildEnabled,
@@ -274,6 +276,7 @@ var (
 		SettingNameSnapshotDataIntegrity:                                    SettingDefinitionSnapshotDataIntegrity,
 		SettingNameSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation: SettingDefinitionSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation,
 		SettingNameSnapshotDataIntegrityCronJob:                             SettingDefinitionSnapshotDataIntegrityCronJob,
+		SettingNameSnapshotMaxCount:                                         SettingDefinitionSnapshotMaxCount,
 		SettingNameRestoreVolumeRecurringJobs:                               SettingDefinitionRestoreVolumeRecurringJobs,
 		SettingNameRemoveSnapshotsDuringFilesystemTrim:                      SettingDefinitionRemoveSnapshotsDuringFilesystemTrim,
 		SettingNameFastReplicaRebuildEnabled:                                SettingDefinitionFastReplicaRebuildEnabled,
@@ -997,6 +1000,16 @@ var (
 		Default:  "0 0 */7 * *",
 	}
 
+	SettingDefinitionSnapshotMaxCount = SettingDefinition{
+		DisplayName: "Snapshot Maximum Count",
+		Description: "Maximum snapshot count for a volume. The value should be between 2 to 250",
+		Category:    SettingCategorySnapshot,
+		Type:        SettingTypeInt,
+		Required:    true,
+		ReadOnly:    false,
+		Default:     "250",
+	}
+
 	SettingDefinitionRemoveSnapshotsDuringFilesystemTrim = SettingDefinition{
 		DisplayName: "Remove Snapshots During Filesystem Trim",
 		Description: "This setting allows Longhorn filesystem trim feature to automatically mark the latest snapshot and its ancestors as removed and stops at the snapshot containing multiple children.\n\n" +

webhook/resources/volume/mutator.go

@@ -170,6 +170,16 @@ func (v *volumeMutator) Create(request *admission.Request, newObj runtime.Object
 		patchOps = append(patchOps, fmt.Sprintf(`{"op": "replace", "path": "/spec/backupCompressionMethod", "value": "%s"}`, defaultCompressionMethod))
 	}
 
+	if volume.Spec.SnapshotMaxCount == 0 {
+		snapshotMaxCount, err := v.getSnapshotMaxCount()
+		if err != nil {
+			err = errors.Wrap(err, "BUG: cannot get valid number for setting snapshot max count")
+			return nil, werror.NewInvalidError(err.Error(), "")
+		}
+		logrus.Infof("Use the default snapshot max count %v", snapshotMaxCount)
+		patchOps = append(patchOps, fmt.Sprintf(`{"op": "replace", "path": "/spec/snapshotMaxCount", "value": %v}`, snapshotMaxCount))
+	}
+
 	// TODO: Remove the mutations below after they are implemented for SPDK volumes
 	if volume.Spec.BackendStoreDriver == longhorn.BackendStoreDriverTypeV2 {
 		if volume.Spec.Encrypted {
@@ -233,6 +243,16 @@ func (v *volumeMutator) Update(request *admission.Request, oldObj runtime.Object
 		patchOps = append(patchOps, fmt.Sprintf(`{"op": "replace", "path": "/spec/backupCompressionMethod", "value": "%s"}`, longhorn.BackupCompressionMethodGzip))
 	}
 
+	if volume.Spec.SnapshotMaxCount == 0 {
+		snapshotMaxCount, err := v.getSnapshotMaxCount()
+		if err != nil {
+			err = errors.Wrap(err, "BUG: cannot get valid number for setting snapshot max count")
+			return nil, werror.NewInvalidError(err.Error(), "")
+		}
+		logrus.Infof("Use the default snapshot max count %v", snapshotMaxCount)
+		patchOps = append(patchOps, fmt.Sprintf(`{"op": "replace", "path": "/spec/snapshotMaxCount", "value": %v}`, snapshotMaxCount))
+	}
+
 	var patchOpsInCommon admission.PatchOps
 	var err error
 	if patchOpsInCommon, err = mutate(newObj, nil); err != nil {
@@ -325,3 +345,11 @@ func (v *volumeMutator) getDefaultReplicaCount() (int, error) {
 	}
 	return int(c), nil
 }
+
+func (v *volumeMutator) getSnapshotMaxCount() (int, error) {
+	c, err := v.ds.GetSettingAsInt(types.SettingNameSnapshotMaxCount)
+	if err != nil {
+		return 0, err
+	}
+	return int(c), nil
+}