diff --git a/engine/schema/src/com/cloud/upgrade/dao/Upgrade410to420.java b/engine/schema/src/com/cloud/upgrade/dao/Upgrade410to420.java
index 4c4faf7fafe..a004a793b23 100644
--- a/engine/schema/src/com/cloud/upgrade/dao/Upgrade410to420.java
+++ b/engine/schema/src/com/cloud/upgrade/dao/Upgrade410to420.java
@@ -89,6 +89,7 @@ public class Upgrade410to420 implements DbUpgrade {
         correctExternalNetworkDevicesSetup(conn);
         removeFirewallServiceFromSharedNetworkOfferingWithSGService(conn);
         fix22xKVMSnapshots(conn);
+	setKVMSnapshotFlag(conn);
         addIndexForAlert(conn);
         fixBaremetalForeignKeys(conn);
         // storage refactor related migration
@@ -296,6 +297,43 @@ public class Upgrade410to420 implements DbUpgrade {
         */
 	}
 
+        //KVM snapshot flag: only turn on if Customers is using snapshot;
+    private void setKVMSnapshotFlag(Connection conn) {
+        s_logger.debug("Verify and set the KVM snapshot flag if snapshot was used. ");
+        PreparedStatement pstmt = null;
+        ResultSet rs = null;
+        try {
+                int numRows = 0;
+                pstmt = conn.prepareStatement("select count(*) from `cloud`.`snapshots` where hypervisor_type = 'KVM'");
+            rs = pstmt.executeQuery();
+            if(rs.next()){
+                numRows = rs.getInt(1);
+            }
+            rs.close();
+            pstmt.close();
+            if (numRows > 0){
+              //Add the configuration flag
+                pstmt = conn.prepareStatement("UPDATE `cloud`.`configuration` SET value = ? WHERE name = 'KVM.snapshot.enabled'");
+                pstmt.setString(1, "true");
+                pstmt.executeUpdate();
+            }
+        } catch (SQLException e) {
+            throw new CloudRuntimeException("Failed to read the snapshot table for KVM upgrade. ", e);
+        } finally {
+            try {
+                if (rs != null) {
+                    rs.close();
+                }
+
+                if (pstmt != null) {
+                    pstmt.close();
+                }
+            } catch (SQLException e) {
+            }
+        }
+        s_logger.debug("Done set KVM snapshot flag. ");
+    }
+
 	private void updatePrimaryStore(Connection conn) {
 	    PreparedStatement sql = null;
 	    PreparedStatement sql2 = null;
diff --git a/server/src/com/cloud/configuration/Config.java b/server/src/com/cloud/configuration/Config.java
index e77f6ce0403..18b240300ab 100755
--- a/server/src/com/cloud/configuration/Config.java
+++ b/server/src/com/cloud/configuration/Config.java
@@ -144,6 +144,7 @@ public enum Config {
     SnapshotPollInterval("Snapshots", SnapshotManager.class, Integer.class, "snapshot.poll.interval", "300", "The time interval in seconds when the management server polls for snapshots to be scheduled.", null),
     SnapshotDeltaMax("Snapshots", SnapshotManager.class, Integer.class, "snapshot.delta.max", "16", "max delta snapshots between two full snapshots.", null),
     BackupSnapshotAferTakingSnapshot("Snapshots", SnapshotManager.class, Boolean.class, "snapshot.backup.rightafter", "true", "backup snapshot right after snapshot is taken", null),
+    KVMSnapshotEnabled("Snapshots", SnapshotManager.class, Boolean.class, "KVM.snapshot.enabled", "false", "whether snapshot is enabled for KVM hosts", null),
 
 	// Advanced
     JobExpireMinutes("Advanced", ManagementServer.class, String.class, "job.expire.minutes", "1440", "Time (in minutes) for async-jobs to be kept in system", null),
diff --git a/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java b/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
index 2c9ef825cba..4ea275d82e6 100755
--- a/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
+++ b/server/src/com/cloud/storage/snapshot/SnapshotManagerImpl.java
@@ -910,6 +910,12 @@ public class SnapshotManagerImpl extends ManagerBase implements SnapshotManager,
 		if (host.getHypervisorType() != HypervisorType.KVM) {
 			return true;
 		}
+
+        //Turn off snapshot by default for KVM, unless it is set in the global flag
+        boolean snapshotEnabled = Boolean.parseBoolean(_configDao.getValue("KVM.snapshot.enabled"));
+        if (!snapshotEnabled) {
+             return false;
+        }
 		// Determine host capabilities
 		String caps = host.getCapabilities();