VM power state for job oritented sync processing

api/src/com/cloud/agent/api/HostVmStateReportEntry.java

@@ -0,0 +1,41 @@
+// 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.agent.api;
+
+import com.cloud.vm.VirtualMachine;
+import com.cloud.vm.VirtualMachine.PowerState;
+
+public class HostVmStateReportEntry {
+    VirtualMachine.PowerState state;
+    String host;
+    
+    public HostVmStateReportEntry() {
+    }
+    
+    public HostVmStateReportEntry(PowerState state, String host) {
+        this.state = state;
+        this.host = host;
+    }
+    
+    public PowerState getState() {
+        return state;
+    }
+    
+    public String getHost() {
+        return host;
+    }
+}

api/src/com/cloud/vm/VirtualMachine.java

@@ -33,6 +33,12 @@ import com.cloud.utils.fsm.StateObject;
  */
 public interface VirtualMachine extends RunningOn, ControlledEntity, Identity, InternalIdentity, StateObject<VirtualMachine.State> {
 
+	public enum PowerState {
+	    PowerUnknown,
+	    PowerOn,
+	    PowerOff,
+    }
+
     public enum State {
         Starting(true, "VM is being started.  At this state, you should find host id filled which means it's being started on that host."),
         Running(false, "VM is running.  host id has the host that it is running on."),
@@ -111,6 +117,15 @@ public interface VirtualMachine extends RunningOn, ControlledEntity, Identity, I
             s_fsm.addTransition(State.Expunging, VirtualMachine.Event.ExpungeOperation, State.Expunging);
             s_fsm.addTransition(State.Error, VirtualMachine.Event.DestroyRequested, State.Expunging);
             s_fsm.addTransition(State.Error, VirtualMachine.Event.ExpungeOperation, State.Expunging);
+            			
+            s_fsm.addTransition(State.Stopping, VirtualMachine.Event.FollowAgentPowerOnReport, State.Running);
+            s_fsm.addTransition(State.Stopped, VirtualMachine.Event.FollowAgentPowerOnReport, State.Running);
+            s_fsm.addTransition(State.Running, VirtualMachine.Event.FollowAgentPowerOnReport, State.Running);
+            s_fsm.addTransition(State.Migrating, VirtualMachine.Event.FollowAgentPowerOnReport, State.Running);
+            s_fsm.addTransition(State.Starting, VirtualMachine.Event.FollowAgentPowerOffReport, State.Stopped);
+            s_fsm.addTransition(State.Stopping, VirtualMachine.Event.FollowAgentPowerOffReport, State.Stopped);
+            s_fsm.addTransition(State.Running, VirtualMachine.Event.FollowAgentPowerOffReport, State.Stopped);
+            s_fsm.addTransition(State.Migrating, VirtualMachine.Event.FollowAgentPowerOffReport, State.Stopped);
         }
         
         public static boolean isVmStarted(State oldState, Event e, State newState) {
@@ -179,6 +194,10 @@ public interface VirtualMachine extends RunningOn, ControlledEntity, Identity, I
         AgentReportMigrated,
         RevertRequested,
         SnapshotRequested,
+        
+        // added for new VMSync logic
+        FollowAgentPowerOnReport,
+        FollowAgentPowerOffReport,        
     };
 
     public enum Type {

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

@@ -46,7 +46,11 @@ import com.cloud.utils.fsm.NoTransitionException;
  * Manages allocating resources to vms.
  */
 public interface VirtualMachineManager extends Manager {
-
+	 
+	public interface Topics {
+        public static final String VM_POWER_STATE = "vm.powerstate";
+	}
+	 
     /**
      * Allocates a new virtual machine instance in the CloudStack DB.  This
      * orchestrates the creation of all virtual resources needed in CloudStack

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

@@ -0,0 +1,32 @@
+// 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.util.Map;
+
+import com.cloud.agent.api.HostVmStateReportEntry;
+import com.cloud.vm.VirtualMachine.PowerState;
+
+public interface VirtualMachinePowerStateSync {
+	
+	void resetHostSyncState(long hostId);
+	
+	void processHostVmStateReport(long hostId, Map<String, HostVmStateReportEntry> report);
+	
+	// to adapt legacy ping report
+	void processHostVmStatePingReport(long hostId, Map<String, PowerState> report);
+}

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

@@ -0,0 +1,125 @@
+// 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.util.HashMap;
+import java.util.Map;
+
+import javax.inject.Inject;
+
+import org.apache.log4j.Logger;
+
+import org.apache.cloudstack.framework.messagebus.MessageBus;
+import org.apache.cloudstack.framework.messagebus.PublishScope;
+
+import com.cloud.agent.api.HostVmStateReportEntry;
+import com.cloud.vm.VirtualMachine.PowerState;
+import com.cloud.vm.dao.VMInstanceDao;
+
+public class VirtualMachinePowerStateSyncImpl implements VirtualMachinePowerStateSync {
+    private static final Logger s_logger = Logger.getLogger(VirtualMachinePowerStateSyncImpl.class);
+
+    @Inject MessageBus _messageBus;
+    @Inject VMInstanceDao _instanceDao;
+    @Inject VirtualMachineManager _vmMgr;
+    
+    public VirtualMachinePowerStateSyncImpl() {
+    }
+    
+    @Override
+	public void resetHostSyncState(long hostId) {
+    	s_logger.info("Reset VM power state sync for host: " + hostId);
+    	_instanceDao.resetHostPowerStateTracking(hostId);
+    }
+    
+    @Override
+	public void processHostVmStateReport(long hostId, Map<String, HostVmStateReportEntry> report) {
+    	if(s_logger.isDebugEnabled())
+    		s_logger.debug("Process host VM state report from ping process. host: " + hostId);
+    	
+    	Map<Long, VirtualMachine.PowerState> translatedInfo = convertToInfos(report);
+    	processReport(hostId, translatedInfo);
+    }
+
+    @Override
+	public void processHostVmStatePingReport(long hostId, Map<String, PowerState> report) {
+    	if(s_logger.isDebugEnabled())
+    		s_logger.debug("Process host VM state report from ping process. host: " + hostId);
+    	
+    	Map<Long, VirtualMachine.PowerState> translatedInfo = convertHostPingInfos(report);
+    	processReport(hostId, translatedInfo);
+    }
+    
+    private void processReport(long hostId, Map<Long, VirtualMachine.PowerState> translatedInfo) {
+    	
+    	for(Map.Entry<Long, VirtualMachine.PowerState> entry : translatedInfo.entrySet()) {
+    		
+        	if(s_logger.isDebugEnabled())
+        		s_logger.debug("VM state report. host: " + hostId + ", vm id: " + entry.getKey() + ", power state: " + entry.getValue());
+
+    		if(_instanceDao.updatePowerState(entry.getKey(), hostId, entry.getValue())) {
+    			
+            	if(s_logger.isDebugEnabled())
+            		s_logger.debug("VM state report is updated. host: " + hostId + ", vm id: " + entry.getKey() + ", power state: " + entry.getValue());
+    			
+                _messageBus.publish(null, VirtualMachineManager.Topics.VM_POWER_STATE, PublishScope.GLOBAL, entry.getKey());
+    		}
+    	}
+    }
+ 
+    private Map<Long, VirtualMachine.PowerState> convertHostPingInfos(Map<String, PowerState> states) {
+        final HashMap<Long, VirtualMachine.PowerState> map = new HashMap<Long, VirtualMachine.PowerState>();
+        if (states == null) {
+            return map;
+        }
+    	
+        for (Map.Entry<String, PowerState> entry : states.entrySet()) {
+        	VMInstanceVO vm = findVM(entry.getKey());
+        	if(vm != null) {
+        		map.put(vm.getId(), entry.getValue());
+        		break;
+        	} else {
+        		s_logger.info("Unable to find matched VM in CloudStack DB. name: " + entry.getKey());
+        	}
+        }
+
+        return map;
+    }
+    	    
+    private Map<Long, VirtualMachine.PowerState> convertToInfos(Map<String, HostVmStateReportEntry> states) {
+        final HashMap<Long, VirtualMachine.PowerState> map = new HashMap<Long, VirtualMachine.PowerState>();
+        if (states == null) {
+            return map;
+        }
+        
+        for (Map.Entry<String, HostVmStateReportEntry> entry : states.entrySet()) {
+        	VMInstanceVO vm = findVM(entry.getKey());
+        	if(vm != null) {
+        		map.put(vm.getId(), entry.getValue().getState());
+        		break;
+        	} else {
+        		s_logger.info("Unable to find matched VM in CloudStack DB. name: " + entry.getKey());
+        	}
+        }
+
+        return map;
+    }
+    
+    private VMInstanceVO findVM(String vmName) {
+        return _instanceDao.findVMByInstanceName(vmName);
+    }
+}

engine/schema/src/com/cloud/vm/VMInstanceVO.java

@@ -163,6 +163,23 @@ public class VMInstanceVO implements VirtualMachine, FiniteStateObject<State, Vi
     @Column(name="disk_offering_id")
     protected Long diskOfferingId;
     
+    //
+    // Power state for VM state sync
+    //
+    @Enumerated(value=EnumType.STRING)
+    @Column(name="power_state", updatable=true)
+    protected PowerState powerState;
+    
+    @Column(name="power_state_update_time", updatable=true, nullable=false)
+    @Temporal(value=TemporalType.TIMESTAMP)
+    protected Date powerStateUpdateTime;
+    
+    @Column(name="power_state_update_count", updatable=true)
+    protected int powerStateUpdateCount;
+    
+    @Column(name="power_host", updatable=true)
+    protected Long powerHostId;
+        
     public VMInstanceVO(long id,
             long serviceOfferingId,
             String name,
@@ -500,5 +517,37 @@ public class VMInstanceVO implements VirtualMachine, FiniteStateObject<State, Vi
     public Boolean isDynamicallyScalable() {
         return this.dynamicallyScalable;
     }
+    
+    public VirtualMachine.PowerState getPowerState() {
+        return powerState;
+    }
+   
+    public void setPowerState(PowerState powerState) {
+        this.powerState = powerState;
+    }
+   
+    public Date getPowerStateUpdateTime() {
+        return powerStateUpdateTime;
+    }
+   
+    public void setPowerStateUpdateTime(Date updateTime) {
+        powerStateUpdateTime = updateTime;
+    }
+   
+    public int getPowerStateUpdateCount() {
+        return powerStateUpdateCount;
+    }
+   
+    public void setPowerStateUpdateCount(int count) {
+        powerStateUpdateCount = count;
+    }
+   
+    public Long getPowerHostId() {
+        return powerHostId;
+    }
+   
+    public void setPowerHostId(Long hostId) {
+        powerHostId = hostId;
+    }
 
 }

engine/schema/src/com/cloud/vm/dao/VMInstanceDao.java

@@ -122,4 +122,9 @@ public interface VMInstanceDao extends GenericDao<VMInstanceVO, Long>, StateDao<
 
     List<VMInstanceVO> listStartingWithNoHostId();
 
+    boolean updatePowerState(long instanceId, long powerHostId, VirtualMachine.PowerState powerState);
+    
+    void resetVmPowerStateTracking(long instanceId);
+    
+    void resetHostPowerStateTracking(long hostId);
 }

engine/schema/src/com/cloud/vm/dao/VMInstanceDaoImpl.java

@@ -37,6 +37,7 @@ import com.cloud.host.HostVO;
 import com.cloud.host.dao.HostDao;
 import com.cloud.server.ResourceTag.ResourceObjectType;
 import com.cloud.tags.dao.ResourceTagDao;
+import com.cloud.utils.DateUtil;
 import com.cloud.utils.Pair;
 import com.cloud.utils.db.Attribute;
 import com.cloud.utils.db.DB;
@@ -63,7 +64,8 @@ import com.cloud.vm.VirtualMachine.Type;
 public class VMInstanceDaoImpl extends GenericDaoBase<VMInstanceVO, Long> implements VMInstanceDao {
 
     public static final Logger s_logger = Logger.getLogger(VMInstanceDaoImpl.class);
-
+    private static final int MAX_CONSECUTIVE_SAME_STATE_UPDATE_COUNT = 3;
+    
     protected SearchBuilder<VMInstanceVO> VMClusterSearch;
     protected SearchBuilder<VMInstanceVO> LHVMClusterSearch;
     protected SearchBuilder<VMInstanceVO> IdStatesSearch;
@@ -638,8 +640,6 @@ public class VMInstanceDaoImpl extends GenericDaoBase<VMInstanceVO, Long> implem
         return listBy(sc);
     }
 
-
-
     @Override
     public List<String> listDistinctHostNames(long networkId, VirtualMachine.Type... types) {
         SearchCriteria<String> sc = DistinctHostNameSearch.create();
@@ -679,5 +679,63 @@ public class VMInstanceDaoImpl extends GenericDaoBase<VMInstanceVO, Long> implem
         sc.setParameters("state", State.Starting);
         return listBy(sc);
     }
-
+    
+    @Override
+    public boolean updatePowerState(long instanceId, long powerHostId, VirtualMachine.PowerState powerState) {
+    	boolean needToUpdate = false;
+    	TransactionLegacy txn = TransactionLegacy.currentTxn();
+    	txn.start();
+         
+        VMInstanceVO instance = findById(instanceId);
+       if(instance != null) {
+           Long savedPowerHostId = instance.getPowerHostId();
+           if(instance.getPowerState() != powerState || savedPowerHostId == null 
+        		   || savedPowerHostId.longValue() != powerHostId) {
+               instance.setPowerState(powerState);
+               instance.setPowerHostId(powerHostId);
+               instance.setPowerStateUpdateCount(1);
+               instance.setPowerStateUpdateTime(DateUtil.currentGMTTime());
+               needToUpdate = true;
+               update(instanceId, instance);
+           } else {
+               // to reduce DB updates, consecutive same state update for more than 3 times
+               if(instance.getPowerStateUpdateCount() < MAX_CONSECUTIVE_SAME_STATE_UPDATE_COUNT) {
+	               instance.setPowerStateUpdateCount(instance.getPowerStateUpdateCount() + 1);
+	               instance.setPowerStateUpdateTime(DateUtil.currentGMTTime());
+	               needToUpdate = true;
+	               update(instanceId, instance);
+               }
+           }
+        }
+        
+        txn.commit();
+        return needToUpdate;
+    }
+    
+    @Override
+    public void resetVmPowerStateTracking(long instanceId) {
+    	TransactionLegacy txn = TransactionLegacy.currentTxn();
+        txn.start();
+        VMInstanceVO instance = findById(instanceId);
+        if(instance != null) {
+               instance.setPowerStateUpdateCount(0);
+               instance.setPowerStateUpdateTime(DateUtil.currentGMTTime());
+               update(instanceId, instance);
+        }
+        
+        txn.commit();
+    }
+    
+    
+    @Override @DB
+    public void resetHostPowerStateTracking(long hostId) {
+       SearchCriteria<VMInstanceVO> sc = createSearchCriteria();
+       sc.addAnd("powerHostId", SearchCriteria.Op.EQ, hostId);
+       
+       VMInstanceVO instance = this.createForUpdate();
+       instance.setPowerStateUpdateCount(0);
+       instance.setPowerStateUpdateTime(DateUtil.currentGMTTime());
+       
+       this.update(instance, sc);
+    }
 }

framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/VmWorkJobDao.java

@@ -0,0 +1,35 @@
+// 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 org.apache.cloudstack.framework.jobs.dao;
+
+import java.util.Date;
+import java.util.List;
+
+import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO;
+import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO.Step;
+
+import com.cloud.utils.db.GenericDao;
+import com.cloud.vm.VirtualMachine;
+
+public interface VmWorkJobDao extends GenericDao<VmWorkJobVO, Long> {
+	VmWorkJobVO findPendingWorkJob(VirtualMachine.Type type, long instanceId);
+	List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId);
+	List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId, String jobCmd);
+	
+	void updateStep(long workJobId, Step step);
+	void expungeCompletedWorkJobs(Date cutDate);
+}

framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/VmWorkJobDaoImpl.java

@@ -0,0 +1,125 @@
+// 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 org.apache.cloudstack.framework.jobs.dao;
+
+import java.util.Date;
+import java.util.List;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO;
+import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO.Step;
+import org.apache.cloudstack.jobs.JobInfo;
+
+import com.cloud.utils.DateUtil;
+import com.cloud.utils.db.Filter;
+import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+import com.cloud.utils.db.SearchCriteria.Op;
+import com.cloud.vm.VirtualMachine;
+
+public class VmWorkJobDaoImpl extends GenericDaoBase<VmWorkJobVO, Long> implements VmWorkJobDao {
+
+    protected SearchBuilder<VmWorkJobVO> PendingWorkJobSearch;
+    protected SearchBuilder<VmWorkJobVO> PendingWorkJobByCommandSearch;
+    protected SearchBuilder<VmWorkJobVO> ExpungeWorkJobSearch;
+	
+	public VmWorkJobDaoImpl() {
+	}
+	
+	@PostConstruct
+	public void init() {
+		PendingWorkJobSearch = createSearchBuilder();
+		PendingWorkJobSearch.and("jobStatus", PendingWorkJobSearch.entity().getStatus(), Op.EQ);
+		PendingWorkJobSearch.and("vmType", PendingWorkJobSearch.entity().getVmType(), Op.EQ);
+		PendingWorkJobSearch.and("vmInstanceId", PendingWorkJobSearch.entity().getVmInstanceId(), Op.EQ);
+		PendingWorkJobSearch.and("step", PendingWorkJobSearch.entity().getStep(), Op.NEQ);
+		PendingWorkJobSearch.done();
+
+		PendingWorkJobByCommandSearch = createSearchBuilder();
+		PendingWorkJobByCommandSearch.and("jobStatus", PendingWorkJobByCommandSearch.entity().getStatus(), Op.EQ);
+		PendingWorkJobByCommandSearch.and("vmType", PendingWorkJobByCommandSearch.entity().getVmType(), Op.EQ);
+		PendingWorkJobByCommandSearch.and("vmInstanceId", PendingWorkJobByCommandSearch.entity().getVmInstanceId(), Op.EQ);
+		PendingWorkJobByCommandSearch.and("step", PendingWorkJobByCommandSearch.entity().getStep(), Op.NEQ);
+		PendingWorkJobByCommandSearch.and("cmd", PendingWorkJobByCommandSearch.entity().getCmd(), Op.EQ);
+		PendingWorkJobByCommandSearch.done();
+		
+		ExpungeWorkJobSearch = createSearchBuilder();
+		ExpungeWorkJobSearch.and("lastUpdated", ExpungeWorkJobSearch.entity().getLastUpdated(), Op.LT);
+		ExpungeWorkJobSearch.and("jobStatus", ExpungeWorkJobSearch.entity().getStatus(), Op.NEQ);
+		ExpungeWorkJobSearch.done();
+	}
+	
+	@Override
+    public VmWorkJobVO findPendingWorkJob(VirtualMachine.Type type, long instanceId) {
+		
+		SearchCriteria<VmWorkJobVO> sc = PendingWorkJobSearch.create();
+		sc.setParameters("jobStatus", JobInfo.	Status.IN_PROGRESS);
+		sc.setParameters("vmType", type);
+		sc.setParameters("vmInstanceId", instanceId);
+		
+		Filter filter = new Filter(VmWorkJobVO.class, "created", true, null, null);
+		List<VmWorkJobVO> result = this.listBy(sc, filter);
+		if(result != null && result.size() > 0)
+			return result.get(0);
+		
+		return null;
+	}
+	
+	@Override
+    public List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId) {
+		
+		SearchCriteria<VmWorkJobVO> sc = PendingWorkJobSearch.create();
+		sc.setParameters("jobStatus", JobInfo.Status.IN_PROGRESS);
+		sc.setParameters("vmType", type);
+		sc.setParameters("vmInstanceId", instanceId);
+		
+		Filter filter = new Filter(VmWorkJobVO.class, "created", true, null, null);
+		return this.listBy(sc, filter);
+	}
+
+	@Override
+    public List<VmWorkJobVO> listPendingWorkJobs(VirtualMachine.Type type, long instanceId, String jobCmd) {
+		
+		SearchCriteria<VmWorkJobVO> sc = PendingWorkJobByCommandSearch.create();
+		sc.setParameters("jobStatus", JobInfo.Status.IN_PROGRESS);
+		sc.setParameters("vmType", type);
+		sc.setParameters("vmInstanceId", instanceId);
+		sc.setParameters("cmd", jobCmd);
+		
+		Filter filter = new Filter(VmWorkJobVO.class, "created", true, null, null);
+		return this.listBy(sc, filter);
+	}
+	
+	@Override
+    public void updateStep(long workJobId, Step step) {
+		VmWorkJobVO jobVo = findById(workJobId);
+		jobVo.setStep(step);
+		jobVo.setLastUpdated(DateUtil.currentGMTTime());
+		update(workJobId, jobVo);
+	}
+	
+	@Override
+    public void expungeCompletedWorkJobs(Date cutDate) {
+		SearchCriteria<VmWorkJobVO> sc = ExpungeWorkJobSearch.create();
+		sc.setParameters("lastUpdated",cutDate);
+        sc.setParameters("jobStatus", JobInfo.Status.IN_PROGRESS);
+		
+		expunge(sc);
+	}
+}

framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/VmWorkJobVO.java

@@ -0,0 +1,101 @@
+// 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 org.apache.cloudstack.framework.jobs.impl;
+
+import javax.persistence.Column;
+import javax.persistence.DiscriminatorValue;
+import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
+import javax.persistence.PrimaryKeyJoinColumn;
+import javax.persistence.Table;
+
+import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
+
+import com.cloud.vm.VirtualMachine;
+
+@Entity
+@Table(name="vm_work_job")
+@DiscriminatorValue(value="VmWork")
+@PrimaryKeyJoinColumn(name="id")
+public class VmWorkJobVO extends AsyncJobVO {
+
+    // These steps are rather arbitrary.  What's recorded depends on the
+    // the operation being performed.
+	public enum Step {
+        Filed(false),
+        Prepare(false),
+        Starting(true),
+        Started(false),
+        Release(false),
+        Done(false),
+        Migrating(true),
+        Reconfiguring(false),
+        Error(false);
+        
+        boolean updateState; // Should the VM State be updated after this step?
+        private Step(boolean updateState) {
+            this.updateState = updateState;
+        }
+        
+        boolean updateState() {
+            return updateState;
+        }
+    }
+	
+    @Column(name="step")
+    Step step;
+    
+    @Column(name="vm_type")
+    @Enumerated(value=EnumType.STRING)
+    VirtualMachine.Type vmType;
+
+    @Column(name="vm_instance_id")
+    long vmInstanceId;
+
+    protected VmWorkJobVO() {
+    }
+
+    public VmWorkJobVO(String related) {
+        step = Step.Filed;
+        setRelated(related);
+    }
+    
+    public Step getStep() {
+    	return step;
+    }
+    
+    public void setStep(Step step) {
+    	this.step = step;
+    }
+    
+    public VirtualMachine.Type getVmType() {
+    	return vmType;
+    }
+    
+    public void setVmType(VirtualMachine.Type vmType) {
+    	this.vmType = vmType;
+    }
+    
+    public long getVmInstanceId() {
+    	return vmInstanceId;
+    }
+    
+    public void setVmInstanceId(long vmInstanceId) {
+    	this.vmInstanceId = vmInstanceId;
+    }
+}