diff --git a/client/pom.xml b/client/pom.xml
index 32ab94a3cc8..d25576a8089 100644
--- a/client/pom.xml
+++ b/client/pom.xml
@@ -329,6 +329,7 @@
         <artifactId>maven-jetty-plugin</artifactId>
         <version>6.1.26</version>
         <configuration>
+          <scanIntervalSeconds>0</scanIntervalSeconds>
           <stopPort>9966</stopPort>
           <stopKey>stop-jetty</stopKey>
           <connectors>
diff --git a/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java b/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java
index 01f02208307..d129fe7f827 100644
--- a/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java
+++ b/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreDao.java
@@ -33,6 +33,7 @@ public interface SnapshotDataStoreDao extends GenericDao<SnapshotDataStoreVO, Lo
     void deletePrimaryRecordsForStore(long id);
 
     SnapshotDataStoreVO findByStoreSnapshot(DataStoreRole role, long storeId, long snapshotId);
+    SnapshotDataStoreVO findParent(DataStoreRole role, Long storeId, Long volumeId);
 
     SnapshotDataStoreVO findBySnapshot(long snapshotId, DataStoreRole role);
 
diff --git a/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreVO.java b/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreVO.java
index 929b2c8254d..300df1e9673 100644
--- a/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreVO.java
+++ b/engine/api/src/org/apache/cloudstack/storage/datastore/db/SnapshotDataStoreVO.java
@@ -95,6 +95,9 @@ public class SnapshotDataStoreVO implements StateObject<ObjectInDataStoreStateMa
     @Column(name = "ref_cnt")
     Long refCnt = 0L;
 
+    @Column(name = "volume_id")
+    Long volumeId;
+
     public String getInstallPath() {
         return installPath;
     }
@@ -257,4 +260,12 @@ public class SnapshotDataStoreVO implements StateObject<ObjectInDataStoreStateMa
     public void decrRefCnt() {
         this.refCnt--;
     }
+
+    public Long getVolumeId() {
+        return volumeId;
+    }
+
+    public void setVolumeId(Long volumeId) {
+        this.volumeId = volumeId;
+    }
 }
