diff --git a/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDao.java b/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDao.java
index 7a00829fd44e..0d86ca0e48c7 100644
--- a/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDao.java
+++ b/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDao.java
@@ -45,4 +45,9 @@ public interface HostTagsDao extends GenericDao<HostTagVO, Long> {
     HostTagResponse newHostTagResponse(HostTagVO hostTag);
 
     List<HostTagVO> searchByIds(Long... hostTagIds);
+
+    /**
+     * List all host tags defined on hosts within a cluster
+     */
+    List<String> listByClusterId(Long clusterId);
 }
diff --git a/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDaoImpl.java b/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDaoImpl.java
index 4aa14a31cfcf..d3fee6a26761 100644
--- a/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDaoImpl.java
+++ b/engine/schema/src/main/java/com/cloud/host/dao/HostTagsDaoImpl.java
@@ -23,6 +23,7 @@
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.Configurable;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Component;
 
@@ -43,9 +44,12 @@ public class HostTagsDaoImpl extends GenericDaoBase<HostTagVO, Long> implements
     private final SearchBuilder<HostTagVO> stSearch;
     private final SearchBuilder<HostTagVO> tagIdsearch;
     private final SearchBuilder<HostTagVO> ImplicitTagsSearch;
+    private final GenericSearchBuilder<HostTagVO, String> tagSearch;
 
     @Inject
     private ConfigurationDao _configDao;
+    @Inject
+    private HostDao hostDao;
 
     public HostTagsDaoImpl() {
         HostSearch = createSearchBuilder();
@@ -72,6 +76,11 @@ public HostTagsDaoImpl() {
         ImplicitTagsSearch.and("hostId", ImplicitTagsSearch.entity().getHostId(), SearchCriteria.Op.EQ);
         ImplicitTagsSearch.and("isImplicit", ImplicitTagsSearch.entity().getIsImplicit(), SearchCriteria.Op.EQ);
         ImplicitTagsSearch.done();
+
+        tagSearch = createSearchBuilder(String.class);
+        tagSearch.selectFields(tagSearch.entity().getTag());
+        tagSearch.and("hostIdIN", tagSearch.entity().getHostId(), SearchCriteria.Op.IN);
+        tagSearch.done();
     }
 
     @Override
@@ -235,4 +244,15 @@ public List<HostTagVO> searchByIds(Long... tagIds) {
 
         return tagList;
     }
+
+    @Override
+    public List<String> listByClusterId(Long clusterId) {
+        List<Long> hostIds = hostDao.listIdsByClusterId(clusterId);
+        if (CollectionUtils.isEmpty(hostIds)) {
+            return new ArrayList<>();
+        }
+        SearchCriteria<String> sc = tagSearch.create();
+        sc.setParameters("hostIdIN", hostIds.toArray());
+        return customSearch(sc, null);
+    }
 }
diff --git a/server/src/main/java/com/cloud/api/query/QueryManagerImpl.java b/server/src/main/java/com/cloud/api/query/QueryManagerImpl.java
index ac9f8ee1433c..fea87b66fed8 100644
--- a/server/src/main/java/com/cloud/api/query/QueryManagerImpl.java
+++ b/server/src/main/java/com/cloud/api/query/QueryManagerImpl.java
@@ -45,6 +45,7 @@
 import com.cloud.storage.dao.StoragePoolAndAccessGroupMapDao;
 import com.cloud.cluster.ManagementServerHostPeerJoinVO;
 
+import com.cloud.vm.UserVmManager;
 import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.acl.ControlledEntity.ACLType;
 import org.apache.cloudstack.acl.SecurityChecker;
