more filesHEADmaster

1 files changed, 622 insertions, 0 deletions

diff --git a/src/orchestration/imagesynchandler.py b/src/orchestration/imagesynchandler.py
new file mode 100644
index 0000000..cddf1ac
--- /dev/null
+++ b/src/orchestration/imagesynchandler.py
@@ -0,0 +1,622 @@
+import django
+import threading
+
+from django.conf import settings
+from django.core.exceptions import AppRegistryNotReady
+
+from multiprocessing import Process, Pool, Queue
+import pexpect
+import traceback
+import logging
+import multiprocessing
+from logging.handlers import QueueHandler
+import re
+import sys
+import time
+import datetime
+from .utils import *
+
+import os
+import signal, psutil
+
+from .db_updater import DBUpdater
+from .vmb_messages import ImageStatus, ImagesStatusMessage
+from .remoteregionhandler import RemoteRegionWorker
+
+try:
+    django.setup()
+    from .models import ImageSync, CentralToRemoteMap, RemoteRegionSetup
+    from caas.models import Cluster
+except django.core.exceptions.AppRegistryNotReady as exp:
+    pass
+
+class ImageSyncWorker(RemoteRegionWorker):
+
+    def __init__(self, loggerQueue, requestQueue, dbQueue, vmbQueue, dbCoordinationQueue, vmbCoordinationQueue, doneQueue):
+        self.loggerQueue = loggerQueue
+        self.requestQueue = requestQueue
+        self.dbQueue = dbQueue
+        self.vmbQueue = vmbQueue
+        self.doneQueue = doneQueue
+        self.dbCoordinationQueue = dbCoordinationQueue
+        self.vmbCoordinationQueue = vmbCoordinationQueue
+        #self.dbQueue = Queue()
+        #self.db_updater = DBUpdater(loggerQueue, self.dbQueue)
+        #self.db_updater_p = Process(target=self.db_updater.run)
+        #self.db_updater_p.start()
+
+    # https://stackoverflow.com/questions/3332043/obtaining-pid-of-child-process
+    # Current not used; Leaving it here if needed in the future
+    def _kill_child_processes(self, parent_pid, sig=signal.SIGTERM):
+        try:
+            parent = psutil.Process(parent_pid)
+        except psutil.NoSuchProcess:
+           return
+        children = parent.children(recursive=True)
+        for process in children:
+            #if not process.is_alive():
+            process.send_signal(sig)
+
+    def _kill_child_process(self, pid):
+        os.kill(pid, signal.SIGKILL)
+
+    def run(self):
+        qh = QueueHandler(self.loggerQueue)
+        self.logger = logging.getLogger()
+        self.logger.addHandler(qh)
+        self.logger.setLevel(logging.DEBUG)
+
+        self.logger.info("ImageSyncWorker started...")
+
+        while True:
+            try:
+                item = self.requestQueue.get(block=True)
+                if item:
+                    self.logger.info(item)
+                    self._handle_request(item)
+            except:
+                pass
+            time.sleep(1)
+
+    def _handle_request(self, request):
+        command = request['command']
+        if command == 'upload':
+            self._handle_request_image_upload(request)
+        if command == 'delete':
+            self._handle_request_image_delete(request)
+        if command == 'cancel':
+            self._handle_request_image_upload_cancel(request)
+
+    def _handle_request_image_delete(self, request):
+        pass
+   
+    def _handle_request_image_upload_cancel(self, request):
+        pass
+
+    def _setup_logger(self):
+        qh = QueueHandler(self.loggerQueue)
+        self.logger = logging.getLogger()
+        self.logger.addHandler(qh)
+        self.logger.setLevel(logging.INFO)
+
+    def get_image_tags(self, remote_region, image_name, logger):
+        remote_region_oam_ip = self._get_remote_region_oam_ip(remote_region, "-1", logger)
+        host_username, host_password, ssh_prefix = get_host_connection_details(remote_region_oam_ip)
+
+        env_string = self._read_remote_openrc(remote_region_oam_ip, logger)
+        basecmd = env_string
+
+        cmds = []
+        cmd = ssh_prefix + " " + basecmd + " system registry-image-tags " + image_name
+        logger.info("cmd:" + cmd)
+        cmds.append(cmd)
+        tags = []
+        output_lines = self._run_command_get_all_lines(cmds, host_password, logger)
+        #logger.info("Returned output lines:")
+        #logger.info(output_lines)
+        print(output_lines)
+        if len(output_lines) > 0:
+            for line in output_lines.split("\n"):
+                #logger.info("Line:" + line)
+                print("Line:" + line)
+                if 'Connection to' not in line and 'Authorization failed' not in line:
+                    parts = line.split(" ")
+                    if len(parts) > 1:
+                        if parts[1] != 'Image':
+                            print("Line:" + line)
+                            tags.append(parts[1])
+                if 'Authorization failed' in line:
+                    tags.append(line)
+        logger.info("Tags:" + str(tags))
+        print("Tags:" + str(tags))
+        return tags
+
+    def delete_image_tag(self, remote_region, image_name, image_tag, logger):
+        remote_region_oam_ip = self._get_remote_region_oam_ip(remote_region, "-1", logger)
+        host_username, host_password, ssh_prefix = get_host_connection_details(remote_region_oam_ip)
+
+        env_string = self._read_remote_openrc(remote_region_oam_ip, logger)
+        basecmd = env_string
+
+        cmds = []
+        cmd = ssh_prefix + " " + basecmd + " system registry-image-delete " + image_name + ":" + image_tag
+        logger.info("cmd:" + cmd)
+        cmds.append(cmd)
+        tags = []
+        output_lines = self._run_command_get_all_lines(cmds, host_password, logger)
+
+        cmds = []
+        cmd = ssh_prefix + " " + basecmd + " system registry-garbage-collect "
+        logger.info("cmd:" + cmd)
+        cmds.append(cmd)
+        tags = []
+        output_lines1 = self._run_command_get_all_lines(cmds, host_password, logger)
+        logger.info("Output lines1:")
+        logger.info(output_lines1)
+
+        lines_to_return = []
+        for line in output_lines.split('\n'):
+            if 'Connection to' not in line:
+                lines_to_return.append(line)
+
+        for line in output_lines1.split('\n'):
+            if 'Connection to' not in line:
+                lines_to_return.append(line)
+
+        logger.info("Output lines:")
+        logger.info(lines_to_return)
+        return lines_to_return
+
+    def _handle_request_image_upload(self, request):
+        transaction_id = request['transaction_id']
+        self.logger.info("Transction ID:" + str(transaction_id))
+        imageList = request['images']
+        remoteRegions = request['remoteRegions']
+        centralRegions = []
+        for region in remoteRegions:
+            centralRegionList = self._get_central_region_oam_ip(region, transaction_id)
+            for centralRegion in centralRegionList:
+	            if centralRegion not in centralRegions:
+		            centralRegions.append(centralRegion)
+            remoteregion_oam_ip = self._get_remote_region_oam_ip(region, transaction_id, self.logger)
+            self.logger.info("Remote region Name:" + region + " Remote region OAM IP:" + remoteregion_oam_ip)
+        self.logger.info(str(transaction_id) + " Downloading images to Central regions.." + ','.join(centralRegions))
+        successful_image_list, failed_image_list = self._handle_central_regions(centralRegions, remoteRegions, imageList, transaction_id, region_type='central')
+
+        if len(successful_image_list) > 0:
+            self.logger.info(str(transaction_id) + " Downloading images to Remote regions.." + ','.join(remoteRegions) + " " + ','.join(successful_image_list))
+            self._handle_regions(remoteRegions, successful_image_list, transaction_id, region_type='remote')
+        else:
+            self.logger.info(str(transaction_id) + " Could not download images to Central region..So not progressing to Remote region")
+        self._wait_and_done(imageList, remoteRegions, transaction_id)
+        return
+
+    def _handle_central_regions(self, regionList, remoteRegionList, imageList, transaction_id, region_type=''):
+        workers = []
+        imageRegionList = get_pairs(imageList, regionList)
+        self.logger.info(str(transaction_id) + " Handling central region")
+        self.logger.info(str(transaction_id) + " ImageList:" + ','.join(imageList))
+        self.logger.info(str(transaction_id) + " RegionList:" + ','.join(regionList))
+        self.logger.info(str(transaction_id) +  " ImageRegionList:" + str(imageRegionList))
+        for region in regionList:
+            self.logger.info(str(transaction_id) + " Region:" + region)
+            successful_image_list, failed_image_list = self._handle_image_list(region, imageList, self.logger, region_type, transaction_id)
+            # We want to download from Artifactory in its own process so that the parent can terminate the process if needed; 
+            # Without doing this in its own process, the parent will block foreever and we won't be able to process subsequent calls.
+            #worker = multiprocessing.Process(target=self._handle_image_list,
+            #                     args=(region, imageList, self.logger, region_type, transaction_id)) 
+            #workers.append(worker)
+            #worker.start()
+            #self.logger.info("Central controller handling process pid:" + str(worker.pid))
+
+        #https://stackoverflow.com/questions/26063877/python-multiprocessing-module-join-processes-with-timeout
+        #TIMEOUT = 300 # 5 minutes 
+        #start = time.time()
+        #while time.time() - start <= TIMEOUT:
+        #    if not any(p.is_alive() for p in workers):
+        #        # All the processes are done, break now.
+        #        break
+        #    time.sleep(1)  # Just to avoid hogging the CPU
+        #else:
+        #    # We only enter this if we didn't 'break' above.
+        #    print("Central downloader timed out, terminating the process...")
+        #    for w in workers:
+        #        self.logger.info("Terminating process handling central region download - pid:" + str(w.pid))
+        #        self._kill_child_process(w.pid)
+        #    return download_to_central_done
+    
+        #for w in workers:
+        #    w.join()
+
+        return successful_image_list, failed_image_list
+
+    def _handle_regions(self, regionList, imageList, transaction_id, region_type=''):
+        self.logger.info(str(transaction_id) + " Handling remote regions")
+        workers = []
+        imageRegionList = get_pairs(imageList, regionList)
+        self.logger.info(str(transaction_id) + " ImageList:" + ','.join(imageList))
+        self.logger.info(str(transaction_id) + " RegionList:" + ','.join(regionList))
+        self.logger.info(str(transaction_id) + " ImageRegionList:" + str(imageRegionList))
+        for item in imageRegionList:
+            self.logger.info(str(transaction_id) + " Item:" + str(item))
+            worker = multiprocessing.Process(target=self._worker_process,
+                                 args=(self.logger, self.loggerQueue, self._worker_configurer, item, region_type, transaction_id))
+            workers.append(worker)
+            worker.start()
+            time.sleep(3) # Stagger the requests
+
+        #https://stackoverflow.com/questions/26063877/python-multiprocessing-module-join-processes-with-timeout
+        TIMEOUT = 3600 # 1 hour 
+        start = time.time()
+        while time.time() - start <= TIMEOUT:
+            if not any(p.is_alive() for p in workers):
+                # All the processes are done, break now.
+                break
+            time.sleep(1)  # Just to avoid hogging the CPU
+        else:
+            # We only enter this if we didn't 'break' above.
+            print("timed out, killing all processes")
+            for w in workers:
+                self.logger.info(str(transaction_id) + " Terminating process handling remote region download - pid:" + str(w.pid))
+                self._kill_child_process(w.pid)
+            return
+        
+        # No need to wait for processes to finish; otherwise we won't be able to pick-up any new incoming requests for an hour.
+        #for w in workers:
+        #    w.join()
+
+    def _wait_and_done(self, imageList, remoteRegions, transaction_id):
+        # Ensure that VMB messages have been sent
+        self.logger.info(str(transaction_id) + " About to be done..waiting for cleanup")
+        num_of_images = len(imageList)
+        num_of_regions = len(remoteRegions)
+        self.logger.info(str(transaction_id) + " Number of images.." + str(num_of_images))
+        self.logger.info(str(transaction_id) + " Number of regions.." + str(num_of_regions))
+        count = 0
+        while count < num_of_images * num_of_regions:
+            vmb_coordination = self.vmbCoordinationQueue.get()
+            self.logger.info(str(transaction_id) + " VMB coordination message:" + vmb_coordination)
+            count = count + 1
+
+            # Ensure that all DB messages have been processed
+            db_coordination = self.dbCoordinationQueue.get()
+            self.logger.info(str(transaction_id) + " DB coordination message:" + db_coordination)
+
+        self.logger.info(str(transaction_id) + " Done")
+        time.sleep(10)
+        self.doneQueue.put("Done")
+
+    def _worker_process(self, lg, queue, configurer, item, region_type, transaction_id):
+        #configurer(queue)
+        name = multiprocessing.current_process().name
+        self._handle_image(item, lg, region_type, transaction_id)
+
+    # Currently not used; left here if we want to configure the loggers to add any extra logging (formats, etc.)
+    def _worker_configurer(self, queue):
+        h = logging.handlers.QueueHandler(queue)
+        root = logging.getLogger()
+        root.addHandler(h)
+        root.setLevel(logging.INFO)
+
+    def _handle_image(self, imageRegionPair, logger, region_type, transaction_id):
+        image = imageRegionPair["image"]
+        regionname = imageRegionPair["region"]
+        region = self._get_remote_region_oam_ip(regionname, transaction_id, logger)
+        name = multiprocessing.current_process().name
+        logger.info(str(transaction_id) + " Process:" + name + " " + image + " " + region)
+
+        logger.info(str(transaction_id) + " Checking if image exist")
+        imageexists = self._check_image_exists(image, region)
+        logger.info(str(transaction_id) + " Image exists status:" + str(imageexists))        
+        if not imageexists:
+            message = 'Starting Image download from Central Controller'
+            status = 'STARTED'
+            self._update_status_async(image, region, message, status, transaction_id, logger)
+            success, err = self._download_image(image, region, logger, region_type, transaction_id)
+            if success:
+                really_there = False
+                dest_image_name_and_tag = self._get_dest_image_name(image)
+                parts = dest_image_name_and_tag.split(":")
+                dest_image_name = ''
+                if len(parts) > 0:
+                    dest_image_name = parts[0]
+                logger.info("Destination image name:" + dest_image_name)
+                dest_image_tags = self.get_image_tags(regionname, dest_image_name, logger)
+                really_there = self._is_it_really_there(image, dest_image_tags, transaction_id, logger)
+                if really_there:
+                    message = 'Image download to Remote Region complete.'
+                    status = 'COMPLETED'
+                else:
+                    message = 'Image download to Remote Region failed. Reason: Could not find image on remote region.'
+                    status = 'FAILED'
+            else:
+                message = 'Image download to Remote Region failed. Reason:' + err
+                status = 'FAILED'
+            image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+            self._update_status_async(image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time, done=True)
+
+            self._notify_vmb(region, regionname, image, status, message, image_upload_completion_time, transaction_id, logger)
+
+    def _is_it_really_there(self, image, dest_image_tags, transaction_id, logger):
+        logger.info(str(transaction_id) + " Comparing image tags..")
+        logger.info(str(transaction_id) + " Image:" + image)
+        logger.info(str(transaction_id) + " Destination Image tags:" + str(dest_image_tags))
+        present = False
+        for tag in dest_image_tags:
+            logger.info(str(transaction_id) + " Tag:" + tag)
+            logger.info(str(transaction_id) + " Image:" + image)
+            if tag in image:
+                present = True
+                break
+        return present
+
+    def _notify_vmb(self, region, regionname, image, status, message, image_upload_completion_time, transaction_id, logger):
+
+        # Notify VMB
+        image_status1 = ImageStatus(
+              cluster=regionname,
+              image=image,
+              action='UPLOAD',
+              status=status,
+              message=message,
+              created_at=str(image_upload_completion_time),
+        )
+
+        logger.info(str(transaction_id) + " Image status::" + str(image_status1))
+
+        site_name, site_location = self.get_fuze_spm_site_details(regionname, logger)
+        logger.info("Site name:" + site_name)
+        logger.info("Site location:" + site_location)
+        reportDescription = 'Image upload status : ' + site_name 
+
+        images_status_list = []
+        images_status_message = ImagesStatusMessage(reportName='vcp_fe_imagestatus',
+                                                    transactionId=transaction_id,
+                                                    reportDescription=reportDescription,
+                                                    reportGeneratedOn=str(datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")))
+        images_status_list.append(image_status1)
+        images_status_message.rowCount = 1
+        images_status_message.reportDataRows = images_status_list
+
+        logger.info(str(transaction_id) + " Image status message::" + str(images_status_message))
+
+        item = {}
+        item['message'] = 'ImageStatus'
+        item['payload'] = images_status_message
+        self.vmbQueue.put(item)
+
+    def _get_central_region_oam_ip(self, remoteregion, transaction_id):
+        remoteclusterObj = Cluster.objects.filter(cluster_name=remoteregion)
+        parent_cluster_id = remoteclusterObj[0].parent_cluster_id
+        self.logger.info(str(transaction_id) + " Parent cluster id.." + str(parent_cluster_id))
+        centralclusterObj = Cluster.objects.filter(id=parent_cluster_id)
+        self.logger.info(str(transaction_id) + " " + str(centralclusterObj))
+        central_region_list = []
+        for central_region in centralclusterObj:
+            central_region_list.append(central_region.oam_vip_address)
+        self.logger.info(str(transaction_id) + " Central Region List:" + ','.join(central_region_list))
+        return central_region_list
+
+    def _get_remote_region_oam_ip(self, regionname, transaction_id, logger):
+        remoteclusterObj = Cluster.objects.filter(cluster_name=regionname)
+        oam_ip = remoteclusterObj[0].oam_vip_address
+        logger.info(str(transaction_id) + " Remote region:" + regionname + " OAM IP:" + str(oam_ip))
+        return oam_ip
+        
+    def _get_central_region_prev(self, remoteregion, transaction_id):
+        centralToRemoteMapObj = CentralToRemoteMap.objects.filter(remote_region_name=remoteregion)
+        self.logger.info(str(transaction_id) + " Inside _get_central_region...")
+        self.logger.info(str(transaction_id) + " " + str(centralToRemoteMapObj))
+        central_region_list = []
+        for central_region in centralToRemoteMapObj:
+        	central_region_list.append(central_region.central_region_name)
+        #central_region_name = centralToRemoteMapObj[0].central_region_name
+        self.logger.info(str(transaction_id) + " Central Region List:" + ','.join(central_region_list))
+        return central_region_list
+
+    def _check_image_exists(self, image, region):
+        # TODO: Query database for a quick check; Query region for accurate check
+        return False
+
+    def _handle_image_list(self, region, imageList, logger, region_type, transaction_id):
+        logger.info(str(transaction_id) + " Handling image list")
+        name = multiprocessing.current_process().name
+        logger.info(str(transaction_id) + " Process:" + name + " " + region)
+
+        central_region_oam_ip = region
+        logger.info("1a")
+        central_region_name = self._get_central_region_name(central_region_oam_ip, logger)
+        logger.info("Central region name:" + central_region_name)
+
+        host_username = settings.HOST_CREDS['username']
+        host_password = settings.HOST_CREDS['password']
+        art_username = settings.ARTIFACTORY_CREDS['username']
+        art_password = settings.ARTIFACTORY_CREDS['password']
+        dr_username = settings.CENTRAL_DR_CREDS['username']
+        dr_password = settings.CENTRAL_DR_CREDS['password']
+       
+        failed_image_list = []
+        successful_image_list = []
+        for image in imageList:
+            image_parts = image.split("/")
+            # Image: http://vnf-twb.vzwnet.com/docker-images/samsung_vdu_vdu_svr20aa5vvzwg06a3_r05_20.a.0-0101_6.0.0/vzw-adpf-rmp:svr20aa5vvzwg06a3_r05
+            artifactory_host = image_parts[0] # ' vsp.vici.verizon.com:7443'
+            logger.info("Artifactory host:" + artifactory_host)
+            ssh_prefix = 'ssh -o StrictHostKeyChecking=no -t ' + host_username + '@' + region
+            login_to_artifactory = []
+            login_to_artifactory.append(ssh_prefix + ' sudo docker login  --username ' + art_username + ' --password ' + art_password + ' ' + artifactory_host)
+            artifactory_login_status, cmd_err = self._run_commands(login_to_artifactory, host_password, logger, transaction_id)
+            if not artifactory_login_status:
+                message = 'Could not download Image from Artifactory. Reason:' + cmd_err
+                status = 'FAILED'
+                failed_image_list.append(image)
+                image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                self._update_status_async(image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time)
+                logger.info(str(transaction_id) + " Notifying VMB")
+                self._notify_vmb(central_region_name, central_region_name, image, status, message, image_upload_completion_time, transaction_id, logger)
+            else:
+                message = 'Starting Image download from Artifactory.'
+                status = 'STARTED'
+                self._update_status_async(image, region, message, status, transaction_id, logger)
+            
+                image_name = self._get_image_name(image)
+                download_from_artifactory = []
+                download_from_artifactory.append(ssh_prefix + ' sudo docker pull ' + image)
+                logger.info(str(transaction_id) + " Downloading " + image)
+                image_download_status, cmd_err1 = self._run_commands(download_from_artifactory, host_password, logger, transaction_id, cmd_timeout=300) # 5 minutes 
+            
+                if image_download_status: 
+                    message = 'Image download from Artifactory complete.'
+                    status = 'COMPLETED'
+                    successful_image_list.append(image)
+                else:
+                    message = 'Could not download Image from Artifactory. Reason:' + cmd_err1
+                    status = 'FAILED'
+                    failed_image_list.append(image)
+                    logger.info(str(transaction_id) + " Notifying VMB")
+                    self._notify_vmb(central_region_name, central_region_name, image, status, message, image_upload_completion_time, transaction_id, logger)
+                image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                self._update_status_async(image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time)
+
+        logger.info("**********")
+        logger.info("Successful Image list:" + str(len(successful_image_list)))
+        logger.info("----------")
+        logger.info("Failed Image list:" + str(len(failed_image_list)))
+        if len(successful_image_list) > 0:
+            login_to_dr_central = [] 
+            login_to_dr_central.append(ssh_prefix + ' sudo docker login --username ' +  dr_username + ' --password ' + dr_password + ' registry.local:9001')
+            self._run_commands(login_to_dr_central, host_password, logger, transaction_id)
+
+            for image in successful_image_list:
+                message = 'Pushing Image to Central Controller Docker Registry.'
+                status = 'STARTED'
+                self._update_status_async(image, region, message, status, transaction_id, logger)
+            
+                image_name = self._get_image_name(image)
+                dest_image = 'registry.local:9001/' + image_name 
+                push_to_local_dr_central = []
+                push_to_local_dr_central.append(ssh_prefix + ' sudo docker tag ' + image + ' ' + dest_image)
+                push_to_local_dr_central.append(ssh_prefix + ' sudo docker push ' + dest_image)
+                logger.info(str(transaction_id) + " Pushing image to Central Docker Registry " + dest_image)
+                self._run_commands(push_to_local_dr_central, host_password, logger, transaction_id)
+
+                message = 'Pushing Image to Central Controller Docker Registry complete.'
+                status = 'COMPLETED'
+                image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                self._update_status_async(image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time)
+
+        return successful_image_list, failed_image_list
+
+    # TODO: Try two times
+    # https://gitlab.verizon.com/vcp/webscale/far-edge/caas/orchestration/-/blob/master/docker_image_sync_steps.txt
+    def _download_image(self, source_image, region, logger, region_type, transaction_id):
+        image_name = self._get_image_name(source_image)
+        central_image = 'registry.central:9001/' + image_name
+        dest_image_name = self._get_dest_image_name(image_name)
+        dest_image = 'registry.local:9001/' + dest_image_name
+        logger.info("Destination Image name:" + dest_image)
+        host_username = settings.HOST_CREDS['username']
+        host_password = settings.HOST_CREDS['password']
+        dr_username = settings.CENTRAL_DR_CREDS['username']
+        dr_password = settings.CENTRAL_DR_CREDS['password']
+
+        ssh_prefix = 'ssh -o StrictHostKeyChecking=no -t ' + host_username + '@' + region
+        download_from_central_dr = []
+        download_from_central_dr.append(ssh_prefix + ' sudo docker login --username ' +  dr_username + ' --password ' + dr_password + ' registry.central:9001')
+        download_from_central_dr.append(ssh_prefix + ' sudo docker pull ' + central_image)
+        push_to_local_dr_remote = []
+        push_to_local_dr_remote.append(ssh_prefix + ' sudo docker tag ' + central_image + ' ' + dest_image)
+        push_to_local_dr_remote.append(ssh_prefix + ' sudo docker login --username ' +  dr_username + ' --password ' + dr_password + ' registry.local:9001')
+        push_to_local_dr_remote.append(ssh_prefix + ' sudo docker push ' + dest_image)
+       
+        image_download_complete = False 
+        if region_type == 'remote':
+            logger.info(str(transaction_id) + " Downloading image from central region")
+            message = 'Downloading Image from Central Controller Docker Registry.'
+            status = 'STARTED'
+            self._update_status_async(central_image, region, message, status, transaction_id, logger)
+            image_download_complete, cmd_err = self._run_commands(download_from_central_dr, host_password, logger, transaction_id, cmd_timeout=3600) # 1 hour 
+            if image_download_complete:
+                message = 'Downloading Image from Central Controller Docker Registry.'
+                status = 'COMPLETED'
+                image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                self._update_status_async(central_image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time)
+
+                message = 'Pushing Image to Remote region Docker Registry.'
+                status = 'STARTED'
+                self._update_status_async(central_image, region, message, status, transaction_id, logger)
+                logger.info(str(transaction_id) + " Push to local docker registry")
+                self._run_commands(push_to_local_dr_remote, host_password, logger, transaction_id)
+                message = 'Pushing Image to Remote region Docker Registry complete.' 
+                status = 'COMPLETED'
+                image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                self._update_status_async(central_image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time)
+            else:
+                message = 'Downloading Image from Central Controller Docker Registry Failed. Reason:' + cmd_err
+                status = 'FAILED'
+                image_upload_completion_time = datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                self._update_status_async(central_image, region, message, status, transaction_id, logger, upload_end_time=image_upload_completion_time)
+        return image_download_complete, cmd_err 
+
+    def _run_commands(self, commands, host_password, logger, transaction_id, cmd_timeout=None):
+        #logger.info(str(transaction_id) + " " + str(commands))
+        image_download_complete = False
+        image_download_error = ''
+        for command in commands:
+            logger.info(command)
+            child = pexpect.spawn(command)
+            #logger.info("-- Child PID:" + str(child.pid))
+            child.timeout=cmd_timeout
+            try:
+                #child.expect(['password: '], timeout=cmd_timeout)
+                #child.expect('\r\n\r\n\w+')#, timeout=cmd_timeout)
+                #child.expect('\w+\r\n')
+                child.expect_exact('password: ', timeout=cmd_timeout)
+                child.sendline(host_password)
+                #child.expect(['Password: '], timeout=cmd_timeout)
+                #child.expect('\w+\r\n')
+                child.expect_exact('Password: ', timeout=cmd_timeout)
+                child.sendline(host_password)
+                #child.sendline("\r\n")
+                all_lines = child.read()
+                all_lines = all_lines.rstrip().lstrip()
+                all_lines = all_lines.decode('utf-8').replace('\r\n', '\n')
+                #logger.info(all_lines)
+                image_download_complete = True # Tentative
+                for line in all_lines.split("\n"):
+                    logger.info(str(transaction_id) + " " + line)
+                    if re.search('error', line, re.IGNORECASE) or 'tag does not exist' in line: #or 'Error response from daemon' in line:
+                        image_download_complete = False
+                        image_download_error = line
+            except:
+                logger.info(str(transaction_id) + " " + str(child))
+        return image_download_complete, image_download_error
+
+    def _get_image_name(self, image):
+        i = image.find("/")
+        image_name = image[i+1:]
+        return image_name
+
+    def _get_dest_image_name(self, image):
+        i = image.rfind("/")
+        image_name = image[i+1:]
+        return image_name
+
+    def _update_status_async(self, image, region, message, status, transaction_id, logger, upload_end_time=None, done=False):
+        item = {}
+        item['type'] = 'image'
+        item['image'] = image
+        item['region'] = region
+        item['message'] = message
+        item['status'] = status
+        item['transaction_id'] = transaction_id
+        item['upload_end_time'] = upload_end_time
+        item['done'] = done # done flag indicates when we are done using db_handler
+        self.dbQueue.put(item)
+
+    def _update_status_sync(self, image, region, message, status, transaction_id, logger):
+        ImageSync.objects.filter(
+                 	         transaction_id=transaction_id,
+                                 docker_image=image).update(upload_message=message,
+                                                      upload_status=status)
+        return