Support multi-scope configuration settings (#10300)

This PR introduces the concept of multi-scope configuration settings. In addition to the Global level, currently all configurations can be set at a single scope level.
It will be useful if a configuration can be set at multiple scopes. For example, a configuration set at the domain level
will apply for all accounts, but it can be set for an account as well. In which case the account level setting will override the domain level setting.

This is done by changing the column `scope` of table `configuration` from string (single scope) to bitmask (multiple scopes).

```
public enum Scope {
    Global(null, 1),
    Zone(Global, 1 << 1),
    Cluster(Zone, 1 << 2),
    StoragePool(Cluster, 1 << 3),
    ManagementServer(Global, 1 << 4),
    ImageStore(Zone, 1 << 5),
    Domain(Global, 1 << 6),
    Account(Domain, 1 << 7);
```
Each scope is also assigned a parent scope. When a configuration for a given scope is not defined but is available for multiple scope types, the value will be retrieved from the parent scope. If there is no parent scope or if the configuration is defined for a single scope only, the value will fall back to the global level.

Hierarchy for different scopes is defined as below :
- Global
    - Zone
        - Cluster
            - Storage Pool
        - Image Store
    - Management Server
    - Domain
        - Account

This PR also updates the scope of the following configurations (Storage Pool scope is added in addition to the existing Zone scope):
- pool.storage.allocated.capacity.disablethreshold
- pool.storage.allocated.resize.capacity.disablethreshold
- pool.storage.capacity.disablethreshold

Doc PR : https://github.com/apache/cloudstack-documentation/pull/476

Signed-off-by: Abhishek Kumar <abhishek.mrt22@gmail.com>
Co-authored-by: Abhishek Kumar <abhishek.mrt22@gmail.com>

engine/components-api/src/main/java/com/cloud/capacity/CapacityManager.java

@@ -16,6 +16,8 @@
 // under the License.
 package com.cloud.capacity;
 
+import java.util.List;
+
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 
@@ -67,7 +69,7 @@ public interface CapacityManager {
                     "0.85",
                     "Percentage (as a value between 0 and 1) of storage utilization above which allocators will disable using the pool for low storage available.",
                     true,
-                    ConfigKey.Scope.Zone);
+                    List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
     static final ConfigKey<Double> StorageOverprovisioningFactor =
             new ConfigKey<>(
                     "Storage",
@@ -85,7 +87,7 @@ public interface CapacityManager {
                     "0.85",
                     "Percentage (as a value between 0 and 1) of allocated storage utilization above which allocators will disable using the pool for low allocated storage available.",
                     true,
-                    ConfigKey.Scope.Zone);
+                    List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
     static final ConfigKey<Boolean> StorageOperationsExcludeCluster =
             new ConfigKey<>(
                     Boolean.class,
@@ -125,7 +127,7 @@ public interface CapacityManager {
                     "Percentage (as a value between 0 and 1) of allocated storage utilization above which allocators will disable using the pool for volume resize. " +
                             "This is applicable only when volume.resize.allowed.beyond.allocation is set to true.",
                     true,
-                    ConfigKey.Scope.Zone);
+                    List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
 
     ConfigKey<Integer> CapacityCalculateWorkers = new ConfigKey<>(ConfigKey.CATEGORY_ADVANCED, Integer.class,
             "capacity.calculate.workers", "1",

engine/components-api/src/main/java/com/cloud/storage/StorageManager.java

@@ -214,7 +214,7 @@ public interface StorageManager extends StorageService {
     ConfigKey<Boolean> AllowVolumeReSizeBeyondAllocation = new ConfigKey<Boolean>("Advanced", Boolean.class, "volume.resize.allowed.beyond.allocation", "false",
             "Determines whether volume size can exceed the pool capacity allocation disable threshold (pool.storage.allocated.capacity.disablethreshold) " +
                     "when resize a volume upto resize capacity disable threshold (pool.storage.allocated.resize.capacity.disablethreshold)",
-            true, ConfigKey.Scope.Zone);
+            true, List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
 
     ConfigKey<Integer> StoragePoolHostConnectWorkers = new ConfigKey<>("Storage", Integer.class,
             "storage.pool.host.connect.workers", "1",

engine/schema/src/main/java/com/cloud/dc/ClusterDetailsDaoImpl.java

@@ -22,11 +22,16 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import javax.inject.Inject;
+
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.ConfigKey.Scope;
 import org.apache.cloudstack.framework.config.ScopedConfigStorage;
 import org.apache.commons.collections.CollectionUtils;
 
+import com.cloud.dc.dao.ClusterDao;
+import com.cloud.org.Cluster;
+import com.cloud.utils.Pair;
 import com.cloud.utils.crypt.DBEncryptionUtil;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
@@ -35,6 +40,9 @@ import org.apache.cloudstack.resourcedetail.ResourceDetailsDaoBase;
 
 public class ClusterDetailsDaoImpl extends ResourceDetailsDaoBase<ClusterDetailsVO> implements ClusterDetailsDao, ScopedConfigStorage {
 
+    @Inject
+    ClusterDao clusterDao;
+
     protected final SearchBuilder<ClusterDetailsVO> ClusterSearch;
     protected final SearchBuilder<ClusterDetailsVO> DetailSearch;
 
@@ -186,4 +194,13 @@ public class ClusterDetailsDaoImpl extends ResourceDetailsDaoBase<ClusterDetails
 
         return name;
     }
+
+    @Override
+    public Pair<Scope, Long> getParentScope(long id) {
+        Cluster cluster = clusterDao.findById(id);
+        if (cluster == null) {
+            return null;
+        }
+        return new Pair<>(getScope().getParent(), cluster.getDataCenterId());
+    }
 }

engine/schema/src/main/java/com/cloud/storage/dao/StoragePoolDetailsDaoImpl.java

@@ -30,6 +30,8 @@ import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailVO;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailsDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 
+import com.cloud.utils.Pair;
+
 public class StoragePoolDetailsDaoImpl extends ResourceDetailsDaoBase<StoragePoolDetailVO> implements StoragePoolDetailsDao, ScopedConfigStorage {
 
     @Inject
@@ -57,4 +59,17 @@ public class StoragePoolDetailsDaoImpl extends ResourceDetailsDaoBase<StoragePoo
         }
         super.addDetail(new StoragePoolDetailVO(resourceId, key, value, display));
     }
+
+    @Override
+    public Pair<Scope, Long> getParentScope(long id) {
+        StoragePoolVO pool = _storagePoolDao.findById(id);
+        if (pool != null) {
+            if (pool.getClusterId() != null) {
+                return new Pair<>(getScope().getParent(), pool.getClusterId());
+            } else {
+                return new Pair<>(ConfigKey.Scope.Zone, pool.getDataCenterId());
+            }
+        }
+        return null;
+    }
 }

engine/schema/src/main/java/com/cloud/upgrade/ConfigurationGroupsAggregator.java

@@ -54,7 +54,7 @@ public class ConfigurationGroupsAggregator {
 
     public void updateConfigurationGroups() {
         LOG.debug("Updating configuration groups");
-        List<ConfigurationVO> configs =  configDao.listAllIncludingRemoved();
+        List<ConfigurationVO> configs =  configDao.searchPartialConfigurations();
         if (CollectionUtils.isEmpty(configs)) {
             return;
         }

engine/schema/src/main/java/com/cloud/upgrade/dao/DatabaseAccessObject.java

@@ -87,6 +87,36 @@ public class DatabaseAccessObject {
         return columnExists;
     }
 
+    public String getColumnType(Connection conn, String tableName, String columnName) {
+        try (PreparedStatement pstmt = conn.prepareStatement(String.format("DESCRIBE %s %s", tableName, columnName));){
+            ResultSet rs = pstmt.executeQuery();
+            if (rs.next()) {
+                return rs.getString("Type");
+            }
+        } catch (SQLException e) {
+            logger.warn("Type for column {} can not be retrieved in {} ignoring exception: {}", columnName, tableName, e.getMessage());
+        }
+        return null;
+    }
+
+    public void addColumn(Connection conn, String tableName, String columnName, String columnDefinition) {
+        try (PreparedStatement pstmt = conn.prepareStatement(String.format("ALTER TABLE %s ADD COLUMN %s %s", tableName, columnName, columnDefinition));){
+            pstmt.executeUpdate();
+            logger.debug("Column {} is added successfully from the table {}", columnName, tableName);
+        } catch (SQLException e) {
+            logger.warn("Unable to add column {} to table {} due to exception", columnName, tableName, e);
+        }
+    }
+
+    public void changeColumn(Connection conn, String tableName, String oldColumnName, String newColumnName, String columnDefinition) {
+        try (PreparedStatement pstmt = conn.prepareStatement(String.format("ALTER TABLE %s CHANGE COLUMN %s %s %s", tableName, oldColumnName, newColumnName, columnDefinition));){
+            pstmt.executeUpdate();
+            logger.debug("Column {} is changed successfully to {} from the table {}", oldColumnName, newColumnName, tableName);
+        } catch (SQLException e) {
+            logger.warn("Unable to add column {} to {} from the table {} due to exception", oldColumnName, newColumnName, tableName, e);
+        }
+    }
+
     public String generateIndexName(String tableName, String... columnName) {
         return String.format("i_%s__%s", tableName, StringUtils.join(columnName, "__"));
     }

engine/schema/src/main/java/com/cloud/upgrade/dao/DbUpgradeUtils.java

@@ -58,4 +58,20 @@ public class DbUpgradeUtils {
         }
     }
 
+    public static String getTableColumnType(Connection conn, String tableName, String columnName) {
+        return dao.getColumnType(conn, tableName, columnName);
+    }
+
+    public static void addTableColumnIfNotExist(Connection conn, String tableName, String columnName, String columnDefinition) {
+        if (!dao.columnExists(conn, tableName, columnName)) {
+            dao.addColumn(conn, tableName, columnName, columnDefinition);
+        }
+    }
+
+    public static void changeTableColumnIfNotExist(Connection conn, String tableName, String oldColumnName, String newColumnName, String columnDefinition) {
+        if (dao.columnExists(conn, tableName, oldColumnName)) {
+            dao.changeColumn(conn, tableName, oldColumnName, newColumnName, columnDefinition);
+        }
+    }
+
 }

engine/schema/src/main/java/com/cloud/upgrade/dao/Upgrade42010to42100.java

@@ -17,10 +17,16 @@
 package com.cloud.upgrade.dao;
 
 import com.cloud.upgrade.SystemVmTemplateRegistration;
+import com.cloud.utils.db.TransactionLegacy;
 import com.cloud.utils.exception.CloudRuntimeException;
 
 import java.io.InputStream;
 import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.cloudstack.framework.config.ConfigKey;
 
 public class Upgrade42010to42100 extends DbUpgradeAbstractImpl implements DbUpgrade, DbUpgradeSystemVmTemplate {
     private SystemVmTemplateRegistration systemVmTemplateRegistration;
@@ -53,6 +59,7 @@ public class Upgrade42010to42100 extends DbUpgradeAbstractImpl implements DbUpgr
 
     @Override
     public void performDataMigration(Connection conn) {
+        migrateConfigurationScopeToBitmask(conn);
     }
 
     @Override
@@ -80,4 +87,35 @@ public class Upgrade42010to42100 extends DbUpgradeAbstractImpl implements DbUpgr
             throw new CloudRuntimeException("Failed to find / register SystemVM template(s)");
         }
     }
+
+    protected void migrateConfigurationScopeToBitmask(Connection conn) {
+        String scopeDataType = DbUpgradeUtils.getTableColumnType(conn, "configuration", "scope");
+        logger.info("Data type of the column scope of table configuration is {}", scopeDataType);
+        if (!"varchar(255)".equals(scopeDataType)) {
+            return;
+        }
+        DbUpgradeUtils.addTableColumnIfNotExist(conn, "configuration", "new_scope", "BIGINT DEFAULT 0");
+        migrateExistingConfigurationScopeValues(conn);
+        DbUpgradeUtils.dropTableColumnsIfExist(conn, "configuration", List.of("scope"));
+        DbUpgradeUtils.changeTableColumnIfNotExist(conn, "configuration", "new_scope", "scope", "BIGINT NOT NULL DEFAULT 0 COMMENT 'Bitmask for scope(s) of this parameter'");
+    }
+
+    protected void migrateExistingConfigurationScopeValues(Connection conn) {
+        StringBuilder sql = new StringBuilder("UPDATE configuration\n" +
+                "SET new_scope = " +
+                "    CASE ");
+        for (ConfigKey.Scope scope : ConfigKey.Scope.values()) {
+            sql.append("        WHEN scope = '").append(scope.name()).append("' THEN ").append(scope.getBitValue()).append(" ");
+        }
+        sql.append("        ELSE 0 " +
+                "    END " +
+                "WHERE scope IS NOT NULL;");
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
+        try (PreparedStatement pstmt = txn.prepareAutoCloseStatement(sql.toString())) {
+            pstmt.executeUpdate();
+        } catch (SQLException e) {
+            logger.error("Failed to migrate existing configuration scope values to bitmask", e);
+            throw new CloudRuntimeException(String.format("Failed to migrate existing configuration scope values to bitmask due to: %s", e.getMessage()));
+        }
+    }
 }

engine/schema/src/main/java/com/cloud/user/AccountDetailsDaoImpl.java

@@ -33,6 +33,7 @@ import com.cloud.domain.DomainVO;
 import com.cloud.domain.dao.DomainDao;
 import com.cloud.domain.dao.DomainDetailsDao;
 import com.cloud.user.dao.AccountDao;
+import com.cloud.utils.Pair;
 import com.cloud.utils.db.QueryBuilder;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
@@ -156,4 +157,13 @@ public class AccountDetailsDaoImpl extends ResourceDetailsDaoBase<AccountDetailV
         }
         return value;
     }
