jobs management update and async support

Major jobs handling update, added support for async jobs queue using
celery lib

monkey_business/cc/admin/ui/index.html

@@ -48,7 +48,7 @@
                 <div id="jobs" class="panel-body panel-collapse collapse in">
                 <table class="table table-bordered table-hover" id="jobs-table">
                   <thead>
-                    <tr><th>Time</th><th>Status</th><th>Data</th></tr>
+                    <tr><th>Id</th><th>Time</th><th>Type</th><th>Status</th><th>Properties</th></tr>
                   </thead>
                   <tbody>
                   </tbody>
@@ -83,7 +83,17 @@
                             <a href="#newJob" data-toggle="collapse">New Job</a>
                         </div>
                         <div id="newJob" style="overflow: visible" class="panel-body panel-collapse collapse in" aria-expanded="true">
-                            <div id="job-config">
+                            <div id="job-config-section">
+                                <div id="job-config">
+                                </div>
+                                <button id="btnSendJob" class="btn btn-default" type="button"
+                                onclick="sendJob()" style="margin-top:-4px; visibility: hidden">
+                                Update
+                                </button>
+                                <button id="btnDeleteJob" class="btn btn-default" type="button"
+                                onclick="deleteJob()" style="margin-top:-4px; visibility: hidden">
+                                Delete
+                                </button>
                             </div>
                         </div>
                     </div>

monkey_business/cc/admin/ui/js/monkeysb-admin.js

@@ -1,18 +1,8 @@
-/*const jsonFile = "/api/jbos";
-var monkeys = null;
-var generationDate = null;*/
-
 // The JSON must be fully loaded before onload() happens for calling draw() on 'monkeys'
 $.ajaxSetup({
     async: false
 });
 
-// Reading the JSON file containing the monkeys' informations
-/*$.getJSON(jsonFile, function(json) {
-    jobs = json.objects;
-    generationDate = json.timestamp;
-});*/
-
 // Images/icons constants
 const ICONS_DIR = "./css/img/objects/";
 const ICONS_EXT = ".png";
@@ -22,15 +12,27 @@ const ICONS_EXT = ".png";
 
 var jobsTable = undefined;
 var vcenterCfg = undefined;
+var jobCfg = undefined;
 
 JSONEditor.defaults.theme = 'bootstrap3';
 
