diff --git a/engine/components-api/src/com/cloud/vm/snapshot/VMSnapshotManager.java b/engine/components-api/src/com/cloud/vm/snapshot/VMSnapshotManager.java
index ce8a818ac1d..82456004cc3 100644
--- a/engine/components-api/src/com/cloud/vm/snapshot/VMSnapshotManager.java
+++ b/engine/components-api/src/com/cloud/vm/snapshot/VMSnapshotManager.java
@@ -19,12 +19,18 @@ package com.cloud.vm.snapshot;
 
 import java.util.List;
 
+import org.apache.cloudstack.framework.config.ConfigKey;
+
 import com.cloud.agent.api.RestoreVMSnapshotCommand;
 import com.cloud.utils.component.Manager;
 import com.cloud.vm.UserVmVO;
 import com.cloud.vm.VMInstanceVO;
 
 public interface VMSnapshotManager extends VMSnapshotService, Manager {
+
+    static final ConfigKey<Integer> VMSnapshotExpireInterval = new ConfigKey<Integer>("Advanced", Integer.class, "vmsnapshot.expire.interval", "-1",
+            "VM Snapshot expire interval in hours", true, ConfigKey.Scope.Account);
+
     public static final int VMSNAPSHOTMAX = 10;
 
     /**
diff --git a/server/src/com/cloud/storage/snapshot/SnapshotManager.java b/server/src/com/cloud/storage/snapshot/SnapshotManager.java
index 606109fe25b..9b6eb8841ea 100644
--- a/server/src/com/cloud/storage/snapshot/SnapshotManager.java
+++ b/server/src/com/cloud/storage/snapshot/SnapshotManager.java
@@ -32,6 +32,10 @@ import org.apache.cloudstack.framework.config.ConfigKey;
  *
  */
 public interface SnapshotManager {
+    public static final int HOURLYMAX = 8;
+    public static final int DAILYMAX = 8;
+    public static final int WEEKLYMAX = 8;
+    public static final int MONTHLYMAX = 12;
     public static final int DELTAMAX = 16;
 
     static final ConfigKey<Integer> SnapshotHourlyMax = new ConfigKey<Integer>(Integer.class, "snapshot.max.hourly", "Snapshots", "8",
diff --git a/server/src/com/cloud/storage/snapshot/SnapshotSchedulerImpl.java b/server/src/com/cloud/storage/snapshot/SnapshotSchedulerImpl.java
index f6c137916fe..7af1c190c93 100644
--- a/server/src/com/cloud/storage/snapshot/SnapshotSchedulerImpl.java
+++ b/server/src/com/cloud/storage/snapshot/SnapshotSchedulerImpl.java
@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.TimeUnit;
 
 import javax.inject.Inject;
 import javax.naming.ConfigurationException;
@@ -64,6 +65,9 @@ import com.cloud.utils.db.DB;
 import com.cloud.utils.db.GlobalLock;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.TransactionLegacy;
+import com.cloud.vm.snapshot.VMSnapshotManager;
+import com.cloud.vm.snapshot.VMSnapshotVO;
+import com.cloud.vm.snapshot.dao.VMSnapshotDao;
 
 @Component
 public class SnapshotSchedulerImpl extends ManagerBase implements SnapshotScheduler {
@@ -87,6 +91,12 @@ public class SnapshotSchedulerImpl extends ManagerBase implements SnapshotSchedu
     protected ApiDispatcher _dispatcher;
     @Inject
     protected AccountDao _acctDao;
+    @Inject
+    protected SnapshotApiService _snapshotService;
+    @Inject
+    protected VMSnapshotDao _vmSnapshotDao;
+    @Inject
+    protected VMSnapshotManager _vmSnaphostManager;
 
     protected AsyncJobDispatcher _asyncDispatcher;
 
@@ -153,6 +163,13 @@ public class SnapshotSchedulerImpl extends ManagerBase implements SnapshotSchedu
         } finally {
             scanLock.releaseRef();
         }
+
+        try {
+            deleteExpiredVMSnapshots();
+        }
+        catch (Exception e) {
+            s_logger.warn("Error in expiring vm snapshots", e);
+        }
     }
 
     private void checkStatusOfCurrentlyExecutingSnapshots() {
@@ -219,6 +236,27 @@ public class SnapshotSchedulerImpl extends ManagerBase implements SnapshotSchedu
         }
     }
 
+    @DB
+    protected void deleteExpiredVMSnapshots() {
+        Date now = new Date();
+        List<VMSnapshotVO> vmSnapshots = _vmSnapshotDao.listAll();
+        for (VMSnapshotVO vmSnapshot : vmSnapshots) {
+            long accountId = vmSnapshot.getAccountId();
+            int expiration_interval_hours = VMSnapshotManager.VMSnapshotExpireInterval.valueIn(accountId);
+            if (expiration_interval_hours < 0 ) {
+                continue;
+            }
+            Date creationTime = vmSnapshot.getCreated();
+            long diffInHours = TimeUnit.MILLISECONDS.toHours(now.getTime() - creationTime.getTime());
+            if (diffInHours >= expiration_interval_hours) {
+                if (s_logger.isDebugEnabled()){
+                    s_logger.debug("Deleting expired VM snapshot id: " + vmSnapshot.getId());
+                }
+                _vmSnaphostManager.deleteVMSnapshot(vmSnapshot.getId());
+            }
+        }
+    }
+
     @DB
     protected void scheduleSnapshots() {
         String displayTime = DateUtil.displayDateInTimezone(DateUtil.GMT_TIMEZONE, _currentTimestamp);
diff --git a/server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java b/server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java
index e4ee4516393..97e46a7ca9d 100644
--- a/server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java
+++ b/server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java
@@ -36,6 +36,7 @@ import org.apache.cloudstack.engine.subsystem.api.storage.VMSnapshotStrategy;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
 import org.apache.cloudstack.framework.config.ConfigKey;
+import org.apache.cloudstack.framework.config.Configurable;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.jobs.AsyncJob;
 import org.apache.cloudstack.framework.jobs.AsyncJobExecutionContext;
@@ -118,7 +119,7 @@ import com.cloud.vm.snapshot.dao.VMSnapshotDao;
 import com.cloud.vm.snapshot.dao.VMSnapshotDetailsDao;
 
 @Component
-public class VMSnapshotManagerImpl extends MutualExclusiveIdsManagerBase implements VMSnapshotManager, VMSnapshotService, VmWorkJobHandler {
+public class VMSnapshotManagerImpl extends MutualExclusiveIdsManagerBase implements VMSnapshotManager, VMSnapshotService, VmWorkJobHandler, Configurable {
     private static final Logger s_logger = Logger.getLogger(VMSnapshotManagerImpl.class);
 
     public static final String VM_WORK_JOB_HANDLER = VMSnapshotManagerImpl.class.getSimpleName();
@@ -1297,4 +1298,14 @@ public class VMSnapshotManagerImpl extends MutualExclusiveIdsManagerBase impleme
         }
         return true;
     }
+
+    @Override
+    public String getConfigComponentName() {
+        return VMSnapshotManager.class.getSimpleName();
+    }
+
+    @Override
+    public ConfigKey<?>[] getConfigKeys() {
+        return new ConfigKey<?>[] {VMSnapshotExpireInterval};
+    }
 }