bank_table: add max-preempt-after column

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 = 29
+DB_SCHEMA_VERSION = 30
 
 PRIORITY_FACTORS = ["fairshare", "queue", "bank"]
 FSHARE_WEIGHT_DEFAULT = 100000
@@ -28,7 +28,7 @@ ASSOCIATION_TABLE = [
     "projects",
     "default_project",
 ]
-BANK_TABLE = ["bank_id", "bank", "active", "parent_bank", "shares", "job_usage", "priority"]
+BANK_TABLE = ["bank_id", "bank", "active", "parent_bank", "shares", "job_usage", "priority", "max_preempt_after"]
 QUEUE_TABLE = [
     "queue",
     "min_nodes_per_job",

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

@@ -16,6 +16,8 @@
 from fluxacct.accounting import formatter as fmt
 from fluxacct.accounting import sql_util as sql
 
+from flux.util import parse_fsd
+
 ###############################################################
 #                                                             #
 #                      Helper Functions                       #
@@ -84,7 +86,7 @@ def reactivate_bank(conn, cur, bank, parent_bank):
 ###############################################################
 
 
-def add_bank(conn, bank, shares, parent_bank="", priority=0.0):
+def add_bank(conn, bank, shares, parent_bank="", priority=0.0, max_preempt_after=None):
     cur = conn.cursor()
 
     if parent_bank == "":
@@ -112,20 +114,25 @@ def add_bank(conn, bank, shares, parent_bank="", priority=0.0):
         reactivate_bank(conn, cur, bank, parent_bank)
         return 0
 
+    if max_preempt_after:
+        # parse Flux Standard Duration (see RFC 23) and add it to the INSERT statement
+        duration = parse_fsd(max_preempt_after)
+        insert_stmt = (
+            "INSERT INTO bank_table "
+            "(bank, parent_bank, shares, priority, max_preempt_after) "
+            "VALUES (?, ?, ?, ?, ?)"
+        )
+        stmt_parameters = (bank, parent_bank, shares, priority, duration)
+    else:
+        insert_stmt = (
+            "INSERT INTO bank_table (bank, parent_bank, shares, priority) "
+            "VALUES (?, ?, ?, ?)"
+        )
+        stmt_parameters = (bank, parent_bank, shares, priority)
+
     # insert the bank values into the database
     try:
-        conn.execute(
-            """
-            INSERT INTO bank_table (
-                bank,
-                parent_bank,
-                shares,
-                priority
-            )
-            VALUES (?, ?, ?, ?)
-            """,
-            (bank, parent_bank, shares, priority),
-        )
+        conn.execute(insert_stmt, stmt_parameters)
         # commit changes
         conn.commit()
 
@@ -235,13 +242,15 @@ def edit_bank(
     shares=None,
     parent_bank=None,
     priority=None,
+    max_preempt_after=None,
 ):
     cur = conn.cursor()
     params = locals()
     editable_fields = [
         "shares",
         "parent_bank",
         "priority",
+        "max_preempt_after",
     ]
     for field in editable_fields:
         if params[field] is not None:
@@ -257,6 +266,13 @@ def edit_bank(
             if field == "shares":
                 if int(shares) <= 0:
                     raise ValueError("new shares amount must be >= 0")
+            if field == "max_preempt_after":
+                if max_preempt_after == "-1":
+                    duration = None
+                else:
+                    # convert FSD to seconds
+                    duration = parse_fsd(max_preempt_after)
+                params[field] = duration
 
             update_stmt = "UPDATE bank_table SET " + field
 

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

@@ -122,13 +122,14 @@ def create_db(
     conn.execute(
         """
             CREATE TABLE IF NOT EXISTS bank_table (
-                bank_id     integer PRIMARY KEY AUTOINCREMENT,
-                bank        text                NOT NULL,
-                active      int(11) DEFAULT 1   NOT NULL,
-                parent_bank text    DEFAULT '',
-                shares      int                 NOT NULL,
-                job_usage   real    DEFAULT 0.0 NOT NULL,
-                priority    real    DEFAULT 0.0 NOT NULL    ON CONFLICT REPLACE DEFAULT 0.0
+                bank_id           integer PRIMARY KEY AUTOINCREMENT,
+                bank              text                NOT NULL,
+                active            int(11) DEFAULT 1   NOT NULL,
+                parent_bank       text    DEFAULT '',
+                shares            int                 NOT NULL,
+                job_usage         real    DEFAULT 0.0 NOT NULL,
+                priority          real    DEFAULT 0.0 NOT NULL    ON CONFLICT REPLACE DEFAULT 0.0,
+                max_preempt_after real
         );"""
     )
     LOGGER.info("Created bank_table successfully")

src/cmd/flux-account-service.py

@@ -309,6 +309,7 @@ def add_bank(self, handle, watcher, msg, arg):
                 msg.payload["shares"],
                 msg.payload.get("parent_bank"),
                 msg.payload.get("priority"),
+                msg.payload.get("max_preempt_after"),
             )
 
             payload = {"add_bank": val}
@@ -341,6 +342,7 @@ def edit_bank(self, handle, watcher, msg, arg):
                 msg.payload.get("shares"),
                 msg.payload.get("parent_bank"),
                 msg.payload.get("priority"),
+                msg.payload.get("max_preempt_after"),
             )
 
             payload = {"edit_bank": val}

src/cmd/flux-account.py

@@ -442,6 +442,15 @@ def add_add_bank_arg(subparsers):
         help="an associated priority for jobs submitted under this bank",
         metavar="PRIORITY",
     )
