(KBOSS) Add synchronous backup compression flag

api/src/main/java/org/apache/cloudstack/api/ApiConstants.java

@@ -1489,6 +1489,7 @@ public class ApiConstants {
     public static final String SCHEDULED = "scheduled";
     public static final String SCHEDULED_DATE = "scheduleddate";
     public static final String BACKUP_PROVIDER = "backupprovider";
+    public static final String COMPRESS_ASYNC = "compressasync";
 
     /**
      * This enum specifies IO Drivers, each option controls specific policies on I/O.

api/src/main/java/org/apache/cloudstack/api/command/user/backup/CreateBackupOfferingCmd.java

@@ -34,7 +34,7 @@
 import org.apache.cloudstack.backup.Backup;
 import org.apache.cloudstack.backup.BackupManager;
 import org.apache.cloudstack.backup.BackupOffering;
-
+import org.apache.commons.lang3.BooleanUtils;
 
 import javax.inject.Inject;
 import java.util.List;
@@ -92,6 +92,10 @@ public class CreateBackupOfferingCmd extends BaseCmd {
             description = "Restrict the backup offering to the Domains identified by these IDs.")
     private List<Long> domainIds;
 
+    @Parameter(name = ApiConstants.COMPRESS_ASYNC, type = CommandType.BOOLEAN, description = "Whether to compress synchronously during backup creation, or asynchronously later. " +
+            "Default true.")
+    private Boolean compressAsync;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -166,6 +170,10 @@ public Boolean getUserDrivenBackups() {
         return userDrivenBackups;
     }
 
+    public boolean isCompressAsync() {
+        return BooleanUtils.toBooleanDefaultIfNull(this.compressAsync, true);
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////

core/src/main/java/org/apache/cloudstack/backup/TakeKbossBackupCommand.java

@@ -42,7 +42,16 @@ public class TakeKbossBackupCommand extends Command {
 
     private boolean isolated;
 
-    public TakeKbossBackupCommand(boolean quiesceVm, boolean runningVM, boolean endChain, String vmName, String imageStoreUrl, List<String> backupChainImageStoreUrls, List<KbossTO> kbossTOS, boolean isolated) {
+    private boolean compress;
+
+    private Backup.CompressionLibrary compressionLib;
+
+    private Integer coroutines;
+
+    private Integer rateLimit;
+
+    public TakeKbossBackupCommand(boolean quiesceVm, boolean runningVM, boolean endChain, String vmName, String imageStoreUrl, List<String> backupChainImageStoreUrls,
+            List<KbossTO> kbossTOS, boolean isolated) {
         this.quiesceVm = quiesceVm;
         this.runningVM = runningVM;
         this.endChain = endChain;
@@ -85,6 +94,38 @@ public boolean isIsolated() {
         return isolated;
     }
 
+    public void setRateLimit(Integer rateLimit) {
+        this.rateLimit = rateLimit;
+    }
+
+    public void setCoroutines(Integer coroutines) {
+        this.coroutines = coroutines;
+    }
+
+    public void setCompressionLib(Backup.CompressionLibrary compressionLib) {
+        this.compressionLib = compressionLib;
+    }
+
+    public void setCompress(boolean compress) {
+        this.compress = compress;
+    }
+
+    public boolean isCompress() {
+        return compress;
+    }
+
+    public Backup.CompressionLibrary getCompressionLib() {
+        return compressionLib;
+    }
+
+    public Integer getCoroutines() {
+        return coroutines;
+    }
+
+    public Integer getRateLimit() {
+        return rateLimit;
+    }
+
     @Override
     public boolean executeInSequence() {
         return false;

plugins/backup/kboss/src/main/java/org/apache/cloudstack/backup/KbossBackupProvider.java

@@ -448,9 +448,13 @@ public Pair<Boolean, Long> orchestrateTakeBackup(Backup backup, boolean quiesceV
                     parentBackupDeltasOnPrimary, volumeUuidToDeltaPrimaryRef, volumeUuidToDeltaSecondaryRef, succeedingVmSnapshot, kbossTO);
         }
 
+        boolean supportsCompression = offeringSupportsCompression(newBackupJoin);
+
         TakeKbossBackupCommand command = new TakeKbossBackupCommand(quiesceVm, runningVm, newBackupJoin.getEndOfChain(), userVm.getInstanceName(), imageStore.getUri(),
                 chainImageStoreUrls, kbossTOs, isolated);
 
+        boolean compressNow = checkSyncCompressionAndConfigureCommand(backupOfferingVO, supportsCompression, command, hostVO);
+
         Answer answer = sendBackupCommand(hostId, command);
 
         if (answer == null || !answer.getResult()) {
@@ -459,20 +463,34 @@ public Pair<Boolean, Long> orchestrateTakeBackup(Backup backup, boolean quiesceV
         }
 
         processBackupSuccess(runningVm, volumeTOs, volumeUuidToDeltaPrimaryRef, volumeUuidToDeltaSecondaryRef, (TakeKbossBackupAnswer)answer, parentBackupDeltasOnPrimary,
-                succeedingVmSnapshotList, backupVO, fullBackup, userVm, hostId, newBackupJoin.getEndOfChain(), isolated);
+                succeedingVmSnapshotList, backupVO, fullBackup, userVm, hostId, newBackupJoin.getEndOfChain(), isolated, compressNow);
 
         if (!isolated) {
             updateCurrentBackup(newBackupJoin);
         }
 
-        if (offeringSupportsCompression(newBackupJoin)) {
+        if (supportsCompression && !compressNow) {
             compressBackupAsync(newBackupJoin, backup.getZoneId(), userVm.getAccountId());
         } else {
             validateBackupAsyncIfHasOfferingSupport(newBackupJoin, backup.getZoneId(), userVm.getAccountId());
         }
         return new Pair<>(Boolean.TRUE, backupVO.getId());
     }
 
+    protected boolean checkSyncCompressionAndConfigureCommand(BackupOfferingVO backupOfferingVO, boolean supportsCompression, TakeKbossBackupCommand command, HostVO hostVO) {
+        BackupOfferingDetailsVO compressAsync = backupOfferingDetailsDao.findDetail(backupOfferingVO.getId(), ApiConstants.COMPRESS_ASYNC);
+        boolean compressNow = compressAsync != null && !Boolean.parseBoolean(compressAsync.getValue()) && supportsCompression;
+
+        if (compressNow) {
+            BackupOfferingDetailsVO detail = backupOfferingDetailsDao.findDetail(backupOfferingVO.getId(), ApiConstants.COMPRESSION_LIBRARY);
+            command.setCompress(true);
+            command.setCompressionLib(detail == null ? Backup.CompressionLibrary.zstd : Backup.CompressionLibrary.valueOf(detail.getValue()));
+            command.setCoroutines(backupCompressionCoroutines.valueIn(hostVO.getClusterId()));
+            command.setRateLimit(backupCompressionRateLimit.valueIn(hostVO.getClusterId()));
+        }
+        return compressNow;
+    }
+
     @Override
     public boolean deleteBackup(Backup backup, boolean forced) {
         logger.debug("Queueing backup [{}] deletion.", backup.getUuid());
@@ -782,7 +800,7 @@ public boolean startBackupCompression(long backupId, long hostId) {
         BackupOfferingDetailsVO detail = backupOfferingDetailsDao.findDetail(backupOfferingVO.getId(), ApiConstants.COMPRESSION_LIBRARY);
         List<InternalBackupJoinVO> backupChain = getBackupJoinParents(backupVO, true);
         List<String> chainImageStoreUrls = getChainImageStoreUrls(backupChain);
-        CompressBackupCommand cmd = new CompressBackupCommand(deltasToCompressAndParents, chainImageStoreUrls, minFreeStorage, detail == null ? null :
+        CompressBackupCommand cmd = new CompressBackupCommand(deltasToCompressAndParents, chainImageStoreUrls, minFreeStorage, detail == null ? Backup.CompressionLibrary.zstd :
                 Backup.CompressionLibrary.valueOf(detail.getValue()), backupCompressionCoroutines.valueIn(hostVO.getClusterId()),
                 backupCompressionRateLimit.valueIn(hostVO.getClusterId()));
         cmd.setWait(backupCompressionTimeout.valueIn(hostVO.getClusterId()));
@@ -2084,7 +2102,8 @@ protected List<VolumeObjectTO> getVolumesThatAreNotPartOfTheBackup(List<VolumeOb
 
     protected void processBackupSuccess(boolean runningVm, List<VolumeObjectTO> volumeTOs, HashMap<String, InternalBackupStoragePoolVO> volumeUuidToDeltaPrimaryRef,
             HashMap<String, InternalBackupDataStoreVO> volumeUuidToDeltaSecondaryRef, TakeKbossBackupAnswer answer, List<InternalBackupStoragePoolVO> parentBackupDeltasOnPrimary,
-            List<VMSnapshotVO> succeedingVmSnapshots, BackupVO backupVO, boolean fullBackup, VirtualMachine userVm, Long hostId, boolean endChain, boolean isolated) {
+            List<VMSnapshotVO> succeedingVmSnapshots, BackupVO backupVO, boolean fullBackup, VirtualMachine userVm, Long hostId, boolean endChain, boolean isolated,
+            boolean compressNow) {
         long physicalBackupSize = 0;
         logger.debug("Processing backup [{}] success.", backupVO.getUuid());
         for (VolumeObjectTO volumeObjectTO : volumeTOs) {
@@ -2094,6 +2113,9 @@ protected void processBackupSuccess(boolean runningVm, List<VolumeObjectTO> volu
 
         expungeOldDeltasAndUpdateVmSnapshotIfNeeded(parentBackupDeltasOnPrimary, succeedingVmSnapshots.isEmpty() ? null : succeedingVmSnapshots.get(0));
 
+        if (compressNow) {
+            backupVO.setCompressionStatus(Backup.CompressionStatus.Compressed);
+        }
         backupVO.setSize(physicalBackupSize);
         backupVO.setStatus(Backup.Status.BackedUp);
         backupVO.setBackedUpVolumes(backupManager.createVolumeInfoFromVolumes(new ArrayList<>(volumeDao.findByInstance(userVm.getId()))));

plugins/backup/kboss/src/test/java/org/apache/cloudstack/backup/KbossBackupProviderTest.java

@@ -787,6 +787,8 @@ public void orchestrateTakeBackupTestIsolatedBackupFailed() {
         doNothing().when(kbossBackupProviderSpy).validateStorages(any(), any());
         doReturn(internalBackupJoinVoMock).when(internalBackupJoinDaoMock).findById(any());
         doReturn(dataStoreMock).when(kbossBackupProviderSpy).getImageStoreForBackup(any(), any());
+        doReturn(false).when(kbossBackupProviderSpy).offeringSupportsCompression(any());
+        doReturn(false).when(kbossBackupProviderSpy).checkSyncCompressionAndConfigureCommand(any(), anyBoolean(), any(), any());
 
         Pair<Boolean, Long> result = kbossBackupProviderSpy.orchestrateTakeBackup(backupVoMock, false, true);
         assertFalse(result.first());
@@ -815,17 +817,18 @@ public void orchestrateTakeBackupTestIsolatedBackupSuccessWithCompression() {
         doReturn(takeKbossBackupAnswerMock).when(kbossBackupProviderSpy).sendBackupCommand(anyLong(), any());
         doReturn(true).when(takeKbossBackupAnswerMock).getResult();
         doNothing().when(kbossBackupProviderSpy).processBackupSuccess(anyBoolean(), any(), any(), any(), any(), any(), any(), any(), anyBoolean(), any(),
-                anyLong(), anyBoolean(), anyBoolean());
-        doReturn(true).when(kbossBackupProviderSpy).offeringSupportsCompression(internalBackupJoinVoMock);
+                anyLong(), anyBoolean(), anyBoolean(), anyBoolean());
         doNothing().when(kbossBackupProviderSpy).compressBackupAsync(internalBackupJoinVoMock, 0, 0);
+        doReturn(true).when(kbossBackupProviderSpy).offeringSupportsCompression(any());
+        doReturn(false).when(kbossBackupProviderSpy).checkSyncCompressionAndConfigureCommand(any(), anyBoolean(), any(), any());
 
         Pair<Boolean, Long> result = kbossBackupProviderSpy.orchestrateTakeBackup(backupVoMock, false, true);
         assertTrue(result.first());
         assertEquals(backupId, result.second());
         verify(kbossBackupProviderSpy, Mockito.times(1)).setBackupAsIsolated(backupVoMock);
         verify(kbossBackupProviderSpy, Mockito.times(2)).createDeltaReferences(Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), any(), any(), any(), any(), any(), any(), any());
         verify(kbossBackupProviderSpy, Mockito.times(1)).processBackupSuccess(anyBoolean(), any(), any(), any(), any(), any(), any(), any(), anyBoolean(), any(),
-                anyLong(), anyBoolean(), anyBoolean());
+                anyLong(), anyBoolean(), anyBoolean(), anyBoolean());
         verify(kbossBackupProviderSpy, Mockito.times(1)).compressBackupAsync(internalBackupJoinVoMock, 0, 0);
     }
 
@@ -850,9 +853,10 @@ public void orchestrateTakeBackupTestBackupSuccessWithValidation() {
         doReturn(takeKbossBackupAnswerMock).when(kbossBackupProviderSpy).sendBackupCommand(anyLong(), any());
         doReturn(true).when(takeKbossBackupAnswerMock).getResult();
         doNothing().when(kbossBackupProviderSpy).processBackupSuccess(anyBoolean(), any(), any(), any(), any(), any(), any(), any(), anyBoolean(), any(),
-                anyLong(), anyBoolean(), anyBoolean());
-        doReturn(false).when(kbossBackupProviderSpy).offeringSupportsCompression(internalBackupJoinVoMock);
+                anyLong(), anyBoolean(), anyBoolean(), anyBoolean());
         doNothing().when(kbossBackupProviderSpy).validateBackupAsyncIfHasOfferingSupport(any(), anyLong(), anyLong());
+        doReturn(false).when(kbossBackupProviderSpy).offeringSupportsCompression(any());
+        doReturn(false).when(kbossBackupProviderSpy).checkSyncCompressionAndConfigureCommand(any(), anyBoolean(), any(), any());
 
         Pair<Boolean, Long> result = kbossBackupProviderSpy.orchestrateTakeBackup(backupVoMock, false, false);
         assertTrue(result.first());
@@ -861,7 +865,7 @@ public void orchestrateTakeBackupTestBackupSuccessWithValidation() {
         verify(internalBackupDataStoreDaoMock).listByBackupId(0);
         verify(kbossBackupProviderSpy, Mockito.times(2)).createDeltaReferences(Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyBoolean(), any(), any(), any(), any(), any(), any(), any());
         verify(kbossBackupProviderSpy, Mockito.times(1)).processBackupSuccess(anyBoolean(), any(), any(), any(), any(), any(), any(), any(), anyBoolean(), any(),
-                anyLong(), anyBoolean(), anyBoolean());
+                anyLong(), anyBoolean(), anyBoolean(), anyBoolean());
         verify(kbossBackupProviderSpy, Mockito.times(1)).validateBackupAsyncIfHasOfferingSupport(internalBackupJoinVoMock, 0, 0);
     }
 

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtCompressBackupCommandWrapper.java

@@ -82,7 +82,7 @@ public Answer execute(CompressBackupCommand command, LibvirtComputingResource se
 
                 HashMap<String, String> options = new HashMap<>();
                 Backup.CompressionLibrary compressionLib = getCompressionLibrary(command, fullDeltaPath);
-                setCompressionTypeOptionIfAvailable(qemuImg, options, compressionLib);
+                qemuImg.setCompressionTypeOptionIfAvailable(options, compressionLib);
                 int coroutines = command.getCoroutines();
                 logger.info("Starting compression for backup delta [{}] with parent [{}] using [{}] coroutines.", child, parent, coroutines);
                 qemuImg.convert(originalBackup, compressedBackup, backingFile, options, null, new QemuImageOptions(originalBackup.getFormat(), originalBackup.getFileName(),
@@ -111,18 +111,6 @@ private Integer validateAndGetRateLimit(CompressBackupCommand command, QemuImg q
         return command.getRateLimit();
     }
 
-    /**
-     * Sets the compression type option if qemu-img is at least in version 5.1. Otherwise, will not set it and qemu will use zlib.
-     * */
-    private void setCompressionTypeOptionIfAvailable(QemuImg qemuImg, HashMap<String, String> options, Backup.CompressionLibrary compressionLib) {
-        if (qemuImg.getVersion() >= QemuImg.QEMU_5_1) {
-            options.put(COMPRESSION_TYPE, compressionLib.name());
-            return;
-        }
-        logger.warn("Qemu is at a lower version than 5.1, we will not be able to use zstd to compress backups. Only zlib is supported for this version. Current version is [{}].",
-                qemuImg.getVersion());
-    }
-
     private Backup.CompressionLibrary getCompressionLibrary(CompressBackupCommand command, String fullDeltaPath) {
         Backup.CompressionLibrary compressionLib = command.getCompressionLib();
         if (compressionLib == Backup.CompressionLibrary.zlib || !Qcow2Inspector.validateQcow2Version(fullDeltaPath, MIN_QCOW_2_VERSION_FOR_ZSTD)) {

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtTakeKbossBackupCommandWrapper.java

@@ -105,7 +105,7 @@ protected void backupVolumes(TakeKbossBackupCommand command, LibvirtComputingRes
 
                 logger.debug("Backing up volume [{}].", volumeUuid);
                 Pair<String, Long> deltaPathOnSecondaryAndSize = copyBackupDeltaToSecondary(storagePoolManager, kbossTO, command.getBackupChainImageStoreUrls(),
-                        command.getImageStoreUrl(), maxWaitInMillis);
+                        command.getImageStoreUrl(), maxWaitInMillis, command);
 
                 mapVolumeUuidToDeltaPathOnSecondaryAndDeltaSize.put(volumeUuid, deltaPathOnSecondaryAndSize);
                 maxWaitInMillis = calculateRemainingTime(maxWaitInMillis, startTimeMillis);
@@ -165,7 +165,7 @@ protected void cleanupVm(TakeKbossBackupCommand command, LibvirtComputingResourc
      * If there were snapshots created after the last backup, they'll be copied alongside and merged in the secondary storage.
      * */
     protected Pair<String, Long> copyBackupDeltaToSecondary(KVMStoragePoolManager storagePoolManager, KbossTO kbossTO, List<String> chainImageStoreUrls, String imageStoreUrl,
-            int waitInMillis) {
+            int waitInMillis, TakeKbossBackupCommand command) {
         VolumeObjectTO delta = kbossTO.getVolumeObjectTO();
         String parentDeltaPathOnSecondary = kbossTO.getPathBackupParentOnSecondary();
         List<String> deltaPathsToCopy = CollectionUtils.isEmpty(kbossTO.getVmSnapshotDeltaPaths()) ? new ArrayList<>() : new ArrayList<>(kbossTO.getVmSnapshotDeltaPaths());
@@ -199,7 +199,7 @@ protected Pair<String, Long> copyBackupDeltaToSecondary(KVMStoragePoolManager st
                 }
 
                 String backupDeltaFullPathOnPrimary = primaryPool.getLocalPathFor(deltaPathsToCopy.remove(0));
-                convertDeltaToSecondary(backupDeltaFullPathOnPrimary, backupDeltaFullPathOnSecondary, parentBackupFullPath, delta.getUuid(), waitInMillis);
+                convertDeltaToSecondary(backupDeltaFullPathOnPrimary, backupDeltaFullPathOnSecondary, parentBackupFullPath, delta.getUuid(), waitInMillis, command);
 
                 if (!deltaPathsToCopy.isEmpty()) {
                     parentDeltaPathOnSecondary = topDelta;
@@ -282,7 +282,8 @@ protected void removeTemporaryDeltas(List<String> temporaryDeltasToRemove, boole
      * @param volumeUuid volume uuid, used for logging.
      * @param waitInMillis timeout in milliseconds.
      * */
-    protected void convertDeltaToSecondary(String pathDeltaOnPrimary, String pathDeltaOnSecondary, String pathParentOnSecondary, String volumeUuid, int waitInMillis)
+    protected void convertDeltaToSecondary(String pathDeltaOnPrimary, String pathDeltaOnSecondary, String pathParentOnSecondary, String volumeUuid, int waitInMillis,
+            TakeKbossBackupCommand command)
             throws QemuImgException, LibvirtException {
         QemuImgFile backupDestination = new QemuImgFile(pathDeltaOnSecondary, QemuImg.PhysicalDiskFormat.QCOW2);
         QemuImgFile backupOrigin = new QemuImgFile(pathDeltaOnPrimary, QemuImg.PhysicalDiskFormat.QCOW2);
@@ -297,8 +298,13 @@ protected void convertDeltaToSecondary(String pathDeltaOnPrimary, String pathDel
         createDirsIfNeeded(pathDeltaOnSecondary, volumeUuid);
 
         QemuImg qemuImg = new QemuImg(waitInMillis);
-        qemuImg.convert(backupOrigin, backupDestination, parentBackup, null, null,  new QemuImageOptions(backupOrigin.getFormat(), backupOrigin.getFileName(), null), null,
-                true, false, false, false, null, null);
+        Map<String, String> options = new HashMap<>();
+        if (command.isCompress()) {
+            qemuImg.setCompressionTypeOptionIfAvailable(options, command.getCompressionLib());
+        }
+        qemuImg.convert(backupOrigin, backupDestination, parentBackup, options, null,  new QemuImageOptions(backupOrigin.getFormat(), backupOrigin.getFileName(), null), null,
+                true, false, false, command.isCompress(), command.getCoroutines(), command.getRateLimit());
+
     }