Merge pull request #1049 from grycap/ssh_conn

Implements #1048

IM/ConfManager.py

@@ -355,6 +355,7 @@ def launch_ctxt_agent(self, vm, tasks):
         """
         Launch the ctxt agent to configure the specified tasks in the specified VM
         """
+        ssh = None
         pid = None
         tmp_dir = None
         try:
@@ -376,7 +377,7 @@ def launch_ctxt_agent(self, vm, tasks):
                 self.log_info("Copy the contextualization agent config file")
 
                 # Copy the contextualization agent config file
-                ssh = vm.get_ssh_ansible_master()
+                ssh = vm.get_ssh_ansible_master(auto_close=False)
                 ssh.sftp_mkdir(remote_dir)
                 ssh.sftp_put(conf_file, remote_dir + "/" + os.path.basename(conf_file))
 
@@ -408,6 +409,8 @@ def launch_ctxt_agent(self, vm, tasks):
             pid = None
             self.log_exception("Error launching the ansible process to configure VM with ID %s" % str(vm.im_id))
         finally:
+            if ssh:
+                ssh.close()
             if tmp_dir:
                 shutil.rmtree(tmp_dir, ignore_errors=True)
 
@@ -762,16 +765,16 @@ def generate_playbook(self, vm, ctxt_elem, tmp_dir):
             if vault_password:
                 vault_edit = self.get_vault_editor(vault_password)
                 if configure.recipes.strip().startswith("$ANSIBLE_VAULT"):
-                    recipes = vault_edit.vault.decrypt(configure.recipes.strip())
+                    recipes = vault_edit.vault.decrypt(configure.recipes.strip()).decode()
                 else:
                     recipes = configure.recipes
                 conf_content = merge_recipes(conf_content, recipes)
-                conf_content = vault_edit.vault.encrypt(conf_content)
+                conf_content = vault_edit.vault.encrypt(conf_content).decode()
             else:
                 conf_content = merge_recipes(conf_content, configure.recipes)
 
             conf_out = open(conf_filename, 'w')
-            conf_out.write(conf_content)
+            conf_out.write(str(conf_content))
             conf_out.close()
             recipe_files.append(ctxt_elem.configure + "_" + ctxt_elem.system + "_task.yml")
 
@@ -802,6 +805,7 @@ def configure_master(self):
                 self.log_info("Sleeping %s secs." % (cont ** 2 * 5))
                 time.sleep(cont ** 2 * 5)
                 cont += 1
+                ssh = None
                 try:
                     self.log_info("Start the contextualization process.")
 
@@ -810,7 +814,7 @@ def configure_master(self):
                     else:
                         if not self.inf.vm_master:
                             raise Exception("No master VM found.")
-                        ssh = self.inf.vm_master.get_ssh(retry=True)
+                        ssh = self.inf.vm_master.get_ssh(retry=True, auto_close=False)
                         if not ssh:
                             raise Exception("Master VM does not have IP.")
                         # Activate tty mode to avoid some problems with sudo in
@@ -882,6 +886,8 @@ def configure_master(self):
                         self.inf.ansible_configured = False
                     success = False
                 finally:
+                    if ssh:
+                        ssh.close()
                     if tmp_dir:
                         shutil.rmtree(tmp_dir, ignore_errors=True)
 
@@ -915,6 +921,7 @@ def wait_master(self):
         success = True
         if not self.inf.ansible_configured:
             # Select the master VM
+            ssh = None
             try:
                 self.inf.add_cont_msg("Select master VM")
                 self.inf.select_vm_master()
@@ -951,7 +958,7 @@ def wait_master(self):
 
                 # Check and change if necessary the credentials of the master
                 # vm
-                ssh = self.inf.vm_master.get_ssh(retry=True)
+                ssh = self.inf.vm_master.get_ssh(retry=True, auto_close=False)
                 # Activate tty mode to avoid some problems with sudo in REL
                 ssh.tty = True
                 self.change_master_credentials(ssh)
@@ -963,6 +970,9 @@ def wait_master(self):
             except Exception:
                 self.log_exception("Error waiting the master VM to be running")
                 self.inf.set_configured(False)
+            finally:
+                if ssh:
+                    ssh.close()
         else:
             self.inf.set_configured(True)
 
