Merge pull request #367 from Kitware/ec2-cluster

Ec2 cluster

Author: cjh1

.gitignore

@@ -5,3 +5,6 @@ bundles
 node_modules/
 npm-debug.log
 *.pyc
+.project
+.pydevproject
+.settings/

server/taskflows/hpccloud/taskflow/paraview.py

@@ -49,17 +49,25 @@ class ParaViewTaskFlow(cumulus.taskflow.TaskFlow):
 
     """
     def start(self, *args, **kwargs):
-
-        # Load the cluster
-        model = ModelImporter.model('cluster', 'cumulus')
         user = getCurrentUser()
-        cluster = model.load(kwargs['cluster']['_id'],
-                             user=user, level=AccessType.ADMIN)
-        cluster = model.filter(cluster, user, passphrase=False)
-        kwargs['cluster'] = cluster
+        # Load the cluster
+        cluster_id = parse('cluster._id').find(kwargs)
+        if cluster_id:
+            model = ModelImporter.model('cluster', 'cumulus')
+            cluster = model.load(kwargs['cluster']['_id'],
+                                 user=user, level=AccessType.ADMIN)
+            cluster = model.filter(cluster, user, passphrase=False)
+            kwargs['cluster'] = cluster
+
+        profile_id = parse('cluster.profileId').find(kwargs)
+        if profile_id:
+            profile_id = profile_id[0].value
+            model = ModelImporter.model('aws', 'cumulus')
+            profile = model.load(profile_id, user=user, level=AccessType.ADMIN)
+            kwargs['profile'] = profile
 
         super(ParaViewTaskFlow, self).start(
-            create_paraview_job.s(self, *args, **kwargs))
+            setup_cluster.s(self, next=create_paraview_job.s(), *args, **kwargs))
 
     def terminate(self):
         self.run_task(paraview_terminate.s())

server/taskflows/hpccloud/taskflow/pyfr.py

@@ -22,6 +22,8 @@
 import subprocess
 import shutil
 from ConfigParser import SafeConfigParser
+from jsonpath_rw import parse
+from bson.objectid import ObjectId
 
 import cumulus.taskflow
 from cumulus.tasks.job import download_job_input_folders, submit_job
@@ -64,18 +66,32 @@ class PyFrTaskFlow(cumulus.taskflow.TaskFlow):
         }
     }
     """
-    def start(self, *args, **kwargs):
+    PYFR_AMI = 'ami-7def1b1d'
 
-        # Load the cluster
-        model = ModelImporter.model('cluster', 'cumulus')
+    def start(self, *args, **kwargs):
         user = getCurrentUser()
-        cluster = model.load(kwargs['cluster']['_id'],
-                             user=user, level=AccessType.ADMIN)
-        cluster = model.filter(cluster, user, passphrase=False)
-        kwargs['cluster'] = cluster
+        # Load the cluster
+        cluster_id = parse('cluster._id').find(kwargs)
+        if cluster_id:
+            cluster_id = cluster_id[0].value
+            model = ModelImporter.model('cluster', 'cumulus')
+            cluster = model.load(cluster_id, user=user, level=AccessType.ADMIN)
+            cluster = model.filter(cluster, user, passphrase=False)
+            kwargs['cluster'] = cluster
+
+        profile_id = parse('cluster.profileId').find(kwargs)
+        if profile_id:
+            profile_id = profile_id[0].value
+            model = ModelImporter.model('aws', 'cumulus')
+            profile = model.load(profile_id, user=user, level=AccessType.ADMIN)
+            kwargs['profile'] = profile
+
+        kwargs['next'] = setup_input.s()
+        kwargs['ami'] = self.PYFR_AMI
 
         super(PyFrTaskFlow, self).start(
-            setup_input.s(self,*args, **kwargs))
+            setup_cluster.s(
+                self, *args, **kwargs))
 
     def terminate(self):
         self.run_task(pyfr_terminate.s())
@@ -167,7 +183,8 @@ def update_config_file(task, client, *args, **kwargs):
                 task.logger.info('%s removed.' % section)
 
         backend_section = 'backend-%s' % kwargs['backend']['type']
-        task.logger.info('Adding backend configuration for %s')
+        task.logger.info('Adding backend configuration for %s'
+                            % kwargs['backend']['type'] )
         # Filter out options with no value
         options = {k: v for k, v in kwargs['backend'].iteritems() if v}
         options.pop('type', None)
@@ -191,8 +208,6 @@ def update_config_file(task, client, *args, **kwargs):
 
 @cumulus.taskflow.task
 def setup_input(task, *args, **kwargs):
-    task.logger.info('Input parameters: %s' % kwargs)
-
     input_folder_id = kwargs['input']['folder']['id']
     mesh_file_id = kwargs['input']['meshFile']['id']
     kwargs['meshFileId'] = mesh_file_id
@@ -201,6 +216,14 @@ def setup_input(task, *args, **kwargs):
     if not number_of_procs:
         number_of_procs = kwargs.get('numberOfNodes')
 
