diff --git a/agent/src/com/cloud/agent/direct/download/DirectTemplateDownloaderImpl.java b/agent/src/com/cloud/agent/direct/download/DirectTemplateDownloaderImpl.java
index e120d847b17..419ab7d1bbd 100644
--- a/agent/src/com/cloud/agent/direct/download/DirectTemplateDownloaderImpl.java
+++ b/agent/src/com/cloud/agent/direct/download/DirectTemplateDownloaderImpl.java
@@ -22,6 +22,7 @@ import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.script.Script;
 import org.apache.cloudstack.utils.security.DigestHelper;
 import org.apache.commons.lang.StringUtils;
+import org.apache.log4j.Logger;
 
 import java.io.File;
 import java.io.FileInputStream;
@@ -37,6 +38,8 @@ public abstract class DirectTemplateDownloaderImpl implements DirectTemplateDown
     private String downloadedFilePath;
     private String installPath;
     private String checksum;
+    private boolean redownload = false;
+    public static final Logger s_logger = Logger.getLogger(DirectTemplateDownloaderImpl.class.getName());
 
     protected DirectTemplateDownloaderImpl(final String url, final String destPoolPath, final Long templateId, final String checksum) {
         this.url = url;
@@ -70,6 +73,10 @@ public abstract class DirectTemplateDownloaderImpl implements DirectTemplateDown
         return url;
     }
 
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
     public String getDestPoolPath() {
         return destPoolPath;
     }
@@ -86,6 +93,18 @@ public abstract class DirectTemplateDownloaderImpl implements DirectTemplateDown
         this.downloadedFilePath = filePath;
     }
 
+    public String getChecksum() {
+        return checksum;
+    }
+
+    public void setChecksum(String checksum) {
+        this.checksum = checksum;
+    }
+
+    public boolean isRedownload() {
+        return redownload;
+    }
+
     /**
      * Return filename from url
      */
@@ -155,14 +174,47 @@ public abstract class DirectTemplateDownloaderImpl implements DirectTemplateDown
     @Override
     public boolean validateChecksum() {
         if (StringUtils.isNotBlank(checksum)) {
+            int retry = 3;
+            boolean valid = false;
             try {
-                return DigestHelper.check(checksum, new FileInputStream(downloadedFilePath));
+                while (!valid && retry > 0) {
+                    retry--;
+                    s_logger.info("Performing checksum validation for downloaded template " + templateId + " using " + checksum + ", retries left: " + retry);
+                    valid = DigestHelper.check(checksum, new FileInputStream(downloadedFilePath));
+                    if (!valid && retry > 0) {
+                        s_logger.info("Checksum validation failded, re-downloading template");
+                        redownload = true;
+                        resetDownloadFile();
+                        downloadTemplate();
+                    }
+                }
+                s_logger.info("Checksum validation for template " + templateId + ": " + (valid ? "succeeded" : "failed"));
+                return valid;
             } catch (IOException e) {
-                throw new CloudRuntimeException("could not check sum for file: " + downloadedFilePath,e);
+                throw new CloudRuntimeException("could not check sum for file: " + downloadedFilePath, e);
             } catch (NoSuchAlgorithmException e) {
                 throw new CloudRuntimeException("Unknown checksum algorithm: " + checksum, e);
             }
         }
+        s_logger.info("No checksum provided, skipping checksum validation");
         return true;
     }
+
+    /**
+     * Delete and create download file
+     */
+    private void resetDownloadFile() {
+        File f = new File(getDownloadedFilePath());
+        s_logger.info("Resetting download file: " + getDownloadedFilePath() + ", in order to re-download and persist template " + templateId + " on it");
+        try {
+            if (f.exists()) {
+                f.delete();
+            }
+            f.createNewFile();
+        } catch (IOException e) {
+            s_logger.error("Error creating file to download on: " + getDownloadedFilePath() + " due to: " + e.getMessage());
+            throw new CloudRuntimeException("Failed to create download file for direct download");
+        }
+    }
+
 }
diff --git a/agent/src/com/cloud/agent/direct/download/HttpDirectTemplateDownloader.java b/agent/src/com/cloud/agent/direct/download/HttpDirectTemplateDownloader.java
index 1f36e43ff12..147ccabf8fc 100644
--- a/agent/src/com/cloud/agent/direct/download/HttpDirectTemplateDownloader.java
+++ b/agent/src/com/cloud/agent/direct/download/HttpDirectTemplateDownloader.java
@@ -22,12 +22,9 @@ package com.cloud.agent.direct.download;
 import com.cloud.utils.exception.CloudRuntimeException;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.httpclient.HttpClient;
-import org.apache.commons.httpclient.HttpMethod;
-import org.apache.commons.httpclient.HttpMethodRetryHandler;
 import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
-import org.apache.commons.httpclient.NoHttpResponseException;
+import org.apache.commons.httpclient.HttpStatus;
 import org.apache.commons.httpclient.methods.GetMethod;
-import org.apache.commons.httpclient.params.HttpMethodParams;
 import org.apache.commons.io.IOUtils;
 import org.apache.log4j.Logger;
 
