Skip to content

Commit c0c9423

Browse files
chu11chu11
committed
job-list: use constraints to filter jobs
Problem: Using RFC31 constraints to match jobs would allow us to support many new filtering and query opportunities in job-list. Solution: Convert job-list queries to use constraints for filtering instead of the earlier solution. This change breaks the old filtering RPC protocol. The "userid", "states", "results", "name", and "queue" RPC fields are no longer supported. Update callers in libjob, flux-job, flux-top, job-archive, python JobList and in the testsuite.
1 parent 6150f1b commit c0c9423

File tree

7 files changed

+114
-128
lines changed

7 files changed

+114
-128
lines changed

‎src/bindings/python/flux/job/list.py‎

Lines changed: 20 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -45,18 +45,31 @@ def job_list(
4545
name=None,
4646
queue=None,
4747
):
48+
# N.B. an "and" operation with no values returns everything
49+
constraint = {"and": []}
50+
if userid != flux.constants.FLUX_USERID_UNKNOWN:
51+
constraint["and"].append({"userid": [userid]})
52+
if name:
53+
constraint["and"].append({"name": [name]})
54+
if queue:
55+
constraint["and"].append({"queue": [queue]})
56+
if states and results:
57+
if states & flux.constants.FLUX_JOB_STATE_INACTIVE:
58+
states &= ~flux.constants.FLUX_JOB_STATE_INACTIVE
59+
tmp = {"or": []}
60+
tmp["or"].append({"states": [states]})
61+
tmp["or"].append({"results": [results]})
62+
constraint["and"].append(tmp)
63+
elif states:
64+
constraint["and"].append({"states": [states]})
65+
elif results:
66+
constraint["and"].append({"results": [results]})
4867
payload = {
4968
"max_entries": int(max_entries),
5069
"attrs": attrs,
51-
"userid": int(userid),
52-
"states": states,
53-
"results": results,
5470
"since": since,
71+
"constraint": constraint,
5572
}
56-
if name:
57-
payload["name"] = name
58-
if queue:
59-
payload["queue"] = queue
6073
return JobListRPC(flux_handle, "job-list.list", payload)
6174

6275

@@ -239,8 +252,6 @@ def add_filter(self, fname):
239252
if fname in self.STATES:
240253
self.states |= self.STATES[fname]
241254
elif fname in self.RESULTS:
242-
# Must specify "inactive" to get results:
243-
self.states |= self.STATES["inactive"]
244255
self.results |= self.RESULTS[fname]
245256
else:
246257
raise ValueError(f"Invalid filter specified: {fname}")

‎src/cmd/flux-job.c‎

Lines changed: 16 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1331,6 +1331,7 @@ int cmd_list (optparse_t *p, int argc, char **argv)
13311331
json_t *value;
13321332
uint32_t userid;
13331333
int states = 0;
1334+
json_t *c;
13341335

13351336
if (isatty (STDOUT_FILENO)) {
13361337
fprintf (stderr,
@@ -1359,16 +1360,20 @@ int cmd_list (optparse_t *p, int argc, char **argv)
13591360
else
13601361
userid = getuid ();
13611362

1363+
if (!(c = json_pack ("{ s:[ {s:[i]}, {s:[i]} ] }",
1364+
"and",
1365+
"userid", userid,
1366+
"states", states)))
1367+
log_msg_exit ("failed to construct constraint object");
1368+
13621369
if (!(f = flux_rpc_pack (h,
13631370
"job-list.list",
13641371
FLUX_NODEID_ANY,
13651372
0,
1366-
"{s:i s:[s] s:i s:i s:i}",
1373+
"{s:i s:[s] s:o}",
13671374
"max_entries", max_entries,
13681375
"attrs", "all",
1369-
"userid", userid,
1370-
"states", states,
1371-
"results", 0)))
1376+
"constraint", c)))
13721377
log_err_exit ("flux_rpc_pack");
13731378
if (flux_rpc_get_unpack (f, "{s:o}", "jobs", &jobs) < 0)
13741379
log_err_exit ("flux job-list.list");
@@ -1396,6 +1401,7 @@ int cmd_list_inactive (optparse_t *p, int argc, char **argv)
13961401
json_t *jobs;
13971402
size_t index;
13981403
json_t *value;
1404+
json_t *c;
13991405

14001406
if (isatty (STDOUT_FILENO)) {
14011407
fprintf (stderr,
@@ -1410,17 +1416,18 @@ int cmd_list_inactive (optparse_t *p, int argc, char **argv)
14101416
if (!(h = flux_open (NULL, 0)))
14111417
log_err_exit ("flux_open");
14121418

1419+
if (!(c = json_pack ("{s:[i]}", "states", FLUX_JOB_STATE_INACTIVE)))
1420+
log_msg_exit ("failed to construct constraint object");
1421+
14131422
if (!(f = flux_rpc_pack (h,
14141423
"job-list.list",
14151424
FLUX_NODEID_ANY,
14161425
0,
1417-
"{s:i s:f s:i s:i s:i s:[s]}",
1426+
"{s:i s:f s:[s] s:o}",
14181427
"max_entries", max_entries,
14191428
"since", since,
1420-
"userid", FLUX_USERID_UNKNOWN,
1421-
"states", FLUX_JOB_STATE_INACTIVE,
1422-
"results", 0,
1423-
"attrs", "all")))
1429+
"attrs", "all",
1430+
"constraint", c)))
14241431
log_err_exit ("flux_rpc_pack");
14251432
if (flux_rpc_get_unpack (f, "{s:o}", "jobs", &jobs) < 0)
14261433
log_err_exit ("flux job-list.list");

