Updated sqswatcher to be more resiliant to errors and improved documentation.

Author: adamrb

docs/source/configuration.rst

@@ -131,7 +131,7 @@ Defaults to false for the default template. ::
 
 scheduler
 """""""""
-Cluster scheduler
+Scheduler to be used with the cluster.  Valid options are sge, openlava, or torque.
 
 Defaults to sge for the default template. ::
 

docs/source/getting_started.rst

@@ -73,7 +73,7 @@ This configure wizard will prompt you for everything you need to create your clu
 
         Cluster Name [mycluster]:
 
-Next, you will be prompted for your AWS Access & Secret Keys.  You can leave these blank to use keys defined in your environment variaables or aws config.  Othewise, set them here to be used by cfncluster.
+Next, you will be prompted for your AWS Access & Secret Keys.  Enter the keys for an IAM user with administrative privledges.  These can also be read from your environment variaables or the  aws CLI config.
 
 ::
 

docs/source/welcome.rst

@@ -1,6 +1,6 @@
 .. _welcome
 
-Working with cfncluster
+Working with CfnCluster
 #######################
 
 .. toctree::

node/sqswatcher/plugins/openlava.py

@@ -35,11 +35,6 @@ def addHost(hostname, cluster_user):
     ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
     hosts_key_file = '/home/' + cluster_user + '/.ssh/known_hosts'
     user_key_file = '/home/' + cluster_user + '/.ssh/id_rsa'
-    try:
-        ssh.load_host_keys(hosts_key_file)
-    except IOError:
-        ssh._host_keys_filename = None
-        pass
     iter=0
     connected=False
     while iter < 3 and connected == False:
@@ -54,6 +49,11 @@ def addHost(hostname, cluster_user):
             if iter == 3:
                print("Unable to provison host")
                return
+    try:
+        ssh.load_host_keys(hosts_key_file)
+    except IOError:
+        ssh._host_keys_filename = None
+        pass
     ssh.save_host_keys(hosts_key_file)
     ssh.close()
 

node/sqswatcher/plugins/sge.py

@@ -59,11 +59,6 @@ def addHost(hostname, cluster_user):
     ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
     hosts_key_file = '/home/' + cluster_user + '/.ssh/known_hosts'
     user_key_file = '/home/' + cluster_user + '/.ssh/id_rsa'
-    try:
-        ssh.load_host_keys(hosts_key_file)
-    except IOError:
-        ssh._host_keys_filename = None
-        pass
     iter=0
     connected=False
     while iter < 3 and connected == False:
@@ -78,6 +73,11 @@ def addHost(hostname, cluster_user):
             if iter == 3:
                print("Unable to provison host")
                return
+    try:
+        ssh.load_host_keys(hosts_key_file)
+    except IOError:
+        ssh._host_keys_filename = None
+        pass
     ssh.save_host_keys(hosts_key_file)
     command = "sudo sh -c \'cd /opt/sge && /opt/sge/inst_sge -x -auto /opt/cfncluster/templates/sge/sge_inst.conf\'"
     stdin, stdout, stderr = ssh.exec_command(command)

node/sqswatcher/plugins/torque.py

@@ -37,11 +37,6 @@ def addHost(hostname,cluster_user):
     ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
     hosts_key_file = '/home/' + cluster_user + '/.ssh/known_hosts'
     user_key_file = '/home/' + cluster_user + '/.ssh/id_rsa'
-    try:
-        ssh.load_host_keys(hosts_key_file)
-    except IOError:
-        ssh._host_keys_filename = None
-        pass
     iter=0
     connected=False
     while iter < 3 and connected == False:
@@ -56,6 +51,11 @@ def addHost(hostname,cluster_user):
             if iter == 3:
                print("Unable to provison host")
                return
+    try:
+        ssh.load_host_keys(hosts_key_file)
+    except IOError:
+        ssh._host_keys_filename = None
+        pass
     ssh.save_host_keys(hosts_key_file)
     ssh.close()
 

node/sqswatcher/sqswatcher.py

@@ -115,8 +115,21 @@ def pollQueue(scheduler, q, t):
                         wait = 15
                         while retry < 3:
                             try:
-                                hostname = ec2.get_all_instances(instance_ids=instanceId)[
-                                           0].instances[0].private_dns_name.split('.')[:1][0]
+                                hostname = ec2.get_all_instances(instance_ids=instanceId)
+
+                                if not hostname:
+                                    print "Unable to find running instance %s." % instanceId
+                                else:
+				    print "Adding Hostname: %s" % hostname
+                                    hostname = hostname[0].instances[0].private_dns_name.split('.')[:1][0]
+                                    s.addHost(hostname,cluster_user)
+
+                                    t.put_item(data={
+                                        'instanceId': instanceId,
+                                        'hostname': hostname
+                                    })
+
+				q.delete_message(result)
                                 break
                             except boto.exception.BotoServerError as e:
                                 if e.error_code == 'RequestLimitExceeded':
@@ -125,15 +138,9 @@ def pollQueue(scheduler, q, t):
                                     wait = (wait*2+retry)
                                 else:
                                     raise e
-
-                        s.addHost(hostname,cluster_user)
-
-                        t.put_item(data={
-                            'instanceId': instanceId,
-                            'hostname': hostname
-                        })
-
-                        q.delete_message(result)
+                            except:
+                                print "Unexpected error:", sys.exc_info()[0]
+                                raise
 
                     elif eventType == 'autoscaling:EC2_INSTANCE_TERMINATE':
                         print eventType, instanceId
@@ -142,12 +149,16 @@ def pollQueue(scheduler, q, t):
                             item = t.get_item(consistent=True, instanceId=instanceId)
                             hostname = item['hostname']
 
-                            s.removeHost(hostname,cluster_user)
+                            if hostname:
+                                s.removeHost(hostname,cluster_user)
 
                             item.delete()
 
                         except boto.dynamodb2.exceptions.ItemNotFound:
                             print ("Did not find %s in the metadb\n" % instanceId)
+                        except:
+                            print "Unexpected error:", sys.exc_info()[0]
+                            raise
 
                         q.delete_message(result)