diff --git a/engine/schema/src/main/resources/META-INF/db/schema-41000to41100.sql b/engine/schema/src/main/resources/META-INF/db/schema-41000to41100.sql
index 70a76d98d86..13a92ae87dd 100644
--- a/engine/schema/src/main/resources/META-INF/db/schema-41000to41100.sql
+++ b/engine/schema/src/main/resources/META-INF/db/schema-41000to41100.sql
@@ -19,8 +19,54 @@
 -- Schema upgrade from 4.10.0.0 to 4.11.0.0
 --;
 
+--;
+-- Stored procedure to do idempotent column add;
+--;
+DROP PROCEDURE IF EXISTS `cloud`.`IDEMPOTENT_ADD_COLUMN`;
+
+CREATE PROCEDURE `cloud`.`IDEMPOTENT_ADD_COLUMN` (
+		IN in_table_name VARCHAR(200)
+    , IN in_column_name VARCHAR(200)
+    , IN in_column_definition VARCHAR(1000)
+)
+BEGIN
+
+    DECLARE CONTINUE HANDLER FOR 1060 BEGIN END; SET @ddl = CONCAT('ALTER TABLE ', in_table_name); SET @ddl = CONCAT(@ddl, ' ', 'ADD COLUMN') ; SET @ddl = CONCAT(@ddl, ' ', in_column_name); SET @ddl = CONCAT(@ddl, ' ', in_column_definition); PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; END;
+
+DROP PROCEDURE IF EXISTS `cloud`.`IDEMPOTENT_DROP_FOREIGN_KEY`;
+
+CREATE PROCEDURE `cloud`.`IDEMPOTENT_DROP_FOREIGN_KEY` (
+		IN in_table_name VARCHAR(200)
+    , IN in_foreign_key_name VARCHAR(200)
+)
+BEGIN
+
+    DECLARE CONTINUE HANDLER FOR 1091 BEGIN END; SET @ddl = CONCAT('ALTER TABLE ', in_table_name); SET @ddl = CONCAT(@ddl, ' ', ' DROP FOREIGN KEY '); SET @ddl = CONCAT(@ddl, ' ', in_foreign_key_name); PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; END;
+
+DROP PROCEDURE IF EXISTS `cloud`.`IDEMPOTENT_DROP_INDEX`;
+
+CREATE PROCEDURE `cloud`.`IDEMPOTENT_DROP_INDEX` (
+		IN in_index_name VARCHAR(200)
+    , IN in_table_name VARCHAR(200)
+)
+BEGIN
+
+    DECLARE CONTINUE HANDLER FOR 1091 BEGIN END; SET @ddl = CONCAT('DROP INDEX ', in_index_name); SET @ddl = CONCAT(@ddl, ' ', ' ON ') ; SET @ddl = CONCAT(@ddl, ' ', in_table_name); PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; END;
+
+DROP PROCEDURE IF EXISTS `cloud`.`IDEMPOTENT_CREATE_UNIQUE_INDEX`;
+
+CREATE PROCEDURE `cloud`.`IDEMPOTENT_CREATE_UNIQUE_INDEX` (
+		IN in_index_name VARCHAR(200)
+    , IN in_table_name VARCHAR(200)
+    , IN in_index_definition VARCHAR(1000)
+)
+BEGIN
+
+    DECLARE CONTINUE HANDLER FOR 1061 BEGIN END; SET @ddl = CONCAT('CREATE UNIQUE INDEX ', in_index_name); SET @ddl = CONCAT(@ddl, ' ', ' ON ') ; SET @ddl = CONCAT(@ddl, ' ', in_table_name); SET @ddl = CONCAT(@ddl, ' ', in_index_definition); PREPARE stmt FROM @ddl; EXECUTE stmt; DEALLOCATE PREPARE stmt; END;
+
 -- Add For VPC flag
-ALTER TABLE cloud.network_offerings ADD COLUMN for_vpc INT(1) NOT NULL DEFAULT 0;
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.network_offerings','for_vpc', 'INT(1) NOT NULL DEFAULT 0');
+
 UPDATE cloud.network_offerings o
 SET for_vpc = 1
 where