+    subparser_add_bank.add_argument(
+        "--max-preempt-after",
+        help=(
+            "max amount of time until a job running under this bank can become "
+            "preemptible; duration must be in Flux Standard Duration"
+        ),
+        type=str,
+        metavar="MAX_PREEMPT_AFTER",
+    )
 
 
 def add_view_bank_arg(subparsers):
@@ -547,6 +556,15 @@ def add_edit_bank_arg(subparsers):
         help="an associated priority for jobs submitted under this bank",
         metavar="PRIORITY",
     )
+    subparser_edit_bank.add_argument(
+        "--max-preempt-after",
+        help=(
+            "max amount of time until a job running under this bank can become "
+            "preemptible; duration must be in Flux Standard Duration"
+        ),
+        type=str,
+        metavar="MAX_PREEMPT_AFTER",
+    )
 
 
 def add_list_banks_arg(subparsers):

t/Makefile.am

@@ -61,6 +61,7 @@ TESTSCRIPTS = \
 	t1056-mf-priority-urgency-factor.t \
 	t1057-flux-account-jobs.t \
 	t1058-flux-account-visuals.t \
+	t1059-issue553.t \
 	t5000-valgrind.t \
 	python/t1000-example.py \
 	python/t1001_db.py \

t/expected/flux_account/A_bank.expected

@@ -1,6 +1,6 @@
-bank_id | bank | active | parent_bank | shares | job_usage | priority
---------+------+--------+-------------+--------+-----------+---------
-2       | A    | 1      | root        | 1      | 0.0       | 0.0     
+bank_id | bank | active | parent_bank | shares | job_usage | priority | max_preempt_after
+--------+------+--------+-------------+--------+-----------+----------+------------------
+2       | A    | 1      | root        | 1      | 0.0       | 0.0      | None             
 
 username | default_bank | shares | job_usage | fairshare
 ---------+--------------+--------+-----------+----------

t/expected/flux_account/F_bank_users.expected

@@ -1,5 +1,5 @@
-bank_id | bank | active | parent_bank | shares | job_usage | priority
---------+------+--------+-------------+--------+-----------+---------
-7       | F    | 1      | D           | 1      | 0.0       | 0.0     
+bank_id | bank | active | parent_bank | shares | job_usage | priority | max_preempt_after
+--------+------+--------+-------------+--------+-----------+----------+------------------
+7       | F    | 1      | D           | 1      | 0.0       | 0.0      | None             
 
 no users under F

t/expected/flux_account/deleted_bank.expected

@@ -1,3 +1,3 @@
-bank_id        bank           active         parent_bank    shares         
-4              C              0              root           50             
+bank_id        bank           active         parent_bank    shares         job_usage      max_preempt_after
+4              C              0              root           50             0.0            None           
 

t/expected/flux_account/root_bank.expected

@@ -6,6 +6,7 @@
     "parent_bank": "",
     "shares": 1,
     "job_usage": 0.0,
-    "priority": 0.0
+    "priority": 0.0,
+    "max_preempt_after": null
   }
 ]

t/python/t1008_banks_output.py

