diff --git a/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/SnapshotDataFactory.java b/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/SnapshotDataFactory.java
index 59e59a60662..6bbeb85d5f9 100644
--- a/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/SnapshotDataFactory.java
+++ b/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/SnapshotDataFactory.java
@@ -29,6 +29,8 @@ public interface SnapshotDataFactory {
 
     SnapshotInfo getSnapshot(long snapshotId, DataStoreRole role);
 
+    List<SnapshotInfo> getSnapshots(long volumeId, DataStoreRole store);
+
     List<SnapshotInfo> listSnapshotOnCache(long snapshotId);
 
     SnapshotInfo getReadySnapshotOnCache(long snapshotId);
diff --git a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotDataFactoryImpl.java b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotDataFactoryImpl.java
index 5dd63e38976..ad58f42a97a 100644
--- a/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotDataFactoryImpl.java
+++ b/engine/storage/snapshot/src/org/apache/cloudstack/storage/snapshot/SnapshotDataFactoryImpl.java
@@ -64,6 +64,24 @@ public class SnapshotDataFactoryImpl implements SnapshotDataFactory {
         return so;
     }
 
+    @Override
+    public List<SnapshotInfo> getSnapshots(long volumeId, DataStoreRole role) {
+
+        SnapshotDataStoreVO snapshotStore = snapshotStoreDao.findByVolume(volumeId, role);
+        if (snapshotStore == null) {
+            return new ArrayList<>();
+        }
+        DataStore store = storeMgr.getDataStore(snapshotStore.getDataStoreId(), role);
+        List<SnapshotVO> volSnapShots = snapshotDao.listByVolumeId(volumeId);
+        List<SnapshotInfo> infos = new ArrayList<>();
+        for(SnapshotVO snapshot: volSnapShots) {
+            SnapshotObject info = SnapshotObject.getSnapshotObject(snapshot, store);
+            infos.add(info);
+        }
+        return infos;
+    }
+
+
     @Override
     public SnapshotInfo getSnapshot(long snapshotId, DataStoreRole role) {
         SnapshotVO snapshot = snapshotDao.findById(snapshotId);
diff --git a/engine/storage/volume/src/org/apache/cloudstack/storage/volume/VolumeServiceImpl.java b/engine/storage/volume/src/org/apache/cloudstack/storage/volume/VolumeServiceImpl.java
index 8818724e722..ad0418f5c9d 100644
--- a/engine/storage/volume/src/org/apache/cloudstack/storage/volume/VolumeServiceImpl.java
+++ b/engine/storage/volume/src/org/apache/cloudstack/storage/volume/VolumeServiceImpl.java
@@ -1543,6 +1543,7 @@ public class VolumeServiceImpl implements VolumeService {
                 future.complete(res);
             } else {
                 srcVolume.processEvent(Event.OperationSuccessed);
+                snapshotMgr.cleanupSnapshotsByVolume(srcVolume.getId());
                 future.complete(res);
             }
         } catch (Exception e) {
@@ -1624,6 +1625,7 @@ public class VolumeServiceImpl implements VolumeService {
             } else {
                 for (Map.Entry<VolumeInfo, DataStore> entry : volumeToPool.entrySet()) {
                     VolumeInfo volume = entry.getKey();
+                    snapshotMgr.cleanupSnapshotsByVolume(volume.getId());
                     volume.processEvent(Event.OperationSuccessed);
                 }
                 future.complete(res);
diff --git a/server/src/com/cloud/storage/snapshot/SnapshotManager.java b/server/src/com/cloud/storage/snapshot/SnapshotManager.java
index c27c6992dcb..f3557d0a670 100644
--- a/server/src/com/cloud/storage/snapshot/SnapshotManager.java
+++ b/server/src/com/cloud/storage/snapshot/SnapshotManager.java
@@ -74,6 +74,8 @@ public interface SnapshotManager extends Configurable {
 
     boolean canOperateOnVolume(Volume volume);
 
+    void cleanupSnapshotsByVolume(Long volumeId);
+
     Answer sendToPool(Volume vol, Command cmd);
 
     SnapshotVO getParentSnapshot(VolumeInfo volume);
diff --git a/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java b/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
index 2c4de7e1042..1f2ca6421be 100755
--- a/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
+++ b/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
@@ -1339,6 +1339,21 @@ public class SnapshotManagerImpl extends MutualExclusiveIdsManagerBase implement
         return true;
     }
 
+    @Override
+    public void cleanupSnapshotsByVolume(Long volumeId) {
+        List<SnapshotInfo> infos = snapshotFactory.getSnapshots(volumeId, DataStoreRole.Primary);
+        for(SnapshotInfo info: infos) {
+            try {
+               if(info != null) {
+                   snapshotSrv.deleteSnapshot(info);
+               }
+            } catch(CloudRuntimeException e) {
+                String msg = "Cleanup of Snapshot with uuid " + info.getUuid() + " in primary storage is failed. Ignoring";
+                s_logger.warn(msg);
+            }
+        }
+    }
+
     @Override
     public Snapshot allocSnapshot(Long volumeId, Long policyId, String snapshotName, Snapshot.LocationType locationType) throws ResourceAllocationException {
         Account caller = CallContext.current().getCallingAccount();
diff --git a/server/test/com/cloud/storage/snapshot/SnapshotManagerTest.java b/server/test/com/cloud/storage/snapshot/SnapshotManagerTest.java
index 685f4954a2b..05eb8b96a3a 100755
--- a/server/test/com/cloud/storage/snapshot/SnapshotManagerTest.java
+++ b/server/test/com/cloud/storage/snapshot/SnapshotManagerTest.java
@@ -16,6 +16,37 @@
 // under the License.
 package com.cloud.storage.snapshot;
 
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.cloudstack.acl.ControlledEntity;
+import org.apache.cloudstack.acl.SecurityChecker.AccessType;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
+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.SnapshotStrategy;
+import org.apache.cloudstack.engine.subsystem.api.storage.StorageStrategyFactory;
+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.SnapshotStrategy.SnapshotOperation;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
+
 import com.cloud.configuration.Resource.ResourceType;
 import com.cloud.exception.InvalidParameterValueException;
 import com.cloud.exception.ResourceAllocationException;
@@ -44,38 +75,10 @@ import com.cloud.vm.dao.UserVmDao;
 import com.cloud.vm.snapshot.VMSnapshot;
 import com.cloud.vm.snapshot.VMSnapshotVO;
 import com.cloud.vm.snapshot.dao.VMSnapshotDao;
-import org.apache.cloudstack.acl.ControlledEntity;
-import org.apache.cloudstack.acl.SecurityChecker.AccessType;
-import org.apache.cloudstack.context.CallContext;
-import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
-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.SnapshotStrategy;
-import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotStrategy.SnapshotOperation;
-import org.apache.cloudstack.engine.subsystem.api.storage.StorageStrategyFactory;
-import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
-import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
-import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
-import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.MockitoAnnotations;
-import org.mockito.Spy;
+import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotService;
 
-import java.util.List;
-import java.util.UUID;
-
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyLong;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 public class SnapshotManagerTest {
     @Spy
@@ -126,6 +129,9 @@ public class SnapshotManagerTest {
     SnapshotDataStoreDao _snapshotStoreDao;
     @Mock
     SnapshotDataStoreVO snapshotStoreMock;
+    @Mock
+    SnapshotService snapshotSrv;
+
 
     private static final long TEST_SNAPSHOT_ID = 3L;
     private static final long TEST_VOLUME_ID = 4L;
@@ -330,5 +336,4 @@ public class SnapshotManagerTest {
         Snapshot snapshot = _snapshotMgr.backupSnapshotFromVmSnapshot(TEST_SNAPSHOT_ID, TEST_VM_ID, TEST_VOLUME_ID, TEST_VM_SNAPSHOT_ID);
         Assert.assertNull(snapshot);
     }
-
 }