@@ -88,7 +134,7 @@ CREATE TABLE IF NOT EXISTS `cloud`.`annotations` (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
 DROP VIEW IF EXISTS `cloud`.`last_annotation_view`;
-CREATE VIEW `last_annotation_view` AS
+CREATE VIEW `cloud`.`last_annotation_view` AS
     SELECT
         `annotations`.`uuid` AS `uuid`,
         `annotations`.`annotation` AS `annotation`,
@@ -405,21 +451,18 @@ UPDATE `cloud`.`monitoring_services` SET pidfile="/var/run/apache2/apache2.pid"
 UPDATE `cloud`.`vm_template` SET guest_os_id=99 WHERE id=8;
 
 -- Network External Ids
-ALTER TABLE `cloud`.`networks` ADD `external_id` varchar(255);
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.networks','external_id', 'varchar(255)');
 
 -- Separate Subnet for CPVM and SSVM (system vms)
-ALTER TABLE `cloud`.`op_dc_ip_address_alloc`
-ADD COLUMN `forsystemvms` TINYINT(1) NOT NULL DEFAULT '0' COMMENT 'Indicates if IP is dedicated for CPVM or SSVM';
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.op_dc_ip_address_alloc','forsystemvms', 'TINYINT(1) NOT NULL DEFAULT 0 COMMENT ''Indicates if IP is dedicated for CPVM or SSVM'' ');
 
-ALTER TABLE `cloud`.`op_dc_ip_address_alloc`
-ADD COLUMN `vlan` INT(10) UNSIGNED NULL COMMENT 'Vlan the management network range is on';
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.op_dc_ip_address_alloc','vlan', 'INT(10) UNSIGNED NULL COMMENT ''Vlan the management network range is on'' ');
 
 -- CLOUDSTACK-4757: Support multidisk OVA
-ALTER TABLE `cloud`.`vm_template` ADD COLUMN `parent_template_id` bigint(20) unsigned DEFAULT NULL COMMENT 'If datadisk template, then id of the root template this template belongs to';
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.vm_template','parent_template_id', 'bigint(20) unsigned DEFAULT NULL COMMENT ''If datadisk template, then id of the root template this template belongs to'' ');
 
 -- CLOUDSTACK-10146: Bypass Secondary Storage for KVM templates
-ALTER TABLE `cloud`.`vm_template`
-ADD COLUMN `direct_download` TINYINT(1) DEFAULT '0' COMMENT 'Indicates if Secondary Storage is bypassed and template is downloaded to Primary Storage';
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.vm_template','direct_download', 'TINYINT(1) DEFAULT 0 COMMENT ''Indicates if Secondary Storage is bypassed and template is downloaded to Primary Storage'' ');
 
 -- Changes to template_view for both multidisk OVA and bypass secondary storage for KVM templates
 DROP VIEW IF EXISTS `cloud`.`template_view`;
@@ -528,8 +571,7 @@ CREATE VIEW `cloud`.`template_view` AS
              OR (`resource_tags`.`resource_type` = 'ISO')))));
 
 -- CLOUDSTACK-10109: Enable dedication of public IPs to SSVM and CPVM