@@ -61,7 +61,8 @@ def test_list_banks_default(self):
             "parent_bank": "",
             "shares": 1,
             "job_usage": 0.0,
-            "priority": 0.0
+            "priority": 0.0,
+            "max_preempt_after": null
           },
           {
             "bank_id": 2,
@@ -70,7 +71,8 @@ def test_list_banks_default(self):
             "parent_bank": "root",
             "shares": 1,
             "job_usage": 0.0,
-            "priority": 0.0
+            "priority": 0.0,
+            "max_preempt_after": null
           }
         ]
         """
@@ -218,10 +220,10 @@ def test_list_banks_custom_six(self):
     def test_list_banks_table_default(self):
         expected = textwrap.dedent(
             """\
-        bank_id | bank | active | parent_bank | shares | job_usage | priority
-        --------+------+--------+-------------+--------+-----------+---------
-        1       | root | 1      |             | 1      | 0.0       | 0.0     
-        2       | A    | 1      | root        | 1      | 0.0       | 0.0       
+        bank_id | bank | active | parent_bank | shares | job_usage | priority | max_preempt_after
+        --------+------+--------+-------------+--------+-----------+----------+------------------
+        1       | root | 1      |             | 1      | 0.0       | 0.0      | None             
+        2       | A    | 1      | root        | 1      | 0.0       | 0.0      | None
         """
         )
         test = b.list_banks(conn)

t/t1059-issue553.t

@@ -0,0 +1,73 @@
+#!/bin/bash
+
+test_description='test max-preempt-after values when adding/editing banks'
+
+. `dirname $0`/sharness.sh
+DB_PATH=$(pwd)/FluxAccountingTest.db
+
+export TEST_UNDER_FLUX_NO_JOB_EXEC=y
+export TEST_UNDER_FLUX_SCHED_SIMPLE_MODE="limited=1"
+test_under_flux 1 job
+
+flux setattr log-stderr-level 1
+
+test_expect_success 'create flux-accounting DB' '
+	flux account -p ${DB_PATH} create-db
+'
+
+test_expect_success 'start flux-accounting service' '
+	flux account-service -p ${DB_PATH} -t
+'
+
+test_expect_success 'add root bank to the DB' '
+	flux account add-bank root 1
+'
+
+test_expect_success 'add a sub bank with max-preempt in seconds' '
+	flux account add-bank --parent-bank=root --max-preempt-after=60s A 1 &&
+	flux account view-bank A > A.test &&
+	grep "\"max_preempt_after\": 60.0" A.test
+'
+
+test_expect_success 'add a sub bank with max-preempt in hours' '
+	flux account add-bank --parent-bank=root --max-preempt-after=1h B 1 &&
+	flux account view-bank B > B.test &&
+	grep "\"max_preempt_after\": 3600.0" B.test
+'
+
+test_expect_success 'add a sub bank with max-preempt in days' '
+	flux account add-bank --parent-bank=root --max-preempt-after=1d C 1 &&
+	flux account view-bank C > C.test &&
+	grep "\"max_preempt_after\": 86400.0" C.test
+'
+
+test_expect_success 'add a sub bank with no max-preempt specified' '
+	flux account add-bank --parent-bank=root D 1 &&
+	flux account view-bank D > D.test &&
+	grep "\"max_preempt_after\": null" D.test
+'
+
+test_expect_success 'trying to add a sub bank with a bad FSD will raise an error' '
+	test_must_fail flux account add-bank \
+		--parent-bank=root \
+		--max-preempt-after=foo E 1 > error.out 2>&1 &&
+	grep "add-bank: ValueError: invalid Flux standard duration" error.out
+'
+
+test_expect_success 'edit max-preempt-after for a bank' '
+	flux account edit-bank A --max-preempt-after=200s &&
+	flux account view-bank A > A_edited.test &&
+	grep "\"max_preempt_after\": 200.0" A_edited.test
+'
+
+test_expect_success 'clear max-preempt-after for a bank' '
+	flux account edit-bank A --max-preempt-after=-1 &&
+	flux account view-bank A > A_cleared.test &&
+	grep "\"max_preempt_after\": null" A_cleared.test
+'
+
+test_expect_success 'shut down flux-accounting service' '
+	flux python -c "import flux; flux.Flux().rpc(\"accounting.shutdown_service\").get()"
+'
+
+test_done