nas: trim whitespace around backup repository mount options

plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java

@@ -589,7 +589,7 @@ public Pair<Boolean, Backup> takeBackup(final VirtualMachine vm, Boolean quiesce
         TakeBackupCommand command = new TakeBackupCommand(vm.getInstanceName(), backupPath);
         command.setBackupRepoType(backupRepository.getType());
         command.setBackupRepoAddress(backupRepository.getAddress());
-        command.setMountOptions(backupRepository.getMountOptions());
+        command.setMountOptions(getMountOptions(backupRepository));
         command.setQuiesce(quiesceVM);
         command.setQuiesceTimeout(NASBackupQuiesceAgentTimeout.valueIn(vm.getDataCenterId()));
         command.setMode(decision.mode);
@@ -735,7 +735,7 @@ private Pair<Boolean, String> restoreVMBackup(VirtualMachine vm, Backup backup)
         restoreCommand.setBackupPath(backup.getExternalId());
         restoreCommand.setBackupRepoType(backupRepository.getType());
         restoreCommand.setBackupRepoAddress(backupRepository.getAddress());
-        restoreCommand.setMountOptions(backupRepository.getMountOptions());
+        restoreCommand.setMountOptions(getMountOptions(backupRepository));
         restoreCommand.setVmName(vm.getName());
         restoreCommand.setBackupVolumesUUIDs(backedVolumesUUIDs);
         Pair<List<PrimaryDataStoreTO>, List<String>> volumePoolsAndPaths = getVolumePoolsAndPaths(restoreVolumes);
@@ -865,7 +865,7 @@ public Pair<Boolean, String> restoreBackedUpVolume(Backup backup, Backup.VolumeI
         DataStore dataStore = dataStoreMgr.getDataStore(pool.getId(), DataStoreRole.Primary);
         restoreCommand.setRestoreVolumePools(Collections.singletonList(dataStore != null ? (PrimaryDataStoreTO)dataStore.getTO() : null));
         restoreCommand.setDiskType(backupVolumeInfo.getType().name().toLowerCase(Locale.ROOT));
-        restoreCommand.setMountOptions(backupRepository.getMountOptions());
+        restoreCommand.setMountOptions(getMountOptions(backupRepository));
         restoreCommand.setVmExists(null);
         restoreCommand.setVmState(vmNameAndState.second());
         restoreCommand.setMountTimeout(NASBackupRestoreMountTimeout.value());
@@ -1006,7 +1006,7 @@ protected void releaseChainDeleteLock(GlobalLock lock) {
      */
     private boolean deleteBackupFileAndRow(Backup backup, BackupRepository repo, Host host) {
         DeleteBackupCommand command = new DeleteBackupCommand(backup.getExternalId(), repo.getType(),
-                repo.getAddress(), repo.getMountOptions());
+                repo.getAddress(), getMountOptions(repo));
         BackupAnswer answer;
         try {
             answer = (BackupAnswer) agentManager.send(host.getId(), command);
@@ -1240,7 +1240,7 @@ public void syncBackupStorageStats(Long zoneId) {
             return;
         }
         for (final BackupRepository repository : repositories) {
-            GetBackupStorageStatsCommand command = new GetBackupStorageStatsCommand(repository.getType(), repository.getAddress(), repository.getMountOptions());
+            GetBackupStorageStatsCommand command = new GetBackupStorageStatsCommand(repository.getType(), repository.getAddress(), getMountOptions(repository));
             BackupStorageStatsAnswer answer;
             try {
                 answer = (BackupStorageStatsAnswer) agentManager.send(host.getId(), command);
@@ -1253,6 +1253,26 @@ public void syncBackupStorageStats(Long zoneId) {
         }
     }
 
+    /**
+     * Returns the repository's mount options without the blanks around each option, so that a value
+     * saved with stray whitespace, such as "vers=4.1 ", reaches mount the same way for every command.
+     * Options that are only whitespace are dropped. An empty option is kept, because cifs reads a
+     * double comma as a comma inside a value such as a password.
+     */
+    protected static String getMountOptions(BackupRepository repository) {
+        String mountOptions = repository.getMountOptions();
+        if (mountOptions == null) {
+            return null;
+        }
+        List<String> options = new ArrayList<>();
+        for (String option : mountOptions.split(",", -1)) {
+            if (option.isEmpty() || StringUtils.isNotBlank(option)) {
+                options.add(option.trim());
+            }
+        }
+        return String.join(",", options);
+    }
+
     @Override
     public List<BackupOffering> listBackupOfferings(Long zoneId) {
         final List<BackupRepository> repositories = backupRepositoryDao.listByZoneAndProvider(zoneId, getName());

plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java

@@ -174,6 +174,64 @@ public void testSyncBackupStorageStats() throws AgentUnavailableException, Opera
         Mockito.verify(backupRepositoryDao, Mockito.times(1)).updateCapacity(backupRepository, 100L, 50L);
     }
 
+    @Test
+    public void testSyncBackupStorageStatsSendsTrimmedMountOptions() throws AgentUnavailableException, OperationTimedoutException {
+        BackupRepositoryVO backupRepository = new BackupRepositoryVO(1L, "nas", "test-repo",
+                "nfs", "address", "vers=4.1 ", 1024L, null);
+
+        HostVO host = mock(HostVO.class);
+        Mockito.when(resourceManager.findOneRandomRunningHostByHypervisor(Hypervisor.HypervisorType.KVM, 1L)).thenReturn(host);
+        Mockito.when(backupRepositoryDao.listByZoneAndProvider(1L, "nas")).thenReturn(Collections.singletonList(backupRepository));
+        ArgumentCaptor<GetBackupStorageStatsCommand> commandCaptor = ArgumentCaptor.forClass(GetBackupStorageStatsCommand.class);
+        BackupStorageStatsAnswer answer = new BackupStorageStatsAnswer(new GetBackupStorageStatsCommand("nfs", "address", null), true, null);
+        Mockito.when(agentManager.send(anyLong(), commandCaptor.capture())).thenReturn(answer);
+
+        nasBackupProvider.syncBackupStorageStats(1L);
+
+        Assert.assertEquals("vers=4.1", commandCaptor.getValue().getMountOptions());
+    }
+
+    private String getMountOptions(String mountOptions) {
+        return NASBackupProvider.getMountOptions(new BackupRepositoryVO(1L, "nas", "test-repo",
+                "nfs", "address", mountOptions, 1024L, null));
+    }
+
+    @Test
+    public void testGetMountOptionsTrimsTrailingWhitespace() {
+        Assert.assertEquals("vers=4.1", getMountOptions("vers=4.1 "));
+    }
+
+    @Test
+    public void testGetMountOptionsTrimsWhitespaceAroundEveryOption() {
+        Assert.assertEquals("vers=4.1,soft", getMountOptions(" vers=4.1 , soft "));
+    }
+
+    @Test
+    public void testGetMountOptionsDropsOptionsThatAreOnlyWhitespace() {
+        Assert.assertEquals("vers=4.1,soft", getMountOptions("vers=4.1, ,soft, "));
+    }
+
+    @Test
+    public void testGetMountOptionsKeepsDoubleCommaInsideAValue() {
+        // cifs reads a double comma as a comma inside the value, here the password "pa,ss"
+        Assert.assertEquals("username=user,password=pa,,ss,vers=3.0", getMountOptions("username=user,password=pa,,ss,vers=3.0 "));
+    }
+
+    @Test
+    public void testGetMountOptionsDoesNotStripWhitespaceInsideAnOption() {
+        Assert.assertEquals("username=some user", getMountOptions("username=some user"));
+    }
+
+    @Test
+    public void testGetMountOptionsReturnsEmptyWhenOnlyWhitespace() {
+        Assert.assertEquals("", getMountOptions("   "));
+    }
+
+    @Test
+    public void testGetMountOptionsKeepsNull() {
+        Assert.assertNull(getMountOptions(null));
+    }
+
     @Test
     public void testListBackupOfferings() {
         BackupRepositoryVO backupRepository = new BackupRepositoryVO(1L, "nas", "test-repo",