-ALTER TABLE `cloud`.`user_ip_address`
-ADD COLUMN `forsystemvms` TINYINT(1) NOT NULL DEFAULT '0' COMMENT 'true if IP is set to system vms, false if not';
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.user_ip_address','forsystemvms', 'TINYINT(1) NOT NULL DEFAULT 0 COMMENT ''true if IP is set to system vms, false if not'' ');
 
 -- ldap binding on domain level
 CREATE TABLE IF NOT EXISTS `cloud`.`domain_details` (
@@ -541,11 +583,11 @@ CREATE TABLE IF NOT EXISTS `cloud`.`domain_details` (
     CONSTRAINT `fk_domain_details__domain_id` FOREIGN KEY (`domain_id`) REFERENCES `domain`(`id`) ON DELETE CASCADE
 )ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
-ALTER TABLE cloud.ldap_configuration ADD COLUMN domain_id BIGINT(20) DEFAULT NULL;
-ALTER TABLE cloud.ldap_trust_map ADD COLUMN account_id BIGINT(20) DEFAULT 0;
-ALTER TABLE cloud.ldap_trust_map DROP FOREIGN KEY fk_ldap_trust_map__domain_id;
-DROP INDEX uk_ldap_trust_map__domain_id ON cloud.ldap_trust_map;
-CREATE UNIQUE INDEX uk_ldap_trust_map__bind_location ON ldap_trust_map (domain_id, account_id);
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.ldap_configuration','domain_id', 'BIGINT(20) DEFAULT NULL');
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.ldap_trust_map','account_id', 'BIGINT(20) DEFAULT 0');
+CALL `cloud`.`IDEMPOTENT_DROP_FOREIGN_KEY`('cloud.ldap_trust_map','fk_ldap_trust_map__domain_id');
+CALL `cloud`.`IDEMPOTENT_DROP_INDEX`('uk_ldap_trust_map__domain_id','cloud.ldap_trust_map');
+CALL `cloud`.`IDEMPOTENT_CREATE_UNIQUE_INDEX`('uk_ldap_trust_map__bind_location','cloud.ldap_trust_map', '(domain_id, account_id)');
 
 CREATE TABLE IF NOT EXISTS `cloud`.`netscaler_servicepackages` (
   `id` bigint unsigned NOT NULL AUTO_INCREMENT COMMENT 'id',
@@ -565,5 +607,5 @@ CREATE TABLE IF NOT EXISTS `cloud`.`external_netscaler_controlcenter` (
   PRIMARY KEY  (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
-ALTER TABLE `cloud`.`sslcerts` ADD COLUMN `name` varchar(255) NULL default NULL COMMENT 'Name of the Certificate';
-ALTER TABLE `cloud`.`network_offerings` ADD COLUMN `service_package_id` varchar(255) NULL default NULL COMMENT 'Netscaler ControlCenter Service Package';
\ No newline at end of file
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.sslcerts','name', 'varchar(255) NULL default NULL COMMENT ''Name of the Certificate'' ');
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.network_offerings','service_package_id', 'varchar(255) NULL default NULL COMMENT ''Netscaler ControlCenter Service Package'' ');
diff --git a/engine/storage/datamotion/src/main/java/org/apache/cloudstack/storage/motion/StorageSystemDataMotionStrategy.java b/engine/storage/datamotion/src/main/java/org/apache/cloudstack/storage/motion/StorageSystemDataMotionStrategy.java
index 30cff850c88..45e951b59a1 100644
--- a/engine/storage/datamotion/src/main/java/org/apache/cloudstack/storage/motion/StorageSystemDataMotionStrategy.java
+++ b/engine/storage/datamotion/src/main/java/org/apache/cloudstack/storage/motion/StorageSystemDataMotionStrategy.java
@@ -43,6 +43,7 @@ import com.cloud.host.HostVO;
 import com.cloud.host.dao.HostDao;
 import com.cloud.host.dao.HostDetailsDao;
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
+import com.cloud.resource.ResourceState;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.Snapshot;
@@ -1933,7 +1934,11 @@ public class StorageSystemDataMotionStrategy implements DataMotionStrategy {
         if (hosts != null && hosts.size() > 0) {
             Collections.shuffle(hosts, RANDOM);
 
-            return hosts.get(0);
+            for (HostVO host : hosts) {
+                if (ResourceState.Enabled.equals(host.getResourceState())) {
+                    return host;
+                }
+            }
         }
 
         throw new CloudRuntimeException("Unable to locate a host");
@@ -1954,6 +1959,10 @@ public class StorageSystemDataMotionStrategy implements DataMotionStrategy {
         Collections.shuffle(hosts, RANDOM);
 
         for (HostVO host : hosts) {
+            if (!ResourceState.Enabled.equals(host.getResourceState())) {
+                continue;
+            }
+
             if (computeClusterMustSupportResign) {
                 long clusterId = host.getClusterId();
 
diff --git a/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/StorageSystemSnapshotStrategy.java b/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/StorageSystemSnapshotStrategy.java
index 88c385b4e61..a6fe50e2581 100644
--- a/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/StorageSystemSnapshotStrategy.java
+++ b/engine/storage/snapshot/src/main/java/org/apache/cloudstack/storage/snapshot/StorageSystemSnapshotStrategy.java
@@ -25,7 +25,6 @@ import com.cloud.event.ActionEvent;
 import com.cloud.event.EventTypes;
 import com.cloud.event.UsageEventUtils;
 import com.cloud.exception.InvalidParameterValueException;
-import com.cloud.exception.ResourceAllocationException;
 import com.cloud.host.HostVO;
 import com.cloud.host.dao.HostDao;
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
@@ -56,7 +55,6 @@ import com.cloud.vm.snapshot.VMSnapshot;
 import com.cloud.vm.snapshot.VMSnapshotService;
 import com.cloud.vm.snapshot.VMSnapshotVO;
 import com.cloud.vm.snapshot.dao.VMSnapshotDao;
-import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 
 import org.apache.cloudstack.engine.subsystem.api.storage.ChapInfo;
@@ -86,6 +84,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
 import java.util.UUID;
 
@@ -102,14 +101,14 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
     @Inject private SnapshotDao snapshotDao;
     @Inject private SnapshotDataFactory snapshotDataFactory;
     @Inject private SnapshotDetailsDao snapshotDetailsDao;
-    @Inject SnapshotDataStoreDao snapshotStoreDao;
+    @Inject private SnapshotDataStoreDao snapshotStoreDao;
     @Inject private VolumeDetailsDao volumeDetailsDao;
     @Inject private VMInstanceDao vmInstanceDao;
     @Inject private VMSnapshotDao vmSnapshotDao;
     @Inject private VMSnapshotService vmSnapshotService;
     @Inject private VolumeDao volumeDao;
     @Inject private VolumeService volService;
-    @Inject private VolumeDetailsDao _volumeDetailsDaoImpl;
+    @Inject private VolumeDetailsDao volumeDetailsDaoImpl;
 
     @Override
     public SnapshotInfo backupSnapshot(SnapshotInfo snapshotInfo) {
@@ -200,7 +199,7 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
         try {
             snapshotObj.processEvent(Snapshot.Event.DestroyRequested);
 
-            List<VolumeDetailVO> volumesFromSnapshot = _volumeDetailsDaoImpl.findDetails("SNAPSHOT_ID", String.valueOf(snapshotId), null);
+            List<VolumeDetailVO> volumesFromSnapshot = volumeDetailsDaoImpl.findDetails("SNAPSHOT_ID", String.valueOf(snapshotId), null);
 
             if (volumesFromSnapshot.size() > 0) {
                 try {
@@ -471,20 +470,11 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
         if (ImageFormat.OVA.equals(volumeInfo.getFormat())) {
             setVmdk(snapshotInfo, volumeInfo);
 
-            try {
-                vmSnapshot = takeHypervisorSnapshot(volumeInfo);
-            }
-            catch (ResourceAllocationException ex) {
-                String errMsg = "Unable to allocate VM snapshot";
-
-                s_logger.error(errMsg, ex);
-
-                throw new CloudRuntimeException(errMsg, ex);
-            }
+            vmSnapshot = takeHypervisorSnapshot(volumeInfo);
         }
 
         SnapshotResult result = null;
-        SnapshotInfo snapshotOnPrimary = null;
+        SnapshotInfo snapshotOnPrimary;
 
         try {
             volumeInfo.stateTransit(Volume.Event.SnapshotRequested);
@@ -549,7 +539,7 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
 
     }
 
-    private VMSnapshot takeHypervisorSnapshot(VolumeInfo volumeInfo) throws ResourceAllocationException {
+    private VMSnapshot takeHypervisorSnapshot(VolumeInfo volumeInfo) {
         VirtualMachine virtualMachine = volumeInfo.getAttachedVM();
 
         if (virtualMachine != null && VirtualMachine.State.Running.equals(virtualMachine.getState())) {
@@ -678,7 +668,8 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
         if (hostId != null) {
             hostVO = hostDao.findById(hostId);
         }
-        else {
+
+        if (hostVO == null || !ResourceState.Enabled.equals(hostVO.getResourceState())) {
             Optional<HostVO> optHostVO = getHost(volumeInfo.getDataCenterId(), false);
 
             if (optHostVO.isPresent()) {
@@ -836,7 +827,7 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
 
         HostVO hostVO = hostDao.findById(hostId);
 
-        if (hostVO != null) {
+        if (hostVO != null && ResourceState.Enabled.equals(hostVO.getResourceState())) {
             return hostVO;
         }
 
@@ -867,7 +858,7 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
                     Collections.shuffle(hosts, new Random(System.nanoTime()));
 
                     for (HostVO host : hosts) {
-                        if (host.getResourceState() == ResourceState.Enabled) {
+                        if (ResourceState.Enabled.equals(host.getResourceState())) {
                             if (computeClusterMustSupportResign) {
                                 if (clusterDao.getSupportsResigning(cluster.getId())) {
                                     return Optional.of(host);
@@ -886,7 +877,7 @@ public class StorageSystemSnapshotStrategy extends SnapshotStrategyBase {
             }
         }
 
-        return Optional.absent();
+        return Optional.empty();
     }
 
     private void markAsBackedUp(SnapshotObject snapshotObj) {