diff --git a/engine/api/src/org/apache/cloudstack/storage/to/SnapshotObjectTO.java b/engine/api/src/org/apache/cloudstack/storage/to/SnapshotObjectTO.java
index 1115aece6ff..4754dcf34e1 100644
--- a/engine/api/src/org/apache/cloudstack/storage/to/SnapshotObjectTO.java
+++ b/engine/api/src/org/apache/cloudstack/storage/to/SnapshotObjectTO.java
@@ -42,8 +42,9 @@ public class SnapshotObjectTO implements DataTO {
         this.setId(snapshot.getId());
         this.volume = (VolumeObjectTO) snapshot.getBaseVolume().getTO();
         this.setVmName(snapshot.getBaseVolume().getAttachedVmName());
-        if (snapshot.getParent() != null) {
-            this.parentSnapshotPath = snapshot.getParent().getPath();
+        SnapshotInfo parentSnapshot = snapshot.getParent();
+        if (parentSnapshot != null) {
+            this.parentSnapshotPath = parentSnapshot.getPath();
         }
         this.dataStore = snapshot.getDataStore().getTO();
         this.setName(snapshot.getName());
diff --git a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotObject.java b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotObject.java
index 1cba96efd40..2fc576b8b8f 100644
--- a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotObject.java
+++ b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotObject.java
@@ -22,12 +22,7 @@ import java.util.Date;
 
 import javax.inject.Inject;
 
-import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
-import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine;
-import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory;
-import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
-import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
-import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
+import org.apache.cloudstack.engine.subsystem.api.storage.*;
 import org.apache.cloudstack.storage.command.CopyCmdAnswer;
 import org.apache.cloudstack.storage.command.CreateObjectAnswer;
 import org.apache.cloudstack.storage.datastore.ObjectInDataStoreManager;
@@ -92,18 +87,18 @@ public class SnapshotObject implements SnapshotInfo {
 
     @Override
     public SnapshotInfo getParent() {
+
         SnapshotDataStoreVO snapStoreVO = this.snapshotStoreDao.findByStoreSnapshot(this.store.getRole(),
                 this.store.getId(), this.snapshot.getId());
-        if (snapStoreVO == null) {
-            return null;
+        Long parentId = null;
+        if (snapStoreVO != null) {
+            parentId = snapStoreVO.getParentSnapshotId();
+            if (parentId != null && parentId != 0) {
+                return this.snapshotFactory.getSnapshot(parentId, store);
+            }
         }
 
-        long parentId = snapStoreVO.getParentSnapshotId();
-        if (parentId == 0) {
-            return null;
-        }
-
-        return this.snapshotFactory.getSnapshot(parentId, store);
+        return null;
     }
 
     @Override
@@ -181,7 +176,11 @@ public class SnapshotObject implements SnapshotInfo {
 
     @Override
     public String getPath() {
-        return this.objectInStoreMgr.findObject(this, getDataStore()).getInstallPath();
+        DataObjectInStore objectInStore = this.objectInStoreMgr.findObject(this, getDataStore());
+        if (objectInStore != null) {
+            return objectInStore.getInstallPath();
+        }
+        return null;
     }
 
     @Override
diff --git a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotServiceImpl.java b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotServiceImpl.java
index 3d7d4f26aa6..81036d04844 100644
--- a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotServiceImpl.java
+++ b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotServiceImpl.java
@@ -199,6 +199,7 @@ public class SnapshotServiceImpl implements SnapshotService {
             s_logger.debug("Failed to take snapshot: " + snapshot.getId(), e);
             try {
                 snapshot.processEvent(Snapshot.Event.OperationFailed);
+                snapshot.processEvent(Event.OperationFailed);
             } catch (NoTransitionException e1) {
                 s_logger.debug("Failed to change state for event: OperationFailed", e);
             }
@@ -237,6 +238,9 @@ public class SnapshotServiceImpl implements SnapshotService {
             // find the image store where the parent snapshot backup is located
             SnapshotDataStoreVO parentSnapshotOnBackupStore = _snapshotStoreDao.findBySnapshot(parentSnapshot.getId(),
                     DataStoreRole.Image);
+            if (parentSnapshotOnBackupStore == null) {
+                return dataStoreMgr.getImageStore(snapshot.getDataCenterId());
+            }
             return dataStoreMgr.getDataStore(parentSnapshotOnBackupStore.getDataStoreId(),
                     parentSnapshotOnBackupStore.getRole());
         }
diff --git a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/XenserverSnapshotStrategy.java b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/XenserverSnapshotStrategy.java
index 92af8c2b071..e30972d09b3 100644
--- a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/XenserverSnapshotStrategy.java
+++ b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/XenserverSnapshotStrategy.java
@@ -18,13 +18,9 @@ package org.apache.cloudstack.storage.snapshot;
 
 import javax.inject.Inject;
 
-import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
-import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
+import org.apache.cloudstack.engine.subsystem.api.storage.*;
 import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine.Event;
 import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine.State;
-import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotDataFactory;
-import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
-import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotService;
 import org.apache.cloudstack.storage.command.CreateObjectAnswer;
 import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
@@ -65,7 +61,8 @@ public class XenserverSnapshotStrategy extends SnapshotStrategyBase {
     @Override
     public SnapshotInfo backupSnapshot(SnapshotInfo snapshot) {
         SnapshotInfo parentSnapshot = snapshot.getParent();
-        if (parentSnapshot != null && parentSnapshot.getPath().equalsIgnoreCase(snapshot.getPath())) {
+
+        if (parentSnapshot != null && snapshot.getPath().equalsIgnoreCase(parentSnapshot.getPath())) {
             s_logger.debug("backup an empty snapshot");
             // don't need to backup this snapshot
             SnapshotDataStoreVO parentSnapshotOnBackupStore = this.snapshotStoreDao.findBySnapshot(
@@ -79,6 +76,7 @@ public class XenserverSnapshotStrategy extends SnapshotStrategyBase {
 
                 SnapshotObjectTO snapTO = new SnapshotObjectTO();
                 snapTO.setPath(parentSnapshotOnBackupStore.getInstallPath());
+
                 CreateObjectAnswer createSnapshotAnswer = new CreateObjectAnswer(snapTO);
 
                 snapshotOnImageStore.processEvent(Event.OperationSuccessed, createSnapshotAnswer);
@@ -109,7 +107,9 @@ public class XenserverSnapshotStrategy extends SnapshotStrategyBase {
             for (i = 1; i < deltaSnap; i++) {
                 parentSnapshotOnBackupStore = this.snapshotStoreDao.findBySnapshot(parentSnapshot.getId(),
                         DataStoreRole.Image);
-
+                if (parentSnapshotOnBackupStore == null) {
+                    break;
+                }
                 Long prevBackupId = parentSnapshotOnBackupStore.getParentSnapshotId();
 
                 if (prevBackupId == 0) {
@@ -193,8 +193,39 @@ public class XenserverSnapshotStrategy extends SnapshotStrategyBase {
 
     @Override
     public SnapshotInfo takeSnapshot(SnapshotInfo snapshot) {
-        snapshot = snapshotSvr.takeSnapshot(snapshot).getSnashot();
-        return this.backupSnapshot(snapshot);
+        SnapshotResult result =  snapshotSvr.takeSnapshot(snapshot);
+        if (result.isFailed()) {
+            s_logger.debug("Failed to take snapshot: " + result.getResult());
+            throw new CloudRuntimeException(result.getResult());
+        }
+        snapshot = result.getSnashot();
+        DataStore primaryStore = snapshot.getDataStore();
+
+        SnapshotInfo backupedSnapshot = this.backupSnapshot(snapshot);
+        try {
+            SnapshotInfo parent = snapshot.getParent();
+            if (backupedSnapshot != null && parent != null) {
+                Long parentSnapshotId = parent.getId();
+                while (parentSnapshotId != null && parentSnapshotId != 0L) {
+                    SnapshotDataStoreVO snapshotDataStoreVO = snapshotStoreDao.findByStoreSnapshot(primaryStore.getRole(),primaryStore.getId(), parentSnapshotId);
+                    if (snapshotDataStoreVO != null) {
+                        parentSnapshotId = snapshotDataStoreVO.getParentSnapshotId();
+                        snapshotStoreDao.remove(snapshotDataStoreVO.getId());
+                    } else {
+                        parentSnapshotId = null;
+                    }
+                }
+                SnapshotDataStoreVO snapshotDataStoreVO = snapshotStoreDao.findByStoreSnapshot(primaryStore.getRole(), primaryStore.getId(),
+                        snapshot.getId());
+                if (snapshotDataStoreVO != null) {
+                    snapshotDataStoreVO.setParentSnapshotId(0L);
+                    snapshotStoreDao.update(snapshotDataStoreVO.getId(), snapshotDataStoreVO);
+                }
+            }
+        } catch (Exception e) {
+            s_logger.debug("Failed to clean up snapshots on primary storage", e);
+        }
+        return backupedSnapshot;
     }
 
     @Override
diff --git a/engine/storage/src/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java b/engine/storage/src/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java
index 427609e3d25..c673776357e 100644
--- a/engine/storage/src/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java
+++ b/engine/storage/src/org/apache/cloudstack/storage/datastore/ObjectInDataStoreManagerImpl.java
@@ -107,10 +107,16 @@ public class ObjectInDataStoreManagerImpl implements ObjectInDataStoreManager {
                 VMTemplateStoragePoolVO vo = new VMTemplateStoragePoolVO(dataStore.getId(), obj.getId());
                 vo = templatePoolDao.persist(vo);
             } else if (obj.getType() == DataObjectType.SNAPSHOT) {
+                SnapshotInfo snapshotInfo = (SnapshotInfo)obj;
                 SnapshotDataStoreVO ss = new SnapshotDataStoreVO();
                 ss.setSnapshotId(obj.getId());
                 ss.setDataStoreId(dataStore.getId());
                 ss.setRole(dataStore.getRole());
+                ss.setVolumeId(snapshotInfo.getVolumeId());
+                SnapshotDataStoreVO snapshotDataStoreVO = snapshotDataStoreDao.findParent(dataStore.getRole(), dataStore.getId(),snapshotInfo.getVolumeId());
+                if (snapshotDataStoreVO != null) {
+                    ss.setParentSnapshotId(snapshotDataStoreVO.getSnapshotId());
+                }
                 ss.setState(ObjectInDataStoreStateMachine.State.Allocated);
                 ss = snapshotDataStoreDao.persist(ss);
             }
@@ -148,6 +154,11 @@ public class ObjectInDataStoreManagerImpl implements ObjectInDataStoreManager {
                 ss.setDataStoreId(dataStore.getId());
                 ss.setRole(dataStore.getRole());
                 ss.setRole(dataStore.getRole());
+                ss.setVolumeId(snapshot.getVolumeId());
+                SnapshotDataStoreVO snapshotDataStoreVO = snapshotDataStoreDao.findParent(dataStore.getRole(), dataStore.getId(),snapshot.getVolumeId());
+                if (snapshotDataStoreVO != null) {
+                    ss.setParentSnapshotId(snapshotDataStoreVO.getSnapshotId());
+                }
                 ss.setInstallPath(TemplateConstants.DEFAULT_SNAPSHOT_ROOT_DIR + "/"
                         + snapshotDao.findById(obj.getId()).getAccountId() + "/" + snapshot.getVolumeId());
                 ss.setState(ObjectInDataStoreStateMachine.State.Allocated);
diff --git a/engine/storage/src/org/apache/cloudstack/storage/image/db/SnapshotDataStoreDaoImpl.java b/engine/storage/src/org/apache/cloudstack/storage/image/db/SnapshotDataStoreDaoImpl.java
index a198004370d..f5e7421dea0 100644
--- a/engine/storage/src/org/apache/cloudstack/storage/image/db/SnapshotDataStoreDaoImpl.java
+++ b/engine/storage/src/org/apache/cloudstack/storage/image/db/SnapshotDataStoreDaoImpl.java
@@ -16,6 +16,9 @@
 // under the License.
 package org.apache.cloudstack.storage.image.db;
 
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
@@ -47,6 +50,12 @@ public class SnapshotDataStoreDaoImpl extends GenericDaoBase<SnapshotDataStoreVO
     private SearchBuilder<SnapshotDataStoreVO> destroyedSearch;
     private SearchBuilder<SnapshotDataStoreVO> snapshotSearch;
     private SearchBuilder<SnapshotDataStoreVO> storeSnapshotSearch;
+    private String parentSearch = "select store_id, store_role, snapshot_id from cloud.snapshot_store_ref where store_id = ? " +
+                                    " and store_role = ? and volume_id = ? and state = 'Ready'" +
+                                    " order by created DESC " +
+                                    " limit 1";
+
+
 
     @Override
     public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
@@ -146,7 +155,32 @@ public class SnapshotDataStoreDaoImpl extends GenericDaoBase<SnapshotDataStoreVO
         sc.setParameters("store_id", storeId);
         sc.setParameters("snapshot_id", snapshotId);
         sc.setParameters("store_role", role);
-        return findOneIncludingRemovedBy(sc);
+        return findOneBy(sc);
+    }
+
+    @Override
+    public SnapshotDataStoreVO findParent(DataStoreRole role, Long storeId, Long volumeId) {
+        Transaction txn = Transaction.currentTxn();
+        PreparedStatement pstmt = null;
+        ResultSet rs = null;
+        try {
+            pstmt = txn.prepareStatement(parentSearch);
+            pstmt.setLong(1, storeId);
+            pstmt.setString(2, role.toString());
+            pstmt.setLong(3, volumeId);
+            rs = pstmt.executeQuery();
+            while (rs.next()) {
+                long sid = rs.getLong(1);
+                String rl = rs.getString(2);
+                long snid = rs.getLong(3);
+                return this.findByStoreSnapshot(role, sid, snid);
+            }
+        } catch (SQLException e) {
+           s_logger.debug("Failed to find parent snapshot: " + e.toString());
+        } finally {
+            txn.close();
+        }
+        return null;
     }
 
     @Override
diff --git a/plugins/storage/volume/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackPrimaryDataStoreDriverImpl.java b/plugins/storage/volume/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackPrimaryDataStoreDriverImpl.java
index a2334074315..9ea91b59ef3 100644
--- a/plugins/storage/volume/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackPrimaryDataStoreDriverImpl.java
+++ b/plugins/storage/volume/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackPrimaryDataStoreDriverImpl.java
@@ -155,6 +155,7 @@ public class CloudStackPrimaryDataStoreDriverImpl implements PrimaryDataStoreDri
         try {
             DataTO snapshotTO = snapshot.getTO();
 
+
             CreateObjectCommand cmd = new CreateObjectCommand(snapshotTO);
             EndPoint ep = this.epSelector.select(snapshot);
             Answer answer = ep.sendMessage(cmd);
diff --git a/setup/db/db/schema-410to420.sql b/setup/db/db/schema-410to420.sql
index b4a3cd57c50..3668618502f 100644
--- a/setup/db/db/schema-410to420.sql
+++ b/setup/db/db/schema-410to420.sql
@@ -198,6 +198,7 @@ CREATE TABLE  `cloud`.`snapshot_store_ref` (
   `update_count` bigint unsigned,
   `ref_cnt` bigint unsigned,
   `updated` datetime,   
+  `volume_id` bigint unsigned,
   PRIMARY KEY  (`id`),
   INDEX `i_snapshot_store_ref__store_id`(`store_id`),
   CONSTRAINT `fk_snapshot_store_ref__snapshot_id` FOREIGN KEY `fk_snapshot_store_ref__snapshot_id` (`snapshot_id`) REFERENCES `snapshots` (`id`),