plugin: add enforcement of max running jobs limit for a queue per-association

src/bindings/python/fluxacct/accounting/create_db.py

@@ -181,6 +181,7 @@ def create_db(
                 max_nodes_per_job   int(11)    DEFAULT 1   NOT NULL    ON CONFLICT REPLACE DEFAULT 1,
                 max_time_per_job    int(11)    DEFAULT 60  NOT NULL    ON CONFLICT REPLACE DEFAULT 60,
                 priority            int(11)    DEFAULT 0   NOT NULL    ON CONFLICT REPLACE DEFAULT 0,
+                max_running_jobs    int(11)    DEFAULT 100 NOT NULL    ON CONFLICT REPLACE DEFAULT 100,
                 PRIMARY KEY (queue)
             );"""
     )

src/bindings/python/fluxacct/accounting/queue_subcommands.py

@@ -37,16 +37,19 @@ def view_queue(conn, queue):
         raise sqlite3.OperationalError(f"an sqlite3.OperationalError occurred: {exc}")
 
 
-def add_queue(conn, queue, min_nodes=1, max_nodes=1, max_time=60, priority=0):
+def add_queue(
+    conn, queue, min_nodes=1, max_nodes=1, max_time=60, priority=0, max_running_jobs=100
+):
     try:
         insert_stmt = """
                       INSERT INTO queue_table (
                         queue,
                         min_nodes_per_job,
                         max_nodes_per_job,
                         max_time_per_job,
-                        priority
-                      ) VALUES (?, ?, ?, ?, ?)
+                        priority,
+                        max_running_jobs
+                      ) VALUES (?, ?, ?, ?, ?, ?)
                       """
         conn.execute(
             insert_stmt,
@@ -56,6 +59,7 @@ def add_queue(conn, queue, min_nodes=1, max_nodes=1, max_time=60, priority=0):
                 max_nodes,
                 max_time,
                 priority,
+                max_running_jobs,
             ),
         )
 
@@ -84,13 +88,15 @@ def edit_queue(
     max_nodes_per_job=None,
     max_time_per_job=None,
     priority=None,
+    max_running_jobs=None,
 ):
     params = locals()
     editable_fields = [
         "min_nodes_per_job",
         "max_nodes_per_job",
         "max_time_per_job",
         "priority",
+        "max_running_jobs",
     ]
 
     for field in editable_fields:

src/cmd/flux-account-priority-update.py

@@ -110,6 +110,7 @@ def bulk_update(path):
             "max_nodes_per_job": int(row[2]),
             "max_time_per_job": int(row[3]),
             "priority": int(row[4]),
+            "max_running_jobs": int(row[5]),
         }
         bulk_q_data.append(single_q_data)
 

src/cmd/flux-account-service.py

@@ -398,6 +398,7 @@ def add_queue(self, handle, watcher, msg, arg):
                 msg.payload["max_nodes_per_job"],
                 msg.payload["max_time_per_job"],
                 msg.payload["priority"],
+                msg.payload["max_running_jobs"],
             )
 
             payload = {"add_queue": val}
@@ -453,6 +454,7 @@ def edit_queue(self, handle, watcher, msg, arg):
                 msg.payload["max_nodes_per_job"],
                 msg.payload["max_time_per_job"],
                 msg.payload["priority"],
+                msg.payload["max_running_jobs"],
             )
 
             payload = {"edit_queue": val}

src/cmd/flux-account.py

@@ -417,6 +417,12 @@ def add_add_queue_arg(subparsers):
         default=0,
         metavar="PRIORITY",
     )
+    subparser_add_queue.add_argument(
+        "--max-running-jobs",
+        help="max number of running jobs an association can have in the queue",
+        default=100,
+        metavar="MAX_RUNNING_JOBS",
+    )
 
 
 def add_view_queue_arg(subparsers):
@@ -467,6 +473,13 @@ def add_edit_queue_arg(subparsers):
         default=None,
         metavar="PRIORITY",
     )
+    subparser_edit_queue.add_argument(
+        "--max-running-jobs",
+        type=int,
+        help="max number of running jobs an association can have in the queue",
+        default=None,
+        metavar="MAX_RUNNING_JOBS",
+    )
 
 
 def add_delete_queue_arg(subparsers):
@@ -763,6 +776,7 @@ def select_accounting_function(args, output_file, parser):
             "max_nodes_per_job": args.max_nodes_per_job,
             "max_time_per_job": args.max_time_per_job,
             "priority": args.priority,
+            "max_running_jobs": args.max_running_jobs,
         }
         return_val = flux.Flux().rpc("accounting.add_queue", data).get()
     elif args.func == "view_queue":
@@ -785,6 +799,7 @@ def select_accounting_function(args, output_file, parser):
             "max_nodes_per_job": args.max_nodes_per_job,
             "max_time_per_job": args.max_time_per_job,
             "priority": args.priority,
+            "max_running_jobs": args.max_running_jobs,
         }
         return_val = flux.Flux().rpc("accounting.edit_queue", data).get()
     elif args.func == "add_project":

src/plugins/accounting.cpp

@@ -84,9 +84,28 @@ json_t* Association::to_json () const
         }
     }
 
+    json_t *queue_usage_json = json_object ();
+    if (!queue_usage_json) {
+        json_decref (held_job_ids);
+        json_decref (user_queues);
+        json_decref (user_projects);
+        return nullptr;
+    }
+    for (const auto &entry : queue_usage) {
+        if (json_object_set_new (queue_usage_json,
+                                 entry.first.c_str (),
+                                 json_integer (entry.second)) < 0) {
+            json_decref (held_job_ids);
+            json_decref (user_queues);
+            json_decref (user_projects);
+            json_decref (queue_usage_json);
+            return nullptr;
+        }
+    }
+
     // 'o' steals the reference for both held_job_ids and user_queues
     json_t *u = json_pack ("{s:s, s:f, s:i, s:i, s:i, s:i,"
-                           " s:o, s:o, s:i, s:o, s:s, s:i, s:i}",
+                           " s:o, s:o, s:i, s:o, s:s, s:i, s:i, s:o}",
                            "bank_name", bank_name.c_str (),
                            "fairshare", fairshare,
                            "max_run_jobs", max_run_jobs,
@@ -99,7 +118,8 @@ json_t* Association::to_json () const
                            "projects", user_projects,
                            "def_project", def_project.c_str (),
                            "max_nodes", max_nodes,
-                           "active", active);
+                           "active", active,
+                           "queue_usage", queue_usage_json);
 
     if (!u)
         return nullptr;
@@ -228,3 +248,14 @@ int get_project_info (const char *project,
 
     return 0;
 }
+
+
+int max_run_jobs_per_queue (const std::map<std::string, Queue> &queues,
+                            const std::string &queue)
+{
+    auto it = queues.find (queue);
+    if (it == queues.end ())
+        return -1;
+
+    return it->second.max_running_jobs;
+}

src/plugins/accounting.hpp

@@ -27,6 +27,7 @@ extern "C" {
 #include <iterator>
 #include <sstream>
 #include <algorithm>
+#include <unordered_map>
 
 // all attributes are per-user/bank
 class Association {
@@ -45,6 +46,11 @@ class Association {
     std::vector<std::string> projects; // list of accessible projects
     std::string def_project;           // default project
     int max_nodes;                     // max num nodes across all running jobs
+    std::unordered_map<std::string, int>
+      queue_usage;                     // track num of running jobs per queue
+    std::unordered_map<std::string,
+                       std::vector<long int>>
+      queue_held_jobs;                // keep track of held job ID's per queue
 
     // methods
     json_t* to_json () const;    // convert object to JSON string
@@ -74,6 +80,7 @@ class Queue {
     int max_nodes_per_job;
     int max_time_per_job;
     int priority;
+    int max_running_jobs;
 };
 
 // get an Association object that points to user/bank in the users map;
@@ -110,4 +117,8 @@ int get_project_info (const char *project,
                       std::vector<std::string> &permissible_projects,
                       std::vector<std::string> projects);
 
+// fetch the max number of running jobs a queue can have per-association
+int max_run_jobs_per_queue (const std::map<std::string, Queue> &queues,
+                            const std::string &queue);
+
 #endif // ACCOUNTING_H