-
 function initAdmin() {
 
     jobsTable = $("#jobs-table").DataTable({
-        "ordering": false,
+        "ordering": true,
+        "order": [[1, "desc"]],
     });
+    jobsTable.on( 'click', 'tr', function () {
+        if ( $(this).hasClass('selected') ) {
+            $(this).removeClass('selected');
+        }
+        else {
+            jobsTable.$('tr.selected').removeClass('selected');
+            $(this).addClass('selected');
+        }
+        jobdata = jobsTable.row(this).data();
+        createNewJob(jobdata[0], jobdata[3]);
+    } );
 
     vcenterCfg = new JSONEditor(document.getElementById('vcenter-config'),{
                         schema: {
@@ -75,7 +77,8 @@ function initAdmin() {
                                   },
                                   datacenter_name: {
                                       title: "Datacenter (opt.)",
-                                      type: "string",                                  },
+                                      type: "string",
+                                  },
                                   cluster_name: {
                                       title: "Cluster (opt.)",
                                       type: "string",
@@ -93,31 +96,26 @@ function initAdmin() {
                         },
                         disable_edit_json: false,
                         disable_properties: true,
-                        startval: $,
                         });
     
-    window.setTimeout(updateJobs, 10000);
+    window.setTimeout(updateJobs, 5000);
     loadVcenterConfig();
     updateJobs();
 
 }
 
-function updateVCenterConf() {
-
-}
-
 function updateJobs() {
     $.getJSON('/job', function(json) {
         jobsTable.clear();
-        var jobs = json.objects;
+        var jobsList = json.objects;
 
-        for (var i = 0; i < jobs.length; i++) {
-            jobsTable.row.add([jobs[i].timestamp, jobs[i].status, JSON.stringify(jobs[i].data)]);
+        for (var i = 0; i < jobsList.length; i++) {
+            jobsTable.row.add([jobsList[i].id, jobsList[i].creation_time, jobsList[i].type,jobsList[i].execution.state, JSON.stringify(jobsList[i].properties)]);
         }
 
         jobsTable.draw();
+        //enableJobsSelect();
     });
-
 }
 
 function loadVcenterConfig() {
@@ -152,60 +150,108 @@ function updateVcenterConfig() {
 
 }
 
-function createNewJob() {
+function createNewJob(id, state) {
+    if (!id) {
+        jobsTable.$('tr.selected').removeClass('selected');
+    }
+
     elem = document.getElementById('job-config');
     elem.innerHTML = ""
     jobCfg = new JSONEditor(elem,{
-                            schema: {
-                              type: "object",
-                              title: "Job",
-                              properties: {
-                                    job: {
-                                        title: "Type",
-                                        $ref: "/jobcreate",
-                                    }
-                              },
-                              options: {
-                                "collapsed": false
-                              },
-                            },
-                            ajax: true,
-                            disable_edit_json: false,
-                            disable_collapse: true,
-                            disable_properties: true,
-                            });
+                                schema: {
+                                  type: "object",
+                                  title: "Job",
+                                  properties: {
+                                        job: {
+                                            title: "Type",
+                                            $ref: "/jobcreate" + ((id)?"?id="+id:""),
+                                        }
+                                  },
+                                  options: {
+                                    "collapsed": false
+                                  },
+                                },
+                                ajax: true,
+                                disable_edit_json: false,
+                                disable_collapse: true,
+                                disable_properties: true,
+                                no_additional_properties: true
+                                });
+
+    jobCfg.on('ready',function() {
+        if (id && state != "pending") {
+            jobCfg.disable();
+            document.getElementById("btnSendJob").style.visibility = "hidden";
+            document.getElementById("btnDeleteJob").style.visibility = "hidden";
+        }
+        else {
+            jobCfg.enable();
+            document.getElementById("btnSendJob").style.visibility = "visible";
+            if (id) {
+                document.getElementById("btnDeleteJob").style.visibility = "visible";
+            }
+            else {
+                document.getElementById("btnDeleteJob").style.visibility = "hidden";
+            }
+        }
+    });
+}
+
+function sendJob() {
+    var job_config = jobCfg.getValue()
+
+    $.ajax({
+            headers : {
+                'Accept' : 'application/json',
+                'Content-Type' : 'application/json'
+            },
+            url : '/jobcreate',
+            type : 'POST',
+            data : JSON.stringify(job_config.job),
+            success : function(response, textStatus, jqXhr) {
+                console.log("Job successfully updated!");
+                updateJobs();
+            },
+            error : function(jqXHR, textStatus, errorThrown) {
+                // log the error to the console
+                console.log("The following error occured: " + textStatus, errorThrown);
+            },
+            complete : function() {
+                console.log("Sending job config...");
+            }
+        });
+}
+
+function deleteJob() {
+    var job_config = jobCfg.getValue();
+    if (job_config.job.id) {
+            $.ajax({
+            headers : {
+                'Accept' : 'application/json',
+                'Content-Type' : 'application/json'
+            },
+            url : '/jobcreate',
+            type : 'GET',
+            data : "action=delete&id=" + job_config.job.id,
+            success : function(response, textStatus, jqXhr) {
+                console.log("Job successfully updated!");
+                updateJobs();
+            },
+            error : function(jqXHR, textStatus, errorThrown) {
+                // log the error to the console
+                console.log("The following error occured: " + textStatus, errorThrown);
+            },
+            complete : function() {
+                console.log("Sending job config...");
+            }
+        });
+    }
 }
 
 function configSched() {
 
 }
 
-/**
- * Manage the event when an object is selected
- */
-function onSelect(properties) {
-
-    /*if (properties.nodes.length > 0) {
-        onNodeSelect(properties.nodes);
-    }
-    else
-    {
-        var content = "<b>No selection</b>"
-        $("#selectionInfo").html(content);
-        $('#monkey-config').hide()
-        $('#btnConfigLoad, #btnConfigUpdate').hide();
-        telemTable.clear();
-        telemTable.draw();
-    }*/
-
-    /*if (properties.edges.length > 0) {
-        onEdgeSelect(properties.edges);
-    }*/
-
-}
-
-
-
 /**
  * Clears the value in the local storage
  */

monkey_business/cc/connectors/__init__.py

@@ -1,17 +1,17 @@
+def _load_prop_dict(self, target, prop):
+    for property in prop:
+        if not target.has_key(property):
+            continue
+        if type(prop[property]) is dict:
+            _load_prop_dict(self, target[property], prop[property])
+        else:
+            target[property] = prop[property]
+
 
 class NetControllerConnector(object):
     def __init__(self):
         self._properties = {}
 
-    def _load_prop_dict(self, target, prop):
-        for property in prop:
-            if not target.has_key(property):
-                continue
-            if type(prop[property]) is dict:
-                self._load_prop_dict(target[property], prop[property])
-            else:
-                target[property] = prop[property]
-
     def is_connected(self):
         return False
 
@@ -22,7 +22,7 @@ class NetControllerConnector(object):
         return self._properties
 
     def load_properties(self, properties):
-        self._load_prop_dict(self._properties, properties)
+        _load_prop_dict(self, self._properties, properties)
 
     def get_vlans_list(self):
         raise NotImplementedError()
@@ -38,17 +38,27 @@ class NetControllerConnector(object):
 
 class NetControllerJob(object):
     connector = NetControllerConnector
+    _properties = {
+        # property: value
+    }
+
+    _enumerations = {
+
+    }
 
     def __init__(self):
-        self._properties = {
-            # property: [value, enumerating_function]
-        }
+        pass
 
     def get_job_properties(self):
         return self._properties
 
-    def set_job_properties(self, properties):
-        return {}
+    def load_job_properties(self, properties):
+        _load_prop_dict(self, self._properties, properties)
+
+    def get_property_function(self, property):
+        if property in self._enumerations.keys():
+            return self._enumerations[property]
+        return None
 
     def run(self):
         raise NotImplementedError()

monkey_business/cc/connectors/demo.py

@@ -36,8 +36,9 @@ class DemoConnector(NetControllerConnector):
 
 class DemoJob(NetControllerJob):
     connector = DemoConnector
-
-    def __init__(self):
-        self._properties = {
-            "vlan": [0, "get_vlans_list"],
-        }
+    _properties = {
+        "vlan": 0,
+    }
+    _enumerations = {
+        "vlan": "get_vlans_list",
+    }

monkey_business/cc/connectors/vcenter.py

@@ -63,7 +63,7 @@ class VCenterConnector(NetControllerConnector):
     def get_entities_on_vlan(self, vlanid):
         return []
 
-    def deploy_monkey(self, vlanid):
+    def deploy_monkey(self, vlanid, vm_name):
         if not self._properties["monkey_template_name"]:
             raise Exception("Monkey template not configured")
 
@@ -72,7 +72,7 @@ class VCenterConnector(NetControllerConnector):
         if not monkey_template:
             raise Exception("Monkey template not found")
 
-        task = self._clone_vm(vcontent, monkey_template)
+        task = self._clone_vm(vcontent, monkey_template, vm_name)
         if not task:
             raise Exception("Error deploying monkey VM")
 
@@ -86,7 +86,7 @@ class VCenterConnector(NetControllerConnector):
         if self._service_instance:
             self.disconnect()
 
-    def _clone_vm(self, vcontent, vm):
+    def _clone_vm(self, vcontent, vm, name):
 
         # get vm target folder
         if self._properties["monkey_vm_info"]["vm_folder"]:
@@ -116,7 +116,7 @@ class VCenterConnector(NetControllerConnector):
         clonespec = vim.vm.CloneSpec()
         clonespec.location = relospec
 
-        task = vm.Clone(folder=destfolder, name=self._properties["monkey_vm_info"]["name"], spec=clonespec)
+        task = vm.Clone(folder=destfolder, name=name, spec=clonespec)
         return self._wait_for_task(task)
 
 
@@ -154,9 +154,11 @@ class VCenterConnector(NetControllerConnector):
 
 class VCenterJob(NetControllerJob):
     connector = VCenterConnector
-
-    def __init__(self):
-        self._properties = {
-            "vlan": [0, "get_vlans_list"],
-        }
+    _properties = {
+        "vlan": 0,
+        "vm_name": "",
+    }
+    _enumerations = {
+        "vlan": "get_vlans_list",
+    }
 

monkey_business/cc/dbconfig.py

@@ -0,0 +1,10 @@
+SRV_ADDRESS = 'localhost:27017'
+
+BROKER_URL = 'mongodb://%(srv)s/monkeybusiness' % {'srv': SRV_ADDRESS}
+MONGO_URI = BROKER_URL
+CELERY_RESULT_BACKEND = 'mongodb://%(srv)s/' % {'srv': SRV_ADDRESS}
+CELERY_MONGODB_BACKEND_SETTINGS = {
+    'database': 'monkeybusiness',
+    'taskmeta_collection': 'celery_taskmeta',
+}
+#CELERYD_LOG_FILE="../celery.log"

monkey_business/cc/main.py

@@ -10,13 +10,10 @@ from datetime import datetime
 import dateutil.parser
 from connectors.vcenter import VCenterJob, VCenterConnector
 from connectors.demo import DemoJob, DemoConnector
-
-MONGO_URL = os.environ.get('MONGO_URL')
-if not MONGO_URL:
-    MONGO_URL = "mongodb://localhost:27017/monkeybusiness"
+import tasks_manager
 
 app = Flask(__name__)
-app.config['MONGO_URI'] = MONGO_URL
+app.config.from_object('dbconfig')
 mongo = PyMongo(app)
 
 available_jobs = [VCenterJob, DemoJob]
@@ -35,16 +32,14 @@ class Job(restful.Resource):
     def get(self, **kw):
         id = kw.get('id')
         timestamp = request.args.get('timestamp')
+        result = {}
 
         if (id):
-            return mongo.db.job.find_one_or_404({"id": id})
+            return mongo.db.job.find_one_or_404({"_id": id})
         else:
-            result = {'timestamp': datetime.now().isoformat()}
+            result['timestamp'] = datetime.now().isoformat()
 
-        find_filter = {}
-        if None != timestamp:
-            find_filter['modifytime'] = {'$gt': dateutil.parser.parse(timestamp)}
-        result['objects'] = [x for x in mongo.db.job.find(find_filter)]
+        result['objects'] = [x for x in mongo.db.job.find().sort("creation_time", -1)]
         return result
 
     def post(self, **kw):
@@ -67,10 +62,11 @@ class Job(restful.Resource):
                                    {"$set": job_json},
                                    upsert=True)
 
+
 class Connector(restful.Resource):
     def get(self, **kw):
         type = request.args.get('type')
-        if (type == 'VCenterConnector'):
+        if type == 'VCenterConnector':
             vcenter = VCenterConnector()
             properties = mongo.db.connector.find_one({"type": 'VCenterConnector'})
             if properties:
@@ -82,7 +78,7 @@ class Connector(restful.Resource):
 
     def post(self, **kw):
         settings_json = json.loads(request.data)
-        if (settings_json.get("type") == 'VCenterConnector'):
+        if settings_json.get("type") == 'VCenterConnector':
 
             # preserve password
             properties = mongo.db.connector.find_one({"type": 'VCenterConnector'})
@@ -93,10 +89,19 @@ class Connector(restful.Resource):
                                                {"$set": settings_json},
                                                upsert=True)
 
