CLOUDSTACK-669: Finalize VM work dispatching mechanism to avoid big switch statement

engine/components-api/src/com/cloud/vm/VmWorkJobHandler.java

@@ -16,11 +16,10 @@
 // under the License.
 package com.cloud.vm;
 
-import org.apache.cloudstack.framework.jobs.AsyncJob;
 import org.apache.cloudstack.jobs.JobInfo;
 
 import com.cloud.utils.Pair;
 
 public interface VmWorkJobHandler {
-    Pair<JobInfo.Status, String> handleVmWorkJob(AsyncJob job, VmWork work) throws Exception;
+    Pair<JobInfo.Status, String> handleVmWorkJob(VmWork work) throws Exception;
 }

engine/components-api/src/com/cloud/vm/VmWorkJobHandlerProxy.java

@@ -0,0 +1,121 @@
+// 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.vm;
+
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.log4j.Logger;
+
+import com.google.gson.Gson;
+
+import org.apache.cloudstack.framework.jobs.impl.JobSerializerHelper;
+import org.apache.cloudstack.jobs.JobInfo;
+
+import com.cloud.serializer.GsonHelper;
+import com.cloud.utils.Pair;
+
+/**
+ * VmWorkJobHandlerProxy can not be used as standalone due to run-time
+ * reflection usage in its implementation, run-time reflection conflicts with Spring proxy mode.
+ * It means that we can not instantiate VmWorkJobHandlerProxy beans directly in Spring and expect
+ * it can handle VmWork directly from there.
+ *
+ */
+public class VmWorkJobHandlerProxy implements VmWorkJobHandler {
+
+    private static final Logger s_logger = Logger.getLogger(VmWorkJobHandlerProxy.class);
+
+    private Object _target;
+    private Map<Class<?>, Method> _handlerMethodMap = new HashMap<Class<?>, Method>();
+
+    private Gson _gsonLogger;
+
+    public VmWorkJobHandlerProxy(Object target) {
+        _gsonLogger = GsonHelper.getGsonLogger();
+
+        buildLookupMap(target.getClass());
+        _target = target;
+    }
+
+    private void buildLookupMap(Class<?> hostClass) {
+        Class<?> clz = hostClass;
+        while (clz != null && clz != Object.class) {
+            Method[] hostHandlerMethods = clz.getDeclaredMethods();
+
+            for (Method method : hostHandlerMethods) {
+                if (isVmWorkJobHandlerMethod(method)) {
+                    Class<?> paramType = method.getParameterTypes()[0];
+                    assert (_handlerMethodMap.get(paramType) == null);
+
+                    method.setAccessible(true);
+                    _handlerMethodMap.put(paramType, method);
+                }
+            }
+
+            clz = clz.getSuperclass();
+        }
+    }
+
+    @SuppressWarnings("deprecation")
+    private boolean isVmWorkJobHandlerMethod(Method method) {
+        if (method.getParameterTypes().length != 1)
+            return false;
+
+        Class<?> returnType = method.getReturnType();
+        if (!Pair.class.isAssignableFrom(returnType))
+            return false;
+
+        Class<?> paramType = method.getParameterTypes()[0];
+        if (!VmWork.class.isAssignableFrom(paramType))
+            return false;
+
+        return true;
+    }
+
+    private Method getHandlerMethod(Class<?> paramType) {
+        return _handlerMethodMap.get(paramType);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public Pair<JobInfo.Status, String> handleVmWorkJob(VmWork work) throws Exception {
+
+        Method method = getHandlerMethod(work.getClass());
+        if (method != null) {
+            if (s_logger.isDebugEnabled())
+                s_logger.debug("Execute VM work job: " + work.getClass().getName() + _gsonLogger.toJson(work));
+
+            Object obj = method.invoke(_target, work);
+
+            if (s_logger.isDebugEnabled())
+                s_logger.debug("Done executing VM work job: " + work.getClass().getName() + _gsonLogger.toJson(work));
+
+            assert (obj instanceof Pair);
+            return (Pair<JobInfo.Status, String>)obj;
+        } else {
+            s_logger.error("Unable to find handler for VM work job: " + work.getClass().getName() + _gsonLogger.toJson(work));
+
+            RuntimeException e = new RuntimeException("Unsupported VM work job: " + work.getClass().getName() + _gsonLogger.toJson(work));
+            String exceptionJson = JobSerializerHelper.toSerializedString(e);
+
+            s_logger.error("Serialize exception object into json: " + exceptionJson);
+            return new Pair<JobInfo.Status, String>(JobInfo.Status.FAILED, exceptionJson);
+        }
+    }
+}

engine/components-api/src/com/cloud/vm/snapshot/VMSnapshotManager.java

@@ -42,5 +42,4 @@ public interface VMSnapshotManager extends VMSnapshotService, Manager {
     boolean syncVMSnapshot(VMInstanceVO vm, Long hostId);
 
     boolean hasActiveVMSnapshotTasks(Long vmId);
-
 }

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

@@ -316,6 +316,8 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
     @Inject
     protected AsyncJobManager _jobMgr;
 
+    VmWorkJobHandlerProxy _jobHandlerProxy = new VmWorkJobHandlerProxy(this);
+
     Map<VirtualMachine.Type, VirtualMachineGuru> _vmGurus = new HashMap<VirtualMachine.Type, VirtualMachineGuru>();
     protected StateMachine2<State, VirtualMachine.Event, VirtualMachine> _stateMachine;
 
@@ -526,6 +528,8 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
 
     @Override
     public boolean start() {
+        // TODO, initial delay is hardcoded
+        _executor.scheduleAtFixedRate(new TransitionTask(), 5000, VmJobStateReportInterval.value(), TimeUnit.SECONDS);
         _executor.scheduleAtFixedRate(new CleanupTask(), VmOpCleanupInterval.value(), VmOpCleanupInterval.value(), TimeUnit.SECONDS);
         cancelWorkItems(_nodeId);
         return true;
@@ -2942,9 +2946,9 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
                 return;
             }
             try {
-                lock.addRef();
-                List<VMInstanceVO> instances =
-                    _vmDao.findVMInTransition(new Date(new Date().getTime() - (AgentManager.Wait.value() * 1000)), State.Starting, State.Stopping);
+                scanStalledVMInTransitionStateOnDisconnectedHosts();
+
+                List<VMInstanceVO> instances = _vmDao.findVMInTransition(new Date(new Date().getTime() - (AgentManager.Wait.value() * 1000)), State.Starting, State.Stopping);
                 for (VMInstanceVO instance : instances) {
                     State state = instance.getState();
                     if (state == State.Stopping) {
@@ -3974,7 +3978,6 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
         }
     }
 
-
     // VMs that in transitional state without recent power state report
     private List<Long> listStalledVMInTransitionStateOnUpHost(long hostId, Date cutTime) {
         String sql = "SELECT i.* FROM vm_instance as i, host as h WHERE h.status = 'UP' " +
@@ -4693,74 +4696,132 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
         return new VmJobVirtualMachineOutcome((VmWorkJobVO)context.getContextParameter("workJob"), vm.getId());
     }
 
-    @Override
-    public Pair<JobInfo.Status, String> handleVmWorkJob(AsyncJob job, VmWork work) throws Exception {
-
+    private Pair<JobInfo.Status, String> orchestrateStart(VmWorkStart work) throws Exception {
         VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
         if (vm == null) {
             s_logger.info("Unable to find vm " + work.getVmId());
         }
         assert (vm != null);
-        if (work instanceof VmWorkStart) {
-            VmWorkStart workStart = (VmWorkStart)work;
-            orchestrateStart(vm.getUuid(), workStart.getParams(), workStart.getPlan(), null);
+
+        orchestrateStart(vm.getUuid(), work.getParams(), work.getPlan(), null);
         return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkStop) {
-            VmWorkStop workStop = (VmWorkStop)work;
-            orchestrateStop(vm.getUuid(), workStop.isCleanup());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkMigrate) {
-            VmWorkMigrate workMigrate = (VmWorkMigrate)work;
-            orchestrateMigrate(vm.getUuid(), workMigrate.getSrcHostId(), workMigrate.getDeployDestination());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkMigrateWithStorage) {
-            VmWorkMigrateWithStorage workMigrateWithStorage = (VmWorkMigrateWithStorage)work;
-            orchestrateMigrateWithStorage(vm.getUuid(),
-                    workMigrateWithStorage.getSrcHostId(),
-                    workMigrateWithStorage.getDestHostId(),
-                    workMigrateWithStorage.getVolumeToPool());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkMigrateForScale) {
-            VmWorkMigrateForScale workMigrateForScale = (VmWorkMigrateForScale)work;
-            orchestrateMigrateForScale(vm.getUuid(),
-                    workMigrateForScale.getSrcHostId(),
-                    workMigrateForScale.getDeployDestination(),
-                    workMigrateForScale.getNewServiceOfferringId());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkReboot) {
-            VmWorkReboot workReboot = (VmWorkReboot)work;
-            orchestrateReboot(vm.getUuid(), workReboot.getParams());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkAddVmToNetwork) {
-            VmWorkAddVmToNetwork workAddVmToNetwork = (VmWorkAddVmToNetwork)work;
-            NicProfile nic = orchestrateAddVmToNetwork(vm, workAddVmToNetwork.getNetwork(),
-                    workAddVmToNetwork.getRequestedNicProfile());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, JobSerializerHelper.toObjectSerializedString(nic));
-        } else if (work instanceof VmWorkRemoveNicFromVm) {
-            VmWorkRemoveNicFromVm workRemoveNicFromVm = (VmWorkRemoveNicFromVm)work;
-            boolean result = orchestrateRemoveNicFromVm(vm, workRemoveNicFromVm.getNic());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
-                    JobSerializerHelper.toObjectSerializedString(new Boolean(result)));
-        } else if (work instanceof VmWorkRemoveVmFromNetwork) {
-            VmWorkRemoveVmFromNetwork workRemoveVmFromNetwork = (VmWorkRemoveVmFromNetwork)work;
-            boolean result = orchestrateRemoveVmFromNetwork(vm,
-                    workRemoveVmFromNetwork.getNetwork(), workRemoveVmFromNetwork.getBroadcastUri());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
-                    JobSerializerHelper.toObjectSerializedString(new Boolean(result)));
-        } else if (work instanceof VmWorkReconfigure) {
-            VmWorkReconfigure workReconfigure = (VmWorkReconfigure)work;
-            reConfigureVm(vm.getUuid(), workReconfigure.getNewServiceOffering(),
-                    workReconfigure.isSameHost());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkStorageMigration) {
-            VmWorkStorageMigration workStorageMigration = (VmWorkStorageMigration)work;
-            orchestrateStorageMigration(vm.getUuid(), workStorageMigration.getDestStoragePool());
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else {
-            RuntimeException e = new RuntimeException("Unsupported VM work command: " + job.getCmd());
-            String exceptionJson = JobSerializerHelper.toSerializedString(e);
-            s_logger.error("Serialize exception object into json: " + exceptionJson);
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.FAILED, exceptionJson);
     }
+
+    private Pair<JobInfo.Status, String> orchestrateStop(VmWorkStop work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+
+        orchestrateStop(vm.getUuid(), work.isCleanup());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateMigrate(VmWorkMigrate work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+
+        orchestrateMigrate(vm.getUuid(), work.getSrcHostId(), work.getDeployDestination());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateMigrateWithStorage(VmWorkMigrateWithStorage work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        orchestrateMigrateWithStorage(vm.getUuid(),
+                work.getSrcHostId(),
+                work.getDestHostId(),
+                work.getVolumeToPool());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateMigrateForScale(VmWorkMigrateForScale work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        orchestrateMigrateForScale(vm.getUuid(),
+                work.getSrcHostId(),
+                work.getDeployDestination(),
+                work.getNewServiceOfferringId());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateReboot(VmWorkReboot work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        orchestrateReboot(vm.getUuid(), work.getParams());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateAddVmToNetwork(VmWorkAddVmToNetwork work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        NicProfile nic = orchestrateAddVmToNetwork(vm, work.getNetwork(),
+                work.getRequestedNicProfile());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, _jobMgr.marshallResultObject(nic));
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateRemoveNicFromVm(VmWorkRemoveNicFromVm work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        boolean result = orchestrateRemoveNicFromVm(vm, work.getNic());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(new Boolean(result)));
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateRemoveVmFromNetwork(VmWorkRemoveVmFromNetwork work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        boolean result = orchestrateRemoveVmFromNetwork(vm,
+                work.getNetwork(), work.getBroadcastUri());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(new Boolean(result)));
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateReconfigure(VmWorkReconfigure work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        reConfigureVm(vm.getUuid(), work.getNewServiceOffering(),
+                work.isSameHost());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateStorageMigration(VmWorkStorageMigration work) throws Exception {
+        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
+        if (vm == null) {
+            s_logger.info("Unable to find vm " + work.getVmId());
+        }
+        assert (vm != null);
+        orchestrateStorageMigration(vm.getUuid(), work.getDestStoragePool());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    @Override
+    public Pair<JobInfo.Status, String> handleVmWorkJob(VmWork work) throws Exception {
+        return _jobHandlerProxy.handleVmWorkJob(work);
     }
 }

engine/orchestration/src/com/cloud/vm/VmWorkJobDispatcher.java

@@ -74,8 +74,8 @@ public class VmWorkJobDispatcher extends AdapterBase implements AsyncJobDispatch
             }
 
             work = VmWorkSerializer.deserialize(workClz, job.getCmdInfo());
-            assert (work != null);
-            if (work == null) {
+            assert(work != null);
+            if(work == null) {
                 s_logger.error("Unable to deserialize VM work " + job.getCmd() + ", job info: " + job.getCmdInfo());
                 _asyncJobMgr.completeAsyncJob(job.getId(), JobInfo.Status.FAILED, 0, "Unable to deserialize VM work");
                 return;
@@ -97,8 +97,9 @@ public class VmWorkJobDispatcher extends AdapterBase implements AsyncJobDispatch
 
             CallContext.register(work.getUserId(), work.getAccountId(), job.getRelated());
 
-            Pair<JobInfo.Status, String> result = handler.handleVmWorkJob(job, work);
+            Pair<JobInfo.Status, String> result = handler.handleVmWorkJob(work);
             _asyncJobMgr.completeAsyncJob(job.getId(), result.first(), 0, result.second());
+
         } catch(Throwable e) {
             s_logger.error("Unable to complete " + job, e);
 

server/src/com/cloud/storage/VolumeApiServiceImpl.java

@@ -61,7 +61,6 @@ import org.apache.cloudstack.framework.jobs.AsyncJobExecutionContext;
 import org.apache.cloudstack.framework.jobs.AsyncJobManager;
 import org.apache.cloudstack.framework.jobs.Outcome;
 import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
-import org.apache.cloudstack.framework.jobs.impl.JobSerializerHelper;
 import org.apache.cloudstack.framework.jobs.impl.OutcomeImpl;
 import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO;
 import org.apache.cloudstack.jobs.JobInfo;
@@ -329,6 +328,8 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
     @Inject
     protected AsyncJobManager _jobMgr;
 
+    VmWorkJobHandlerProxy _jobHandlerProxy = new VmWorkJobHandlerProxy(this);
+
     // TODO
     static final ConfigKey<Boolean> VmJobEnabled = new ConfigKey<Boolean>("Advanced", Boolean.class, "vm.job.enabled", "false",
             "True to enable new VM sync model. false to use the old way", false);
@@ -2353,92 +2354,38 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                 snapshotId);
     }
 
+    private Pair<JobInfo.Status, String> orchestrateAttachVolumeToVM(VmWorkAttachVolume work) throws Exception {
+        orchestrateAttachVolumeToVM(work.getVmId(), work.getVolumeId(), work.getDeviceId());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateDetachVolumeFromVM(VmWorkAttachVolume work) throws Exception {
+        orchestrateDetachVolumeFromVM(work.getVmId(), work.getVolumeId());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateResizeVolume(VmWorkResizeVolume work) throws Exception {
+        orchestrateResizeVolume(work.getVolumeId(), work.getCurrentSize(), work.getNewSize(),
+                work.getNewServiceOfferingId(), work.isShrinkOk());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateMigrateVolume(VmWorkMigrateVolume work) throws Exception {
+        Volume newVol = orchestrateMigrateVolume(work.getVolumeId(), work.getDestPoolId(), work.isLiveMigrate());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(new Long(newVol.getId())));
+    }
+
+    private Pair<JobInfo.Status, String> orchestrateTakeVolumeSnapshot(VmWorkTakeVolumeSnapshot work) throws Exception {
+        Account account = _accountDao.findById(work.getAccountId());
+        orchestrateTakeVolumeSnapshot(work.getVolumeId(), work.getPolicyId(), work.getSnapshotId(),
+                account, work.isQuiesceVm());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(work.getSnapshotId()));
+    }
+
     @Override
-    public Pair<JobInfo.Status, String> handleVmWorkJob(AsyncJob job, VmWork work) throws Exception {
-        VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, work.getVmId());
-        if (vm == null) {
-            s_logger.info("Unable to find vm " + work.getVmId());
-        }
-        assert (vm != null);
-
-        if (work instanceof VmWorkAttachVolume) {
-
-            VmWorkAttachVolume attachWork = (VmWorkAttachVolume)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Attach-Volume within VM work job context. vmId: " + attachWork.getVmId() + ", volId: " + attachWork.getVolumeId() + ", deviceId: "
-                        + attachWork.getDeviceId());
-
-            orchestrateAttachVolumeToVM(attachWork.getVmId(), attachWork.getVolumeId(), attachWork.getDeviceId());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Attach-Volume within VM work job context. vmId: " + attachWork.getVmId() + ", volId: " + attachWork.getVolumeId() + ", deviceId: "
-                        + attachWork.getDeviceId());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkDetachVolume) {
-            VmWorkDetachVolume detachWork = (VmWorkDetachVolume)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Detach-Volume within VM work job context. vmId: " + detachWork.getVmId() + ", volId: " + detachWork.getVolumeId());
-
-            orchestrateDetachVolumeFromVM(detachWork.getVmId(), detachWork.getVolumeId());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Detach-Volume within VM work job context. vmId: " + detachWork.getVmId() + ", volId: " + detachWork.getVolumeId());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-        } else if (work instanceof VmWorkResizeVolume) {
-            VmWorkResizeVolume resizeWork = (VmWorkResizeVolume)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Resize-Volume within VM work job context. vmId: " + resizeWork.getVmId()
-                        + ", volId: " + resizeWork.getVolumeId() + ", size " + resizeWork.getCurrentSize() + " -> " + resizeWork.getNewSize());
-
-            orchestrateResizeVolume(resizeWork.getVolumeId(), resizeWork.getCurrentSize(), resizeWork.getNewSize(),
-                    resizeWork.getNewServiceOfferingId(), resizeWork.isShrinkOk());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Resize-Volume within VM work job context. vmId: " + resizeWork.getVmId()
-                        + ", volId: " + resizeWork.getVolumeId() + ", size " + resizeWork.getCurrentSize() + " -> " + resizeWork.getNewSize());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-
-        } else if (work instanceof VmWorkMigrateVolume) {
-            VmWorkMigrateVolume migrateWork = (VmWorkMigrateVolume)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Migrate-Volume within VM work job context. vmId: " + migrateWork.getVmId()
-                        + ", volId: " + migrateWork.getVolumeId() + ", destPoolId: " + migrateWork.getDestPoolId() + ", live: " + migrateWork.isLiveMigrate());
-
-            Volume newVol = orchestrateMigrateVolume(migrateWork.getVolumeId(), migrateWork.getDestPoolId(), migrateWork.isLiveMigrate());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Migrate-Volume within VM work job context. vmId: " + migrateWork.getVmId()
-                        + ", volId: " + migrateWork.getVolumeId() + ", destPoolId: " + migrateWork.getDestPoolId() + ", live: " + migrateWork.isLiveMigrate());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, JobSerializerHelper.toObjectSerializedString(new Long(newVol.getId())));
-        } else if (work instanceof VmWorkTakeVolumeSnapshot) {
-            VmWorkTakeVolumeSnapshot snapshotWork = (VmWorkTakeVolumeSnapshot)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Take-Volume-Snapshot within VM work job context. vmId: " + snapshotWork.getVmId()
-                        + ", volId: " + snapshotWork.getVolumeId() + ", policyId: " + snapshotWork.getPolicyId() + ", quiesceVm: " + snapshotWork.isQuiesceVm());
-
-            Account account = _accountDao.findById(snapshotWork.getAccountId());
-            orchestrateTakeVolumeSnapshot(snapshotWork.getVolumeId(), snapshotWork.getPolicyId(), snapshotWork.getSnapshotId(),
-                    account, snapshotWork.isQuiesceVm());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Take-Volume-Snapshot within VM work job context. vmId: " + snapshotWork.getVmId()
-                        + ", volId: " + snapshotWork.getVolumeId() + ", policyId: " + snapshotWork.getPolicyId() + ", quiesceVm: " + snapshotWork.isQuiesceVm());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, JobSerializerHelper.toObjectSerializedString(snapshotWork.getSnapshotId()));
-        } else {
-            RuntimeException e = new RuntimeException("Unsupported VM work command: " + job.getCmd());
-            String exceptionJson = JobSerializerHelper.toSerializedString(e);
-            s_logger.error("Serialize exception object into json: " + exceptionJson);
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.FAILED, exceptionJson);
-        }
+    public Pair<JobInfo.Status, String> handleVmWorkJob(VmWork work) throws Exception {
+        return _jobHandlerProxy.handleVmWorkJob(work);
     }
 }

