Merge pull request #560 from cmoussa1/add.max-cores.attr

`association_table`: add `max_cores` attribute, send information to plugin

src/bindings/python/fluxacct/accounting/__init__.py.in

@@ -1,6 +1,6 @@
 DB_DIR = "@X_LOCALSTATEDIR@/lib/flux/"
 DB_PATH = "@X_LOCALSTATEDIR@/lib/flux/FluxAccounting.db"
-DB_SCHEMA_VERSION = 24
+DB_SCHEMA_VERSION = 25
 
 # flux-accounting DB table column names
 ASSOCIATION_TABLE = [
@@ -17,6 +17,7 @@ ASSOCIATION_TABLE = [
     "max_running_jobs",
     "max_active_jobs",
     "max_nodes",
+    "max_cores",
     "queues",
     "projects",
     "default_project",

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

@@ -105,6 +105,7 @@ def create_db(
                 max_running_jobs int(11)     DEFAULT 5              NOT NULL    ON CONFLICT REPLACE DEFAULT 5,
                 max_active_jobs  int(11)     DEFAULT 7              NOT NULL    ON CONFLICT REPLACE DEFAULT 7,
                 max_nodes        int(11)     DEFAULT 2147483647     NOT NULL    ON CONFLICT REPLACE DEFAULT 2147483647,
+                max_cores        int(11)     DEFAULT 2147483647     NOT NULL    ON CONFLICT REPLACE DEFAULT 2147483647,
                 queues           tinytext    DEFAULT ''             NOT NULL    ON CONFLICT REPLACE DEFAULT '',
                 projects         tinytext    DEFAULT '*'            NOT NULL    ON CONFLICT REPLACE DEFAULT '*',
                 default_project  tinytext    DEFAULT '*'            NOT NULL    ON CONFLICT REPLACE DEFAULT '*',

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

@@ -274,6 +274,7 @@ def add_user(
     max_running_jobs=5,
     max_active_jobs=7,
     max_nodes=2147483647,
+    max_cores=2147483647,
     queues="",
     projects="*",
 ):
@@ -330,8 +331,9 @@ def add_user(
             INSERT INTO association_table (creation_time, mod_time, username,
                                            userid, bank, default_bank, shares,
                                            max_running_jobs, max_active_jobs,
-                                           max_nodes, queues, projects, default_project)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                                           max_nodes, max_cores, queues, projects,
+                                           default_project)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 int(time.time()),
@@ -344,6 +346,7 @@ def add_user(
                 max_running_jobs,
                 max_active_jobs,
                 max_nodes,
+                max_cores,
                 queues,
                 projects,
                 default_project,
@@ -417,6 +420,8 @@ def edit_user(conn, username, bank=None, **kwargs):
             can have at any given time.
         max_nodes: The man number of nodes an association can have across all of their
             running jobs.
+        max_cores: The max number of cores an association can have across all of their
+            running jobs.
         queues: A comma-separated list of all of the queues an association can run jobs
             under.
         projects: A comma-separated list of all of the projects an association can run jobs
@@ -437,6 +442,7 @@ def edit_user(conn, username, bank=None, **kwargs):
         "max_running_jobs",
         "max_active_jobs",
         "max_nodes",
+        "max_cores",
         "queues",
         "projects",
         "default_project",

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

@@ -78,7 +78,7 @@ def bulk_update(path):
     for row in cur.execute(
         """SELECT userid, bank, default_bank,
            fairshare, max_running_jobs, max_active_jobs,
-           queues, active, projects, default_project, max_nodes
+           queues, active, projects, default_project, max_nodes, max_cores
            FROM association_table"""
     ):
         # create a JSON payload with the results of the query
@@ -94,6 +94,7 @@ def bulk_update(path):
             "projects": str(row[8]),
             "def_project": str(row[9]),
             "max_nodes": int(row[10]),
+            "max_cores": int(row[11]),
         }
         bulk_user_data.append(single_user_data)
 
@@ -159,6 +160,7 @@ def send_instance_owner_info():
         "projects": "*",
         "def_project": "*",
         "max_nodes": 1000000,
+        "max_cores": 1000000,
     }
 
     flux.Flux().rpc(

src/cmd/flux-account-service.py

@@ -179,6 +179,7 @@ def add_user(self, handle, watcher, msg, arg):
                 max_running_jobs=msg.payload["max_running_jobs"],
                 max_active_jobs=msg.payload["max_active_jobs"],
                 max_nodes=msg.payload["max_nodes"],
+                max_cores=msg.payload["max_cores"],
                 queues=msg.payload["queues"],
                 projects=msg.payload["projects"],
             )
@@ -223,6 +224,7 @@ def edit_user(self, handle, watcher, msg, arg):
                 max_running_jobs=msg.payload["max_running_jobs"],
                 max_active_jobs=msg.payload["max_active_jobs"],
                 max_nodes=msg.payload["max_nodes"],
+                max_cores=msg.payload["max_cores"],
                 queues=msg.payload["queues"],
                 projects=msg.payload["projects"],
                 default_project=msg.payload["default_project"],

src/cmd/flux-account.py

@@ -54,7 +54,7 @@ def add_view_user_arg(subparsers):
         metavar=(
             "CREATION_TIME,MOD_TIME,ACTIVE,USERNAME,USERID,BANK,DEFAULT_BANK,"
             "SHARES,JOB_USAGE,FAIRSHARE,MAX_RUNNING_JOBS,MAX_ACTIVE_JOBS,MAX_NODES,"
-            "QUEUES,PROJECTS,DEFAULT_PROJECT"
+            "MAX_CORES,QUEUES,PROJECTS,DEFAULT_PROJECT"
         ),
     )
 
@@ -108,6 +108,12 @@ def add_add_user_arg(subparsers):
         default=2147483647,
         metavar="MAX_NODES",
     )