‎src/cmd/top/joblist_pane.c‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -356,11 +356,10 @@ void joblist_pane_query (struct joblist_pane *joblist)
356356
"job-list.list",
357357
0,
358358
0,
359-
"{s:i s:i s:i s:i s:[s,s,s,s,s,s,s,s]}",
359+
"{s:i s:{s:[i]} s:[s,s,s,s,s,s,s,s]}",
360360
"max_entries", win_dim.y_length - 1,
361-
"userid", FLUX_USERID_UNKNOWN,
361+
"constraint",
362362
"states", FLUX_JOB_STATE_RUNNING,
363-
"results", 0,
364363
"attrs",
365364
"annotations",
366365
"userid",

‎src/common/libjob/list.c‎

Lines changed: 22 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ flux_future_t *flux_job_list (flux_t *h,
2525
{
2626
flux_future_t *f;
2727
json_t *o = NULL;
28+
json_t *c = NULL;
2829
int valid_states = (FLUX_JOB_STATE_PENDING
2930
| FLUX_JOB_STATE_RUNNING
3031
| FLUX_JOB_STATE_INACTIVE);
@@ -37,15 +38,22 @@ flux_future_t *flux_job_list (flux_t *h,
3738
errno = EINVAL;
3839
return NULL;
3940
}
41+
if (!(c = json_pack ("{ s:[ {s:[i]}, {s:[i]} ] }",
42+
"and",
43+
"userid", userid,
44+
"states", states ? states : valid_states))) {
45+
json_decref (o);
46+
errno = ENOMEM;
47+
return NULL;
48+
}
4049
if (!(f = flux_rpc_pack (h, "job-list.list", FLUX_NODEID_ANY, 0,
41-
"{s:i s:o s:i s:i s:i}",
50+
"{s:i s:o s:o}",
4251
"max_entries", max_entries,
4352
"attrs", o,
44-
"userid", userid,
45-
"states", states,
46-
"results", 0))) {
53+
"constraint", c))) {
4754
saved_errno = errno;
4855
json_decref (o);
56+
json_decref (c);
4957
errno = saved_errno;
5058
return NULL;
5159
}
@@ -59,23 +67,28 @@ flux_future_t *flux_job_list_inactive (flux_t *h,
5967
{
6068
flux_future_t *f;
6169
json_t *o = NULL;
70+
json_t *c = NULL;
6271
int saved_errno;
6372

6473
if (!h || max_entries < 0 || since < 0. || !attrs_json_str
6574
|| !(o = json_loads (attrs_json_str, 0, NULL))) {
6675
errno = EINVAL;
6776
return NULL;
6877
}
78+
if (!(c = json_pack ("{s:[i]}", "states", FLUX_JOB_STATE_INACTIVE))) {
79+
json_decref (o);
80+
errno = ENOMEM;
81+
return NULL;
82+
}
6983
if (!(f = flux_rpc_pack (h, "job-list.list", FLUX_NODEID_ANY, 0,
70-
"{s:i s:f s:i s:i s:i s:o}",
84+
"{s:i s:f s:o s:o}",
7185
"max_entries", max_entries,
7286
"since", since,
73-
"userid", FLUX_USERID_UNKNOWN,
74-
"states", FLUX_JOB_STATE_INACTIVE,
75-
"results", 0,
76-
"attrs", o))) {
87+
"attrs", o,
88+
"constraint", c))) {
7789
saved_errno = errno;
7890
json_decref (o);
91+
json_decref (c);
7992
errno = saved_errno;
8093
return NULL;
8194
}

‎src/modules/job-archive/job-archive.c‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -505,19 +505,18 @@ void job_archive_cb (flux_reactor_t *r,
505505
"job-list.list",
506506
FLUX_NODEID_ANY,
507507
0,
508-
"{s:i s:f s:i s:i s:i s:[ssssss]}",
508+
"{s:i s:f s:[ssssss] s:{s:[i]}}",
509509
"max_entries", 0,
510510
"since", ctx->since,
511-
"userid", FLUX_USERID_UNKNOWN,
512-
"states", FLUX_JOB_STATE_INACTIVE,
513-
"results", 0,
514511
"attrs",
515512
"userid",
516513
"ranks",
517514
"t_submit",
518515
"t_run",
519516
"t_cleanup",
520-
"t_inactive"))) {
517+
"t_inactive",
518+
"constraint",
519+
"states", FLUX_JOB_STATE_INACTIVE))) {
521520
flux_log_error (ctx->h, "%s: flux_rpc_pack", __FUNCTION__);
522521
return;
523522
}

0 commit comments

Comments
 (0)