server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java

@@ -42,7 +42,6 @@ import org.apache.cloudstack.framework.jobs.AsyncJobExecutionContext;
 import org.apache.cloudstack.framework.jobs.AsyncJobManager;
 import org.apache.cloudstack.framework.jobs.Outcome;
 import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
-import org.apache.cloudstack.framework.jobs.impl.JobSerializerHelper;
 import org.apache.cloudstack.framework.jobs.impl.OutcomeImpl;
 import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO;
 import org.apache.cloudstack.jobs.JobInfo;
@@ -91,6 +90,7 @@ import com.cloud.vm.VirtualMachineProfile;
 import com.cloud.vm.VmWork;
 import com.cloud.vm.VmWorkConstants;
 import com.cloud.vm.VmWorkJobHandler;
+import com.cloud.vm.VmWorkJobHandlerProxy;
 import com.cloud.vm.VmWorkSerializer;
 import com.cloud.vm.dao.UserVmDao;
 import com.cloud.vm.dao.VMInstanceDao;
@@ -124,6 +124,8 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
     @Inject
     AsyncJobManager _jobMgr;
 
+    VmWorkJobHandlerProxy _jobHandlerProxy = new VmWorkJobHandlerProxy(this);
+
     int _vmSnapshotMax;
     int _wait;
 
