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

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

controller/volume_controller.go

@@ -459,6 +459,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
 	}
@@ -1188,6 +1192,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() {
@@ -3191,6 +3212,8 @@ func (c *VolumeController) createEngine(v *longhorn.Volume, currentEngineName st
 			ReplicaAddressMap:         map[string]string{},
 			UpgradedReplicaAddressMap: map[string]string{},
 			RevisionCounterDisabled:   v.Spec.RevisionCounterDisabled,
+			SnapshotMaxCount:          v.Spec.SnapshotMaxCount,
+			SnapshotMaxSize:           v.Spec.SnapshotMaxSize,
 		},
 	}
 
@@ -3242,6 +3265,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

@@ -411,6 +411,14 @@ func (s *DataStore) ValidateSetting(name, value string) (err error) {
 		if value == "true" && autoCleanupValue {
 			return errors.Errorf("cannot set %v setting to true when %v setting is true", name, types.SettingNameAutoCleanupSystemGeneratedSnapshot)
 		}
+	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

@@ -290,6 +290,26 @@ func (e *EngineBinary) VolumeUnmapMarkSnapChainRemovedSet(engine *longhorn.Engin
 	return nil
 }
 
+// VolumeSnapshotMaxCountSet calls engine binary
+// TODO: Deprecated, replaced by gRPC proxy
+func (e *EngineBinary) VolumeSnapshotMaxCountSet(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
+}
+
+// VolumeSnapshotMaxSizeSet calls engine binary
+// TODO: Deprecated, replaced by gRPC proxy
+func (e *EngineBinary) VolumeSnapshotMaxSizeSet(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) VolumeSnapshotMaxCountSet(*longhorn.Engine) error {
+	return fmt.Errorf(ErrNotImplement)
+}
+
+func (e *EngineSimulator) VolumeSnapshotMaxSizeSet(*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

@@ -313,6 +313,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 {
@@ -354,6 +356,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.DataEngine), e.Name, e.Spec.VolumeName,
 		p.DirectToURL(e), e.Spec.UnmapMarkSnapChainRemovedEnabled)
 }
+
+func (p *Proxy) VolumeSnapshotMaxCountSet(e *longhorn.Engine) error {
+	return p.grpcClient.VolumeSnapshotMaxCountSet(string(e.Spec.BackendStoreDriver), e.Name, e.Spec.VolumeName,
+		p.DirectToURL(e), e.Spec.SnapshotMaxCount)
+}
+
+func (p *Proxy) VolumeSnapshotMaxSizeSet(e *longhorn.Engine) error {
+	return p.grpcClient.VolumeSnapshotMaxSizeSet(string(e.Spec.BackendStoreDriver), e.Name, e.Spec.VolumeName,
+		p.DirectToURL(e), e.Spec.SnapshotMaxSize)
+}

engineapi/types.go

@@ -75,6 +75,8 @@ type EngineClient interface {
 	VolumeFrontendStart(*longhorn.Engine) error
 	VolumeFrontendShutdown(*longhorn.Engine) error
 	VolumeUnmapMarkSnapChainRemovedSet(engine *longhorn.Engine) error
+	VolumeSnapshotMaxCountSet(engine *longhorn.Engine) error
+	VolumeSnapshotMaxSizeSet(engine *longhorn.Engine) error
 
 	ReplicaList(*longhorn.Engine) (map[string]*Replica, error)
 	ReplicaAdd(engine *longhorn.Engine, replicaName, url string, isRestoreVolume, fastSync bool, replicaFileSyncHTTPClientTimeout int64) error
@@ -128,6 +130,8 @@ type Volume struct {
 	LastExpansionError        string `json:"lastExpansionError"`
 	LastExpansionFailedAt     string `json:"lastExpansionFailedAt"`
 	UnmapMarkSnapChainRemoved bool   `json:"unmapMarkSnapChainRemoved"`
+	SnapshotMaxCount          int    `json:"snapshotMaxCount"`
+	SnapshotMaxSize           int64  `json:"SnapshotMaxSize"`
 }
 
 type BackupTarget struct {

manager/volume.go

@@ -1143,3 +1143,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

@@ -29,7 +29,7 @@ const (
 	ValueUnknown = "unknown"
 
 	// From `maximumChainLength` in longhorn-engine/pkg/replica/replica.go
-	maxSnapshotNum = 250
+	MaxSnapshotNum = 250
 )
 
 type SettingType string
@@ -104,6 +104,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")
@@ -185,6 +186,7 @@ var (
 		SettingNameSnapshotDataIntegrity,
 		SettingNameSnapshotDataIntegrityCronJob,
 		SettingNameSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation,
+		SettingNameSnapshotMaxCount,
 		SettingNameRestoreVolumeRecurringJobs,
 		SettingNameRemoveSnapshotsDuringFilesystemTrim,
 		SettingNameFastReplicaRebuildEnabled,
@@ -292,6 +294,7 @@ var (
 		SettingNameSnapshotDataIntegrity:                                    SettingDefinitionSnapshotDataIntegrity,
 		SettingNameSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation: SettingDefinitionSnapshotDataIntegrityImmediateCheckAfterSnapshotCreation,
 		SettingNameSnapshotDataIntegrityCronJob:                             SettingDefinitionSnapshotDataIntegrityCronJob,
+		SettingNameSnapshotMaxCount:                                         SettingDefinitionSnapshotMaxCount,
 		SettingNameRestoreVolumeRecurringJobs:                               SettingDefinitionRestoreVolumeRecurringJobs,
 		SettingNameRemoveSnapshotsDuringFilesystemTrim:                      SettingDefinitionRemoveSnapshotsDuringFilesystemTrim,
 		SettingNameFastReplicaRebuildEnabled:                                SettingDefinitionFastReplicaRebuildEnabled,
@@ -1052,6 +1055,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:     strconv.Itoa(MaxSnapshotNum),
+	}
+
 	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" +
@@ -1455,8 +1468,8 @@ func ValidateSetting(name, value string) (err error) {
 		if err != nil {
 			return errors.Wrapf(err, "value %v is not a number", value)
 		}
-		if maxNumber < 1 || maxNumber > maxSnapshotNum {
-			return fmt.Errorf("the value %v should be between 1 and %v", maxNumber, maxSnapshotNum)
+		if maxNumber < 1 || maxNumber > MaxSnapshotNum {
+			return fmt.Errorf("the value %v should be between 1 and %v", maxNumber, MaxSnapshotNum)
 		}
 	case SettingNameEngineReplicaTimeout:
 		timeout, err := strconv.Atoi(value)

upgrade/v15xto160/upgrade.go

@@ -75,6 +75,8 @@ func upgradeVolumes(namespace string, lhClient *lhclientset.Clientset, resourceM
 		} else {
 			v.Spec.DataEngine = longhorn.DataEngineType(v.Spec.BackendStoreDriver)
 		}
+
+		v.Spec.SnapshotMaxCount = types.MaxSnapshotNum
 	}
 
 	return nil