+
+    @Override
+    public Pair<Scope, Long> getParentScope(long id) {
+        Account account = _accountDao.findById(id);
+        if (account == null) {
+            return null;
+        }
+        return new Pair<>(getScope().getParent(), account.getDomainId());
+    }
 }

engine/schema/src/main/java/org/apache/cloudstack/storage/datastore/db/ImageStoreDetailsDaoImpl.java

@@ -20,6 +20,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import javax.inject.Inject;
+
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.ConfigKey.Scope;
@@ -27,6 +29,8 @@ import org.apache.cloudstack.framework.config.ScopedConfigStorage;
 import org.apache.cloudstack.resourcedetail.ResourceDetailsDaoBase;
 import org.springframework.stereotype.Component;
 
+import com.cloud.storage.ImageStore;
+import com.cloud.utils.Pair;
 import com.cloud.utils.crypt.DBEncryptionUtil;
 import com.cloud.utils.db.QueryBuilder;
 import com.cloud.utils.db.SearchBuilder;
@@ -36,6 +40,9 @@ import com.cloud.utils.db.TransactionLegacy;
 
 @Component
 public class ImageStoreDetailsDaoImpl extends ResourceDetailsDaoBase<ImageStoreDetailVO> implements ImageStoreDetailsDao, ScopedConfigStorage {
+    @Inject
+    ImageStoreDao imageStoreDao;
+
     protected final SearchBuilder<ImageStoreDetailVO> storeSearch;
 
     public ImageStoreDetailsDaoImpl() {
@@ -113,10 +120,20 @@ public class ImageStoreDetailsDaoImpl extends ResourceDetailsDaoBase<ImageStoreD
     public String getConfigValue(long id, ConfigKey<?> key) {
         ImageStoreDetailVO vo = findDetail(id, key.key());
         return vo == null ? null : getActualValue(vo);
-        }
+    }
 
     @Override
     public void addDetail(long resourceId, String key, String value, boolean display) {
         super.addDetail(new ImageStoreDetailVO(resourceId, key, value, display));
     }
+
+    @Override
+    public Pair<Scope, Long> getParentScope(long id) {
+        ImageStore store = imageStoreDao.findById(id);
+        if (store == null) {
+            return null;
+        }
+        return new Pair<>(getScope().getParent(), store.getDataCenterId());
+    }
+
 }

engine/schema/src/test/java/com/cloud/upgrade/ConfigurationGroupsAggregatorTest.java

@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.upgrade;
+
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.framework.config.dao.ConfigurationGroupDao;
+import org.apache.cloudstack.framework.config.dao.ConfigurationSubGroupDao;
+import org.apache.cloudstack.framework.config.impl.ConfigurationSubGroupVO;
+import org.apache.cloudstack.framework.config.impl.ConfigurationVO;
+import org.apache.logging.log4j.Logger;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ConfigurationGroupsAggregatorTest {
+    @InjectMocks
+    private ConfigurationGroupsAggregator configurationGroupsAggregator = new ConfigurationGroupsAggregator();
+
+    @Mock
+    private ConfigurationDao configDao;
+
+    @Mock
+    private ConfigurationGroupDao configGroupDao;
+
+    @Mock
+    private ConfigurationSubGroupDao configSubGroupDao;
+
+    @Mock
+    private Logger logger;
+
+    @Test
+    public void testUpdateConfigurationGroups() {
+        ConfigurationVO config = new ConfigurationVO("Advanced", "DEFAULT", "management-server",
+                "test.config.name", null, "description");
+        config.setGroupId(1L);
+        config.setSubGroupId(1L);
+
+        when(configDao.searchPartialConfigurations()).thenReturn(Collections.singletonList(config));
+
+        ConfigurationSubGroupVO configSubGroup = Mockito.mock(ConfigurationSubGroupVO.class);
+        when(configSubGroupDao.findByName("name")).thenReturn(configSubGroup);
+        Mockito.when(configSubGroup.getId()).thenReturn(10L);
+        Mockito.when(configSubGroup.getGroupId()).thenReturn(5L);
+
+        configurationGroupsAggregator.updateConfigurationGroups();
+
+        Assert.assertEquals(Long.valueOf(5), config.getGroupId());
+        Assert.assertEquals(Long.valueOf(10), config.getSubGroupId());
+        Mockito.verify(configDao, Mockito.times(1)).persist(config);
+        Mockito.verify(logger, Mockito.times(1)).debug("Updating configuration groups");
+        Mockito.verify(logger, Mockito.times(1)).debug("Successfully updated configuration groups.");
+    }
+}

engine/schema/src/test/java/com/cloud/upgrade/dao/DatabaseAccessObjectTest.java

@@ -511,4 +511,57 @@ public class DatabaseAccessObjectTest {
         verify(loggerMock, times(1)).warn(anyString(), eq(sqlException));
     }
 
+    @Test
+    public void testGetColumnType() throws Exception {
+        when(connectionMock.prepareStatement(contains("DESCRIBE"))).thenReturn(preparedStatementMock);
+        when(preparedStatementMock.executeQuery()).thenReturn(resultSetMock);
+        when(resultSetMock.next()).thenReturn(true);
+        when(resultSetMock.getString("Type")).thenReturn("type");
+
+        Connection conn = connectionMock;
+        String tableName = "tableName";
+        String columnName = "columnName";
+
+        Assert.assertEquals("type", dao.getColumnType(conn, tableName, columnName));
+
+        verify(connectionMock, times(1)).prepareStatement(anyString());
+        verify(preparedStatementMock, times(1)).executeQuery();
+        verify(preparedStatementMock, times(1)).close();
+        verify(loggerMock, times(0)).debug(anyString());
+    }
+
+    @Test
+    public void testAddColumn() throws Exception {
+        when(connectionMock.prepareStatement(contains("ADD COLUMN"))).thenReturn(preparedStatementMock);
+        when(preparedStatementMock.executeUpdate()).thenReturn(1);
+
+        Connection conn = connectionMock;
+        String tableName = "tableName";
+        String columnName = "columnName";
+        String columnType = "columnType";
+
+        dao.addColumn(conn, tableName, columnName, columnType);
+
+        verify(connectionMock, times(1)).prepareStatement(anyString());
+        verify(preparedStatementMock, times(1)).executeUpdate();
+        verify(preparedStatementMock, times(1)).close();
+    }
+
+    @Test
+    public void testChangeColumn() throws Exception {
+        when(connectionMock.prepareStatement(contains("CHANGE COLUMN"))).thenReturn(preparedStatementMock);
+        when(preparedStatementMock.executeUpdate()).thenReturn(1);
+
+        Connection conn = connectionMock;
+        String tableName = "tableName";
+        String columnName = "columnName";
+        String newColumnName = "columnName2";
+        String columnDefinition = "columnDefinition";
+
+        dao.changeColumn(conn, tableName, columnName, newColumnName, columnDefinition);
+
+        verify(connectionMock, times(1)).prepareStatement(anyString());
+        verify(preparedStatementMock, times(1)).executeUpdate();
+        verify(preparedStatementMock, times(1)).close();
+    }
 }

engine/schema/src/test/java/com/cloud/upgrade/dao/DbUpgradeUtilsTest.java

@@ -159,4 +159,33 @@ public class DbUpgradeUtilsTest {
         verify(daoMock, times(1)).columnExists(conn, tableName, column3);
         verify(daoMock, times(1)).dropColumn(conn, tableName, column3);
     }