@@ -478,7 +480,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
         }
     }
 
-    public boolean orchestrateDeleteVMSnapshot(Long vmSnapshotId) {
+    private boolean orchestrateDeleteVMSnapshot(Long vmSnapshotId) {
         Account caller = getCaller();
 
         VMSnapshotVO vmSnapshot = _vmSnapshotDao.findById(vmSnapshotId);
@@ -585,7 +587,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
         }
     }
 
-    public UserVm orchestrateRevertToVMSnapshot(Long vmSnapshotId) throws InsufficientCapacityException, ResourceUnavailableException, ConcurrentOperationException {
+    private UserVm orchestrateRevertToVMSnapshot(Long vmSnapshotId) throws InsufficientCapacityException, ResourceUnavailableException, ConcurrentOperationException {
 
         // check if VM snapshot exists in DB
         VMSnapshotVO vmSnapshotVo = _vmSnapshotDao.findById(vmSnapshotId);
@@ -984,72 +986,31 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
                 vmId);
     }
 
-    @Override
-    public Pair<JobInfo.Status, String> handleVmWorkJob(AsyncJob job, VmWork work) throws Exception {
-
-        if (work instanceof VmWorkCreateVMSnapshot) {
-            VmWorkCreateVMSnapshot createWork = (VmWorkCreateVMSnapshot)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Create-VM-Snapshot within VM work job context. vmId: " + createWork.getVmId()
-                        + ", VM snapshotId: " + createWork.getVmSnapshotId() + "quiesce: " + createWork.isQuiesceVm());
-
-            VMSnapshot vmSnapshot = orchestrateCreateVMSnapshot(createWork.getVmId(), createWork.getVmSnapshotId(), createWork.isQuiesceVm());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Create-VM-Snapshot within VM work job context. vmId: " + createWork.getVmId()
-                        + ", VM snapshotId: " + createWork.getVmSnapshotId() + "quiesce: " + createWork.isQuiesceVm());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, JobSerializerHelper.toObjectSerializedString(new Long(vmSnapshot.getId())));
-        } else if (work instanceof VmWorkDeleteVMSnapshot) {
-            VmWorkDeleteVMSnapshot deleteWork = (VmWorkDeleteVMSnapshot)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Delete-VM-Snapshot within VM work job context. vmId: " + deleteWork.getVmId()
-                        + ", VM snapshotId: " + deleteWork.getVmSnapshotId());
-
-            boolean result = orchestrateDeleteVMSnapshot(deleteWork.getVmSnapshotId());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Delete-VM-Snapshot within VM work job context. vmId: " + deleteWork.getVmId()
-                        + ", VM snapshotId: " + deleteWork.getVmSnapshotId());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, JobSerializerHelper.toObjectSerializedString(new Boolean(result)));
-
-        } else if (work instanceof VmWorkRevertToVMSnapshot) {
-            VmWorkRevertToVMSnapshot revertWork = (VmWorkRevertToVMSnapshot)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Revert-VM-Snapshot within VM work job context. vmId: " + revertWork.getVmId()
-                        + ", VM snapshotId: " + revertWork.getVmSnapshotId());
-
-            orchestrateRevertToVMSnapshot(revertWork.getVmSnapshotId());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Done executing Revert-VM-Snapshot within VM work job context. vmId: " + revertWork.getVmId()
-                        + ", VM snapshotId: " + revertWork.getVmSnapshotId());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
-
-        } else if (work instanceof VmWorkDeleteAllVMSnapshots) {
-            VmWorkDeleteAllVMSnapshots deleteAllWork = (VmWorkDeleteAllVMSnapshots)work;
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Delete-All-VM-Snapshot within VM work job context. vmId: " + deleteAllWork.getVmId());
-
-            boolean result = orchestrateDeleteAllVMSnapshots(deleteAllWork.getVmId(), deleteAllWork.getSnapshotType());
-
-            if (s_logger.isDebugEnabled())
-                s_logger.debug("Execute Delete-All-VM-Snapshot within VM work job context. vmId: " + deleteAllWork.getVmId());
-
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, JobSerializerHelper.toObjectSerializedString(new Boolean(result)));
-
-        } else {
-
-            RuntimeException e = new RuntimeException("Unsupported VM work command: " + job.getCmd());
-            String exceptionJson = JobSerializerHelper.toSerializedString(e);
-            s_logger.error("Serialize exception object into json: " + exceptionJson);
-            return new Pair<JobInfo.Status, String>(JobInfo.Status.FAILED, exceptionJson);
+    public Pair<JobInfo.Status, String> orchestrateCreateVMSnapshot(VmWorkCreateVMSnapshot work) throws Exception {
+        VMSnapshot snapshot = orchestrateCreateVMSnapshot(work.getVmId(), work.getVmSnapshotId(), work.isQuiesceVm());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(new Long(snapshot.getId())));
     }
+
+    public Pair<JobInfo.Status, String> orchestrateDeleteVMSnapshot(VmWorkDeleteVMSnapshot work) {
+        boolean result = orchestrateDeleteVMSnapshot(work.getVmSnapshotId());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(new Boolean(result)));
+    }
+
+    public Pair<JobInfo.Status, String> orchestrateRevertToVMSnapshot(VmWorkRevertToVMSnapshot work) throws Exception {
+        orchestrateRevertToVMSnapshot(work.getVmSnapshotId());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, null);
+    }
+
+    public Pair<JobInfo.Status, String> orchestrateDeleteAllVMSnapshots(VmWorkDeleteAllVMSnapshots work) {
+        boolean result = orchestrateDeleteAllVMSnapshots(work.getVmId(), work.getSnapshotType());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED,
+                _jobMgr.marshallResultObject(new Boolean(result)));
+    }
+
+    @Override
+    public Pair<JobInfo.Status, String> handleVmWorkJob(VmWork work) throws Exception {
+        return _jobHandlerProxy.handleVmWorkJob(work);
     }
 }