Submitting multiple dynamic VM Scaling API commands for the same instance can result in two usage events in the same second causing a compound key violation in usage service

Root cause:
Even though dynamic scaling job is handled in vmworkjob queue which ensures serilizing multiple jobs but the database updating and generating usage events are out of the job queue.

Solution:
Moved all updations into the job queue

Author: harikrishna-patnala

engine/api/src/main/java/com/cloud/vm/VirtualMachineManager.java

@@ -152,7 +152,7 @@ void advanceReboot(String vmUuid, Map<VirtualMachineProfile.Param, Object> param
      * @param serviceOfferingId
      * @return
      */
-    boolean upgradeVmDb(long vmId, long serviceOfferingId);
+    boolean upgradeVmDb(long vmId, ServiceOffering newServiceOffering, ServiceOffering currentServiceOffering);
 
     /**
      * @param vm
@@ -201,7 +201,7 @@ NicProfile addVmToNetwork(VirtualMachine vm, Network network, NicProfile request
     boolean replugNic(Network network, NicTO nic, VirtualMachineTO vm, ReservationContext context, DeployDestination dest) throws ConcurrentOperationException,
             ResourceUnavailableException, InsufficientCapacityException;
 
-    VirtualMachine reConfigureVm(String vmUuid, ServiceOffering newServiceOffering, boolean sameHost) throws ResourceUnavailableException, ConcurrentOperationException,
+    VirtualMachine reConfigureVm(String vmUuid, ServiceOffering oldServiceOffering, ServiceOffering newServiceOffering, Map<String, String> customParameters, boolean sameHost) throws ResourceUnavailableException, ConcurrentOperationException,
             InsufficientServerCapacityException;
 
     void findHostAndMigrate(String vmUuid, Long newSvcOfferingId, DeploymentPlanner.ExcludeList excludeHostList) throws InsufficientCapacityException,

engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java

@@ -41,6 +41,7 @@
 
 import com.cloud.agent.api.PrepareForMigrationAnswer;
 import com.cloud.agent.api.to.DpdkTO;
+import com.cloud.event.UsageEventVO;
 import com.cloud.offering.NetworkOffering;
 import com.cloud.offerings.dao.NetworkOfferingDetailsDao;
 import org.apache.cloudstack.affinity.dao.AffinityGroupVMMapDao;
@@ -232,6 +233,8 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
 
     private static final String VM_SYNC_ALERT_SUBJECT = "VM state sync alert";
 
+    @Inject
+    private UserVmManager _userVmMgr;
     @Inject
     private DataStoreManager dataStoreMgr;
     @Inject
@@ -3406,13 +3409,20 @@ public void checkIfCanUpgrade(final VirtualMachine vmInstance, final ServiceOffe
     }
 
     @Override
-    public boolean upgradeVmDb(final long vmId, final long serviceOfferingId) {
-        final VMInstanceVO vmForUpdate = _vmDao.createForUpdate();
-        vmForUpdate.setServiceOfferingId(serviceOfferingId);
-        final ServiceOffering newSvcOff = _entityMgr.findById(ServiceOffering.class, serviceOfferingId);
+    public boolean upgradeVmDb(final long vmId, final ServiceOffering newServiceOffering, ServiceOffering currentServiceOffering) {
+
+        final VMInstanceVO vmForUpdate = _vmDao.findById(vmId);
+        vmForUpdate.setServiceOfferingId(newServiceOffering.getId());
+        final ServiceOffering newSvcOff = _entityMgr.findById(ServiceOffering.class, newServiceOffering.getId());
         vmForUpdate.setHaEnabled(newSvcOff.isOfferHA());
         vmForUpdate.setLimitCpuUse(newSvcOff.getLimitCpuUse());
         vmForUpdate.setServiceOfferingId(newSvcOff.getId());
+        if (newServiceOffering.isDynamic()) {
+            saveCustomOfferingDetails(vmId, newServiceOffering);
+        }
+        if (currentServiceOffering.isDynamic() && !newServiceOffering.isDynamic()) {
+            removeCustomOfferingDetails(vmId);
+        }
         return _vmDao.update(vmId, vmForUpdate);
     }
 
@@ -4087,8 +4097,8 @@ public boolean unplugNic(final Network network, final NicTO nic, final VirtualMa
     }
 
     @Override
-    public VMInstanceVO reConfigureVm(final String vmUuid, final ServiceOffering oldServiceOffering,
-            final boolean reconfiguringOnExistingHost)
+    public VMInstanceVO reConfigureVm(final String vmUuid, final ServiceOffering oldServiceOffering, final ServiceOffering newServiceOffering,
+            Map<String, String> customParameters, final boolean reconfiguringOnExistingHost)
                     throws ResourceUnavailableException, InsufficientServerCapacityException, ConcurrentOperationException {
 
         final AsyncJobExecutionContext jobContext = AsyncJobExecutionContext.getCurrentExecutionContext();
@@ -4098,14 +4108,14 @@ public VMInstanceVO reConfigureVm(final String vmUuid, final ServiceOffering old
             final VirtualMachine vm = _vmDao.findByUuid(vmUuid);
             placeHolder = createPlaceHolderWork(vm.getId());
             try {
-                return orchestrateReConfigureVm(vmUuid, oldServiceOffering, reconfiguringOnExistingHost);
+                return orchestrateReConfigureVm(vmUuid, oldServiceOffering, newServiceOffering, reconfiguringOnExistingHost);
             } finally {
                 if (placeHolder != null) {
                     _workJobDao.expunge(placeHolder.getId());
                 }
             }
         } else {
-            final Outcome<VirtualMachine> outcome = reconfigureVmThroughJobQueue(vmUuid, oldServiceOffering, reconfiguringOnExistingHost);
+            final Outcome<VirtualMachine> outcome = reconfigureVmThroughJobQueue(vmUuid, oldServiceOffering, newServiceOffering, customParameters, reconfiguringOnExistingHost);
 
             VirtualMachine vm = null;
             try {
@@ -4134,14 +4144,12 @@ public VMInstanceVO reConfigureVm(final String vmUuid, final ServiceOffering old
         }
     }
 
-    private VMInstanceVO orchestrateReConfigureVm(final String vmUuid, final ServiceOffering oldServiceOffering, final boolean reconfiguringOnExistingHost) throws ResourceUnavailableException,
-    ConcurrentOperationException {
+    private VMInstanceVO orchestrateReConfigureVm(final String vmUuid, final ServiceOffering oldServiceOffering, final ServiceOffering newServiceOffering,
+                                                  final boolean reconfiguringOnExistingHost) throws ResourceUnavailableException, ConcurrentOperationException {
         final VMInstanceVO vm = _vmDao.findByUuid(vmUuid);
+        upgradeVmDb(vm.getId(), newServiceOffering, oldServiceOffering);
 
-        final long newServiceofferingId = vm.getServiceOfferingId();
-        final ServiceOffering newServiceOffering = _offeringDao.findById(vm.getId(), newServiceofferingId);
         final HostVO hostVo = _hostDao.findById(vm.getHostId());
-
         final Float memoryOvercommitRatio = CapacityManager.MemOverprovisioningFactor.valueIn(hostVo.getClusterId());
         final Float cpuOvercommitRatio = CapacityManager.CpuOverprovisioningFactor.valueIn(hostVo.getClusterId());
         final long minMemory = (long)(newServiceOffering.getRamSize() / memoryOvercommitRatio);
@@ -4168,7 +4176,7 @@ private VMInstanceVO orchestrateReConfigureVm(final String vmUuid, final Service
             if (reconfiguringOnExistingHost) {
                 vm.setServiceOfferingId(oldServiceOffering.getId());
                 _capacityMgr.releaseVmCapacity(vm, false, false, vm.getHostId()); //release the old capacity
-                vm.setServiceOfferingId(newServiceofferingId);
+                vm.setServiceOfferingId(newServiceOffering.getId());
                 _capacityMgr.allocateVmCapacity(vm, false); // lock the new capacity
             }
 
@@ -4177,7 +4185,9 @@ private VMInstanceVO orchestrateReConfigureVm(final String vmUuid, final Service
                 s_logger.error("Unable to scale vm due to " + (reconfigureAnswer == null ? "" : reconfigureAnswer.getDetails()));
                 throw new CloudRuntimeException("Unable to scale vm due to " + (reconfigureAnswer == null ? "" : reconfigureAnswer.getDetails()));
             }
-
+            if (vm.getType().equals(VirtualMachine.Type.User)) {
+                _userVmMgr.generateUsageEvent(vm, vm.isDisplayVm(), EventTypes.EVENT_VM_DYNAMIC_SCALE);
+            }
             success = true;
         } catch (final OperationTimedoutException e) {
             throw new AgentUnavailableException("Operation timed out on reconfiguring " + vm, dstHostId);
@@ -4186,7 +4196,7 @@ private VMInstanceVO orchestrateReConfigureVm(final String vmUuid, final Service
         } finally {
             if (!success) {
                 _capacityMgr.releaseVmCapacity(vm, false, false, vm.getHostId()); // release the new capacity
-                vm.setServiceOfferingId(oldServiceOffering.getId());
+                upgradeVmDb(vm.getId(), oldServiceOffering, newServiceOffering); // rollback
                 _capacityMgr.allocateVmCapacity(vm, false); // allocate the old capacity
             }
         }
@@ -4195,6 +4205,33 @@ private VMInstanceVO orchestrateReConfigureVm(final String vmUuid, final Service
 
     }
 
+    private void removeCustomOfferingDetails(long vmId) {
+        Map<String, String> details = userVmDetailsDao.listDetailsKeyPairs(vmId);
+        details.remove(UsageEventVO.DynamicParameters.cpuNumber.name());
+        details.remove(UsageEventVO.DynamicParameters.cpuSpeed.name());
+        details.remove(UsageEventVO.DynamicParameters.memory.name());
+        List<UserVmDetailVO> detailList = new ArrayList<UserVmDetailVO>();
+        for(Map.Entry<String, String> entry: details.entrySet()) {
+            UserVmDetailVO detailVO = new UserVmDetailVO(vmId, entry.getKey(), entry.getValue(), true);
+            detailList.add(detailVO);
+        }
+        userVmDetailsDao.saveDetails(detailList);
+    }
+
+    private void saveCustomOfferingDetails(long vmId, ServiceOffering serviceOffering) {
+        //save the custom values to the database.
+        Map<String, String> details = userVmDetailsDao.listDetailsKeyPairs(vmId);
+        details.put(UsageEventVO.DynamicParameters.cpuNumber.name(), serviceOffering.getCpu().toString());
+        details.put(UsageEventVO.DynamicParameters.cpuSpeed.name(), serviceOffering.getSpeed().toString());
+        details.put(UsageEventVO.DynamicParameters.memory.name(), serviceOffering.getRamSize().toString());
+        List<UserVmDetailVO> detailList = new ArrayList<UserVmDetailVO>();
+        for (Map.Entry<String, String> entry: details.entrySet()) {
+            UserVmDetailVO detailVO = new UserVmDetailVO(vmId, entry.getKey(), entry.getValue(), true);
+            detailList.add(detailVO);
+        }
+        userVmDetailsDao.saveDetails(detailList);
+    }
+
     @Override
     public String getConfigComponentName() {
         return VirtualMachineManager.class.getSimpleName();
@@ -5090,7 +5127,7 @@ public Outcome<VirtualMachine> removeVmFromNetworkThroughJobQueue(
     }
 
     public Outcome<VirtualMachine> reconfigureVmThroughJobQueue(
-            final String vmUuid, final ServiceOffering newServiceOffering, final boolean reconfiguringOnExistingHost) {
+            final String vmUuid, final ServiceOffering oldServiceOffering, final ServiceOffering newServiceOffering, Map<String, String> customParameters, final boolean reconfiguringOnExistingHost) {
 
         final CallContext context = CallContext.current();
         final User user = context.getCallingUser();
@@ -5121,7 +5158,7 @@ public Outcome<VirtualMachine> reconfigureVmThroughJobQueue(
 
             // save work context info (there are some duplications)
             final VmWorkReconfigure workInfo = new VmWorkReconfigure(user.getId(), account.getId(), vm.getId(),
-                    VirtualMachineManagerImpl.VM_WORK_JOB_HANDLER, newServiceOffering.getId(), reconfiguringOnExistingHost);
+                    VirtualMachineManagerImpl.VM_WORK_JOB_HANDLER, oldServiceOffering.getId(), newServiceOffering.getId(), customParameters, reconfiguringOnExistingHost);
             workJob.setCmdInfo(VmWorkSerializer.serialize(workInfo));
 
             _jobMgr.submitAsyncJob(workJob, VmWorkConstants.VM_WORK_QUEUE, vm.getId());
@@ -5280,10 +5317,14 @@ private Pair<JobInfo.Status, String> orchestrateReconfigure(final VmWorkReconfig
             s_logger.info("Unable to find vm " + work.getVmId());
         }
         assert vm != null;
+        ServiceOfferingVO oldServiceOffering = _offeringDao.findById(work.getOldServiceOfferingId());
+        ServiceOfferingVO newServiceOffering = _offeringDao.findById(work.getNewServiceOfferingId());
+        if (newServiceOffering.isDynamic()) {
+            // update the service offering object with the custom parameters like cpu, memory
+            newServiceOffering = _offeringDao.getcomputeOffering(newServiceOffering, work.getCustomParameters());
+        }
 
-        final ServiceOffering newServiceOffering = _offeringDao.findById(vm.getId(), work.getNewServiceOfferingId());
-
-        reConfigureVm(vm.getUuid(), newServiceOffering,
+        reConfigureVm(vm.getUuid(), oldServiceOffering, newServiceOffering, work.getCustomParameters(),
                 work.isSameHost());
         return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
     }

engine/orchestration/src/main/java/com/cloud/vm/VmWorkReconfigure.java

@@ -17,25 +17,40 @@
 package com.cloud.vm;
 
 
+import java.util.Map;
+
 public class VmWorkReconfigure extends VmWork {
     private static final long serialVersionUID = -4517030323758086615L;
 
+    Long oldServiceOfferingId;
     Long newServiceOfferingId;
+
+    Map<String, String> customParameters;
     boolean sameHost;
 
-    public VmWorkReconfigure(long userId, long accountId, long vmId, String handlerName,
-            Long newServiceOfferingId, boolean sameHost) {
+    public VmWorkReconfigure(long userId, long accountId, long vmId, String handlerName, Long oldServiceOfferingId,
+            Long newServiceOfferingId, Map<String, String> customParameters, boolean sameHost) {
 
         super(userId, accountId, vmId, handlerName);
 
+        this.oldServiceOfferingId = oldServiceOfferingId;
         this.newServiceOfferingId = newServiceOfferingId;
+        this.customParameters = customParameters;
         this.sameHost = sameHost;
     }
 
+    public Long getOldServiceOfferingId() {
+        return oldServiceOfferingId;
+    }
+
     public Long getNewServiceOfferingId() {
         return newServiceOfferingId;
     }
 
+    public Map<String, String> getCustomParameters() {
+        return customParameters;
+    }
+
     public boolean isSameHost() {
         return sameHost;
     }

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

@@ -4086,15 +4086,7 @@ private VirtualMachine upgradeStoppedSystemVm(final Long systemVmId, final Long
         }
         _itMgr.checkIfCanUpgrade(systemVm, newServiceOffering);
 
-        final boolean result = _itMgr.upgradeVmDb(systemVmId, serviceOfferingId);
-
-        if (newServiceOffering.isDynamic()) {
-            //save the custom values to the database.
-            _userVmMgr.saveCustomOfferingDetails(systemVmId, newServiceOffering);
-        }
-        if (currentServiceOffering.isDynamic() && !newServiceOffering.isDynamic()) {
-            _userVmMgr.removeCustomOfferingDetails(systemVmId);
-        }
+        final boolean result = _itMgr.upgradeVmDb(systemVmId, newServiceOffering, currentServiceOffering);
 
         if (result) {
             return _vmInstanceDao.findById(systemVmId);

server/src/main/java/com/cloud/vm/UserVmManager.java

@@ -32,7 +32,6 @@
 import com.cloud.exception.ManagementServerException;
 import com.cloud.exception.ResourceUnavailableException;
 import com.cloud.exception.VirtualMachineMigrationException;
-import com.cloud.offering.ServiceOffering;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.user.Account;
@@ -117,10 +116,6 @@ UserVm updateVirtualMachine(long id, String displayName, String group, Boolean h
     //find a common place for all the scaling and upgrading code of both user and systemvms.
     void validateCustomParameters(ServiceOfferingVO serviceOffering, Map<String, String> customParameters);
 
-    public void saveCustomOfferingDetails(long vmId, ServiceOffering serviceOffering);
-
-    public void removeCustomOfferingDetails(long vmId);
-
     void generateUsageEvent(VirtualMachine vm, boolean isDisplay, String eventType);
 
     void persistDeviceBusInfo(UserVmVO paramUserVmVO, String paramString);