+    subparser_add_user.add_argument(
+        "--max-cores",
+        help="max number of cores a user can have across all of their running jobs",
+        default=2147483647,
+        metavar="MAX_CORES",
+    )
     subparser_add_user.add_argument(
         "--queues",
         help="queues the user is allowed to run jobs in",
@@ -191,6 +197,12 @@ def add_edit_user_arg(subparsers):
         default=None,
         metavar="MAX_NODES",
     )
+    subparser_edit_user.add_argument(
+        "--max-cores",
+        help="max number of cores a user can have across all of their running jobs",
+        default=None,
+        metavar="MAX_CORES",
+    )
     subparser_edit_user.add_argument(
         "--queues",
         help="queues the user is allowed to run jobs in",

src/plugins/accounting.cpp

@@ -85,8 +85,8 @@ json_t* Association::to_json () const
     }
 
     // '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}",
+    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:i}",
                            "bank_name", bank_name.c_str (),
                            "fairshare", fairshare,
                            "max_run_jobs", max_run_jobs,
@@ -99,6 +99,7 @@ json_t* Association::to_json () const
                            "projects", user_projects,
                            "def_project", def_project.c_str (),
                            "max_nodes", max_nodes,
+                           "max_cores", max_cores,
                            "active", active);
 
     if (!u)

src/plugins/accounting.hpp

@@ -45,6 +45,7 @@ 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
+    int max_cores;                     // max num cores across all running jobs
 
     // methods
     json_t* to_json () const;    // convert object to JSON string

src/plugins/mf_priority.cpp

@@ -187,6 +187,7 @@ static void add_special_association (flux_plugin_t *p, flux_t *h, int userid)
     a->active = 1;
     a->held_jobs = std::vector<long int>();
     a->max_nodes = INT16_MAX;
+    a->max_cores = INT16_MAX;
 
     if (flux_jobtap_job_aux_set (p,
                                  FLUX_JOBTAP_CURRENT_JOB,
@@ -277,7 +278,7 @@ static void rec_update_cb (flux_t *h,
                            void *arg)
 {
     char *bank, *def_bank, *assoc_queues, *assoc_projects, *def_project = NULL;
-    int uid, max_running_jobs, max_active_jobs, max_nodes = 0;
+    int uid, max_running_jobs, max_active_jobs, max_nodes, max_cores = 0;
     double fshare = 0.0;
     json_t *data, *jtemp = NULL;
     json_error_t error;
@@ -301,7 +302,7 @@ static void rec_update_cb (flux_t *h,
 
         if (json_unpack_ex (el, &error, 0,
                             "{s:i, s:s, s:s, s:F, s:i,"
-                            " s:i, s:s, s:i, s:s, s:s, s:i}",
+                            " s:i, s:s, s:i, s:s, s:s, s:i, s:i}",
                             "userid", &uid,
                             "bank", &bank,
                             "def_bank", &def_bank,
@@ -312,7 +313,8 @@ static void rec_update_cb (flux_t *h,
                             "active", &active,
                             "projects", &assoc_projects,
                             "def_project", &def_project,
-                            "max_nodes", &max_nodes) < 0)
+                            "max_nodes", &max_nodes,
+                            "max_cores", &max_cores) < 0)
             flux_log (h, LOG_ERR, "mf_priority unpack: %s", error.text);
 
         Association *b;
@@ -325,6 +327,7 @@ static void rec_update_cb (flux_t *h,
         b->active = active;
         b->def_project = def_project;
         b->max_nodes = max_nodes;
+        b->max_cores = max_cores;
 
         // split queues comma-delimited string and add it to b->queues vector
         b->queues.clear ();

src/plugins/test/accounting_test01.cpp

@@ -55,7 +55,8 @@ void add_user_to_map (
         a.active,
         a.projects,
         a.def_project,
-        a.max_nodes
+        a.max_nodes,
+        a.max_cores
     };
 }
 
@@ -67,9 +68,9 @@ void initialize_map (
     std::map<int, std::map<std::string, Association>> &users)
 {
     Association user1 = {"bank_A", 0.5, 5, 0, 7, 0, {},
-                         {}, 0, 1, {"*"}, "*", 2147483647};
+                         {}, 0, 1, {"*"}, "*", 2147483647, 2147483647};
     Association user2 = {"bank_A", 0.5, 5, 0, 7, 0, {},
-                         {}, 0, 1, {"*"}, "*", 2147483647};
+                         {}, 0, 1, {"*"}, "*", 2147483647, 2147483647};
 
     add_user_to_map (users, 1001, "bank_A", user1);
     users_def_bank[1001] = "bank_A";
@@ -271,7 +272,7 @@ static void test_check_map_dne_true ()
     users_def_bank.clear ();
 
     Association tmp_user = {"DNE", 0.5, 5, 0, 7, 0, {},
-                            {}, 0, 1, {"*"}, "*", 2147483647};
+                            {}, 0, 1, {"*"}, "*", 2147483647, 2147483647};
     add_user_to_map (users, 9999, "DNE", tmp_user);
     users_def_bank[9999] = "DNE";
 

t/t1017-update-db.t

@@ -83,6 +83,7 @@ test_expect_success 'get all the columns of the updated table in the DB and chec
 	max_running_jobs
 	max_active_jobs
 	max_nodes
+	max_cores
 	queues
 	projects
 	default_project