@@ -1016,8 +1026,7 @@ def generate_playbooks_and_hosts(self):
                 for ctxt_elem in contextualizes[ctxt_num]:
                     if ctxt_elem.system in vm_group and ctxt_elem.get_ctxt_tool() == "Ansible":
                         vm = vm_group[ctxt_elem.system][0]
-                        filenames.extend(self.generate_playbook(
-                            vm, ctxt_elem, tmp_dir))
+                        filenames.extend(self.generate_playbook(vm, ctxt_elem, tmp_dir))
 
             filenames.append(self.generate_etc_hosts(tmp_dir))
             filenames.append(self.generate_inventory(tmp_dir))
@@ -1035,8 +1044,7 @@ def generate_playbooks_and_hosts(self):
             if self.inf.radl.ansible_hosts:
                 for ansible_host in self.inf.radl.ansible_hosts:
                     (user, passwd, private_key) = ansible_host.getCredentialValues()
-                    ssh = SSHRetry(ansible_host.getHost(),
-                                   user, passwd, private_key)
+                    ssh = SSHRetry(ansible_host.getHost(), user, passwd, private_key)
                     ssh.sftp_mkdir(remote_dir)
                     ssh.sftp_put_files(recipe_files)
             else:
@@ -1062,7 +1070,7 @@ def wait_vm_running(self, vm, timeout):
            - timeout(int): Max time to wait the VM to be running.
         Returns: True if all the VMs are running or false otherwise
         """
-        delay = 10
+        delay = Config.CHECK_CTXT_PROCESS_INTERVAL
         wait = 0
         while not self._stop_thread and wait < timeout:
             if not vm.destroy:

IM/SSH.py

@@ -91,10 +91,14 @@ def run(self):
 class SSH:
     """ Class to encapsulate SSH operations using paramiko """
 
-    def __init__(self, host, user, passwd=None, private_key=None, port=22, proxy_host=None):
+    def __init__(self, host, user, passwd=None, private_key=None, port=22, proxy_host=None, auto_close=True):
         # Atributo para la version "thread"
         self.thread = None
 
+        self.client = None
+        self.proxy = None
+        self.auto_close = auto_close
+
         self.proxy_host = proxy_host
         self.tty = False
         self.port = port
@@ -117,6 +121,17 @@ def __init__(self, host, user, passwd=None, private_key=None, port=22, proxy_hos
             self.private_key_obj = paramiko.RSAKey.from_private_key(
                 private_key_obj)
 
+    def close(self):
+        """
+        Close the SSH client connection
+        """
+        if self.client:
+            self.client.close()
+            self.client = None
+        if self.proxy:
+            self.proxy.close()
+            self.proxy = None
+
     def __str__(self):
         res = "SSH: host: " + self.host + ", port: " + \
             str(self.port) + ", user: " + self.username
@@ -136,6 +151,9 @@ def connect(self, time_out=None):
 
             Returns: a paramiko SSHClient connected with the server.
         """
+        if self.client and self.client.get_transport() and self.client.get_transport().is_authenticated():
+            return self.client, self.proxy
+
         client = paramiko.SSHClient()
         client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 
@@ -172,6 +190,9 @@ def connect(self, time_out=None):
                            password=self.password, timeout=time_out, sock=proxy_channel,
                            pkey=self.private_key_obj)
 
+        self.client = client
+        self.proxy = proxy
+
         return client, proxy
 
     def test_connectivity(self, time_out=None):
@@ -228,10 +249,11 @@ def execute(self, command, timeout=None):
         for line in stderr:
             res_stderr += line
 
-        channel.close()
-        client.close()
-        if proxy:
-            proxy.close()
+        if self.auto_close:
+            channel.close()
+            client.close()
+            if proxy:
+                proxy.close()
         return (res_stdout, res_stderr, exit_status)
 
     def sftp_get(self, src, dest):
@@ -253,10 +275,11 @@ def sftp_get(self, src, dest):
             sftp = scp.SCPClient(transport)
 
         sftp.get(src, dest)
-        sftp.close()
-        if proxy:
-            proxy.close()
-        transport.close()
+        if self.auto_close:
+            sftp.close()
+            if proxy:
+                proxy.close()
+            transport.close()
 
     def sftp_get_files(self, src, dest):
         """ Gets a list of files from the remote server
@@ -277,10 +300,11 @@ def sftp_get_files(self, src, dest):
 
         for file0, file1 in zip(src, dest):
             sftp.get(file0, file1)
-        sftp.close()
-        if proxy:
-            proxy.close()
-        transport.close()
+        if self.auto_close:
+            sftp.close()
+            if proxy:
+                proxy.close()
+            transport.close()
 
     def sftp_put_files(self, files):
         """ Puts a list of files to the remote server
