Allow backups to be made incrementally (#114)

README.md

@@ -123,6 +123,14 @@ Manila backup methods:
 - **Snapshot** - Create a snapshot using Manila.
 - **Clone** - Create a snapshot using Manila, but immediatelly create a volume from this snapshot and afterwards cleanup original snapshot.
 
+#### Incremental backups
+
+For backup method `backups`, incremental backups are supported. This will however lead to the inability to delete backups in a reverse order, since it is not possible to delete a (older) backup that has dependant (newer) backups. 
+
+**Important: TTL is not working correctly with expiration, so set a large value such as (43800h, approx. 5 years)** since TTL inherently use a `First In, First Out` algorithm, where incremental backups should be deleted on a `Last In, First Out` basis. To cleanup these incremental backups, it's suggested to use the Velero CLI to perform deletion on this `Last In, First Out` basis.
+
+Next to that, when incremental backups are enabled, the first backup of a volume will always be a full backup, since this is needed to create increments on.
+
 ### Consistency and Durability
 
 Please note two facts regarding volume backups:

docs/installation-using-cli.md

@@ -83,6 +83,8 @@ spec:
     # deletes all dependent volume resources (i.e. snapshots) before deleting
     # the clone volume (works only, when a snapshot method is set to clone)
     cascadeDelete: "true"
+    # backups will be created incrementally (works only when snapshot method is set to backup)
+    incrementalBackup: "true"
 ```
 
 For backups of Manila shares create another configuration of `volumesnapshotlocations.velero.io`:

docs/installation-using-helm.md

@@ -65,6 +65,8 @@ configuration:
       # deletes all dependent volume resources (i.e. snapshots) before deleting
       # the clone volume (works only, when a snapshot method is set to clone)
       cascadeDelete: "true"
+      # backups will be created incrementally (works only when snapshot method is set to backup)
+      incrementalBackup: "true"
   # for Manila shared filesystem storage
   - name: manila
     provider: community.openstack.org/openstack-manila

src/cinder/block_store.go

@@ -99,6 +99,7 @@ type BlockStore struct {
 	cascadeDelete      bool
 	containerName      string
 	log                logrus.FieldLogger
+	backupIncremental  bool
 }
 
 // NewBlockStore instantiates a Cinder Volume Snapshotter.
@@ -158,6 +159,10 @@ func (b *BlockStore) Init(config map[string]string) error {
 	if err != nil {
 		return fmt.Errorf("cannot parse cascadeDelete config variable: %w", err)
 	}
+	b.backupIncremental, err = strconv.ParseBool(utils.GetConf(b.config, "backupIncremental", "false"))
+	if err != nil {
+		return fmt.Errorf("cannot parse backupIncremental config variable: %w", err)
+	}
 
 	// load optional containerName
 	b.containerName = utils.GetConf(b.config, "containerName", "")
@@ -321,12 +326,13 @@ func (b *BlockStore) createVolumeFromClone(cloneID, volumeType, volumeAZ string)
 
 func (b *BlockStore) createVolumeFromBackup(backupID, volumeType, volumeAZ string) (string, error) {
 	logWithFields := b.log.WithFields(logrus.Fields{
-		"backupID":      backupID,
-		"volumeType":    volumeType,
-		"volumeAZ":      volumeAZ,
-		"backupTimeout": b.backupTimeout,
-		"volumeTimeout": b.volumeTimeout,
-		"method":        b.config["method"],
+		"backupID":          backupID,
+		"volumeType":        volumeType,
+		"volumeAZ":          volumeAZ,
+		"backupTimeout":     b.backupTimeout,
+		"backupIncremental": b.backupIncremental,
+		"volumeTimeout":     b.volumeTimeout,
+		"method":            b.config["method"],
 	})
 	logWithFields.Info("BlockStore.CreateVolumeFromSnapshot called")
 
@@ -588,12 +594,13 @@ func (b *BlockStore) createClone(volumeID, volumeAZ string, tags map[string]stri
 func (b *BlockStore) createBackup(volumeID, volumeAZ string, tags map[string]string) (string, error) {
 	backupName := fmt.Sprintf("%s.backup.%s", volumeID, strconv.FormatUint(utils.Rand.Uint64(), 10))
 	logWithFields := b.log.WithFields(logrus.Fields{
-		"backupName":    backupName,
-		"volumeID":      volumeID,
-		"volumeAZ":      volumeAZ,
-		"tags":          tags,
-		"backupTimeout": b.backupTimeout,
-		"method":        b.config["method"],
+		"backupName":        backupName,
+		"volumeID":          volumeID,
+		"volumeAZ":          volumeAZ,
+		"tags":              tags,
+		"backupTimeout":     b.backupTimeout,
+		"backupIncremental": b.backupIncremental,
+		"method":            b.config["method"],
 	})
 	logWithFields.Info("BlockStore.CreateSnapshot called")
 
@@ -603,20 +610,41 @@ func (b *BlockStore) createBackup(volumeID, volumeAZ string, tags map[string]str
 		return "", fmt.Errorf("failed to get volume %v from cinder: %w", volumeID, err)
 	}
 
+	existingBackups, err := b.getVolumeBackups(logWithFields, volumeID)
+	if err != nil {
+		logWithFields.Error("failed to retrieve existing volume backups.")
+		return "", fmt.Errorf("failed to retrieve existing backups %v from cinder: %w", volumeID, err)
+	}
+
+	var existingBackup *backups.Backup
+	for _, b := range existingBackups {
+		if b.VolumeID == volumeID && utils.SliceContains(backupStatuses, b.Status) {
+			existingBackup = &b
+			break
+		}
+	}
+
 	opts := &backups.CreateOpts{
 		Name:        backupName,
 		VolumeID:    volumeID,
 		Description: "Velero volume backup",
 		Container:   backupName,
 		Metadata:    utils.Merge(originVolume.Metadata, tags),
 		Force:       true,
+		Incremental: b.backupIncremental,
 	}
 
 	// Override container if one was passed by the user
 	if b.containerName != "" {
 		opts.Container = b.containerName
 	}
 
+	// Disable incremental backup for volume where no backup exists yet.
+	if b.backupIncremental && existingBackup == nil {
+		logWithFields.Infof("No backup exists yet for volume %s, will first run a full backup.", volumeID)
+		opts.Incremental = false
+	}
+
 	backup, err := backups.Create(b.client, opts).Extract()
 	if err != nil {
 		logWithFields.Error("failed to create backup from volume")
@@ -1093,3 +1121,18 @@ func expandVolumeProperties(log logrus.FieldLogger, volume *volumes.Volume) imag
 	}
 	return imgAttrUpdateOpts
 }
+
+func (b *BlockStore) getVolumeBackups(logWithFields *logrus.Entry, volumeID string) ([]backups.Backup, error) {
+	// use detail and a non-volumeid search to allow usage of later microversions
+	pages, err := backups.ListDetail(b.client, nil).AllPages()
+	if err != nil {
+		return nil, fmt.Errorf("failed to list backups: %w", err)
+	}
+
+	allBackups, err := backups.ExtractBackups(pages)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract backups: %w", err)
+	}
+
+	return allBackups, nil
+}