+    if not number_of_procs:
+        size = parse('cluster.config.launch.params.node_instance_count').find(kwargs)
+        if size:
+            number_of_procs = size[0].value + 1
+        else:
+            raise Exception('Unable to extract number of nodes in cluster')
+
+
     if not number_of_procs:
         raise Exception('Unable to determine number of mpi processes to run.')
 
@@ -270,6 +293,24 @@ def setup_input(task, *args, **kwargs):
             if os.path.exists(output_dir):
                 shutil.rmtree(output_dir)
 
+    # If we are running in the cloud determine backend to use
+    if kwargs['cluster']['type'] == 'ec2':
+        machine_spec = kwargs.get('machine')
+        # If we have GPUs use cuda
+        if int(machine_spec['gpu']) == 1:
+            backend = {
+                'type': 'cuda',
+                'device-id': 'round-robin'
+            }
+        # Use OpenMP
+        else:
+            backend = {
+                'type': 'openmp',
+                'cblas': '/usr/lib/openblas-base/libblas.so'
+            }
+
+        kwargs['backend'] = backend
+
     update_config_file(task, client, *args, **kwargs)
 
     ini_file_id = kwargs['input']['iniFile']['id']
@@ -302,7 +343,10 @@ def create_job(task, *args, **kwargs):
               'path': 'input'
             }
         ],
-        'output': []
+        'output': [],
+        'params': {
+            'numberOfSlots': kwargs['numberOfProcs']
+        }
     }
 
     client = _create_girder_client(
@@ -334,7 +378,8 @@ def submit_pyfr_job(task, cluster,  job, *args, **kwargs):
     task.logger.info('Submitting job %s to cluster.' % job['_id'])
     girder_token = task.taskflow.girder_token
 
-    job['params'] = kwargs
+    job['params'].update(kwargs)
+
     submit_job(cluster, job, log_write_url=None,
                           girder_token=girder_token, monitor=False)
 
@@ -391,7 +436,10 @@ def create_export_job(task, job_name, files, job_dir, mesh_filename):
         'name': job_name,
         'commands': commands,
         'input': [],
-        'output': []
+        'output': [],
+        'params': {
+            'numberOfSlots': 1
+        }
     }
 
     client = _create_girder_client(

server/taskflows/hpccloud/taskflow/utility/__init__.py

@@ -1,9 +1,12 @@
 import json
 from jsonpath_rw import parse
+from celery.canvas import Signature
 
+import cumulus
 from cumulus.tasks.job import terminate_job
 from cumulus.constants import JobState
 
+from girder_client import GirderClient, HttpError
 
 def terminate_jobs(task, client, cluster, jobs):
     for job in jobs:
@@ -32,3 +35,98 @@ def get_cluster_job_output_dir(cluster):
 
     return job_output_dir
 
+def create_girder_client(girder_api_url, girder_token):
+    client = GirderClient(apiUrl=girder_api_url)
+    client.token = girder_token
+
+    return client
+
+def create_ec2_cluster(task, cluster, profile, ami):
+    machine_type = cluster['machine']['id']
+    nodeCount = cluster['clusterSize']-1
+    launch_spec = 'ec2'
+    launch_params = {
+        'master_instance_type': machine_type,
+        'master_instance_ami': ami,
+        'node_instance_count': nodeCount,
+        'node_instance_type': machine_type,
+        'node_instance_ami': ami
+    }
+    provision_spec = 'gridengine/site'
+    provision_params = {
+        'ansible_ssh_user': 'ubuntu'
+    }
+
+    body = {
+        'type': 'ec2',
+        'name': cluster['name'],
+         'config': {
+            'launch': {
+                'spec': launch_spec,
+                'params': launch_params
+            },
+            'provision': {
+                'spec': provision_spec
+            }
+        },
+        'profileId': cluster['profileId']
+    }
+    client = create_girder_client(
+        task.taskflow.girder_api_url, task.taskflow.girder_token)
+
+    try:
+        cluster = client.post('clusters',  data=json.dumps(body))
+    except HttpError as he:
+        task.logger.exception(he.responseText)
+        raise
+
+    msg = 'Created cluster: %s' % cluster['_id']
+    task.taskflow.logger.info(msg)
+    task.logger.info(msg)
+
+    # Now save cluster id in metadata
+    task.taskflow.set_metadata('cluster', cluster)
+
+    task.logger.info('Starting cluster.')
+
+    body = {
+        'status': 'launching'
+    }
+    client.patch('clusters/%s' % cluster['_id'], data=json.dumps(body))
+
+    secret_key = profile['secretAccessKey']
+    log_write_url = '%s/clusters/%s/log' % (task.taskflow.girder_api_url,
+                                            cluster['_id'])
+    provision_params['cluster_state'] = 'running'
+    launch_params['cluster_state'] = 'running'
+    girder_token = task.taskflow.girder_token
+    cumulus.ansible.tasks.cluster.start_cluster(
+        launch_spec, provision_spec, cluster, profile, secret_key,
+        launch_params, provision_params, girder_token, log_write_url,
+        master_name='head')
+
+    # Get the update to date cluster
+    cluster = client.get('clusters/%s' % cluster['_id'])
+
+    return cluster
+
+@cumulus.taskflow.task
+def setup_cluster(task, *args,**kwargs):
+    cluster = kwargs['cluster']
+
+    if '_id' in cluster:
+        task.taskflow.logger.info('We are using an existing cluster: %s' % cluster['name'])
+    else:
+        task.taskflow.logger.info('We are creating an EC2 cluster.')
+        task.logger.info('Cluster name %s' % cluster['name'])
+        kwargs['machine'] = cluster.get('machine')
+        ami = kwargs.get('ami')
+        profile = kwargs.get('profile')
+        cluster = create_ec2_cluster(task, cluster, profile, ami)
+        task.logger.info('Cluster started.')
+
+    # Call any follow on task
+    if 'next' in kwargs:
+        kwargs['cluster'] = cluster
+        next = Signature.from_dict(kwargs['next'])
+        next.delay(*args, **kwargs)

src/StateTransitionBehavior.js

@@ -77,6 +77,12 @@ export function handleTaskflowChange(state, taskflow) {
     }
   }
 