+
+def get_jobclass_by_name(name):
+    for jobclass in available_jobs:
+        if jobclass.__name__ == name:
+            return jobclass()
+
+
 class JobCreation(restful.Resource):
     def get(self, **kw):
         jobtype = request.args.get('type')
-        if not jobtype:
+        action = request.args.get('action')
+        jobid = request.args.get('id')
+        if not (jobtype or jobid):
             res = []
             update_connectors()
             for con in available_jobs:
@@ -105,24 +110,49 @@ class JobCreation(restful.Resource):
             return {"oneOf": res}
 
         job = None
-        for jobclass in available_jobs:
-            if jobclass.__name__ == jobtype:
-                job = jobclass()
+        if not jobid:
+            job = get_jobclass_by_name(jobtype)
+        else:
+            loaded_job = mongo.db.job.find_one({"_id": bson.ObjectId(jobid)})
+            if loaded_job:
+                job = get_jobclass_by_name(loaded_job.get("type"))
+                job.load_job_properties(loaded_job.get("properties"))
+
+        if action == "delete":
+            if loaded_job.get("execution")["state"] == "pending":
+                mongo.db.job.remove({"_id": bson.ObjectId(jobid)})
+                return {'status': 'ok'}
+            else:
+                return {'status': 'bad state'}
 
         if job and job.connector.__name__ in active_connectors.keys():