@@ -4330,6 +4331,9 @@ private Pair<List<Long>, Integer> searchForServiceOfferingIdsAndCount(ListServic
         List<String> hostTags = new ArrayList<>();
         if (currentVmOffering != null) {
             hostTags.addAll(com.cloud.utils.StringUtils.csvTagsToList(currentVmOffering.getHostTag()));
+            if (UserVmManager.AllowDifferentHostTagsOfferingsForVmScale.value()) {
+                addVmCurrentClusterHostTags(vmInstance, hostTags);
+            }
         }
 
         if (!hostTags.isEmpty()) {
@@ -4341,7 +4345,7 @@ private Pair<List<Long>, Integer> searchForServiceOfferingIdsAndCount(ListServic
                     flag = false;
                     serviceOfferingSearch.op("hostTag" + tag, serviceOfferingSearch.entity().getHostTag(), Op.FIND_IN_SET);
                 } else {
-                    serviceOfferingSearch.and("hostTag" + tag, serviceOfferingSearch.entity().getHostTag(), Op.FIND_IN_SET);
+                    serviceOfferingSearch.or("hostTag" + tag, serviceOfferingSearch.entity().getHostTag(), Op.FIND_IN_SET);
                 }
             }
             serviceOfferingSearch.cp().cp();
@@ -4486,6 +4490,30 @@ private Pair<List<Long>, Integer> searchForServiceOfferingIdsAndCount(ListServic
         return new Pair<>(offeringIds, count);
     }
 
+    protected void addVmCurrentClusterHostTags(VMInstanceVO vmInstance, List<String> hostTags) {
+        if (vmInstance == null) {
+            return;
+        }
+        Long hostId = vmInstance.getHostId() == null ? vmInstance.getLastHostId() : vmInstance.getHostId();
+        if (hostId == null) {
+            return;
+        }
+        HostVO host = hostDao.findById(hostId);
+        if (host == null) {
+            logger.warn("Unable to find host with id " + hostId);
+            return;
+        }
+        List<String> clusterTags = _hostTagDao.listByClusterId(host.getClusterId());
+        if (CollectionUtils.isEmpty(clusterTags)) {
+            logger.debug("No host tags defined for hosts in the cluster " + host.getClusterId());
+            return;
+        }
+        Set<String> existingTagsSet = new HashSet<>(hostTags);
+        clusterTags.stream()
+                .filter(tag -> !existingTagsSet.contains(tag))
+                .forEach(hostTags::add);
+    }
+
     @Override
     public ListResponse<ZoneResponse> listDataCenters(ListZonesCmd cmd) {
         Pair<List<DataCenterJoinVO>, Integer> result = listDataCentersInternal(cmd);
diff --git a/server/src/main/java/com/cloud/vm/UserVmManager.java b/server/src/main/java/com/cloud/vm/UserVmManager.java
index 0a744709644c..38cb6d2db46b 100644
--- a/server/src/main/java/com/cloud/vm/UserVmManager.java
+++ b/server/src/main/java/com/cloud/vm/UserVmManager.java
@@ -108,6 +108,9 @@ public interface UserVmManager extends UserVmService {
             "Comma separated list of allowed additional VM settings if VM instance settings are read from OVA.",
             true, ConfigKey.Scope.Zone, null, null, null, null, null, ConfigKey.Kind.CSV, null);
 
+    ConfigKey<Boolean> AllowDifferentHostTagsOfferingsForVmScale = new ConfigKey<>("Advanced", Boolean.class, "allow.different.host.tags.offerings.for.vm.scale", "false",
+            "Enables/Disable allowing to change a VM offering to offerings with different host tags", true);
+
     static final int MAX_USER_DATA_LENGTH_BYTES = 2048;
 
     public  static  final String CKS_NODE = "cksnode";
diff --git a/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java b/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
index 4cb721666bd3..650eeca7114d 100644
--- a/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
+++ b/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
@@ -9394,7 +9394,7 @@ public ConfigKey<?>[] getConfigKeys() {
                 VmIpFetchThreadPoolMax, VmIpFetchTaskWorkers, AllowDeployVmIfGivenHostFails, EnableAdditionalVmConfig, DisplayVMOVFProperties,
                 KvmAdditionalConfigAllowList, XenServerAdditionalConfigAllowList, VmwareAdditionalConfigAllowList, DestroyRootVolumeOnVmDestruction,
                 EnforceStrictResourceLimitHostTagCheck, StrictHostTags, AllowUserForceStopVm, VmDistinctHostNameScope,
-                VmwareAdditionalDetailsFromOvaEnabled, VmwareAllowedAdditionalDetailsFromOva};
+                VmwareAdditionalDetailsFromOvaEnabled, VmwareAllowedAdditionalDetailsFromOva, AllowDifferentHostTagsOfferingsForVmScale};
     }
 
     @Override
diff --git a/server/src/test/java/com/cloud/api/query/QueryManagerImplTest.java b/server/src/test/java/com/cloud/api/query/QueryManagerImplTest.java
index 892cd1e7def6..750f4d8655b1 100644
--- a/server/src/test/java/com/cloud/api/query/QueryManagerImplTest.java
+++ b/server/src/test/java/com/cloud/api/query/QueryManagerImplTest.java
@@ -18,6 +18,7 @@
 package com.cloud.api.query;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -33,6 +34,7 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import com.cloud.host.dao.HostTagsDao;
 import org.apache.cloudstack.acl.SecurityChecker;
 import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.api.ResponseObject;
@@ -156,6 +158,9 @@ public class QueryManagerImplTest {
     @Mock
     HostDao hostDao;
 
+    @Mock
+    HostTagsDao hostTagsDao;
+
     @Mock
     ClusterDao clusterDao;
 
@@ -622,4 +627,39 @@ public void updateHostsExtensions_withHostResponses_setsExtension() {
         verify(host1).setExtensionId("a");
         verify(host2).setExtensionId("b");
     }
+
+    @Test
+    public void testAddVmCurrentClusterHostTags() {
+        String tag1 = "tag1";
+        String tag2 = "tag2";
+        VMInstanceVO vmInstance = mock(VMInstanceVO.class);
+        HostVO host = mock(HostVO.class);
+        when(vmInstance.getHostId()).thenReturn(null);
+        when(vmInstance.getLastHostId()).thenReturn(1L);
+        when(hostDao.findById(1L)).thenReturn(host);
+        when(host.getClusterId()).thenReturn(1L);
+        when(hostTagsDao.listByClusterId(1L)).thenReturn(Arrays.asList(tag1, tag2));
+
+        List<String> hostTags = new ArrayList<>(Collections.singleton(tag1));
+        queryManagerImplSpy.addVmCurrentClusterHostTags(vmInstance, hostTags);
+        assertEquals(2, hostTags.size());
+        assertTrue(hostTags.contains(tag2));
+    }
+
+    @Test
+    public void testAddVmCurrentClusterHostTagsEmptyHostTagsInCluster() {
+        String tag1 = "tag1";
+        VMInstanceVO vmInstance = mock(VMInstanceVO.class);
+        HostVO host = mock(HostVO.class);
+        when(vmInstance.getHostId()).thenReturn(null);
+        when(vmInstance.getLastHostId()).thenReturn(1L);
+        when(hostDao.findById(1L)).thenReturn(host);
+        when(host.getClusterId()).thenReturn(1L);
+        when(hostTagsDao.listByClusterId(1L)).thenReturn(null);
+
+        List<String> hostTags = new ArrayList<>(Collections.singleton(tag1));
+        queryManagerImplSpy.addVmCurrentClusterHostTags(vmInstance, hostTags);
+        assertEquals(1, hostTags.size());
+        assertTrue(hostTags.contains(tag1));
+    }
 }