+  // for taskflows on ec2 the meta object is not as readily available
+  // this is due to fewer jobs coming through SSE which triggers a fetch for trad clusters.
+  if (!taskflow.flow.meta) {
+    dispatch(TaskflowActions.fetchTaskflow(taskflow.flow._id));
+  }
+
   // Update taslkfow meta
   if (allComplete !== taskflow.allComplete ||
     outputDirectory[0] !== taskflow.outputDirectory ||

src/pages/Preferences/AWS/AWSForm.js

@@ -43,6 +43,10 @@ export default React.createClass({
     const data = nextProps.data,
       oldData = this.props.data;
 
+    if (!nextProps.data._id) {
+      this.refs.nameInput.focus();
+    }
+
     if (!deepEquals(data, oldData)) {
       this.setState({ data });
     }
@@ -90,6 +94,7 @@ export default React.createClass({
               onChange={this.formChange}
               disabled={this.state.data._id}
               required
+              ref="nameInput"
             />
           </section>
           <section className={style.group}>

src/pages/Preferences/Cluster/ClusterForm.js

@@ -38,6 +38,12 @@ export default React.createClass({
     onChange: React.PropTypes.func,
   },
 
+  componentWillReceiveProps(nextProps) {
+    if (!nextProps.data._id) {
+      this.refs.nameInput.focus();
+    }
+  },
+
   formChange(event) {
     const propName = event.target.dataset.key;
     const value = event.target.value;
@@ -81,6 +87,7 @@ export default React.createClass({
                 data-key="name"
                 onChange={this.formChange}
                 required
+                ref="nameField"
               />
           </section>
           <section className={style.group}>

src/pages/Preferences/Cluster/index.js

@@ -84,7 +84,7 @@ const ClusterPrefs = React.createClass({
 
   clusterHasSimulation(id) {
     for (let i = 0; i < this.props.taskflows.length; i++) {
-      if (this.props.taskflows[i].flow.meta.cluster._id === id) {
+      if (this.props.taskflows[i].flow && this.props.taskflows[i].flow.meta.cluster._id === id) {
         return this.props.taskflows[i].simulation;
       }
     }

src/panels/JobMonitor/index.js

@@ -166,7 +166,7 @@ export default connect(
     // Sort the tasks by created timestamp
     tasks.sort((task1, task2) => Date.parse(task1.created) > Date.parse(task2.created));
 
-    if (cluster.log && cluster.log.length) {
+    if (cluster && cluster.log && cluster.log.length) {
       clusterLog = cluster.log.sort((task1, task2) => Date.parse(task1.created) > Date.parse(task2.created));
       clusterLogStreamState = cluster.logStream ? cluster.logStream.readyState : CLOSED;
     }

src/panels/run/RunEC2.js

@@ -51,6 +51,8 @@ export default React.createClass({
 
     if (key === 'profile') {
       value = this.state.profiles[value];
+    } else if (key === 'machine') {
+      value = machines[value];
     }
 
     if (this.props.onChange) {
@@ -67,7 +69,7 @@ export default React.createClass({
     var machineMapper = (machine, index) =>
       <option
         key={machine.id}
-        value={machine.id}
+        value={index}
       >
         { `${machine.name} - ${machine.cpu} core${machine.cpu > 1 ? 's' : ''} - ${machine.memory}GB ${machine.gpu ? ' + GPU' : ''} - ${machine.storage}` }
       </option>;
@@ -108,7 +110,7 @@ export default React.createClass({
               <select
                 onChange={this.dataChange} className={style.input}
                 data-key="machine"
-                defaultValue={machines[0].id}
+                defaultValue={machines[0]}
               >
                 {machines.map(machineMapper)}
               </select>