-            properties = dict()
-            job_prop = job.get_job_properties()
+            properties = {
+                "type": {
+                    "type": "enum",
+                    "enum": [job.__class__.__name__],
+                    "options": {"hidden": True}
+                }
+            }
+            if (jobid):
+                properties["_id"] = {
+                    "type": "enum",
+                    "enum": [jobid],
+                    "name": "ID",
+                }
 
+            job_prop = job.get_job_properties()
             for prop in job_prop:
                 properties[prop] = dict({})
-                if type(job_prop[prop][0]) is int:
+                properties[prop]["default"] = job_prop[prop]
+                if type(job_prop[prop]) is int:
                     properties[prop]["type"] = "number"
-                elif type(job_prop[prop][0]) is bool:
+                elif type(job_prop[prop]) is bool:
                     properties[prop]["type"] = "boolean"
                 else:
                     properties[prop]["type"] = "string"
-                if job_prop[prop][1]:
-                    properties[prop]["enum"] = list(active_connectors[job.connector.__name__].__getattribute__(job_prop[prop][1])())
+                enum = job.get_property_function(prop)
+                if enum:
+                    properties[prop]["enum"] = list(active_connectors[job.connector.__name__].__getattribute__(enum)())
 
             res = dict({
                 "title": "%s Job" % jobtype,
@@ -137,6 +167,45 @@ class JobCreation(restful.Resource):
 
         return {}
 
+    def post(self, **kw):
+        settings_json = json.loads(request.data)
+        jobtype = settings_json.get("type")
+        jobid = settings_json.get("id")
+        job = None
+        for jobclass in available_jobs:
+            if jobclass.__name__ == jobtype:
+                job = jobclass()
+        if not job:
+            return {'status': 'bad type'}
+
+        # params validation
+        job.load_job_properties(settings_json)
+        parsed_prop = job.get_job_properties()
+        if jobid:
+            res = mongo.db.job.update({"_id": bson.ObjectId(jobid)},
+                                      {"$set": {"properties": parsed_prop}})
+            if res and (res["ok"] == 1):
+                return {'status': 'ok', 'updated': res["nModified"]}
+            else:
+                return {'status': 'failed'}
+
+        else:
+            execution_state = {"taskid": "",
+                               "state" : "pending"}
+            new_job = {
+                "creation_time": datetime.now(),
+                "type": jobtype,
+                "properties": parsed_prop,
+                "execution": execution_state,
+            }
+            jobid = mongo.db.job.insert(new_job)
+            async = tasks_manager.run_task.delay(jobid)
+            execution_state["taskid"] = async.id
+            mongo.db.job.update({"_id": jobid},
+                                {"$set": {"execution": execution_state}})
+
+            return {'status': 'created'}
+
 
 def normalize_obj(obj):
     if obj.has_key('_id') and not obj.has_key('id'):
@@ -185,7 +254,9 @@ def update_connectors():
                 active_connectors.pop(connector_name)
                 app.logger.info("Error activating connector: %s, reason: %s" % (connector_name, e))
 
-
+@app.before_first_request
+def init():
+    update_connectors()
 
 @app.route('/admin/<path:path>')
 def send_admin(path):

monkey_business/cc/tasks_manager.py

@@ -0,0 +1,45 @@
+import os
+import time
+from flask import Flask
+from flask.ext.pymongo import PyMongo
+from celery import Celery
+
+def make_celery(app):
+    celery = Celery(main='MONKEY_TASKS', backend=app.config['CELERY_RESULT_BACKEND'],
+                    broker=app.config['BROKER_URL'])
+    celery.conf.update(app.config)
+    TaskBase = celery.Task
+    class ContextTask(TaskBase):
+        abstract = True
+        def __call__(self, *args, **kwargs):
+            with app.app_context():
+                return TaskBase.__call__(self, *args, **kwargs)
+    celery.Task = ContextTask
+    return celery
+
+fapp = Flask(__name__)
+fapp.config.from_object('dbconfig')
+celery = make_celery(fapp)
+mongo = PyMongo(fapp)
+
+@celery.task
+def run_task(jobid):
+    task_id = run_task.request.id
+    print "searching for ", jobid
+    job = mongo.db.job.find_one({"_id": jobid})
+    if not job:
+        return False
+    job["execution"]["state"] = "processing"
+    mongo.db.job.update({"_id": jobid}, job)
+
+    time.sleep(30)
+
+    job["execution"]["state"] = "done"
+    mongo.db.job.update({"_id": jobid}, job)
+    return "task: " + task_id
+
+
+@celery.task
+def update_cache(connector):
+    time.sleep(30)
+    return "job: " + repr(job)

monkey_business/readme.txt

@@ -0,0 +1,4 @@
+dependencies:
+sudo pip install pyVmomi
+sudo pip install celery
+sudo pip install -U celery[mongodb]