+
+    @Test
+    public void testAddTableColumnIfNotExist() throws Exception {
+        Connection conn = connectionMock;
+        String tableName = "tableName";
+        String columnName = "columnName";
+        String columnDefinition = "columnDefinition";
+        when(daoMock.columnExists(conn, tableName, columnName)).thenReturn(false);
+
+        DbUpgradeUtils.addTableColumnIfNotExist(conn, tableName, columnName, columnDefinition);
+
+        verify(daoMock, times(1)).columnExists(conn, tableName, columnName);
+        verify(daoMock, times(1)).addColumn(conn, tableName, columnName, columnDefinition);
+    }
+
+    @Test
+    public void testChangeTableColumnIfNotExist() throws Exception {
+        Connection conn = connectionMock;
+        String tableName = "tableName";
+        String oldColumnName = "oldColumnName";
+        String newColumnName = "newColumnName";
+        String columnDefinition = "columnDefinition";
+        when(daoMock.columnExists(conn, tableName, oldColumnName)).thenReturn(true);
+
+        DbUpgradeUtils.changeTableColumnIfNotExist(conn, tableName, oldColumnName, newColumnName, columnDefinition);
+
+        verify(daoMock, times(1)).columnExists(conn, tableName, oldColumnName);
+        verify(daoMock, times(1)).changeColumn(conn, tableName, oldColumnName, newColumnName, columnDefinition);
+    }
 }

engine/schema/src/test/java/com/cloud/upgrade/dao/Upgrade42010to42100Test.java

@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.upgrade.dao;
+
+import static org.mockito.Mockito.when;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import com.cloud.utils.db.TransactionLegacy;
+
+@RunWith(MockitoJUnitRunner.class)
+public class Upgrade42010to42100Test {
+    @Spy
+    Upgrade42010to42100 upgrade;
+
+    @Mock
+    private Connection conn;
+
+    @Test
+    public void testPerformDataMigration() throws SQLException {
+        try (MockedStatic<DbUpgradeUtils> ignored = Mockito.mockStatic(DbUpgradeUtils.class)) {
+            DbUpgradeUtils dbUpgradeUtils = Mockito.mock(DbUpgradeUtils.class);
+            when(dbUpgradeUtils.getTableColumnType(conn, "configuration", "scope")).thenReturn("varchar(255)");
+
+            try (MockedStatic<TransactionLegacy> ignored2 = Mockito.mockStatic(TransactionLegacy.class)) {
+                TransactionLegacy txn = Mockito.mock(TransactionLegacy.class);
+                when(TransactionLegacy.currentTxn()).thenReturn(txn);
+                PreparedStatement pstmt = Mockito.mock(PreparedStatement.class);
+                String sql = "UPDATE configuration\n" +
+                        "SET new_scope =" +
+                        "     CASE" +
+                        "         WHEN scope = 'Global' THEN 1" +
+                        "         WHEN scope = 'Zone' THEN 2" +
+                        "         WHEN scope = 'Cluster' THEN 4" +
+                        "         WHEN scope = 'StoragePool' THEN 8" +
+                        "         WHEN scope = 'ManagementServer' THEN 16" +
+                        "         WHEN scope = 'ImageStore' THEN 32" +
+                        "         WHEN scope = 'Domain' THEN 64" +
+                        "         WHEN scope = 'Account' THEN 128" +
+                        "         ELSE 0" +
+                        "     END WHERE scope IS NOT NULL;";
+                when(txn.prepareAutoCloseStatement(sql)).thenReturn(pstmt);
+                upgrade.performDataMigration(conn);
+
+                Mockito.verify(pstmt, Mockito.times(1)).executeUpdate();
+            }
+        }
+    }
+}

framework/config/src/main/java/org/apache/cloudstack/config/Configuration.java

@@ -17,6 +17,9 @@
 package org.apache.cloudstack.config;
 
 import java.util.Date;
+import java.util.List;
+
+import org.apache.cloudstack.framework.config.ConfigKey;
 
 /**
  * Configuration represents one global configuration parameter for CloudStack.
@@ -74,7 +77,9 @@ public interface Configuration {
      * always global.  A non-null value indicates that this parameter can be
      * set at a certain organization level.
      */
