Worker group dashboard + misc fixes

Author: sylvinus

mrq/agent.py

@@ -2,7 +2,7 @@
 
 from .context import get_current_config, connections, log
 import time
-import json
+import datetime
 import gevent
 from bson import ObjectId
 from collections import defaultdict
@@ -33,6 +33,8 @@ def work(self):
 
         self.pool.wait()
 
+        connections.mongodb_jobs.mrq_agents.delete_one({"_id": self.id})
+
     def shutdown_now(self):
         self.pool.terminate()
 
@@ -78,7 +80,8 @@ def get_agent_report(self):
             "current_workers": [p["command"] for p in self.pool.processes],
             "available_cpu": get_current_config()["available_cpu"],
             "available_memory": get_current_config()["available_memory"],
-            "worker_group": self.worker_group
+            "worker_group": self.worker_group,
+
         }
         return report
 
@@ -181,7 +184,8 @@ def orchestrate(self):
                 connections.mongodb_jobs.mrq_agents.update_one({"_id": agent["_id"]}, {"$set": {
                     "desired_workers": agent["new_desired_workers"],
                     "free_cpu": agent["free_cpu"],
-                    "free_memory": agent["free_memory"]
+                    "free_memory": agent["free_memory"],
+                    "datereported": datetime.datetime.utcnow()
                 }})
 
     def get_desired_workers_for_group(self, group):

mrq/dashboard/app.py

@@ -15,6 +15,7 @@
 import json
 import argparse
 from werkzeug.serving import run_simple
+from future.builtins import str
 
 sys.path.insert(0, os.getcwd())
 
@@ -122,6 +123,24 @@ def get_workers():
     return jsonify(data)
 
 
+@app.route('/api/workergroups', methods=["GET"])
+@requires_auth
+def get_workergroups():
+    collection = connections.mongodb_jobs.mrq_workergroups
+    data = {"workergroups": {str(row.pop("_id")): row for row in collection.find(sort=[("_id", 1)])}}
+    return jsonify(data)
+
+
+@app.route('/api/workergroups', methods=["POST"])
+@requires_auth
+def post_workergroups():
+    workergroups = json.loads(request.form["workergroups"])
+    for k, v in workergroups.iteritems():
+        connections.mongodb_jobs.mrq_workergroups.update_one({"_id": k}, {"$set": v}, upsert=True)
+
+    return jsonify({"status": "ok"})
+
+
 def build_api_datatables_query(req):
     query = {}
 

mrq/dashboard/static/js/router.js