@@ -301,10 +325,11 @@ def sftp_put_files(self, files):
 
         for src, dest in files:
             sftp.put(src, dest)
-        sftp.close()
-        if proxy:
-            proxy.close()
-        transport.close()
+        if self.auto_close:
+            sftp.close()
+            if proxy:
+                proxy.close()
+            transport.close()
 
     def sftp_put(self, src, dest):
         """ Puts a file to the remote server
@@ -323,10 +348,11 @@ def sftp_put(self, src, dest):
             # in case of failure try to use scp
             sftp = scp.SCPClient(transport)
         sftp.put(src, dest)
-        sftp.close()
-        if proxy:
-            proxy.close()
-        transport.close()
+        if self.auto_close:
+            sftp.close()
+            if proxy:
+                proxy.close()
+            transport.close()
 
     def sftp_get_dir(self, src, dest):
         """ Gets recursively a directory from the remote server
@@ -348,10 +374,11 @@ def sftp_get_dir(self, src, dest):
             full_dest = filename.replace(src, dest)
             sftp.get(filename, full_dest)
 
-        sftp.close()
-        if proxy:
-            proxy.close()
-        transport.close()
+        if self.auto_close:
+            sftp.close()
+            if proxy:
+                proxy.close()
+            transport.close()
 
     def sftp_walk(self, src, files=None, sftp=None):
         """ Gets recursively the list of items in a directory from the remote server
@@ -474,10 +501,11 @@ def sftp_mkdir(self, directory, mode=0o777):
                 sftp.mkdir(directory, mode)
                 res = True
 
-            sftp.close()
-            if proxy:
-                proxy.close()
-            transport.close()
+            if self.auto_close:
+                sftp.close()
+                if proxy:
+                    proxy.close()
+                transport.close()
         else:
             # use mkdir over ssh to create the directory
             _, _, status = self.execute("mkdir -p %s" % directory)
@@ -498,10 +526,11 @@ def sftp_list(self, directory):
         transport = client.get_transport()
         sftp = paramiko.SFTPClient.from_transport(transport)
         res = sftp.listdir(directory)
-        sftp.close()
-        if proxy:
-            proxy.close()
-        transport.close()
+        if self.auto_close:
+            sftp.close()
+            if proxy:
+                proxy.close()
+            transport.close()
         return res
 
     def sftp_list_attr(self, directory):
@@ -518,10 +547,11 @@ def sftp_list_attr(self, directory):
         transport = client.get_transport()
         sftp = paramiko.SFTPClient.from_transport(transport)
         res = sftp.listdir_attr(directory)
-        sftp.close()
-        transport.close()
-        if proxy:
-            proxy.close()
+        if self.auto_close:
+            sftp.close()
+            transport.close()
+            if proxy:
+                proxy.close()
         return res
 
     def getcwd(self):
@@ -539,10 +569,11 @@ def getcwd(self):
 
         if sftp_avail:
             cwd = sftp.getcwd()
-            sftp.close()
-            if proxy:
-                proxy.close()
-            transport.close()
+            if self.auto_close:
+                sftp.close()
+                if proxy:
+                    proxy.close()
+                transport.close()
         else:
             # use rm over ssh to delete the file
             cwd, _, _ = self.execute("pwd")
@@ -603,10 +634,11 @@ def sftp_remove(self, path):
 
         if sftp_avail:
             res = sftp.remove(path)
-            sftp.close()
-            if proxy:
-                proxy.close()
-            transport.close()
+            if self.auto_close:
+                sftp.close()
+                if proxy:
+                    proxy.close()
+                transport.close()
         else:
             # use rm over ssh to delete the file
             _, _, status = self.execute("rm -f %s" % path)
@@ -635,10 +667,11 @@ def sftp_chmod(self, path, mode):
         if sftp_avail:
             sftp.chmod(path, mode)
             res = True
-            sftp.close()
-            if proxy:
-                proxy.close()
-            transport.close()
+            if self.auto_close:
+                sftp.close()
+                if proxy:
+                    proxy.close()
+                transport.close()
         else:
             # use chmod over ssh to change permissions
             _, _, status = self.execute("chmod %s %s" % (oct(mode), path))