-    String getScope();
+    int getScope();
+
+    List<ConfigKey.Scope> getScopes();
 
     /**
      * @return can the configuration parameter be changed without restarting the server.

framework/config/src/main/java/org/apache/cloudstack/framework/config/ConfigDepot.java

@@ -18,6 +18,8 @@ package org.apache.cloudstack.framework.config;
 
 import java.util.Set;
 
+import com.cloud.utils.Pair;
+
 /**
  * ConfigDepot is a repository of configurations.
  *
@@ -34,4 +36,5 @@ public interface ConfigDepot {
     boolean isNewConfig(ConfigKey<?> configKey);
     String getConfigStringValue(String key, ConfigKey.Scope scope, Long scopeId);
     void invalidateConfigCache(String key, ConfigKey.Scope scope, Long scopeId);
+    Pair<ConfigKey.Scope, Long> getParentScope(ConfigKey.Scope scope, Long id);
 }

framework/config/src/main/java/org/apache/cloudstack/framework/config/ConfigKey.java

@@ -17,8 +17,14 @@
 package org.apache.cloudstack.framework.config;
 
 import java.sql.Date;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 import org.apache.cloudstack.framework.config.impl.ConfigDepotImpl;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import com.cloud.utils.Pair;
 import com.cloud.utils.Ternary;
@@ -30,6 +36,7 @@ import com.cloud.utils.exception.CloudRuntimeException;
  *
  */
 public class ConfigKey<T> {
+    private static final Logger logger = LogManager.getLogger(ConfigKey.class);
 
     public static final String CATEGORY_ADVANCED = "Advanced";
     public static final String CATEGORY_ALERT = "Alert";
@@ -37,7 +44,89 @@ public class ConfigKey<T> {
     public static final String CATEGORY_SYSTEM = "System";
 
     public enum Scope {
-        Global, Zone, Cluster, StoragePool, Account, ManagementServer, ImageStore, Domain
+        Global(null, 1),
+        Zone(Global, 1 << 1),
+        Cluster(Zone, 1 << 2),
+        StoragePool(Cluster, 1 << 3),
+        ManagementServer(Global, 1 << 4),
+        ImageStore(Zone, 1 << 5),
+        Domain(Global, 1 << 6),
+        Account(Domain, 1 << 7);
+
+        private final Scope parent;
+        private final int bitValue;
+
+        Scope(Scope parent, int bitValue) {
+            this.parent = parent;
+            this.bitValue = bitValue;
+        }
+
+        public Scope getParent() {
+            return parent;
+        }
+
+        public int getBitValue() {
+            return bitValue;
+        }
+
+        public boolean isDescendantOf(Scope other) {
+            Scope parent = this.getParent();
+            while (parent != null) {
+                if (parent == other) {
+                    return true;
+                }
+                parent = parent.getParent();
+            }
+            return false;
+        }
+
+        public static List<Scope> getAllDescendants(String str) {
+            Scope s1 = Scope.valueOf(str);
+            List<Scope> scopes = new ArrayList<>();
+            for (Scope s : Scope.values()) {
+                if (s.isDescendantOf(s1)) {
+                    scopes.add(s);
+                }
+            }
+            return scopes;
+        }
+
+        public static List<Scope> decode(int bitmask) {
+            if (bitmask == 0) {
+                return Collections.emptyList();
+            }
+            List<Scope> scopes = new ArrayList<>();
+            for (Scope scope : Scope.values()) {
+                if ((bitmask & scope.getBitValue()) != 0) {
+                    scopes.add(scope);
+                }
+            }
+            return scopes;
+        }
+
+        public static String decodeAsCsv(int bitmask) {
+            if (bitmask == 0) {
+                return null;
+            }
+            StringBuilder builder = new StringBuilder();
+            for (Scope scope : Scope.values()) {
+                if ((bitmask & scope.getBitValue()) != 0) {
+                    builder.append(scope.name()).append(", ");
+                }
+            }
+            if (builder.length() > 0) {
+                builder.setLength(builder.length() - 2);
+            }
+            return builder.toString();
+        }
+
+        public static int getBitmask(Scope... scopes) {
+            int bitmask = 0;
+            for (Scope scope : scopes) {
+                bitmask |= scope.getBitValue();
+            }
+            return bitmask;
+        }
     }
 
     public enum Kind {
@@ -70,8 +159,8 @@ public class ConfigKey<T> {
         return _displayText;
     }
 
-    public Scope scope() {
+    public List<Scope> getScopes() {
-        return _scope;
+        return scopes;
     }
 
     public boolean isDynamic() {
@@ -108,7 +197,7 @@ public class ConfigKey<T> {
     private final String _defaultValue;
     private final String _description;
     private final String _displayText;
-    private final Scope _scope; // Parameter can be at different levels (Zone/cluster/pool/account), by default every parameter is at global
+    private final List<Scope> scopes; // Parameter can be at different levels (Zone/cluster/pool/account), by default every parameter is at global
     private final boolean _isDynamic;
     private final String _parent;
     private final Ternary<String, String, Long> _group; // Group name, description with precedence
@@ -128,6 +217,10 @@ public class ConfigKey<T> {
         this(type, name, category, defaultValue, description, isDynamic, scope, null);
     }
 
+    public ConfigKey(String category, Class<T> type, String name, String defaultValue, String description, boolean isDynamic, List<Scope> scopes) {
+        this(type, name, category, defaultValue, description, isDynamic, scopes, null);
+    }
+
     public ConfigKey(String category, Class<T> type, String name, String defaultValue, String description, boolean isDynamic, Scope scope, String parent) {
         this(type, name, category, defaultValue, description, isDynamic, scope, null, null, parent, null, null, null, null);
     }
@@ -148,6 +241,10 @@ public class ConfigKey<T> {
         this(type, name, category, defaultValue, description, isDynamic, scope, multiplier, null, null, null, null, null, null);
     }
 
+    public ConfigKey(Class<T> type, String name, String category, String defaultValue, String description, boolean isDynamic, List<Scope> scopes, T multiplier) {
+        this(type, name, category, defaultValue, description, isDynamic, scopes, multiplier, null, null, null, null, null, null);
+    }
+
     public ConfigKey(Class<T> type, String name, String category, String defaultValue, String description, boolean isDynamic, Scope scope, T multiplier, String parent) {
         this(type, name, category, defaultValue, description, isDynamic, scope, multiplier, null, parent, null, null, null, null);
     }
@@ -159,13 +256,22 @@ public class ConfigKey<T> {
 
     public ConfigKey(Class<T> type, String name, String category, String defaultValue, String description, boolean isDynamic, Scope scope, T multiplier,
                      String displayText, String parent, Ternary<String, String, Long> group, Pair<String, Long> subGroup, Kind kind, String options) {
+        this(type, name, category, defaultValue, description, isDynamic, scope == null ? null : List.of(scope), multiplier,
+                displayText, parent, group, subGroup, kind, options);
+    }
+
+    public ConfigKey(Class<T> type, String name, String category, String defaultValue, String description, boolean isDynamic, List<Scope> scopes, T multiplier,
+                     String displayText, String parent, Ternary<String, String, Long> group, Pair<String, Long> subGroup, Kind kind, String options) {
         _category = category;
         _type = type;
         _name = name;
         _defaultValue = defaultValue;
         _description = description;
         _displayText = displayText;
-        _scope = scope;
+        this.scopes = new ArrayList<>();
+        if (scopes != null) {
+            this.scopes.addAll(scopes);
+        }
         _isDynamic = isDynamic;
         _multiplier = multiplier;
         _parent = parent;
@@ -218,28 +324,45 @@ public class ConfigKey<T> {
             String value = s_depot != null ? s_depot.getConfigStringValue(_name, Scope.Global, null) : null;
             _value = valueOf((value == null) ? defaultValue() : value);
         }
-
         return _value;
     }
 
-    protected T valueInScope(Scope scope, Long id) {
+    protected T valueInGlobalOrAvailableParentScope(Scope scope, Long id) {
+        if (scopes.size() <= 1) {
+            return value();
+        }
+        Pair<Scope, Long> s = new Pair<>(scope, id);
+        do {
+            s = s_depot != null ? s_depot.getParentScope(s.first(), s.second()) : null;
+            if (s != null && scopes.contains(s.first())) {
+                return valueInScope(s.first(), s.second());
+            }
+        } while (s != null);
+        logger.trace("Global value for config ({}): {}", _name, _value);
+        return value();
+    }
+
+    public T valueInScope(Scope scope, Long id) {
         if (id == null) {
             return value();
         }
-
         String value = s_depot != null ? s_depot.getConfigStringValue(_name, scope, id) : null;
         if (value == null) {
-            return value();
+            return valueInGlobalOrAvailableParentScope(scope, id);
         }
+        logger.trace("Scope({}) value for config ({}): {}", scope, _name, _value);
         return valueOf(value);
     }
 
-    public T valueIn(Long id) {
+    protected Scope getPrimaryScope() {
-        return valueInScope(_scope, id);
+        if (CollectionUtils.isNotEmpty(scopes)) {
+            return scopes.get(0);
+        }
+        return null;
     }
 
-    public T valueInDomain(Long domainId) {
+    public T valueIn(Long id) {
-        return valueInScope(Scope.Domain, domainId);
+        return valueInScope(getPrimaryScope(), id);
     }
 
     @SuppressWarnings("unchecked")
@@ -277,4 +400,20 @@ public class ConfigKey<T> {
         }
     }
 
+    public boolean isGlobalOrEmptyScope() {
+        return CollectionUtils.isEmpty(scopes) ||
+                (scopes.size() == 1 && scopes.get(0) == Scope.Global);
+    }
+
+    public int getScopeBitmask() {
+        int bitmask = 0;
+        if (CollectionUtils.isEmpty(scopes)) {
+            return bitmask;
+        }
+        for (Scope scope : scopes) {
+            bitmask |= scope.getBitValue();
+        }
+        return bitmask;
+    }
+
 }

framework/config/src/main/java/org/apache/cloudstack/framework/config/ScopedConfigStorage.java

@@ -18,6 +18,8 @@ package org.apache.cloudstack.framework.config;
 
 import org.apache.cloudstack.framework.config.ConfigKey.Scope;
 
+import com.cloud.utils.Pair;
+
 /**
  *
  * This method is used by individual storage for configuration
@@ -31,4 +33,7 @@ public interface ScopedConfigStorage {
     default String getConfigValue(long id, ConfigKey<?> key) {
         return getConfigValue(id, key.key());
     }
+    default Pair<Scope, Long> getParentScope(long id) {
+        return null;
+    }
 }

framework/config/src/main/java/org/apache/cloudstack/framework/config/dao/ConfigurationDao.java

@@ -16,6 +16,7 @@
 // under the License.
 package org.apache.cloudstack.framework.config.dao;
 
+import java.util.List;
 import java.util.Map;
 
 import org.apache.cloudstack.framework.config.impl.ConfigurationVO;
@@ -67,4 +68,6 @@ public interface ConfigurationDao extends GenericDao<ConfigurationVO, String> {
     boolean update(String name, String category, String value);
 
     void invalidateCache();
+
+    List<ConfigurationVO> searchPartialConfigurations();
 }

framework/config/src/main/java/org/apache/cloudstack/framework/config/dao/ConfigurationDaoImpl.java

@@ -43,6 +43,7 @@ public class ConfigurationDaoImpl extends GenericDaoBase<ConfigurationVO, String
 
     final SearchBuilder<ConfigurationVO> InstanceSearch;
     final SearchBuilder<ConfigurationVO> NameSearch;
+    final SearchBuilder<ConfigurationVO> PartialSearch;
 
     public static final String UPDATE_CONFIGURATION_SQL = "UPDATE configuration SET value = ? WHERE name = ?";
 
@@ -53,6 +54,11 @@ public class ConfigurationDaoImpl extends GenericDaoBase<ConfigurationVO, String
         NameSearch = createSearchBuilder();
         NameSearch.and("name", NameSearch.entity().getName(), SearchCriteria.Op.EQ);
         setRunLevel(ComponentLifecycle.RUN_LEVEL_SYSTEM_BOOTSTRAP);
+
+        PartialSearch = createSearchBuilder();
+        PartialSearch.select("name", SearchCriteria.Func.NATIVE, PartialSearch.entity().getName());
+        PartialSearch.select("groupId", SearchCriteria.Func.NATIVE, PartialSearch.entity().getGroupId());
+        PartialSearch.select("subGroupId", SearchCriteria.Func.NATIVE, PartialSearch.entity().getSubGroupId());
     }
 
     @Override
@@ -207,4 +213,9 @@ public class ConfigurationDaoImpl extends GenericDaoBase<ConfigurationVO, String
         return findOneIncludingRemovedBy(sc);
     }
 
+    @Override
+    public List<ConfigurationVO> searchPartialConfigurations() {
+        SearchCriteria<ConfigurationVO> sc = PartialSearch.create();
+        return searchIncludingRemoved(sc, null, null, false);
+    }
 }

framework/config/src/main/java/org/apache/cloudstack/framework/config/impl/ConfigDepotImpl.java

@@ -144,9 +144,11 @@ public class ConfigDepotImpl implements ConfigDepot, ConfigDepotAdmin {
 
             createOrupdateConfigObject(date, configurable.getConfigComponentName(), key, null);
 
-            if ((key.scope() != null) && (key.scope() != ConfigKey.Scope.Global)) {
+            if (!key.isGlobalOrEmptyScope()) {
-                Set<ConfigKey<?>> currentConfigs = _scopeLevelConfigsMap.get(key.scope());
+                for (ConfigKey.Scope scope : key.getScopes()) {
-                currentConfigs.add(key);
+                    Set<ConfigKey<?>> currentConfigs = _scopeLevelConfigsMap.get(scope);
+                    currentConfigs.add(key);
+                }
             }
         }
 
@@ -204,12 +206,12 @@ public class ConfigDepotImpl implements ConfigDepot, ConfigDepotAdmin {
         } else {
             boolean configUpdated = false;
             if (vo.isDynamic() != key.isDynamic() || !ObjectUtils.equals(vo.getDescription(), key.description()) || !ObjectUtils.equals(vo.getDefaultValue(), key.defaultValue()) ||
-                !ObjectUtils.equals(vo.getScope(), key.scope().toString()) ||
+                !ObjectUtils.equals(vo.getScope(), key.getScopeBitmask()) ||
                 !ObjectUtils.equals(vo.getComponent(), componentName)) {
                 vo.setDynamic(key.isDynamic());
                 vo.setDescription(key.description());
                 vo.setDefaultValue(key.defaultValue());
-                vo.setScope(key.scope().toString());
+                vo.setScope(key.getScopeBitmask());
                 vo.setComponent(componentName);
                 vo.setUpdated(date);
                 configUpdated = true;
@@ -283,12 +285,7 @@ public class ConfigDepotImpl implements ConfigDepot, ConfigDepotAdmin {
             scopeId = Long.valueOf(parts[2]);
         } catch (IllegalArgumentException ignored) {}
         if (!ConfigKey.Scope.Global.equals(scope) && scopeId != null) {
-            ScopedConfigStorage scopedConfigStorage = null;
+            ScopedConfigStorage scopedConfigStorage = getScopedStorage(scope);
-            for (ScopedConfigStorage storage : _scopedStorages) {
-                if (storage.getScope() == scope) {
-                    scopedConfigStorage = storage;
-                }
-            }
             if (scopedConfigStorage == null) {
                 throw new CloudRuntimeException("Unable to find config storage for this scope: " + scope + " for " + key);
             }
@@ -315,26 +312,6 @@ public class ConfigDepotImpl implements ConfigDepot, ConfigDepotAdmin {
         configCache.invalidate(getConfigCacheKey(key, scope, scopeId));
     }
 
-    public ScopedConfigStorage findScopedConfigStorage(ConfigKey<?> config) {
-        for (ScopedConfigStorage storage : _scopedStorages) {
-            if (storage.getScope() == config.scope()) {
-                return storage;
-            }
-        }
-
-        throw new CloudRuntimeException("Unable to find config storage for this scope: " + config.scope() + " for " + config.key());
-    }
-
-    public ScopedConfigStorage getDomainScope(ConfigKey<?> config) {
-        for (ScopedConfigStorage storage : _scopedStorages) {
-            if (storage.getScope() == ConfigKey.Scope.Domain) {
-                return storage;
-            }
-        }
-
-        throw new CloudRuntimeException("Unable to find config storage for this scope: " + ConfigKey.Scope.Domain + " for " + config.key());
-    }
-
     public List<ScopedConfigStorage> getScopedStorages() {
         return _scopedStorages;
     }
@@ -398,4 +375,27 @@ public class ConfigDepotImpl implements ConfigDepot, ConfigDepotAdmin {
     public boolean isNewConfig(ConfigKey<?> configKey) {
         return newConfigs.contains(configKey.key());
     }
+
+    protected ScopedConfigStorage getScopedStorage(ConfigKey.Scope scope) {
+        ScopedConfigStorage scopedConfigStorage = null;
+        for (ScopedConfigStorage storage : _scopedStorages) {
+            if (storage.getScope() == scope) {
+                scopedConfigStorage = storage;
+                break;
+            }
+        }
+        return scopedConfigStorage;
+    }
+
+    @Override
+    public Pair<ConfigKey.Scope, Long> getParentScope(ConfigKey.Scope scope, Long id) {
+        if (scope.getParent() == null) {
+            return null;
+        }
+        ScopedConfigStorage scopedConfigStorage = getScopedStorage(scope);
+        if (scopedConfigStorage == null) {
+            return null;
+        }
+        return scopedConfigStorage.getParentScope(id);
+    }
 }

framework/config/src/main/java/org/apache/cloudstack/framework/config/impl/ConfigurationVO.java

@@ -17,6 +17,7 @@
 package org.apache.cloudstack.framework.config.impl;
 
 import java.util.Date;
+import java.util.List;
 
 import javax.persistence.Column;
 import javax.persistence.Entity;
@@ -60,7 +61,7 @@ public class ConfigurationVO implements Configuration {
     private boolean dynamic;
 
     @Column(name = "scope")
-    private String scope;
+    private Integer scope;
 
     @Column(name = "updated")
     @Temporal(value = TemporalType.TIMESTAMP)
@@ -102,6 +103,7 @@ public class ConfigurationVO implements Configuration {
         this.name = name;
         this.description = description;
         this.parent = parentConfigName;
+        this.scope = 0;
         setValue(value);
         setDisplayText(displayText);
         setGroupId(groupId);
@@ -112,7 +114,7 @@ public class ConfigurationVO implements Configuration {
         this(key.category(), "DEFAULT", component, key.key(), key.defaultValue(), key.description(), key.displayText(), key.parent());
         defaultValue = key.defaultValue();
         dynamic = key.isDynamic();
-        scope = key.scope() != null ? key.scope().toString() : null;
+        scope = key.getScopeBitmask();
     }
 
     @Override
@@ -183,10 +185,15 @@ public class ConfigurationVO implements Configuration {
     }
 
     @Override
-    public String getScope() {
+    public int getScope() {
         return scope;
     }
 
+    @Override
+    public List<ConfigKey.Scope> getScopes() {
+        return ConfigKey.Scope.decode(scope);
+    }
+
     @Override
     public boolean isDynamic() {
         return dynamic;
@@ -205,7 +212,7 @@ public class ConfigurationVO implements Configuration {
         this.defaultValue = defaultValue;
     }
 
-    public void setScope(String scope) {
+    public void setScope(int scope) {
         this.scope = scope;
     }
 

framework/config/src/test/java/org/apache/cloudstack/framework/config/ConfigKeyTest.java

@@ -16,6 +16,8 @@
 // under the License.
 package org.apache.cloudstack.framework.config;
 
+import java.util.List;
+
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -47,4 +49,31 @@ public class ConfigKeyTest {
         ConfigKey key = new ConfigKey("hond", Boolean.class, "naam", "truus", "thrown name", false);
         Assert.assertFalse("zero and 0L should be considered the same address", key.isSameKeyAs(0L));
     }
+
+    @Test
+    public void testDecode() {
+        ConfigKey key = new ConfigKey("testcategoey", Boolean.class, "test", "true", "test descriptuin", false, List.of(Scope.Zone, Scope.StoragePool));
+        int bitmask = key.getScopeBitmask();
+        List<Scope> scopes = ConfigKey.Scope.decode(bitmask);
+        Assert.assertEquals(bitmask, ConfigKey.Scope.getBitmask(scopes.toArray(new Scope[0])));
+        for (Scope scope : scopes) {
+            Assert.assertTrue(scope == Scope.Zone || scope == Scope.StoragePool);
+        }
+    }
+
+    @Test
+    public void testDecodeAsCsv() {
+        ConfigKey key = new ConfigKey("testcategoey", Boolean.class, "test", "true", "test descriptuin", false, List.of(Scope.Zone, Scope.StoragePool));
+        int bitmask = key.getScopeBitmask();
+        String scopes = ConfigKey.Scope.decodeAsCsv(bitmask);
+        Assert.assertTrue("Zone, StoragePool".equals(scopes));
+    }
+
+    @Test
+    public void testGetDescendants() {
+        List<Scope> descendants = ConfigKey.Scope.getAllDescendants(Scope.Zone.name());
+        for (Scope descendant : descendants) {
+            Assert.assertTrue(descendant == Scope.Cluster || descendant == Scope.StoragePool || descendant == Scope.ImageStore);
+        }
+    }
 }

framework/config/src/test/java/org/apache/cloudstack/framework/config/impl/ConfigDepotImplTest.java

@@ -20,10 +20,14 @@ package org.apache.cloudstack.framework.config.impl;
 
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 
 import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.config.Configurable;
+import org.apache.cloudstack.framework.config.ScopedConfigStorage;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.framework.config.dao.ConfigurationSubGroupDao;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -33,11 +37,15 @@ import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import com.cloud.utils.Pair;
+
 @RunWith(MockitoJUnitRunner.class)
 public class ConfigDepotImplTest {
 
     @Mock
     ConfigurationDao _configDao;
+    @Mock
+    ConfigurationSubGroupDao configSubGroupDao;
 
     @InjectMocks
     private ConfigDepotImpl configDepotImpl = new ConfigDepotImpl();
@@ -107,4 +115,76 @@ public class ConfigDepotImplTest {
         runTestGetConfigStringValueExpiry(((ConfigDepotImpl.CONFIG_CACHE_EXPIRE_SECONDS) + 5) * 1000,
                 2);
     }
+
+    @Test
+    public void testPopulateConfigurationNewVO() {
+        ConfigKey StorageDisableThreshold = new ConfigKey<>(ConfigKey.CATEGORY_ALERT, Double.class, "pool.storage.capacity.disablethreshold", "0.85",
+                "Percentage (as a value between 0 and 1) of storage utilization above which allocators will disable using the pool for low storage available.",
+                true, List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
+        Configurable configurable = new Configurable() {
+            @Override
+            public String getConfigComponentName() {
+                return "test";
+            }
+
+            @Override
+            public ConfigKey<?>[] getConfigKeys() {
+                return new ConfigKey<?>[] { StorageDisableThreshold };
+            }
+        };
+        configDepotImpl.setConfigurables(List.of(configurable));
+        configDepotImpl.populateConfigurations();
+
+        Assert.assertEquals("pool.storage.capacity.disablethreshold",
+                configDepotImpl._scopeLevelConfigsMap.get(ConfigKey.Scope.Zone).iterator().next().key());
+        Assert.assertEquals("pool.storage.capacity.disablethreshold",
+                configDepotImpl._scopeLevelConfigsMap.get(ConfigKey.Scope.StoragePool).iterator().next().key());
+        Assert.assertEquals(0, configDepotImpl._scopeLevelConfigsMap.get(ConfigKey.Scope.Cluster).size());
+    }
+
+    @Test
+    public void testPopulateConfiguration() {
+        ConfigKey StorageDisableThreshold = new ConfigKey<>(ConfigKey.CATEGORY_ALERT, Double.class, "pool.storage.capacity.disablethreshold", "0.85",
+                "Percentage (as a value between 0 and 1) of storage utilization above which allocators will disable using the pool for low storage available.",
+                true, List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
+        Configurable configurable = new Configurable() {
+            @Override
+            public String getConfigComponentName() {
+                return "test";
+            }
+
+            @Override
+            public ConfigKey<?>[] getConfigKeys() {
+                return new ConfigKey<?>[]{StorageDisableThreshold};
+            }
+        };
+        configDepotImpl.setConfigurables(List.of(configurable));
+
+        ConfigurationVO configurationVO = new ConfigurationVO(StorageDisableThreshold.category(), "DEFAULT", "component",
+                StorageDisableThreshold.key(), StorageDisableThreshold.defaultValue(), StorageDisableThreshold.description(),
+                StorageDisableThreshold.displayText(), StorageDisableThreshold.parent(), 1L, 10L);
+        Mockito.when(_configDao.findById("pool.storage.capacity.disablethreshold")).thenReturn(configurationVO);
+        configDepotImpl.populateConfigurations();
+
+        Mockito.verify(_configDao, Mockito.times(1)).persist(configurationVO);
+    }
+
+    @Test
+    public void getParentScopeWithValidScope() {
+        ConfigKey.Scope scope = ConfigKey.Scope.Cluster;
+        ScopedConfigStorage scopedConfigStorage = Mockito.mock(ScopedConfigStorage.class);
+        Long id = 1L;
+        ConfigKey.Scope parentScope = ConfigKey.Scope.Zone;
+        Long parentId = 2L;
+
+        Mockito.when(scopedConfigStorage.getScope()).thenReturn(scope);
+        Mockito.when(scopedConfigStorage.getParentScope(id)).thenReturn(new Pair<>(parentScope, parentId));
+
+        configDepotImpl.setScopedStorages(Collections.singletonList(scopedConfigStorage));
+        Pair<ConfigKey.Scope, Long> result = configDepotImpl.getParentScope(scope, id);
+
+        Assert.assertNotNull(result);
+        Assert.assertEquals(parentScope, result.first());
+        Assert.assertEquals(parentId, result.second());
+    }
 }

framework/db/src/main/java/com/cloud/utils/db/SearchBase.java

@@ -484,6 +484,9 @@ public abstract class SearchBase<J extends SearchBase<?, T, K>, T, K> {
                     tableAlias = attr.table;
                 }
             }
+            if (op == Op.BINARY_OR) {
+                sql.append("(");
+            }
 
             sql.append(tableAlias).append(".").append(attr.columnName).append(op.toString());
             if (op == Op.IN && params.length == 1) {

framework/db/src/main/java/com/cloud/utils/db/SearchCriteria.java

@@ -38,7 +38,7 @@ public class SearchCriteria<K> {
                 " NOT BETWEEN ? AND ? ",
                 2), IN(" IN () ", -1), NOTIN(" NOT IN () ", -1), LIKE(" LIKE ? ", 1), NLIKE(" NOT LIKE ? ", 1), NIN(" NOT IN () ", -1), NULL(" IS NULL ", 0), NNULL(
                 " IS NOT NULL ",
-                0), SC(" () ", 1), TEXT("  () ", 1), RP("", 0), AND(" AND ", 0), OR(" OR ", 0), NOT(" NOT ", 0), FIND_IN_SET(" ) ", 1);
+                0), SC(" () ", 1), TEXT("  () ", 1), RP("", 0), AND(" AND ", 0), OR(" OR ", 0), NOT(" NOT ", 0), FIND_IN_SET(" ) ", 1), BINARY_OR(" & ?) > 0", 1);
 
         private final String op;
         int params;

plugins/metrics/src/main/java/org/apache/cloudstack/metrics/MetricsServiceImpl.java

@@ -80,6 +80,7 @@ import org.apache.cloudstack.response.ZoneMetricsResponse;
 import org.apache.cloudstack.storage.datastore.db.ImageStoreDao;
 import org.apache.cloudstack.storage.datastore.db.ObjectStoreDao;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.utils.bytescale.ByteScaleUtils;
 import org.apache.commons.beanutils.BeanUtils;
 import org.apache.commons.collections.CollectionUtils;
@@ -627,6 +628,7 @@ public class MetricsServiceImpl extends MutualExclusiveIdsManagerBase implements
     public List<StoragePoolMetricsResponse> listStoragePoolMetrics(List<StoragePoolResponse> poolResponses) {
         final List<StoragePoolMetricsResponse> metricsResponses = new ArrayList<>();
         Map<String, Long> clusterUuidToIdMap = clusterDao.findByUuids(poolResponses.stream().map(StoragePoolResponse::getClusterId).toArray(String[]::new)).stream().collect(Collectors.toMap(ClusterVO::getUuid, ClusterVO::getId));
+        Map<String, Long> poolUuidToIdMap = storagePoolDao.findByUuids(poolResponses.stream().map(StoragePoolResponse::getId).toArray(String[]::new)).stream().collect(Collectors.toMap(StoragePoolVO::getUuid, StoragePoolVO::getId));
         for (final StoragePoolResponse poolResponse: poolResponses) {
             StoragePoolMetricsResponse metricsResponse = new StoragePoolMetricsResponse();
 
@@ -637,8 +639,9 @@ public class MetricsServiceImpl extends MutualExclusiveIdsManagerBase implements
             }
 
             Long poolClusterId = clusterUuidToIdMap.get(poolResponse.getClusterId());
+            Long poolId = poolUuidToIdMap.get(poolResponse.getId());
             final Double storageThreshold = AlertManager.StorageCapacityThreshold.valueIn(poolClusterId);
-            final Double storageDisableThreshold = CapacityManager.StorageCapacityDisableThreshold.valueIn(poolClusterId);
+            final Double storageDisableThreshold = CapacityManager.StorageCapacityDisableThreshold.valueIn(poolId);
 
             metricsResponse.setHasAnnotation(poolResponse.hasAnnotation());
             metricsResponse.setDiskSizeUsedGB(poolResponse.getDiskSizeUsed());

plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/storage/LinstorStorageAdaptor.java

@@ -507,7 +507,7 @@ public class LinstorStorageAdaptor implements StorageAdaptor {
 
             // if there is only one template-for property left for templates, the template isn't needed anymore
             // or if it isn't a template anyway, it will not have this Aux property
-            // _cs-template-for- poperties work like a ref-count.
+            // _cs-template-for- properties work like a ref-count.
             if (rd.getProps().keySet().stream()
                     .filter(key -> key.startsWith("Aux/" + LinstorUtil.CS_TEMPLATE_FOR_PREFIX))
                     .count() == expectedProps) {

server/src/main/java/com/cloud/configuration/Config.java

@@ -19,7 +19,6 @@ package com.cloud.configuration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
-import java.util.StringTokenizer;
 
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
 import org.apache.cloudstack.engine.subsystem.api.storage.StoragePoolAllocator;
@@ -1807,26 +1806,25 @@ public enum Config {
     private final String _defaultValue;
     private final String _description;
     private final String _range;
-    private final String _scope; // Parameter can be at different levels (Zone/cluster/pool/account), by default every parameter is at global
+    private final int _scope; // Parameter can be at different levels (Zone/cluster/pool/account), by default every parameter is at global
     private final ConfigKey.Kind _kind;
     private final String _options;
 
-    private static final HashMap<String, List<Config>> s_scopeLevelConfigsMap = new HashMap<String, List<Config>>();
+    private static final HashMap<Integer, List<Config>> s_scopeLevelConfigsMap = new HashMap<>();
     static {
-        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Zone.toString(), new ArrayList<Config>());
+        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Zone.getBitValue(), new ArrayList<Config>());
-        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Cluster.toString(), new ArrayList<Config>());
+        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Cluster.getBitValue(), new ArrayList<Config>());
-        s_scopeLevelConfigsMap.put(ConfigKey.Scope.StoragePool.toString(), new ArrayList<Config>());
+        s_scopeLevelConfigsMap.put(ConfigKey.Scope.StoragePool.getBitValue(), new ArrayList<Config>());
-        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Account.toString(), new ArrayList<Config>());
+        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Account.getBitValue(), new ArrayList<Config>());
-        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Global.toString(), new ArrayList<Config>());
+        s_scopeLevelConfigsMap.put(ConfigKey.Scope.Global.getBitValue(), new ArrayList<Config>());
 
         for (Config c : Config.values()) {
             //Creating group of parameters per each level (zone/cluster/pool/account)
-            StringTokenizer tokens = new StringTokenizer(c.getScope(), ",");
+            List<ConfigKey.Scope> scopes = ConfigKey.Scope.decode(c.getScope());
-            while (tokens.hasMoreTokens()) {
+            for (ConfigKey.Scope scope : scopes) {
-                String scope = tokens.nextToken().trim();
+                List<Config> currentConfigs = s_scopeLevelConfigsMap.get(scope.getBitValue());
-                List<Config> currentConfigs = s_scopeLevelConfigsMap.get(scope);
                 currentConfigs.add(c);
-                s_scopeLevelConfigsMap.put(scope, currentConfigs);
+                s_scopeLevelConfigsMap.put(scope.getBitValue(), currentConfigs);
             }
         }
     }
@@ -1870,7 +1868,7 @@ public enum Config {
         _defaultValue = defaultValue;
         _description = description;
         _range = range;
-        _scope = ConfigKey.Scope.Global.toString();
+        _scope = ConfigKey.Scope.Global.getBitValue();
         _kind = kind;
         _options = options;
     }
@@ -1895,7 +1893,7 @@ public enum Config {
         return _type;
     }
 
-    public String getScope() {
+    public int getScope() {
         return _scope;
     }
 

server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java

@@ -1072,7 +1072,7 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
         Optional optionalValue;
         String defaultValue;
         String category;
-        String configScope;
+        List<ConfigKey.Scope> configScope;
         final ConfigurationVO config = _configDao.findByName(name);
         if (config == null) {
             configKey = _configDepot.get(name);
@@ -1082,11 +1082,11 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
             }
             defaultValue = configKey.defaultValue();
             category = configKey.category();
-            configScope = configKey.scope().toString();
+            configScope = configKey.getScopes();
         } else {
             defaultValue = config.getDefaultValue();
             category = config.getCategory();
-            configScope = config.getScope();
+            configScope = config.getScopes();
         }
 
         String scope = "";
@@ -1111,8 +1111,11 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
             throw new InvalidParameterValueException("cannot handle multiple IDs, provide only one ID corresponding to the scope");
         }
 
-        if (scope != null && !scope.equals(ConfigKey.Scope.Global.toString()) && !configScope.contains(scope)) {
+        if (scope != null) {
-            throw new InvalidParameterValueException("Invalid scope id provided for the parameter " + name);
+            ConfigKey.Scope scopeVal = ConfigKey.Scope.valueOf(scope);
+            if (!scope.equals(ConfigKey.Scope.Global.toString()) && !configScope.contains(scopeVal)) {
+                throw new InvalidParameterValueException("Invalid scope id provided for the parameter " + name);
+            }
         }
 
         String newValue = null;
@@ -1222,10 +1225,11 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
             return "Invalid configuration variable.";
         }
 
-        String configScope = cfg.getScope();
+        List<ConfigKey.Scope> configScope = cfg.getScopes();
         if (scope != null) {
-            if (!configScope.contains(scope) &&
+            ConfigKey.Scope scopeVal = ConfigKey.Scope.valueOf(scope);
-                    !(ENABLE_ACCOUNT_SETTINGS_FOR_DOMAIN.value() && configScope.contains(ConfigKey.Scope.Account.toString()) &&
+            if (!configScope.contains(scopeVal) &&
+                    !(ENABLE_ACCOUNT_SETTINGS_FOR_DOMAIN.value() && configScope.contains(ConfigKey.Scope.Account) &&
                             scope.equals(ConfigKey.Scope.Domain.toString()))) {
                 logger.error("Invalid scope id provided for the parameter " + name);
                 return "Invalid scope id provided for the parameter " + name;

server/src/main/java/com/cloud/network/IpAddressManagerImpl.java

@@ -2476,7 +2476,7 @@ public class IpAddressManagerImpl extends ManagerBase implements IpAddressManage
 
     @Override
     public PublicIpQuarantine addPublicIpAddressToQuarantine(IpAddress publicIpAddress, Long domainId) {
-        Integer quarantineDuration = PUBLIC_IP_ADDRESS_QUARANTINE_DURATION.valueInDomain(domainId);
+        Integer quarantineDuration = PUBLIC_IP_ADDRESS_QUARANTINE_DURATION.valueIn(domainId);
         if (quarantineDuration <= 0) {
             logger.debug(String.format("Not adding IP [%s] to quarantine because configuration [%s] has value equal or less to 0.", publicIpAddress.getAddress(),
                     PUBLIC_IP_ADDRESS_QUARANTINE_DURATION.key()));

server/src/main/java/com/cloud/server/ManagementServerImpl.java

@@ -892,7 +892,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
     @Inject
     protected UserVmDao _userVmDao;
     @Inject
-    private ConfigurationDao _configDao;
+    protected ConfigurationDao _configDao;
     @Inject
     private ConfigurationGroupDao _configGroupDao;
     @Inject
@@ -904,7 +904,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
     @Inject
     private DiskOfferingDao _diskOfferingDao;
     @Inject
-    private DomainDao _domainDao;
+    protected DomainDao _domainDao;
     @Inject
     private AccountDao _accountDao;
     @Inject
@@ -960,7 +960,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
     @Inject
     private HostTagsDao _hostTagsDao;
     @Inject
-    private ConfigDepot _configDepot;
+    protected ConfigDepot _configDepot;
     @Inject
     private UserVmManager _userVmMgr;
     @Inject
@@ -2192,7 +2192,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
         final String groupName = cmd.getGroupName();
         final String subGroupName = cmd.getSubGroupName();
         final String parentName = cmd.getParentName();
-        String scope = null;
+        ConfigKey.Scope scope = null;
         Long id = null;
         int paramCountCheck = 0;
 
@@ -2208,35 +2208,35 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
         }
 
         if (zoneId != null) {
-            scope = ConfigKey.Scope.Zone.toString();
+            scope = ConfigKey.Scope.Zone;
             id = zoneId;
             paramCountCheck++;
         }
         if (clusterId != null) {
-            scope = ConfigKey.Scope.Cluster.toString();
+            scope = ConfigKey.Scope.Cluster;
             id = clusterId;
             paramCountCheck++;
         }
         if (accountId != null) {
             Account account = _accountMgr.getAccount(accountId);
             _accountMgr.checkAccess(caller, null, false, account);
-            scope = ConfigKey.Scope.Account.toString();
+            scope = ConfigKey.Scope.Account;
             id = accountId;
             paramCountCheck++;
         }
         if (domainId != null) {
             _accountMgr.checkAccess(caller, _domainDao.findById(domainId));
-            scope = ConfigKey.Scope.Domain.toString();
+            scope = ConfigKey.Scope.Domain;
             id = domainId;
             paramCountCheck++;
         }
         if (storagepoolId != null) {
-            scope = ConfigKey.Scope.StoragePool.toString();
+            scope = ConfigKey.Scope.StoragePool;
             id = storagepoolId;
             paramCountCheck++;
         }
         if (imageStoreId != null) {
-            scope = ConfigKey.Scope.ImageStore.toString();
+            scope = ConfigKey.Scope.ImageStore;
             id = imageStoreId;
             paramCountCheck++;
         }
@@ -2291,19 +2291,19 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
         // hidden configurations are not displayed using the search API
         sc.addAnd("category", SearchCriteria.Op.NEQ, "Hidden");
 
-        if (scope != null && !scope.isEmpty()) {
+        if (scope != null) {
             // getting the list of parameters at requested scope
             if (ConfigurationManagerImpl.ENABLE_ACCOUNT_SETTINGS_FOR_DOMAIN.value()
-                && scope.equals(ConfigKey.Scope.Domain.toString())) {
+                && scope.equals(ConfigKey.Scope.Domain)) {
-                sc.addAnd("scope", SearchCriteria.Op.IN, ConfigKey.Scope.Domain.toString(), ConfigKey.Scope.Account.toString());
+                sc.addAnd("scope", SearchCriteria.Op.BINARY_OR, (ConfigKey.Scope.Domain.getBitValue() | ConfigKey.Scope.Account.getBitValue()));
             } else {
-                sc.addAnd("scope", SearchCriteria.Op.EQ, scope);
+                sc.addAnd("scope", SearchCriteria.Op.BINARY_OR, scope.getBitValue());
             }
         }
 
         final Pair<List<ConfigurationVO>, Integer> result = _configDao.searchAndCount(sc, searchFilter);
 
-        if (scope != null && !scope.isEmpty()) {
+        if (scope != null) {
             // Populate values corresponding the resource id
             final List<ConfigurationVO> configVOList = new ArrayList<>();
             for (final ConfigurationVO param : result.first()) {
@@ -2311,13 +2311,8 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
                 if (configVo != null) {
                     final ConfigKey<?> key = _configDepot.get(param.getName());
                     if (key != null) {
-                        if (scope.equals(ConfigKey.Scope.Domain.toString())) {
+                        Object value = key.valueInScope(scope, id);
-                            Object value = key.valueInDomain(id);
+                        configVo.setValue(value == null ? null : value.toString());
-                            configVo.setValue(value == null ? null : value.toString());
-                        } else {
-                            Object value = key.valueIn(id);
-                            configVo.setValue(value == null ? null : value.toString());
-                        }
                         configVOList.add(configVo);
                     } else {
                         logger.warn("ConfigDepot could not find parameter " + param.getName() + " for scope " + scope);

server/src/main/java/com/cloud/server/StatsCollector.java

@@ -2011,7 +2011,7 @@ public class StatsCollector extends ManagerBase implements ComponentMethodInterc
         Integer maxRetentionTime = vmStatsMaxRetentionTime.value();
         if (maxRetentionTime <= 0) {
             logger.debug(String.format("Skipping VM stats cleanup. The [%s] parameter [%s] is set to 0 or less than 0.",
-                    vmStatsMaxRetentionTime.scope(), vmStatsMaxRetentionTime.toString()));
+                    ConfigKey.Scope.decodeAsCsv(vmStatsMaxRetentionTime.getScopeBitmask()), vmStatsMaxRetentionTime.toString()));
             return;
         }
         logger.trace("Removing older VM stats records.");
@@ -2029,7 +2029,7 @@ public class StatsCollector extends ManagerBase implements ComponentMethodInterc
         if (maxRetentionTime <= 0) {
             if (logger.isDebugEnabled()) {
                 logger.debug(String.format("Skipping Volume stats cleanup. The [%s] parameter [%s] is set to 0 or less than 0.",
-                        vmDiskStatsMaxRetentionTime.scope(), vmDiskStatsMaxRetentionTime.toString()));
+                        ConfigKey.Scope.decodeAsCsv(vmDiskStatsMaxRetentionTime.getScopeBitmask()), vmDiskStatsMaxRetentionTime.toString()));
             }
             return;
         }

server/src/main/java/com/cloud/storage/StorageManagerImpl.java

@@ -3010,7 +3010,7 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
         long totalSize = pool.getCapacityBytes();
         long usedSize = getUsedSize(pool);
         double usedPercentage = ((double)usedSize / (double)totalSize);
-        double storageUsedThreshold = CapacityManager.StorageCapacityDisableThreshold.valueIn(pool.getDataCenterId());
+        double storageUsedThreshold = CapacityManager.StorageCapacityDisableThreshold.valueIn(pool.getId());
         if (logger.isDebugEnabled()) {
             logger.debug("Checking pool {} for storage, totalSize: {}, usedBytes: {}, usedPct: {}, disable threshold: {}", pool, pool.getCapacityBytes(), pool.getUsedBytes(), usedPercentage, storageUsedThreshold);
         }
@@ -3282,7 +3282,7 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
 
         logger.debug("Total capacity of the pool {} is {}", poolVO, toHumanReadableSize(totalOverProvCapacity));
 
-        double storageAllocatedThreshold = CapacityManager.StorageAllocatedCapacityDisableThreshold.valueIn(pool.getDataCenterId());
+        double storageAllocatedThreshold = CapacityManager.StorageAllocatedCapacityDisableThreshold.valueIn(pool.getId());
 
         if (logger.isDebugEnabled()) {
             logger.debug("Checking pool: {} for storage allocation , maxSize : {}, " +
@@ -3302,12 +3302,12 @@ public class StorageManagerImpl extends ManagerBase implements StorageManager, C
             if (!forVolumeResize) {
                 return false;
             }
-            if (!AllowVolumeReSizeBeyondAllocation.valueIn(pool.getDataCenterId())) {
+            if (!AllowVolumeReSizeBeyondAllocation.valueIn(pool.getId())) {
                 logger.debug(String.format("Skipping the pool %s as %s is false", pool, AllowVolumeReSizeBeyondAllocation.key()));
                 return false;
             }
 
-            double storageAllocatedThresholdForResize = CapacityManager.StorageAllocatedCapacityDisableThresholdForVolumeSize.valueIn(pool.getDataCenterId());
+            double storageAllocatedThresholdForResize = CapacityManager.StorageAllocatedCapacityDisableThresholdForVolumeSize.valueIn(pool.getId());
             if (usedPercentage > storageAllocatedThresholdForResize) {
                 logger.debug(String.format("Skipping the pool %s since its allocated percentage: %s has crossed the allocated %s: %s",
                         pool, usedPercentage, CapacityManager.StorageAllocatedCapacityDisableThresholdForVolumeSize.key(), storageAllocatedThresholdForResize));

server/src/test/java/com/cloud/configuration/ConfigurationManagerImplTest.java

@@ -46,14 +46,17 @@ import com.cloud.storage.dao.VMTemplateZoneDao;
 import com.cloud.storage.dao.VolumeDao;
 import com.cloud.user.Account;
 import com.cloud.user.User;
+import com.cloud.utils.Pair;
 import com.cloud.utils.db.EntityManager;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.net.NetUtils;
 import com.cloud.vm.dao.VMInstanceDao;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
+import org.apache.cloudstack.api.command.admin.config.ResetCfgCmd;
 import org.apache.cloudstack.api.command.admin.network.CreateNetworkOfferingCmd;
 import org.apache.cloudstack.api.command.admin.offering.UpdateDiskOfferingCmd;
 import org.apache.cloudstack.api.command.admin.zone.DeleteZoneCmd;
+import org.apache.cloudstack.config.Configuration;
 import org.apache.cloudstack.engine.subsystem.api.storage.ZoneScope;
 import org.apache.cloudstack.framework.config.ConfigDepot;
 import org.apache.cloudstack.framework.config.ConfigKey;
@@ -61,6 +64,9 @@ import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.config.impl.ConfigurationVO;
 import org.apache.cloudstack.storage.datastore.db.ImageStoreDao;
 import org.apache.cloudstack.resourcedetail.DiskOfferingDetailVO;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolDetailsDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.vm.UnmanagedVMsManager;
 import org.junit.Assert;
 import org.junit.Before;
@@ -159,6 +165,10 @@ public class ConfigurationManagerImplTest {
     NetworkService networkService;
     @Mock
     NetworkModel networkModel;
+    @Mock
+    PrimaryDataStoreDao storagePoolDao;
+    @Mock
+    StoragePoolDetailsDao storagePoolDetailsDao;
 
     DeleteZoneCmd deleteZoneCmd;
     CreateNetworkOfferingCmd createNetworkOfferingCmd;
@@ -175,7 +185,7 @@ public class ConfigurationManagerImplTest {
 
     @Before
     public void setUp() throws Exception {
-        Mockito.when(configurationVOMock.getScope()).thenReturn(ConfigKey.Scope.Global.name());
+        Mockito.when(configurationVOMock.getScopes()).thenReturn(List.of(ConfigKey.Scope.Global));
         Mockito.when(configDao.findByName(Mockito.anyString())).thenReturn(configurationVOMock);
         Mockito.when(configDepot.get(Mockito.anyString())).thenReturn(configKeyMock);
 
@@ -442,6 +452,7 @@ public class ConfigurationManagerImplTest {
         Assert.assertNotNull(offering);
     }
 
+    @Test
     public void testValidateInvalidConfiguration() {
         Mockito.doReturn(null).when(configDao).findByName(Mockito.anyString());
         String msg = configurationManagerImplSpy.validateConfigurationValue("test.config.name", "testvalue", ConfigKey.Scope.Global.toString());
@@ -451,7 +462,7 @@ public class ConfigurationManagerImplTest {
     @Test
     public void testValidateInvalidScopeForConfiguration() {
         ConfigurationVO cfg = mock(ConfigurationVO.class);
-        when(cfg.getScope()).thenReturn(ConfigKey.Scope.Account.toString());
+        when(cfg.getScopes()).thenReturn(List.of(ConfigKey.Scope.Account));
         Mockito.doReturn(cfg).when(configDao).findByName(Mockito.anyString());
         String msg = configurationManagerImplSpy.validateConfigurationValue("test.config.name", "testvalue", ConfigKey.Scope.Domain.toString());
         Assert.assertEquals("Invalid scope id provided for the parameter test.config.name", msg);
@@ -460,12 +471,12 @@ public class ConfigurationManagerImplTest {
     @Test
     public void testValidateConfig_ThreadsOnKVMHostToTransferVMwareVMFiles_Failure() {
         ConfigurationVO cfg = mock(ConfigurationVO.class);
-        when(cfg.getScope()).thenReturn(ConfigKey.Scope.Global.toString());
+        when(cfg.getScopes()).thenReturn(List.of(ConfigKey.Scope.Global));
         ConfigKey<Integer> configKey = UnmanagedVMsManager.ThreadsOnKVMHostToImportVMwareVMFiles;
         Mockito.doReturn(cfg).when(configDao).findByName(Mockito.anyString());
         Mockito.doReturn(configKey).when(configurationManagerImplSpy._configDepot).get(configKey.key());
 
-        String result = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "11", configKey.scope().toString());
+        String result = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "11", configKey.getScopes().get(0).name());
 
         Assert.assertNotNull(result);
     }
@@ -473,24 +484,24 @@ public class ConfigurationManagerImplTest {
     @Test
     public void testValidateConfig_ThreadsOnKVMHostToTransferVMwareVMFiles_Success() {
         ConfigurationVO cfg = mock(ConfigurationVO.class);
-        when(cfg.getScope()).thenReturn(ConfigKey.Scope.Global.toString());
+        when(cfg.getScopes()).thenReturn(List.of(ConfigKey.Scope.Global));
         ConfigKey<Integer> configKey = UnmanagedVMsManager.ThreadsOnKVMHostToImportVMwareVMFiles;
         Mockito.doReturn(cfg).when(configDao).findByName(Mockito.anyString());
         Mockito.doReturn(configKey).when(configurationManagerImplSpy._configDepot).get(configKey.key());
-        String msg = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "10", configKey.scope().toString());
+        String msg = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "10", configKey.getScopes().get(0).name());
         Assert.assertNull(msg);
     }
 
     @Test
     public void testValidateConfig_ConvertVmwareInstanceToKvmTimeout_Failure() {
         ConfigurationVO cfg = mock(ConfigurationVO.class);
-        when(cfg.getScope()).thenReturn(ConfigKey.Scope.Global.toString());
+        when(cfg.getScopes()).thenReturn(List.of(ConfigKey.Scope.Global));
         ConfigKey<Integer> configKey = UnmanagedVMsManager.ConvertVmwareInstanceToKvmTimeout;
         Mockito.doReturn(cfg).when(configDao).findByName(Mockito.anyString());
         Mockito.doReturn(configKey).when(configurationManagerImplSpy._configDepot).get(configKey.key());
         configurationManagerImplSpy.populateConfigValuesForValidationSet();
 
-        String result = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "0", configKey.scope().toString());
+        String result = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "0", configKey.getScopes().get(0).name());
 
         Assert.assertNotNull(result);
     }
@@ -498,12 +509,12 @@ public class ConfigurationManagerImplTest {
     @Test
     public void testValidateConfig_ConvertVmwareInstanceToKvmTimeout_Success() {
         ConfigurationVO cfg = mock(ConfigurationVO.class);
-        when(cfg.getScope()).thenReturn(ConfigKey.Scope.Global.toString());
+        when(cfg.getScopes()).thenReturn(List.of(ConfigKey.Scope.Global));
         ConfigKey<Integer> configKey = UnmanagedVMsManager.ConvertVmwareInstanceToKvmTimeout;
         Mockito.doReturn(cfg).when(configDao).findByName(Mockito.anyString());
         Mockito.doReturn(configKey).when(configurationManagerImplSpy._configDepot).get(configKey.key());
         configurationManagerImplSpy.populateConfigValuesForValidationSet();
-        String msg = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "9", configKey.scope().toString());
+        String msg = configurationManagerImplSpy.validateConfigurationValue(configKey.key(), "9", configKey.getScopes().get(0).name());
         Assert.assertNull(msg);
     }
 
@@ -851,4 +862,26 @@ public class ConfigurationManagerImplTest {
         boolean result = configurationManagerImplSpy.shouldValidateConfigRange(Config.ConsoleProxySessionMax.name(), "test", Config.ConsoleProxyUrlDomain);
         Assert.assertTrue(result);
     }
+
+    @Test
+    public void testResetConfigurations() {
+        Long poolId = 1L;
+        ResetCfgCmd cmd = Mockito.mock(ResetCfgCmd.class);
+        Mockito.when(cmd.getCfgName()).thenReturn("pool.storage.capacity.disablethreshold");
+        Mockito.when(cmd.getStoragepoolId()).thenReturn(poolId);
+        Mockito.when(cmd.getZoneId()).thenReturn(null);
+        Mockito.when(cmd.getClusterId()).thenReturn(null);
+        Mockito.when(cmd.getAccountId()).thenReturn(null);
+        Mockito.when(cmd.getDomainId()).thenReturn(null);
+        Mockito.when(cmd.getImageStoreId()).thenReturn(null);
+
+        ConfigurationVO cfg = new ConfigurationVO("Advanced", "DEFAULT", "test", "pool.storage.capacity.disablethreshold", null, "description");
+        cfg.setScope(10);
+        cfg.setDefaultValue(".85");
+        Mockito.when(configDao.findByName("pool.storage.capacity.disablethreshold")).thenReturn(cfg);
+        Mockito.when(storagePoolDao.findById(poolId)).thenReturn(Mockito.mock(StoragePoolVO.class));
+
+        Pair<Configuration, String> result = configurationManagerImplSpy.resetConfiguration(cmd);
+        Assert.assertEquals(".85", result.second());
+    }
 }

server/src/test/java/com/cloud/server/ManagementServerImplTest.java

@@ -17,6 +17,7 @@
 package com.cloud.server;
 
 import com.cloud.dc.Vlan.VlanType;
+import com.cloud.domain.dao.DomainDao;
 import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.host.DetailVO;
 import com.cloud.host.Host;
@@ -48,15 +49,20 @@ import com.cloud.vm.dao.UserVmDetailsDao;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.BaseCmd;
+import org.apache.cloudstack.api.command.admin.config.ListCfgsByCmd;
 import org.apache.cloudstack.api.command.user.address.ListPublicIpAddressesCmd;
 import org.apache.cloudstack.api.command.user.ssh.RegisterSSHKeyPairCmd;
 import org.apache.cloudstack.api.command.user.userdata.DeleteUserDataCmd;
 import org.apache.cloudstack.api.command.user.userdata.ListUserDataCmd;
 import org.apache.cloudstack.api.command.user.userdata.RegisterUserDataCmd;
+import org.apache.cloudstack.config.Configuration;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStoreDriver;
+import org.apache.cloudstack.framework.config.ConfigDepot;
 import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.framework.config.impl.ConfigurationVO;
 import org.apache.cloudstack.userdata.UserDataManager;
 import org.junit.After;
 import org.junit.Assert;
@@ -131,6 +137,16 @@ public class ManagementServerImplTest {
 
     @Mock
     HostDetailsDao hostDetailsDao;
+
+    @Mock
+    ConfigurationDao configDao;
+
+    @Mock
+    ConfigDepot configDepot;
+
+    @Mock
+    DomainDao domainDao;
+
     private AutoCloseable closeable;
 
     @Before
@@ -145,6 +161,9 @@ public class ManagementServerImplTest {
         spy._UserVmDetailsDao = userVmDetailsDao;
         spy._detailsDao = hostDetailsDao;
         spy.userDataManager = userDataManager;
+        spy._configDao = configDao;
+        spy._configDepot = configDepot;
+        spy._domainDao = domainDao;
     }
 
     @After
@@ -684,4 +703,41 @@ public class ManagementServerImplTest {
         Mockito.when(driver.zoneWideVolumesAvailableWithoutClusterMotion()).thenReturn(false);
         Assert.assertTrue(spy.zoneWideVolumeRequiresStorageMotion(dataStore, host1, host2));
     }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testSearchForConfigurationsMultipleIds() {
+        ListCfgsByCmd cmd = Mockito.mock(ListCfgsByCmd.class);
+        Mockito.when(cmd.getConfigName()).thenReturn("pool.storage.capacity.disablethreshold");
+        Mockito.when(cmd.getZoneId()).thenReturn(1L);
+        Mockito.when(cmd.getStoragepoolId()).thenReturn(2L);
+        spy.searchForConfigurations(cmd);
+    }
+
+    @Test
+    public void testSearchForConfigurations() {
+        Long poolId = 1L;
+        ListCfgsByCmd cmd = Mockito.mock(ListCfgsByCmd.class);
+        Mockito.when(cmd.getConfigName()).thenReturn("pool.storage.capacity.disablethreshold");
+        Mockito.when(cmd.getStoragepoolId()).thenReturn(poolId);
+        Mockito.when(cmd.getZoneId()).thenReturn(null);
+        Mockito.when(cmd.getClusterId()).thenReturn(null);
+        Mockito.when(cmd.getAccountId()).thenReturn(null);
+        Mockito.when(cmd.getDomainId()).thenReturn(null);
+        Mockito.when(cmd.getImageStoreId()).thenReturn(null);
+
+        SearchCriteria<ConfigurationVO> sc = Mockito.mock(SearchCriteria.class);
+        Mockito.when(configDao.createSearchCriteria()).thenReturn(sc);
+        ConfigurationVO cfg = new ConfigurationVO("Advanced", "DEFAULT", "test", "pool.storage.capacity.disablethreshold", null, "description");
+        Mockito.when(configDao.searchAndCount(any(), any())).thenReturn(new Pair<>(List.of(cfg), 1));
+        Mockito.when(configDao.findByName("pool.storage.capacity.disablethreshold")).thenReturn(cfg);
+
+        ConfigKey storageDisableThreshold = new ConfigKey<>(ConfigKey.CATEGORY_ALERT, Double.class, "pool.storage.capacity.disablethreshold", "0.85",
+                "Percentage (as a value between 0 and 1) of storage utilization above which allocators will disable using the pool for low storage available.",
+                true, List.of(ConfigKey.Scope.StoragePool, ConfigKey.Scope.Zone));
+        when(configDepot.get("pool.storage.capacity.disablethreshold")).thenReturn(storageDisableThreshold);
+
+        Pair<List<? extends Configuration>, Integer> result = spy.searchForConfigurations(cmd);
+
+        Assert.assertEquals("0.85", result.first().get(0).getValue());
+    }
 }

server/src/test/java/com/cloud/storage/StorageManagerImplTest.java

@@ -780,7 +780,6 @@ public class StorageManagerImplTest {
 
         Mockito.when(pool.getId()).thenReturn(poolId);
         Mockito.when(pool.getCapacityBytes()).thenReturn(capacityBytes);
-        Mockito.when(pool.getDataCenterId()).thenReturn(zoneId);
         Mockito.when(storagePoolDao.findById(poolId)).thenReturn(pool);
         Mockito.when(pool.getPoolType()).thenReturn(Storage.StoragePoolType.NetworkFilesystem);
 

server/src/test/java/com/cloud/vpc/dao/MockConfigurationDaoImpl.java

@@ -16,12 +16,14 @@
 // under the License.
 package com.cloud.vpc.dao;
 
-import com.cloud.utils.db.GenericDaoBase;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.config.impl.ConfigurationVO;
 
-import java.util.HashMap;
+import com.cloud.utils.db.GenericDaoBase;
-import java.util.Map;
 
 public class MockConfigurationDaoImpl extends GenericDaoBase<ConfigurationVO, String> implements ConfigurationDao {
 
@@ -117,4 +119,8 @@ public class MockConfigurationDaoImpl extends GenericDaoBase<ConfigurationVO, St
     public void invalidateCache() {
     }
 
+    @Override
+    public List<ConfigurationVO> searchPartialConfigurations() {
+        return List.of();
+    }
 }