@@ -12,7 +12,8 @@ define(["backbone", "underscore", "jquery"],function(Backbone, _, $) {
       'jobs': 'jobs',
       'io': 'io',
       'scheduledjobs': 'scheduledjobs',
-      'status': 'status'
+      'status': 'status',
+      'workergroups': 'workergroups'
     },
 
     /**
@@ -125,6 +126,11 @@ define(["backbone", "underscore", "jquery"],function(Backbone, _, $) {
       this.app.rootView.showChildPage('io', {"options": {"params": params || {}}});
     },
 
+    workergroups: function(params) {
+      this.setNavbar("workergroups");
+      this.app.rootView.showChildPage('workergroups', {"options": {"params": params || {}}});
+    },
+
     taskpaths: function(params) {
       this.setNavbar("taskpaths");
       this.app.rootView.showChildPage('taskpaths', {"options": {"params": params || {}}});

mrq/dashboard/static/js/views/root.js

@@ -2,11 +2,11 @@
  * @fileoverview Defines the view container view that contains all views
  */
 define(["views/generic/page", "jquery",
-        "views/queues", "views/workers", "views/jobs", "views/scheduledjobs", "views/index", "views/taskpaths", "views/status", "views/taskexceptions", "views/io"],
+        "views/queues", "views/workers", "views/jobs", "views/scheduledjobs", "views/index", "views/taskpaths", "views/status", "views/taskexceptions", "views/io", "views/workergroups"],
 
       function(
         Page, $,
-        QueuesView, WorkersView, JobsView, ScheduledJobsView, IndexView, TaskPathsView, StatusView, TaskExceptionsView, IOView
+        QueuesView, WorkersView, JobsView, ScheduledJobsView, IndexView, TaskPathsView, StatusView, TaskExceptionsView, IOView, WorkerGroupsView
   ) {
 
   return Page.extend({
@@ -116,6 +116,7 @@ define(["views/generic/page", "jquery",
       this.addChildPage('taskexceptions', new TaskExceptionsView());
       this.addChildPage('index', new IndexView());
       this.addChildPage('status', new StatusView());
+      this.addChildPage('workergroups', new WorkerGroupsView());
 
       return this;
     }

mrq/dashboard/static/js/views/workergroups.js

@@ -0,0 +1,37 @@
+define(["jquery", "underscore", "models", "views/generic/page"],function($, _, Models, Page) {
+
+  return Page.extend({
+
+    el: '.js-page-workergroups',
+
+    template:"#tpl-page-workergroups",
+
+    events:{
+      "click .submit": "submit"
+    },
+
+    render: function() {
+      var self = this;
+      $.get("/api/workergroups").done(function(data) {
+        self.renderTemplate();
+        self.$("textarea").val(JSON.stringify(data["workergroups"], null, 8));
+      });
+    },
+
+    submit: function(el) {
+      var self = this;
+
+      self.$("button")[0].innerHTML = "Wait...";
+
+      var val = self.$("textarea").val();
+
+      $.post("/api/workergroups", {"workergroups": val}).done(function(data) {
+        if (data.status != "ok") {
+          return alert("There was an error while saving!");
+        }
+        self.$("button")[0].innerHTML = "Save";
+      });
+    }
+  });
+
+});

mrq/dashboard/templates/index.html

@@ -38,6 +38,9 @@
             <li class="js-nav-workers">
               <a href="/#workers">Workers</a>
             </li>
+            <li class="js-nav-workergroups">
+              <a href="/#workergroups">Worker groups</a>
+            </li>
             <li class="js-nav-taskpaths">
               <a href="/#taskpaths">Paths</a>
             </li>
@@ -114,6 +117,7 @@
         <div class="js-page-taskpaths"></div>
         <div class="js-page-taskexceptions"></div>
         <div class="js-page-io"></div>
+        <div class="js-page-workergroups"></div>
 
 
       </script>
@@ -246,6 +250,17 @@ <h3>Current I/O operations</h3>
         </div>
       </script>
 
+      <script id="tpl-page-workergroups" type="text/template">
+
+        <h3>Worker groups</h3>
+        <div class="container">
+          <div class="row">
+            <textarea style="width:100%;height:400px;font-family:monospace;" onkeydown="if(event.keyCode===9){var v=this.value,s=this.selectionStart,e=this.selectionEnd;this.value=v.substring(0, s)+'\t'+v.substring(e);this.selectionStart=this.selectionEnd=s+1;return false;}"></textarea>
+            <button class="submit">Save</button>
+          </div>
+        </div>
+      </script>
+
       <script id="tpl-page-index" type="text/template">
 
         <h3>Overall</h3>

mrq/worker.py

@@ -148,7 +148,12 @@ def ensure_indexes(self):
             [("datequeued", 1)], sparse=True, background=True)
 
         self.mongodb_jobs.mrq_scheduled_jobs.ensure_index(
-            [("hash", 1)], unique=True, background=False, drop_dups=True)
+            [("hash", 1)], unique=True, background=False)
+
+        self.mongodb_jobs.mrq_agents.ensure_index(
+            [("datereported", 1)], background=False, expireAfterSeconds=300)
+        self.mongodb_jobs.mrq_agents.ensure_index(
+            [("worker_group", 1)], background=False)
 
         try:
             # This will be default in MongoDB 2.6

tests/test_agent.py

@@ -163,14 +163,22 @@ def test_agent_process(worker):
     time.sleep(3)
 
     assert connections.mongodb_jobs.mrq_workers.count() == 1
-    worker = connections.mongodb_jobs.mrq_workers.find_one()
-    assert worker["status"] == "wait"
+    w = connections.mongodb_jobs.mrq_workers.find_one()
+    assert w["status"] == "wait"
 
     connections.mongodb_jobs.mrq_workergroups.update_one({"_id": "xxx"}, {"$set": {"profiles": [
 
     ]}})
 
     time.sleep(4)
 
-    worker = connections.mongodb_jobs.mrq_workers.find_one()
-    assert worker["status"] == "stop"
+    w = connections.mongodb_jobs.mrq_workers.find_one()
+    assert w["status"] == "stop"
+
+    assert connections.mongodb_jobs.mrq_agents.count() == 1
+
+    worker.stop(deps=False)
+
+    assert connections.mongodb_jobs.mrq_agents.count() == 0
+
+    worker.stop_deps()