@@ -36,21 +33,22 @@ import java.io.FileOutputStream;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Map;
 
 public class HttpDirectTemplateDownloader extends DirectTemplateDownloaderImpl {
 
-    private HttpClient client;
+    protected HttpClient client;
     private static final MultiThreadedHttpConnectionManager s_httpClientManager = new MultiThreadedHttpConnectionManager();
-    private static final int CHUNK_SIZE = 1024 * 1024; //1M
-    protected HttpMethodRetryHandler myretryhandler;
     public static final Logger s_logger = Logger.getLogger(HttpDirectTemplateDownloader.class.getName());
     protected GetMethod request;
+    protected Map<String, String> reqHeaders = new HashMap<>();
 
     public HttpDirectTemplateDownloader(String url, Long templateId, String destPoolPath, String checksum, Map<String, String> headers) {
         super(url, destPoolPath, templateId, checksum);
+        s_httpClientManager.getParams().setConnectionTimeout(5000);
+        s_httpClientManager.getParams().setSoTimeout(5000);
         client = new HttpClient(s_httpClientManager);
-        myretryhandler = createRetryTwiceHandler();
         request = createRequest(url, headers);
         String downloadDir = getDirectDownloadTempPath(templateId);
         createTemporaryDirectoryAndFile(downloadDir);
@@ -64,50 +62,34 @@ public class HttpDirectTemplateDownloader extends DirectTemplateDownloaderImpl {
 
     protected GetMethod createRequest(String downloadUrl, Map<String, String> headers) {
         GetMethod request = new GetMethod(downloadUrl);
-        request.getParams().setParameter(HttpMethodParams.RETRY_HANDLER, myretryhandler);
         request.setFollowRedirects(true);
         if (MapUtils.isNotEmpty(headers)) {
             for (String key : headers.keySet()) {
                 request.setRequestHeader(key, headers.get(key));
+                reqHeaders.put(key, headers.get(key));
             }
         }
         return request;
     }
 
-    protected HttpMethodRetryHandler createRetryTwiceHandler() {
-        return new HttpMethodRetryHandler() {
-            @Override
-            public boolean retryMethod(final HttpMethod method, final IOException exception, int executionCount) {
-                if (executionCount >= 2) {
-                    // Do not retry if over max retry count
-                    return false;
-                }
-                if (exception instanceof NoHttpResponseException) {
-                    // Retry if the server dropped connection on us
-                    return true;
-                }
-                if (!method.isRequestSent()) {
-                    // Retry if the request has not been sent fully or
-                    // if it's OK to retry methods that have been sent
-                    return true;
-                }
-                // otherwise do not retry
-                return false;
-            }
-        };
-    }
-
     @Override
     public boolean downloadTemplate() {
         try {
-            client.executeMethod(request);
+            int status = client.executeMethod(request);
+            if (status != HttpStatus.SC_OK) {
+                s_logger.warn("Not able to download template, status code: " + status);
+                return false;
+            }
+            return performDownload();
         } catch (IOException e) {
             throw new CloudRuntimeException("Error on HTTP request: " + e.getMessage());
+        } finally {
+            request.releaseConnection();
         }
-        return performDownload();
     }
 
     protected boolean performDownload() {
+        s_logger.info("Downloading template " + getTemplateId() + " from " + getUrl() + " to: " + getDownloadedFilePath());
         try (
                 InputStream in = request.getResponseBodyAsStream();
                 OutputStream out = new FileOutputStream(getDownloadedFilePath());
diff --git a/agent/src/com/cloud/agent/direct/download/HttpsDirectTemplateDownloader.java b/agent/src/com/cloud/agent/direct/download/HttpsDirectTemplateDownloader.java
index 664181fbda3..38f59837cd8 100644
--- a/agent/src/com/cloud/agent/direct/download/HttpsDirectTemplateDownloader.java
+++ b/agent/src/com/cloud/agent/direct/download/HttpsDirectTemplateDownloader.java
@@ -24,6 +24,8 @@ import com.cloud.utils.script.Script;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpEntity;
 import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.commons.collections.MapUtils;
+import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
@@ -44,14 +46,15 @@ import java.security.KeyStore;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
 import java.security.cert.CertificateException;
+import java.util.Map;
 
 public class HttpsDirectTemplateDownloader extends HttpDirectTemplateDownloader {
 
     private CloseableHttpClient httpsClient;
     private HttpUriRequest req;
 
-    public HttpsDirectTemplateDownloader(String url, Long templateId, String destPoolPath, String checksum) {
-        super(url, templateId, destPoolPath, checksum, null);
+    public HttpsDirectTemplateDownloader(String url, Long templateId, String destPoolPath, String checksum, Map<String, String> headers) {
+        super(url, templateId, destPoolPath, checksum, headers);
         SSLContext sslcontext = null;
         try {
             sslcontext = getSSLContext();
@@ -59,12 +62,21 @@ public class HttpsDirectTemplateDownloader extends HttpDirectTemplateDownloader
             throw new CloudRuntimeException("Failure getting SSL context for HTTPS downloader: " + e.getMessage());
         }
         SSLConnectionSocketFactory factory = new SSLConnectionSocketFactory(sslcontext, SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
-        httpsClient = HttpClients.custom().setSSLSocketFactory(factory).build();
-        req = createUriRequest(url);
+        RequestConfig config = RequestConfig.custom()
+                .setConnectTimeout(5000)
+                .setConnectionRequestTimeout(5000)
+                .setSocketTimeout(5000).build();
+        httpsClient = HttpClients.custom().setSSLSocketFactory(factory).setDefaultRequestConfig(config).build();
+        createUriRequest(url, headers);
     }
 
-    protected HttpUriRequest createUriRequest(String downloadUrl) {
-        return new HttpGet(downloadUrl);
+    protected void createUriRequest(String downloadUrl, Map<String, String> headers) {
+        req = new HttpGet(downloadUrl);
+        if (MapUtils.isNotEmpty(headers)) {
+            for (String headerKey: headers.keySet()) {
+                req.setHeader(headerKey, headers.get(headerKey));
+            }
+        }
     }
 
     private SSLContext getSSLContext() throws KeyStoreException, NoSuchAlgorithmException, CertificateException, IOException, KeyManagementException {
@@ -98,6 +110,7 @@ public class HttpsDirectTemplateDownloader extends HttpDirectTemplateDownloader
      * Consume response and persist it on getDownloadedFilePath() file
      */
     protected boolean consumeResponse(CloseableHttpResponse response) {
+        s_logger.info("Downloading template " + getTemplateId() + " from " + getUrl() + " to: " + getDownloadedFilePath());
         if (response.getStatusLine().getStatusCode() != 200) {
             throw new CloudRuntimeException("Error on HTTPS response");
         }
@@ -113,4 +126,4 @@ public class HttpsDirectTemplateDownloader extends HttpDirectTemplateDownloader
         return true;
     }
 
-}
\ No newline at end of file
+}
diff --git a/agent/src/com/cloud/agent/direct/download/MetalinkDirectTemplateDownloader.java b/agent/src/com/cloud/agent/direct/download/MetalinkDirectTemplateDownloader.java
index e4ecd6d9c5c..2fd8ba03611 100644
--- a/agent/src/com/cloud/agent/direct/download/MetalinkDirectTemplateDownloader.java
+++ b/agent/src/com/cloud/agent/direct/download/MetalinkDirectTemplateDownloader.java
@@ -18,32 +18,81 @@
 //
 package com.cloud.agent.direct.download;
 
-import com.cloud.utils.script.Script;
+import com.cloud.utils.UriUtils;
+import com.cloud.utils.exception.CloudRuntimeException;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.log4j.Logger;
 
 import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
 
-public class MetalinkDirectTemplateDownloader extends DirectTemplateDownloaderImpl {
+public class MetalinkDirectTemplateDownloader extends HttpDirectTemplateDownloader {
 
-    private String downloadDir;
+    private String metalinkUrl;
+    private List<String> metalinkUrls;
+    private List<String> metalinkChecksums;
+    private Random random = new Random();
+    private static final Logger s_logger = Logger.getLogger(MetalinkDirectTemplateDownloader.class.getName());
 
-    public MetalinkDirectTemplateDownloader(String url, String destPoolPath, Long templateId, String checksum) {
-        super(url, destPoolPath, templateId, checksum);
-        String relativeDir = getDirectDownloadTempPath(templateId);
-        downloadDir = getDestPoolPath() + File.separator + relativeDir;
-        createFolder(downloadDir);
+    public MetalinkDirectTemplateDownloader(String url, String destPoolPath, Long templateId, String checksum, Map<String, String> headers) {
+        super(url, templateId, destPoolPath, checksum, headers);
+        metalinkUrl = url;
+        metalinkUrls = UriUtils.getMetalinkUrls(metalinkUrl);
+        metalinkChecksums = UriUtils.getMetalinkChecksums(metalinkUrl);
+        if (CollectionUtils.isEmpty(metalinkUrls)) {
+            throw new CloudRuntimeException("No urls found on metalink file: " + metalinkUrl + ". Not possible to download template " + templateId);
+        }
+        setUrl(metalinkUrls.get(0));
+        s_logger.info("Metalink downloader created, metalink url: " + metalinkUrl + " parsed - " +
+                metalinkUrls.size() + " urls and " +
+                (CollectionUtils.isNotEmpty(metalinkChecksums) ? metalinkChecksums.size() : "0") + " checksums found");
     }
 
     @Override
     public boolean downloadTemplate() {
-        String downloadCommand = "aria2c " + getUrl() + " -d " + downloadDir + " --check-integrity=true";
-        Script.runSimpleBashScript(downloadCommand);
-        //Remove .metalink file
-        Script.runSimpleBashScript("rm -f " + downloadDir + File.separator + getFileNameFromUrl());
-        String fileName = Script.runSimpleBashScript("ls " + downloadDir);
-        if (fileName == null) {
-            return false;
+        if (StringUtils.isBlank(getUrl())) {
+            throw new CloudRuntimeException("Download url has not been set, aborting");
         }
-        setDownloadedFilePath(downloadDir + File.separator + fileName);
-        return true;
+        String downloadDir = getDirectDownloadTempPath(getTemplateId());
+        boolean downloaded = false;
+        int i = 0;
+        do {
+            if (!isRedownload()) {
+                setUrl(metalinkUrls.get(i));
+            }
+            s_logger.info("Trying to download template from url: " + getUrl());
+            try {
+                File f = new File(getDestPoolPath() + File.separator + downloadDir + File.separator + getFileNameFromUrl());
+                if (f.exists()) {
+                    f.delete();
+                    f.createNewFile();
+                }
+                setDownloadedFilePath(f.getAbsolutePath());
+                request = createRequest(getUrl(), reqHeaders);
+                downloaded = super.downloadTemplate();
+                if (downloaded) {
+                    s_logger.info("Successfully downloaded template from url: " + getUrl());
+                }
+
+            } catch (Exception e) {
+                s_logger.error("Error downloading template: " + getTemplateId() + " from " + getUrl() + ": " + e.getMessage());
+            }
+            i++;
+        }
+        while (!downloaded && !isRedownload() && i < metalinkUrls.size());
+        return downloaded;
+    }
+
+    @Override
+    public boolean validateChecksum() {
+        if (StringUtils.isBlank(getChecksum()) && CollectionUtils.isNotEmpty(metalinkChecksums)) {
+            String chk = metalinkChecksums.get(random.nextInt(metalinkChecksums.size()));
+            setChecksum(chk);
+            s_logger.info("Checksum not provided but " + metalinkChecksums.size() + " found on metalink file, performing checksum using one of them: " + chk);
+        }
+        return super.validateChecksum();
     }
 }
diff --git a/api/src/com/cloud/event/EventTypes.java b/api/src/com/cloud/event/EventTypes.java
index b7454693a74..907b93eca10 100644
--- a/api/src/com/cloud/event/EventTypes.java
+++ b/api/src/com/cloud/event/EventTypes.java
@@ -581,6 +581,8 @@ public class EventTypes {
     public static final String EVENT_ANNOTATION_CREATE = "ANNOTATION.CREATE";
     public static final String EVENT_ANNOTATION_REMOVE = "ANNOTATION.REMOVE";
 
+    public static final String EVENT_TEMPLATE_DIRECT_DOWNLOAD_FAILURE = "TEMPLATE.DIRECT.DOWNLOAD.FAILURE";
+    public static final String EVENT_ISO_DIRECT_DOWNLOAD_FAILURE = "ISO.DIRECT.DOWNLOAD.FAILURE";
 
     static {
 
@@ -972,6 +974,9 @@ public class EventTypes {
 
         entityEventDetails.put(EVENT_ANNOTATION_CREATE, Annotation.class);
         entityEventDetails.put(EVENT_ANNOTATION_REMOVE, Annotation.class);
+
+        entityEventDetails.put(EVENT_TEMPLATE_DIRECT_DOWNLOAD_FAILURE, VirtualMachineTemplate.class);
+        entityEventDetails.put(EVENT_ISO_DIRECT_DOWNLOAD_FAILURE, "Iso");
     }
 
     public static String getEntityForEvent(String eventName) {
diff --git a/api/src/org/apache/cloudstack/api/command/admin/direct/download/UploadTemplateDirectDownloadCertificate.java b/api/src/org/apache/cloudstack/api/command/admin/direct/download/UploadTemplateDirectDownloadCertificate.java
index cd95e10d608..89c0c25c9ca 100755
--- a/api/src/org/apache/cloudstack/api/command/admin/direct/download/UploadTemplateDirectDownloadCertificate.java
+++ b/api/src/org/apache/cloudstack/api/command/admin/direct/download/UploadTemplateDirectDownloadCertificate.java
@@ -67,9 +67,12 @@ public class UploadTemplateDirectDownloadCertificate extends BaseCmd {
             throw new ServerApiException(ApiErrorCode.PARAM_ERROR, "Currently supporting KVM hosts only");
         }
 
+        SuccessResponse response = new SuccessResponse(getCommandName());
         try {
-            directDownloadManager.uploadCertificateToHosts(certificate, name);;
-            setResponseObject(new SuccessResponse(getCommandName()));
+            LOG.debug("Uploading certificate " + name + " to agents for Direct Download");
+            boolean result = directDownloadManager.uploadCertificateToHosts(certificate, name, hypervisor);
+            response.setSuccess(result);
+            setResponseObject(response);
         } catch (Exception e) {
             throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, e.getMessage());
         }
@@ -77,7 +80,7 @@ public class UploadTemplateDirectDownloadCertificate extends BaseCmd {
 
     @Override
     public String getCommandName() {
-        return UploadTemplateDirectDownloadCertificate.APINAME;
+        return APINAME.toLowerCase() + BaseCmd.RESPONSE_SUFFIX;
     }
 
     @Override
diff --git a/core/src/com/cloud/storage/template/MetalinkTemplateDownloader.java b/core/src/com/cloud/storage/template/MetalinkTemplateDownloader.java
index 149c6a1e826..39c42e04196 100644
--- a/core/src/com/cloud/storage/template/MetalinkTemplateDownloader.java
+++ b/core/src/com/cloud/storage/template/MetalinkTemplateDownloader.java
@@ -19,37 +19,138 @@
 package com.cloud.storage.template;
 
 import com.cloud.storage.StorageLayer;
-import com.cloud.utils.script.Script;
+import com.cloud.utils.UriUtils;
+import org.apache.commons.httpclient.HttpClient;
+import org.apache.commons.httpclient.HttpMethod;
+import org.apache.commons.httpclient.HttpMethodRetryHandler;
+import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
+import org.apache.commons.httpclient.NoHttpResponseException;
+import org.apache.commons.httpclient.methods.GetMethod;
+import org.apache.commons.httpclient.params.HttpMethodParams;
+import org.apache.commons.io.IOUtils;
 import org.apache.log4j.Logger;
+import org.springframework.util.CollectionUtils;
 
 import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
 
 public class MetalinkTemplateDownloader extends TemplateDownloaderBase implements TemplateDownloader {
 
     private TemplateDownloader.Status status = TemplateDownloader.Status.NOT_STARTED;
+    protected HttpClient client;
+    private static final MultiThreadedHttpConnectionManager s_httpClientManager = new MultiThreadedHttpConnectionManager();
+    protected HttpMethodRetryHandler myretryhandler;
+    protected GetMethod request;
+    private boolean toFileSet = false;
 
     private static final Logger LOGGER = Logger.getLogger(MetalinkTemplateDownloader.class.getName());
 
     public MetalinkTemplateDownloader(StorageLayer storageLayer, String downloadUrl, String toDir, DownloadCompleteCallback callback, long maxTemplateSize) {
         super(storageLayer, downloadUrl, toDir, maxTemplateSize, callback);
-        String[] parts = _downloadUrl.split("/");
-        String filename = parts[parts.length - 1];
-        _toFile = toDir + File.separator + filename;
+        s_httpClientManager.getParams().setConnectionTimeout(5000);
+        client = new HttpClient(s_httpClientManager);
+        myretryhandler = createRetryTwiceHandler();
+        request = createRequest(downloadUrl);
     }
 
+    protected GetMethod createRequest(String downloadUrl) {
+        GetMethod request = new GetMethod(downloadUrl);
+        request.getParams().setParameter(HttpMethodParams.RETRY_HANDLER, myretryhandler);
+        request.setFollowRedirects(true);
+        if (!toFileSet) {
+            String[] parts = downloadUrl.split("/");
+            String filename = parts[parts.length - 1];
+            _toFile = _toDir + File.separator + filename;
+            toFileSet = true;
+        }
+        return request;
+    }
+
+    protected HttpMethodRetryHandler createRetryTwiceHandler() {
+        return new HttpMethodRetryHandler() {
+            @Override
+            public boolean retryMethod(final HttpMethod method, final IOException exception, int executionCount) {
+                if (executionCount >= 2) {
+                    // Do not retry if over max retry count
+                    return false;
+                }
+                if (exception instanceof NoHttpResponseException) {
+                    // Retry if the server dropped connection on us
+                    return true;
+                }
+                if (!method.isRequestSent()) {
+                    // Retry if the request has not been sent fully or
+                    // if it's OK to retry methods that have been sent
+                    return true;
+                }
+                // otherwise do not retry
+                return false;
+            }
+        };
+    }
+
+    private boolean downloadTemplate() {
+        try {
+            client.executeMethod(request);
+        } catch (IOException e) {
+            LOGGER.error("Error on HTTP request: " + e.getMessage());
+            return false;
+        }
+        return performDownload();
+    }
+
+    private boolean performDownload() {
+        try (
+                InputStream in = request.getResponseBodyAsStream();
+                OutputStream out = new FileOutputStream(_toFile);
+        ) {
+            IOUtils.copy(in, out);
+        } catch (IOException e) {
+            LOGGER.error("Error downloading template from: " + _downloadUrl + " due to: " + e.getMessage());
+            return false;
+        }
+        return true;
+    }
     @Override
     public long download(boolean resume, DownloadCompleteCallback callback) {
-        if (!status.equals(Status.NOT_STARTED)) {
-            // Only start downloading if we haven't started yet.
-            LOGGER.debug("Template download is already started, not starting again. Template: " + _downloadUrl);
+        if (_status == Status.ABORTED || _status == Status.UNRECOVERABLE_ERROR || _status == Status.DOWNLOAD_FINISHED) {
             return 0;
         }
+
+        LOGGER.info("Starting metalink download from: " + _downloadUrl);
+        _start = System.currentTimeMillis();
+
         status = Status.IN_PROGRESS;
-        Script.runSimpleBashScript("aria2c " + _downloadUrl + " -d " + _toDir);
+        List<String> metalinkUrls = UriUtils.getMetalinkUrls(_downloadUrl);
+        if (CollectionUtils.isEmpty(metalinkUrls)) {
+            LOGGER.error("No URLs found for metalink: " + _downloadUrl);
+            status = Status.UNRECOVERABLE_ERROR;
+            return 0;
+        }
+        boolean downloaded = false;
+        int i = 0;
+        while (!downloaded && i < metalinkUrls.size()) {
+            String url = metalinkUrls.get(i);
+            request = createRequest(url);
+            downloaded = downloadTemplate();
+            i++;
+        }
+        if (!downloaded) {
+            LOGGER.error("Template couldnt be downloaded");
+            status = Status.UNRECOVERABLE_ERROR;
+            return 0;
+        }
+        LOGGER.info("Template downloaded successfully on: " + _toFile);
         status = Status.DOWNLOAD_FINISHED;
-        String sizeResult = Script.runSimpleBashScript("ls -als " + _toFile + " | awk '{print $1}'");
-        long size = Long.parseLong(sizeResult);
-        return size;
+        _downloadTime = System.currentTimeMillis() - _start;
+        if (_callback != null) {
+            _callback.downloadComplete(status);
+        }
+        return _totalBytes;
     }
 
     @Override
@@ -63,4 +164,13 @@ public class MetalinkTemplateDownloader extends TemplateDownloaderBase implement
         }
     }
 
+    @Override
+    public Status getStatus() {
+        return status;
+    }
+
+    @Override
+    public void setStatus(Status status) {
+        this.status = status;
+    }
 }
diff --git a/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadAnswer.java b/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadAnswer.java
index e4e559d8102..0ba9797bfe5 100644
--- a/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadAnswer.java
+++ b/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadAnswer.java
@@ -24,11 +24,13 @@ public class DirectDownloadAnswer extends Answer {
 
     private Long templateSize;
     private String installPath;
+    private boolean retryOnOtherHosts;
 
-    public DirectDownloadAnswer(final boolean result, final String msg) {
+    public DirectDownloadAnswer(final boolean result, final String msg, final boolean retry) {
         super(null);
         this.result = result;
         this.details = msg;
+        this.retryOnOtherHosts = retry;
     }
 
     public DirectDownloadAnswer(final boolean result, final Long size, final String installPath) {
@@ -45,4 +47,8 @@ public class DirectDownloadAnswer extends Answer {
     public String getInstallPath() {
         return installPath;
     }
+
+    public boolean isRetryOnOtherHosts() {
+        return retryOnOtherHosts;
+    }
 }
diff --git a/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadCommand.java b/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadCommand.java
index 140ad99bb5e..7a05d6144e4 100644
--- a/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadCommand.java
+++ b/core/src/org/apache/cloudstack/agent/directdownload/DirectDownloadCommand.java
@@ -22,6 +22,8 @@ package org.apache.cloudstack.agent.directdownload;
 import org.apache.cloudstack.storage.command.StorageSubSystemCommand;
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 
+import java.util.Map;
+
 public abstract class DirectDownloadCommand extends StorageSubSystemCommand {
 
     public enum DownloadProtocol {
@@ -32,12 +34,14 @@ public abstract class DirectDownloadCommand extends StorageSubSystemCommand {
     private Long templateId;
     private PrimaryDataStoreTO destPool;
     private String checksum;
+    private Map<String, String> headers;
 
-    protected DirectDownloadCommand (final String url, final Long templateId, final PrimaryDataStoreTO destPool, final String checksum) {
+    protected DirectDownloadCommand (final String url, final Long templateId, final PrimaryDataStoreTO destPool, final String checksum, final Map<String, String> headers) {
         this.url = url;
         this.templateId = templateId;
         this.destPool = destPool;
         this.checksum = checksum;
+        this.headers = headers;
     }
 
     public String getUrl() {
@@ -56,6 +60,10 @@ public abstract class DirectDownloadCommand extends StorageSubSystemCommand {
         return checksum;
     }
 
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
+
     @Override
     public void setExecuteInSequence(boolean inSeq) {
     }
diff --git a/core/src/org/apache/cloudstack/agent/directdownload/HttpDirectDownloadCommand.java b/core/src/org/apache/cloudstack/agent/directdownload/HttpDirectDownloadCommand.java
index 525c8bf50fb..7e32688154c 100644
--- a/core/src/org/apache/cloudstack/agent/directdownload/HttpDirectDownloadCommand.java
+++ b/core/src/org/apache/cloudstack/agent/directdownload/HttpDirectDownloadCommand.java
@@ -24,15 +24,8 @@ import java.util.Map;
 
 public class HttpDirectDownloadCommand extends DirectDownloadCommand {
 
-    private Map<String, String> headers;
-
     public HttpDirectDownloadCommand(String url, Long templateId, PrimaryDataStoreTO destPool, String checksum, Map<String, String> headers) {
-        super(url, templateId, destPool, checksum);
-        this.headers = headers;
-    }
-
-    public Map<String, String> getHeaders() {
-        return headers;
+        super(url, templateId, destPool, checksum, headers);
     }
 
 }
diff --git a/core/src/org/apache/cloudstack/agent/directdownload/HttpsDirectDownloadCommand.java b/core/src/org/apache/cloudstack/agent/directdownload/HttpsDirectDownloadCommand.java
index 26ed59e41cb..ca926f1c761 100644
--- a/core/src/org/apache/cloudstack/agent/directdownload/HttpsDirectDownloadCommand.java
+++ b/core/src/org/apache/cloudstack/agent/directdownload/HttpsDirectDownloadCommand.java
@@ -26,6 +26,6 @@ import java.util.Map;
 public class HttpsDirectDownloadCommand extends DirectDownloadCommand {
 
     public HttpsDirectDownloadCommand(String url, Long templateId, PrimaryDataStoreTO destPool, String checksum, Map<String, String> headers) {
-        super(url, templateId, destPool, checksum);
+        super(url, templateId, destPool, checksum, headers);
     }
 }
\ No newline at end of file
diff --git a/core/src/org/apache/cloudstack/agent/directdownload/MetalinkDirectDownloadCommand.java b/core/src/org/apache/cloudstack/agent/directdownload/MetalinkDirectDownloadCommand.java
index 92ec7453425..da528d96694 100644
--- a/core/src/org/apache/cloudstack/agent/directdownload/MetalinkDirectDownloadCommand.java
+++ b/core/src/org/apache/cloudstack/agent/directdownload/MetalinkDirectDownloadCommand.java
@@ -20,10 +20,12 @@ package org.apache.cloudstack.agent.directdownload;
 
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 
+import java.util.Map;
+
 public class MetalinkDirectDownloadCommand extends DirectDownloadCommand {
 
-    public MetalinkDirectDownloadCommand(String url, Long templateId, PrimaryDataStoreTO destPool, String checksum) {
-        super(url, templateId, destPool, checksum);
+    public MetalinkDirectDownloadCommand(String url, Long templateId, PrimaryDataStoreTO destPool, String checksum, Map<String, String> headers) {
+        super(url, templateId, destPool, checksum, headers);
     }
 
 }
diff --git a/core/src/org/apache/cloudstack/agent/directdownload/NfsDirectDownloadCommand.java b/core/src/org/apache/cloudstack/agent/directdownload/NfsDirectDownloadCommand.java
index 55cef2a2f6a..abc0137dfbd 100644
--- a/core/src/org/apache/cloudstack/agent/directdownload/NfsDirectDownloadCommand.java
+++ b/core/src/org/apache/cloudstack/agent/directdownload/NfsDirectDownloadCommand.java
@@ -20,10 +20,12 @@ package org.apache.cloudstack.agent.directdownload;
 
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 
+import java.util.Map;
+
 public class NfsDirectDownloadCommand extends DirectDownloadCommand {
 
-    public NfsDirectDownloadCommand(final String url, final Long templateId, final PrimaryDataStoreTO destPool, final String checksum) {
-        super(url, templateId, destPool, checksum);
+    public NfsDirectDownloadCommand(final String url, final Long templateId, final PrimaryDataStoreTO destPool, final String checksum, final Map<String, String> headers) {
+        super(url, templateId, destPool, checksum, headers);
     }
 
 }
diff --git a/framework/direct-download/src/org/apache/cloudstack/framework/agent/direct/download/DirectDownloadService.java b/framework/direct-download/src/org/apache/cloudstack/framework/agent/direct/download/DirectDownloadService.java
index a9a96cc3483..f3153e3470e 100644
--- a/framework/direct-download/src/org/apache/cloudstack/framework/agent/direct/download/DirectDownloadService.java
+++ b/framework/direct-download/src/org/apache/cloudstack/framework/agent/direct/download/DirectDownloadService.java
@@ -27,5 +27,5 @@ public interface DirectDownloadService {
     /**
      * Upload client certificate to each running host
      */
-    boolean uploadCertificateToHosts(String certificateCer, String certificateName);
+    boolean uploadCertificateToHosts(String certificateCer, String certificateName, String hypervisor);
 }
diff --git a/plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java b/plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
index f09e8f7b0ea..36be2d39a2e 100644
--- a/plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
+++ b/plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
@@ -42,6 +42,7 @@ import com.cloud.agent.direct.download.HttpDirectTemplateDownloader;
 import com.cloud.agent.direct.download.MetalinkDirectTemplateDownloader;
 import com.cloud.agent.direct.download.NfsDirectTemplateDownloader;
 import com.cloud.agent.direct.download.HttpsDirectTemplateDownloader;
+import com.cloud.exception.InvalidParameterValueException;
 import org.apache.cloudstack.agent.directdownload.HttpsDirectDownloadCommand;
 import org.apache.cloudstack.agent.directdownload.DirectDownloadCommand;
 import org.apache.cloudstack.agent.directdownload.HttpDirectDownloadCommand;
@@ -1071,19 +1072,9 @@ public class KVMStorageProcessor implements StorageProcessor {
         final DiskTO disk = cmd.getDisk();
         final TemplateObjectTO isoTO = (TemplateObjectTO)disk.getData();
         final DataStoreTO store = isoTO.getDataStore();
-        String dataStoreUrl = null;
-        if (store instanceof NfsTO) {
-            NfsTO nfsStore = (NfsTO)store;
-            dataStoreUrl = nfsStore.getUrl();
-        } else if (store instanceof PrimaryDataStoreTO && ((PrimaryDataStoreTO) store).getPoolType().equals(StoragePoolType.NetworkFilesystem)) {
-            //In order to support directly downloaded ISOs
-            String psHost = ((PrimaryDataStoreTO) store).getHost();
-            String psPath = ((PrimaryDataStoreTO) store).getPath();
-            dataStoreUrl = "nfs://" + psHost + File.separator + psPath;
-        } else {
-            return new AttachAnswer("unsupported protocol");
-        }
+
         try {
+            String dataStoreUrl = getDataStoreUrlFromStore(store);
             final Connect conn = LibvirtConnection.getConnectionByVmName(cmd.getVmName());
             attachOrDetachISO(conn, cmd.getVmName(), dataStoreUrl + File.separator + isoTO.getPath(), true);
         } catch (final LibvirtException e) {
@@ -1092,6 +1083,8 @@ public class KVMStorageProcessor implements StorageProcessor {
             return new Answer(cmd, false, e.toString());
         } catch (final InternalErrorException e) {
             return new Answer(cmd, false, e.toString());
+        } catch (final InvalidParameterValueException e) {
+            return new Answer(cmd, false, e.toString());
         }
 
         return new Answer(cmd);
@@ -1102,24 +1095,45 @@ public class KVMStorageProcessor implements StorageProcessor {
         final DiskTO disk = cmd.getDisk();
         final TemplateObjectTO isoTO = (TemplateObjectTO)disk.getData();
         final DataStoreTO store = isoTO.getDataStore();
-        if (!(store instanceof NfsTO)) {
-            return new AttachAnswer("unsupported protocol");
-        }
-        final NfsTO nfsStore = (NfsTO)store;
+
         try {
+            String dataStoreUrl = getDataStoreUrlFromStore(store);
             final Connect conn = LibvirtConnection.getConnectionByVmName(cmd.getVmName());
-            attachOrDetachISO(conn, cmd.getVmName(), nfsStore.getUrl() + File.separator + isoTO.getPath(), false);
+            attachOrDetachISO(conn, cmd.getVmName(), dataStoreUrl + File.separator + isoTO.getPath(), false);
         } catch (final LibvirtException e) {
             return new Answer(cmd, false, e.toString());
         } catch (final URISyntaxException e) {
             return new Answer(cmd, false, e.toString());
         } catch (final InternalErrorException e) {
             return new Answer(cmd, false, e.toString());
+        } catch (final InvalidParameterValueException e) {
+            return new Answer(cmd, false, e.toString());
         }
 
         return new Answer(cmd);
     }
 
+    /**
+     * Return data store URL from store
+     */
+    private String getDataStoreUrlFromStore(DataStoreTO store) {
+        if (!(store instanceof NfsTO) && (!(store instanceof PrimaryDataStoreTO) ||
+                store instanceof PrimaryDataStoreTO && !((PrimaryDataStoreTO) store).getPoolType().equals(StoragePoolType.NetworkFilesystem))) {
+            throw new InvalidParameterValueException("unsupported protocol");
+        }
+
+        if (store instanceof NfsTO) {
+            NfsTO nfsStore = (NfsTO)store;
+            return nfsStore.getUrl();
+        } else if (store instanceof PrimaryDataStoreTO && ((PrimaryDataStoreTO) store).getPoolType().equals(StoragePoolType.NetworkFilesystem)) {
+            //In order to support directly downloaded ISOs
+            String psHost = ((PrimaryDataStoreTO) store).getHost();
+            String psPath = ((PrimaryDataStoreTO) store).getPath();
+            return "nfs://" + psHost + File.separator + psPath;
+        }
+        return store.getUrl();
+    }
+
     protected synchronized String attachOrDetachDevice(final Connect conn, final boolean attach, final String vmName, final String xml) throws LibvirtException, InternalErrorException {
         Domain dm = null;
         try {
@@ -1582,36 +1596,57 @@ public class KVMStorageProcessor implements StorageProcessor {
         return new Answer(cmd, false, "not implememented yet");
     }
 
+    /**
+     * Get direct template downloader from direct download command and destination pool
+     */
+    private DirectTemplateDownloader getDirectTemplateDownloaderFromCommand(DirectDownloadCommand cmd, KVMStoragePool destPool) {
+        if (cmd instanceof HttpDirectDownloadCommand) {
+            return new HttpDirectTemplateDownloader(cmd.getUrl(), cmd.getTemplateId(), destPool.getLocalPath(), cmd.getChecksum(), cmd.getHeaders());
+        } else if (cmd instanceof HttpsDirectDownloadCommand) {
+            return new HttpsDirectTemplateDownloader(cmd.getUrl(), cmd.getTemplateId(), destPool.getLocalPath(), cmd.getChecksum(), cmd.getHeaders());
+        } else if (cmd instanceof NfsDirectDownloadCommand) {
+            return new NfsDirectTemplateDownloader(cmd.getUrl(), destPool.getLocalPath(), cmd.getTemplateId(), cmd.getChecksum());
+        } else if (cmd instanceof MetalinkDirectDownloadCommand) {
+            return new MetalinkDirectTemplateDownloader(cmd.getUrl(), destPool.getLocalPath(), cmd.getTemplateId(), cmd.getChecksum(), cmd.getHeaders());
+        } else {
+            throw new IllegalArgumentException("Unsupported protocol, please provide HTTP(S), NFS or a metalink");
+        }
+    }
+
     @Override
     public Answer handleDownloadTemplateToPrimaryStorage(DirectDownloadCommand cmd) {
         final PrimaryDataStoreTO pool = cmd.getDestPool();
         if (!pool.getPoolType().equals(StoragePoolType.NetworkFilesystem)) {
-            return new DirectDownloadAnswer(false, "Unsupported pool type " + pool.getPoolType().toString());
+            return new DirectDownloadAnswer(false, "Unsupported pool type " + pool.getPoolType().toString(), true);
         }
         KVMStoragePool destPool = storagePoolMgr.getStoragePool(pool.getPoolType(), pool.getUuid());
         DirectTemplateDownloader downloader;
 
-        if (cmd instanceof HttpDirectDownloadCommand) {
-            downloader = new HttpDirectTemplateDownloader(cmd.getUrl(), cmd.getTemplateId(), destPool.getLocalPath(), cmd.getChecksum(), ((HttpDirectDownloadCommand) cmd).getHeaders());
-        } else if (cmd instanceof HttpsDirectDownloadCommand) {
-            downloader = new HttpsDirectTemplateDownloader(cmd.getUrl(), cmd.getTemplateId(), destPool.getLocalPath(), cmd.getChecksum());
-        } else if (cmd instanceof NfsDirectDownloadCommand) {
-            downloader = new NfsDirectTemplateDownloader(cmd.getUrl(), destPool.getLocalPath(), cmd.getTemplateId(), cmd.getChecksum());
-        } else if (cmd instanceof MetalinkDirectDownloadCommand) {
-            downloader = new MetalinkDirectTemplateDownloader(cmd.getUrl(), destPool.getLocalPath(), cmd.getTemplateId(), cmd.getChecksum());
-        } else {
-            return new DirectDownloadAnswer(false, "Unsupported protocol, please provide HTTP(S), NFS or a metalink");
+        try {
+            downloader = getDirectTemplateDownloaderFromCommand(cmd, destPool);
+        } catch (IllegalArgumentException e) {
+            return new DirectDownloadAnswer(false, "Unable to create direct downloader: " + e.getMessage(), true);
         }
 
-        if (!downloader.downloadTemplate()) {
-            return new DirectDownloadAnswer(false, "Could not download template " + cmd.getTemplateId() + " on " + destPool.getLocalPath());
-        }
-        if (!downloader.validateChecksum()) {
-            return new DirectDownloadAnswer(false, "Checksum validation failed for template " + cmd.getTemplateId());
-        }
-        if (!downloader.extractAndInstallDownloadedTemplate()) {
-            return new DirectDownloadAnswer(false, "Template downloaded but there was an error on installation");
+        try {
+            s_logger.info("Trying to download template");
+            if (!downloader.downloadTemplate()) {
+                s_logger.warn("Couldn't download template");
+                return new DirectDownloadAnswer(false, "Unable to download template", true);
+            }
+            if (!downloader.validateChecksum()) {
+                s_logger.warn("Couldn't validate template checksum");
+                return new DirectDownloadAnswer(false, "Checksum validation failed", false);
+            }
+            if (!downloader.extractAndInstallDownloadedTemplate()) {
+                s_logger.warn("Couldn't extract and install template");
+                return new DirectDownloadAnswer(false, "Extraction and installation failed", false);
+            }
+        } catch (CloudRuntimeException e) {
+            s_logger.warn("Error downloading template " + cmd.getTemplateId() + " due to: " + e.getMessage());
+            return new DirectDownloadAnswer(false, "Unable to download template: " + e.getMessage(), true);
         }
+
         DirectTemplateInformation info = downloader.getTemplateInformation();
         return new DirectDownloadAnswer(true, info.getSize(), info.getInstallPath());
     }
diff --git a/server/src/com/cloud/vm/UserVmManagerImpl.java b/server/src/com/cloud/vm/UserVmManagerImpl.java
index dab741c3c27..a5bfc47584d 100644
--- a/server/src/com/cloud/vm/UserVmManagerImpl.java
+++ b/server/src/com/cloud/vm/UserVmManagerImpl.java
@@ -313,6 +313,7 @@ import com.cloud.vm.snapshot.VMSnapshotManager;
 import com.cloud.vm.snapshot.VMSnapshotVO;
 import com.cloud.vm.snapshot.dao.VMSnapshotDao;
 import com.cloud.storage.snapshot.SnapshotApiService;
+import com.cloud.storage.VMTemplateStorageResourceAssoc;
 
 public class UserVmManagerImpl extends ManagerBase implements UserVmManager, VirtualMachineGuru, UserVmService, Configurable {
     private static final Logger s_logger = Logger.getLogger(UserVmManagerImpl.class);
@@ -6100,10 +6101,8 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
                     throw ex;
                 }
             }
-            TemplateDataStoreVO tmplStore = _templateStoreDao.findByTemplateZoneReady(template.getId(), vm.getDataCenterId());
-            if (tmplStore == null) {
-                throw new InvalidParameterValueException("Cannot restore the vm as the template " + template.getUuid() + " isn't available in the zone");
-            }
+
+            checkRestoreVmFromTemplate(vm, template);
 
             if (needRestart) {
                 try {
@@ -6217,6 +6216,27 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
     }
 
+    /**
+     * Perform basic checkings to make sure restore is possible. If not, #InvalidParameterValueException is thrown
+     * @param vm vm
+     * @param template template
+     * @throws InvalidParameterValueException if restore is not possible
+     */
+    private void checkRestoreVmFromTemplate(UserVmVO vm, VMTemplateVO template) {
+        TemplateDataStoreVO tmplStore;
+        if (!template.isDirectDownload()) {
+            tmplStore = _templateStoreDao.findByTemplateZoneReady(template.getId(), vm.getDataCenterId());
+            if (tmplStore == null) {
+                throw new InvalidParameterValueException("Cannot restore the vm as the template " + template.getUuid() + " isn't available in the zone");
+            }
+        } else {
+            tmplStore = _templateStoreDao.findByTemplate(template.getId(), DataStoreRole.Image);
+            if (tmplStore == null || (tmplStore != null && !tmplStore.getDownloadState().equals(VMTemplateStorageResourceAssoc.Status.BYPASSED))) {
+                throw new InvalidParameterValueException("Cannot restore the vm as the bypassed template " + template.getUuid() + " isn't available in the zone");
+            }
+        }
+    }
+
     private void handleManagedStorage(UserVmVO vm, VolumeVO root) {
         if (Volume.State.Allocated.equals(root.getState())) {
             return;
diff --git a/server/src/org/apache/cloudstack/direct/download/DirectDownloadManagerImpl.java b/server/src/org/apache/cloudstack/direct/download/DirectDownloadManagerImpl.java
index 6aa7ad1b914..c1ffc5ef473 100755
--- a/server/src/org/apache/cloudstack/direct/download/DirectDownloadManagerImpl.java
+++ b/server/src/org/apache/cloudstack/direct/download/DirectDownloadManagerImpl.java
@@ -18,13 +18,18 @@
 //
 package org.apache.cloudstack.direct.download;
 
+import static com.cloud.storage.Storage.ImageFormat;
+
 import com.cloud.agent.AgentManager;
 import com.cloud.agent.api.Answer;
+import com.cloud.event.ActionEventUtils;
+import com.cloud.event.EventTypes;
+import com.cloud.event.EventVO;
 import com.cloud.host.Host;
 import com.cloud.host.HostVO;
 import com.cloud.host.Status;
 import com.cloud.host.dao.HostDao;
-import com.cloud.hypervisor.Hypervisor;
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.VMTemplateStoragePoolVO;
 import com.cloud.storage.VMTemplateStorageResourceAssoc;
@@ -40,6 +45,8 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.stream.Collectors;
 import javax.inject.Inject;
 
@@ -51,6 +58,7 @@ import org.apache.cloudstack.agent.directdownload.MetalinkDirectDownloadCommand;
 import org.apache.cloudstack.agent.directdownload.NfsDirectDownloadCommand;
 import org.apache.cloudstack.agent.directdownload.HttpsDirectDownloadCommand;
 import org.apache.cloudstack.agent.directdownload.SetupDirectDownloadCertificate;
+import org.apache.cloudstack.context.CallContext;
 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.ObjectInDataStoreStateMachine;
@@ -58,6 +66,7 @@ import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.log4j.Logger;
 
@@ -137,6 +146,46 @@ public class DirectDownloadManagerImpl extends ManagerBase implements DirectDown
         return headers;
     }
 
+    /**
+     * Get running host IDs within the same hypervisor, cluster and datacenter than hostId. ID hostId is not included on the returned list
+     */
+    protected List<Long> getRunningHostIdsInTheSameCluster(Long clusterId, long dataCenterId, HypervisorType hypervisorType, long hostId) {
+        List<Long> list = hostDao.listByDataCenterIdAndHypervisorType(dataCenterId, hypervisorType)
+                .stream()
+                .filter(x -> x.getHypervisorType().equals(hypervisorType) && x.getStatus().equals(Status.Up) &&
+                        x.getType().equals(Host.Type.Routing) && x.getClusterId().equals(clusterId) &&
+                        x.getId() != hostId)
+                .map(x -> x.getId())
+                .collect(Collectors.toList());
+        Collections.shuffle(list);
+        return list;
+    }
+
+    /**
+     * Create host IDs array having hostId as the first element
+     */
+    protected Long[] createHostIdsList(List<Long> hostIds, long hostId) {
+        if (CollectionUtils.isEmpty(hostIds)) {
+            return Arrays.asList(hostId).toArray(new Long[1]);
+        }
+        Long[] ids = new Long[hostIds.size() + 1];
+        ids[0] = hostId;
+        int i = 1;
+        for (Long id : hostIds) {
+            ids[i] = id;
+            i++;
+        }
+        return ids;
+    }
+
+    /**
+     * Get hosts to retry download having hostId as the first element
+     */
+    protected Long[] getHostsToRetryOn(Long clusterId, long dataCenterId, HypervisorType hypervisorType, long hostId) {
+        List<Long> hostIds = getRunningHostIdsInTheSameCluster(clusterId, dataCenterId, hypervisorType, hostId);
+        return createHostIdsList(hostIds, hostId);
+    }
+
     @Override
     public void downloadTemplate(long templateId, long poolId, long hostId) {
         VMTemplateVO template = vmTemplateDao.findById(templateId);
@@ -167,11 +216,8 @@ public class DirectDownloadManagerImpl extends ManagerBase implements DirectDown
 
         DownloadProtocol protocol = getProtocolFromUrl(url);
         DirectDownloadCommand cmd = getDirectDownloadCommandFromProtocol(protocol, url, templateId, to, checksum, headers);
-        Answer answer = agentManager.easySend(hostId, cmd);
-        if (answer == null || !answer.getResult()) {
-            throw new CloudRuntimeException("Host " + hostId + " could not download template " +
-                    templateId + " on pool " + poolId);
-        }
+
+        Answer answer = sendDirectDownloadCommand(cmd, template, poolId, host);
 
         VMTemplateStoragePoolVO sPoolRef = vmTemplatePoolDao.findByPoolTemplate(poolId, templateId);
         if (sPoolRef == null) {
@@ -190,13 +236,56 @@ public class DirectDownloadManagerImpl extends ManagerBase implements DirectDown
         }
     }
 
+    /**
+     * Send direct download command for downloading template with ID templateId on storage pool with ID poolId.<br/>
+     * At first, cmd is sent to host, in case of failure it will retry on other hosts before failing
+     * @param cmd direct download command
+     * @param template template
+     * @param poolId pool id
+     * @param host first host to which send the command
+     * @return download answer from any host which could handle cmd
+     */
+    private Answer sendDirectDownloadCommand(DirectDownloadCommand cmd, VMTemplateVO template, long poolId, HostVO host) {
+        boolean downloaded = false;
+        int retry = 3;
+        Long[] hostsToRetry = getHostsToRetryOn(host.getClusterId(), host.getDataCenterId(), host.getHypervisorType(), host.getId());
+        int hostIndex = 0;
+        Answer answer = null;
+        Long hostToSendDownloadCmd = hostsToRetry[hostIndex];
+        boolean continueRetrying = true;
+        while (!downloaded && retry > 0 && continueRetrying) {
+            s_logger.debug("Sending Direct download command to host " + hostToSendDownloadCmd);
+            answer = agentManager.easySend(hostToSendDownloadCmd, cmd);
+            if (answer != null) {
+                DirectDownloadAnswer ans = (DirectDownloadAnswer)answer;
+                downloaded = answer.getResult();
+                continueRetrying = ans.isRetryOnOtherHosts();
+            }
+            hostToSendDownloadCmd = hostsToRetry[(hostIndex + 1) % hostsToRetry.length];
+            retry --;
+        }
+        if (!downloaded) {
+            logUsageEvent(template, poolId);
+            throw new CloudRuntimeException("Template " + template.getId() + " could not be downloaded on pool " + poolId + ", failing after trying on several hosts");
+        }
+        return answer;
+    }
+
+    /**
+     * Log and persist event for direct download failure
+     */
+    private void logUsageEvent(VMTemplateVO template, long poolId) {
+        String event = EventTypes.EVENT_TEMPLATE_DIRECT_DOWNLOAD_FAILURE;
+        if (template.getFormat() == ImageFormat.ISO) {
+            event = EventTypes.EVENT_ISO_DIRECT_DOWNLOAD_FAILURE;
+        }
+        String description = "Direct Download for template Id: " + template.getId() + " on pool Id: " + poolId + " failed";
+        s_logger.error(description);
+        ActionEventUtils.onCompletedActionEvent(CallContext.current().getCallingUserId(), template.getAccountId(), EventVO.LEVEL_INFO, event, description, 0);
+    }
+
     /**
      * Return DirectDownloadCommand according to the protocol
-     * @param protocol
-     * @param url
-     * @param templateId
-     * @param destPool
-     * @return
      */
     private DirectDownloadCommand getDirectDownloadCommandFromProtocol(DownloadProtocol protocol, String url, Long templateId, PrimaryDataStoreTO destPool,
                                                                        String checksum, Map<String, String> httpHeaders) {
@@ -205,24 +294,34 @@ public class DirectDownloadManagerImpl extends ManagerBase implements DirectDown
         } else if (protocol.equals(DownloadProtocol.HTTPS)) {
             return new HttpsDirectDownloadCommand(url, templateId, destPool, checksum, httpHeaders);
         } else if (protocol.equals(DownloadProtocol.NFS)) {
-            return new NfsDirectDownloadCommand(url, templateId, destPool, checksum);
+            return new NfsDirectDownloadCommand(url, templateId, destPool, checksum, httpHeaders);
         } else if (protocol.equals(DownloadProtocol.METALINK)) {
-            return new MetalinkDirectDownloadCommand(url, templateId, destPool, checksum);
+            return new MetalinkDirectDownloadCommand(url, templateId, destPool, checksum, httpHeaders);
         } else {
             return null;
         }
     }
 
-    @Override
-    public boolean uploadCertificateToHosts(String certificateCer, String certificateName) {
-        List<HostVO> hosts = hostDao.listAllHostsByType(Host.Type.Routing)
+    /**
+     * Return the list of running hosts to which upload certificates for Direct Download
+     */
+    private List<HostVO> getRunningHostsToUploadCertificate(HypervisorType hypervisorType) {
+        return hostDao.listAllHostsByType(Host.Type.Routing)
                 .stream()
                 .filter(x -> x.getStatus().equals(Status.Up) &&
-                            x.getHypervisorType().equals(Hypervisor.HypervisorType.KVM))
+                        x.getHypervisorType().equals(hypervisorType))
                 .collect(Collectors.toList());
-        for (HostVO host : hosts) {
-            if (!uploadCertificate(certificateCer, certificateName, host.getId())) {
-                throw new CloudRuntimeException("Uploading certificate " + certificateName + " failed on host: " + host.getId());
+    }
+
+    @Override
+    public boolean uploadCertificateToHosts(String certificateCer, String certificateName, String hypervisor) {
+        HypervisorType hypervisorType = HypervisorType.getType(hypervisor);
+        List<HostVO> hosts = getRunningHostsToUploadCertificate(hypervisorType);
+        if (CollectionUtils.isNotEmpty(hosts)) {
+            for (HostVO host : hosts) {
+                if (!uploadCertificate(certificateCer, certificateName, host.getId())) {
+                    throw new CloudRuntimeException("Uploading certificate " + certificateName + " failed on host: " + host.getId());
+                }
             }
         }
         return true;
diff --git a/utils/src/main/java/com/cloud/utils/UriUtils.java b/utils/src/main/java/com/cloud/utils/UriUtils.java
index d7fbb89e47e..b3ec4643fb1 100644
--- a/utils/src/main/java/com/cloud/utils/UriUtils.java
+++ b/utils/src/main/java/com/cloud/utils/UriUtils.java
@@ -36,6 +36,7 @@ import java.util.ListIterator;
 import java.util.StringTokenizer;
 import java.util.Map;
 import java.util.HashMap;
+import java.util.Comparator;
 
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -63,6 +64,7 @@ import org.w3c.dom.Document;
 import org.w3c.dom.Element;
 import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
+import org.w3c.dom.NamedNodeMap;
 
 public class UriUtils {
 
@@ -297,6 +299,74 @@ public class UriUtils {
         }
     }
 
+    /**
+     * Add element to priority list examining node attributes: priority (for urls) and type (for checksums)
+     */
+    protected static void addPriorityListElementExaminingNode(String tagName, Node node, List<Pair<String, Integer>> priorityList) {
+        Integer priority = Integer.MAX_VALUE;
+        String first = node.getTextContent();
+        if (node.hasAttributes()) {
+            NamedNodeMap attributes = node.getAttributes();
+            for (int k=0; k<attributes.getLength(); k++) {
+                Node attr = attributes.item(k);
+                if (tagName.equals("url") && attr.getNodeName().equals("priority")) {
+                    String prio = attr.getNodeValue().replace("\"", "");
+                    priority = Integer.parseInt(prio);
+                    break;
+                } else if (tagName.equals("hash") && attr.getNodeName().equals("type")) {
+                    first = "{" + attr.getNodeValue() + "}" + first;
+                    break;
+                }
+            }
+        }
+        priorityList.add(new Pair<>(first, priority));
+    }
+
+    /**
+     * Return the list of first elements on the list of pairs
+     */
+    protected static List<String> getListOfFirstElements(List<Pair<String, Integer>> priorityList) {
+        List<String> values = new ArrayList<>();
+        for (Pair<String, Integer> pair : priorityList) {
+            values.add(pair.first());
+        }
+        return values;
+    }
+
+    /**
+     * Return HttpClient with connection timeout
+     */
+    private static HttpClient getHttpClient() {
+        MultiThreadedHttpConnectionManager s_httpClientManager = new MultiThreadedHttpConnectionManager();
+        s_httpClientManager.getParams().setConnectionTimeout(5000);
+        return new HttpClient(s_httpClientManager);
+    }
+
+    public static List<String> getMetalinkChecksums(String url) {
+        HttpClient httpClient = getHttpClient();
+        GetMethod getMethod = new GetMethod(url);
+        try {
+            if (httpClient.executeMethod(getMethod) == HttpStatus.SC_OK) {
+                InputStream is = getMethod.getResponseBodyAsStream();
+                Map<String, List<String>> checksums = getMultipleValuesFromXML(is, new String[] {"hash"});
+                if (checksums.containsKey("hash")) {
+                    List<String> listChksum = new ArrayList<>();
+                    for (String chk : checksums.get("hash")) {
+                        listChksum.add(chk.replaceAll("\n", "").replaceAll(" ", "").trim());
+                    }
+                    return listChksum;
+                }
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        } finally {
+            getMethod.releaseConnection();
+        }
+        return null;
+    }
+    /**
+     * Retrieve values from XML documents ordered by ascending priority for each tag name
+     */
     protected static Map<String, List<String>> getMultipleValuesFromXML(InputStream is, String[] tagNames) {
         Map<String, List<String>> returnValues = new HashMap<String, List<String>>();
         try {
@@ -307,14 +377,15 @@ public class UriUtils {
             for (int i = 0; i < tagNames.length; i++) {
                 NodeList targetNodes = rootElement.getElementsByTagName(tagNames[i]);
                 if (targetNodes.getLength() <= 0) {
-                    s_logger.error("no " + tagNames[i] + " tag in XML response...returning null");
+                    s_logger.error("no " + tagNames[i] + " tag in XML response...");
                 } else {
-                    List<String> valueList = new ArrayList<String>();
+                    List<Pair<String, Integer>> priorityList = new ArrayList<>();
                     for (int j = 0; j < targetNodes.getLength(); j++) {
                         Node node = targetNodes.item(j);
-                        valueList.add(node.getTextContent());
+                        addPriorityListElementExaminingNode(tagNames[i], node, priorityList);
                     }
-                    returnValues.put(tagNames[i], valueList);
+                    priorityList.sort(Comparator.comparing(x -> x.second()));
+                    returnValues.put(tagNames[i], getListOfFirstElements(priorityList));
                 }
             }
         } catch (Exception ex) {
@@ -329,7 +400,7 @@ public class UriUtils {
      * @return true if at least one existent URL defined on metalink, false if not
      */
     protected static boolean checkUrlExistenceMetalink(String url) {
-        HttpClient httpClient = new HttpClient(new MultiThreadedHttpConnectionManager());
+        HttpClient httpClient = getHttpClient();
         GetMethod getMethod = new GetMethod(url);
         try {
             if (httpClient.executeMethod(getMethod) == HttpStatus.SC_OK) {
@@ -356,23 +427,30 @@ public class UriUtils {
             }
         } catch (IOException e) {
             s_logger.warn(e.getMessage());
+        } finally {
+            getMethod.releaseConnection();
         }
         return false;
     }
 
     /**
-     * Get list of urls on metalink
-     * @param metalinkUrl
-     * @return
+     * Get list of urls on metalink ordered by ascending priority (for those which priority tag is not defined, highest priority value is assumed)
      */
     public static List<String> getMetalinkUrls(String metalinkUrl) {
-        HttpClient httpClient = new HttpClient(new MultiThreadedHttpConnectionManager());
+        HttpClient httpClient = getHttpClient();
         GetMethod getMethod = new GetMethod(metalinkUrl);
         List<String> urls = new ArrayList<>();
-        try (
-                InputStream is = getMethod.getResponseBodyAsStream()
-        ) {
-            if (httpClient.executeMethod(getMethod) == HttpStatus.SC_OK) {
+        int status;
+        try {
+            status = httpClient.executeMethod(getMethod);
+        } catch (IOException e) {
+            s_logger.error("Error retrieving urls form metalink: " + metalinkUrl);
+            getMethod.releaseConnection();
+            return null;
+        }
+        try {
+            InputStream is = getMethod.getResponseBodyAsStream();
+            if (status == HttpStatus.SC_OK) {
                 Map<String, List<String>> metalinkUrlsMap = getMultipleValuesFromXML(is, new String[] {"url"});
                 if (metalinkUrlsMap.containsKey("url")) {
                     List<String> metalinkUrls = metalinkUrlsMap.get("url");
@@ -381,6 +459,8 @@ public class UriUtils {
             }
         } catch (IOException e) {
             s_logger.warn(e.getMessage());
+        } finally {
+            getMethod.releaseConnection();
         }
         return urls;
     }
@@ -388,7 +468,7 @@ public class UriUtils {
     // use http HEAD method to validate url
     public static void checkUrlExistence(String url) {
         if (url.toLowerCase().startsWith("http") || url.toLowerCase().startsWith("https")) {
-            HttpClient httpClient = new HttpClient(new MultiThreadedHttpConnectionManager());
+            HttpClient httpClient = getHttpClient();
             HeadMethod httphead = new HeadMethod(url);
             try {
                 if (httpClient.executeMethod(httphead) != HttpStatus.SC_OK) {
@@ -398,9 +478,11 @@ public class UriUtils {
                     throw new IllegalArgumentException("Invalid URLs defined on metalink: " + url);
                 }
             } catch (HttpException hte) {
-                throw new IllegalArgumentException("Cannot reach URL: " + url);
+                throw new IllegalArgumentException("Cannot reach URL: " + url + " due to: " + hte.getMessage());
             } catch (IOException ioe) {
-                throw new IllegalArgumentException("Cannot reach URL: " + url);
+                throw new IllegalArgumentException("Cannot reach URL: " + url + " due to: " + ioe.getMessage());
+            } finally {
+                httphead.releaseConnection();
             }
         }
     }
@@ -444,7 +526,8 @@ public class UriUtils {
                 && (!uripath.toLowerCase().endsWith("ova")
                         && !uripath.toLowerCase().endsWith("ova.zip")
                         && !uripath.toLowerCase().endsWith("ova.bz2")
-                        && !uripath.toLowerCase().endsWith("ova.gz")))
+                        && !uripath.toLowerCase().endsWith("ova.gz")
+                        && !uripath.toLowerCase().endsWith("metalink")))
                 || (format.equalsIgnoreCase("tar")
                 && (!uripath.toLowerCase().endsWith("tar")
                         && !uripath.toLowerCase().endsWith("tar.zip")
@@ -468,7 +551,8 @@ public class UriUtils {
                 && (!uripath.toLowerCase().endsWith("iso")
                         && !uripath.toLowerCase().endsWith("iso.zip")
                         && !uripath.toLowerCase().endsWith("iso.bz2")
-                        && !uripath.toLowerCase().endsWith("iso.gz")))) {
+                        && !uripath.toLowerCase().endsWith("iso.gz"))
+                        && !uripath.toLowerCase().endsWith("metalink"))) {
             throw new IllegalArgumentException("Please specify a valid URL. URL:" + uripath + " is an invalid for the format " + format.toLowerCase());
         }