Merge pull request #5712 from chu11/issue5688_job_list_missing_queue_stats

job-list: initialize queue stats

Author: mergify[bot]

src/bindings/python/flux/kvs.py

@@ -795,6 +795,7 @@ def __del__(self):
         try:
             super().__del__()
         except AttributeError:
+            # not an error if super did not implement
             pass
 
     def __init__(self, future_handle):

src/modules/job-list/job-list.c

@@ -107,6 +107,30 @@ static void disconnect_cb (flux_t *h,
     job_stats_disconnect (ctx->jsctx->statsctx, msg);
 }
 
+static void config_reload_cb (flux_t *h,
+                              flux_msg_handler_t *mh,
+                              const flux_msg_t *msg,
+                              void *arg)
+{
+    struct list_ctx *ctx = arg;
+    const flux_conf_t *conf;
+    flux_error_t error;
+    const char *errstr = NULL;
+
+    if (flux_conf_reload_decode (msg, &conf) < 0)
+        goto error;
+    if (job_state_config_reload (ctx->jsctx, conf, &error) < 0) {
+        errstr = error.text;
+        goto error;
+    }
+    if (flux_respond (h, msg, NULL) < 0)
+        flux_log_error (h, "error responding to config-reload request");
+    return;
+error:
+    if (flux_respond_error (h, msg, errno, errstr) < 0)
+        flux_log_error (h, "error responding to config-reload request");
+}
+
 static const struct flux_msg_handler_spec htab[] = {
     { .typemask     = FLUX_MSGTYPE_REQUEST,
       .topic_glob   = "job-list.list",
@@ -148,6 +172,12 @@ static const struct flux_msg_handler_spec htab[] = {
       .cb           = disconnect_cb,
       .rolemask     = FLUX_ROLE_USER,
     },
+    {
+      .typemask     = FLUX_MSGTYPE_REQUEST,
+      .topic_glob   = "job-list.config-reload",
+      .cb           = config_reload_cb,
+      .rolemask     = 0
+    },
     FLUX_MSGHANDLER_TABLE_END,
 };
 

src/modules/job-list/job_state.c

@@ -1798,6 +1798,13 @@ void job_state_destroy (void *data)
     }
 }
 
+int job_state_config_reload (struct job_state_ctx *jsctx,
+                             const flux_conf_t *conf,
+                             flux_error_t *errp)
+{
+    return job_stats_config_reload (jsctx->statsctx, conf, errp);
+}
+
 /*
  * vi:tabstop=4 shiftwidth=4 expandtab
  */

src/modules/job-list/job_state.h

@@ -72,6 +72,10 @@ void job_state_unpause_cb (flux_t *h, flux_msg_handler_t *mh,
 
 int job_state_init_from_kvs (struct job_state_ctx *jsctx);
 
+int job_state_config_reload (struct job_state_ctx *jsctx,
+                             const flux_conf_t *conf,
+                             flux_error_t *errp);
+
 #endif /* ! _FLUX_JOB_LIST_JOB_STATE_H */
 
 /*

src/modules/job-list/stats.c

@@ -45,18 +45,19 @@ static void free_wrapper (void **item)
 }
 
 static struct job_stats *queue_stats_lookup (struct job_stats_ctx *statsctx,
-                                             struct job *job)
+                                             const char *name,
+                                             bool create_if_missing)
 {
     struct job_stats *stats = NULL;
 
-    if (!job->queue)
+    if (!name)
         return NULL;
 
-    stats = zhashx_lookup (statsctx->queue_stats, job->queue);
-    if (!stats) {
+    stats = zhashx_lookup (statsctx->queue_stats, name);
+    if (!stats && create_if_missing) {
         if (!(stats = calloc (1, sizeof (*stats))))
             return NULL;
-        (void)zhashx_insert (statsctx->queue_stats, job->queue, stats);
+        (void)zhashx_insert (statsctx->queue_stats, name, stats);
     }
     return stats;
 }
@@ -127,7 +128,7 @@ void job_stats_update (struct job_stats_ctx *statsctx,
 
     stats_update (&statsctx->all, job, newstate);
 
-    if ((stats = queue_stats_lookup (statsctx, job)))
+    if ((stats = queue_stats_lookup (statsctx, job->queue, true)))
         stats_update (stats, job, newstate);
 
     arm_timer (statsctx);
@@ -138,7 +139,7 @@ void job_stats_add_queue (struct job_stats_ctx *statsctx,
 {
     struct job_stats *stats;
 
-    if ((stats = queue_stats_lookup (statsctx, job)))
+    if ((stats = queue_stats_lookup (statsctx, job->queue, true)))
         stats_add (stats, job, job->state);
 
     arm_timer (statsctx);
@@ -174,9 +175,16 @@ void job_stats_remove_queue (struct job_stats_ctx *statsctx,
 {
     struct job_stats *stats;
 
-    if ((stats = queue_stats_lookup (statsctx, job)))
-        stats_remove (stats, job);
+    if (!(stats = queue_stats_lookup (statsctx, job->queue, false))) {
+        if (job->queue)
+            flux_log (statsctx->h,
+                      LOG_DEBUG,
+                      "no queue stats for %s",
+                      job->queue);
+        return;
+    }
 
+    stats_remove (stats, job);
     arm_timer (statsctx);
 }
 
@@ -211,9 +219,16 @@ void job_stats_purge (struct job_stats_ctx *statsctx, struct job *job)
 
     stats_purge (&statsctx->all, job);
 
-    if ((stats = queue_stats_lookup (statsctx, job)))
-        stats_purge (stats, job);
+    if (!(stats = queue_stats_lookup (statsctx, job->queue, false))) {
+        if (job->queue)
+            flux_log (statsctx->h,
+                      LOG_DEBUG,
+                      "no queue stats for %s",
+                      job->queue);
+        return;
+    }
 
+    stats_purge (stats, job);
     arm_timer (statsctx);
 }
 
@@ -408,6 +423,36 @@ static void job_stats_cb (flux_t *h,
         flux_log_error (h, "error responding to job-stats request");
 }
 
+static int config_parse_queues (struct job_stats_ctx *statsctx,
+                                const flux_conf_t *conf,
+                                flux_error_t *errp)
+{
+    json_t *queues;
+
+    if (flux_conf_unpack (conf, NULL, "{s:o}", "queues", &queues) == 0
+        && json_object_size (queues) > 0) {
+        const char *name;
+        json_t *value;
+        json_object_foreach (queues, name, value) {
+            /* setup initial queue stats, so that user gets initial
+             * stats before first job is submitted to the queue */
+            if (!queue_stats_lookup (statsctx, name, true)) {
+                flux_log_error (statsctx->h, "queue_stats_lookup");
+                return -1;
+            }
+        }
+    }
+
+    return 0;
+}
+
+int job_stats_config_reload (struct job_stats_ctx *statsctx,
+                             const flux_conf_t *conf,
+                             flux_error_t *errp)
+{
+    return config_parse_queues (statsctx, conf, errp);
+}
+
 static const struct flux_msg_handler_spec htab[] = {
     { .typemask     = FLUX_MSGTYPE_REQUEST,
       .topic_glob   = "job-list.job-stats",
@@ -420,6 +465,7 @@ static const struct flux_msg_handler_spec htab[] = {
 struct job_stats_ctx *job_stats_ctx_create (flux_t *h)
 {
     struct job_stats_ctx *statsctx = NULL;
+    flux_error_t error;
 
     if (!(statsctx = calloc (1, sizeof (*statsctx))))
         return NULL;
@@ -441,6 +487,13 @@ struct job_stats_ctx *job_stats_ctx_create (flux_t *h)
                                                        statsctx)))
         goto error;
 
+    if (config_parse_queues (statsctx,
+                             flux_get_conf (statsctx->h),
+                             &error) < 0) {
+        flux_log (statsctx->h, LOG_ERR, "%s", error.text);
+        goto error;
+    }
+
     return statsctx;
 
 error:

src/modules/job-list/stats.h

@@ -51,6 +51,10 @@ void job_stats_disconnect (struct job_stats_ctx *statsctx,
  */
 int job_stats_watchers (struct job_stats_ctx *statsctx);
 
+int job_stats_config_reload (struct job_stats_ctx *statsctx,
+                             const flux_conf_t *conf,
+                             flux_error_t *errp);
+
 #endif /* ! _FLUX_JOB_LIST_JOB_STATS_H */
 
 // vi: ts=4 sw=4 expandtab

t/t2260-job-list.t

@@ -2132,10 +2132,14 @@ test_expect_success 'job-list parses flux-restart events' '
 # jobspec-update event testing
 #
 
+# N.B. "defaultqueue" will habe jobs "submitted" to it, but updates
+# make it appear that jobs were not submitted to it.  "nosubmitqueue"
+# will never have jobs submitted to it.
 test_expect_success 'configure update queues' '
 	flux config load <<-EOT &&
 	[queues.defaultqueue]
 	[queues.updatequeue]
+	[queues.nosubmitqueue]
 	EOT
 	flux queue start --all
 '
@@ -2173,9 +2177,15 @@ test_expect_success 'job-list returns expected jobspec changes' '
 	flux job list -s inactive | grep $(cat update1.id) | jq -e ".duration == 1000.0"
 '
 
+#
+# "nosubmitqueue" never has job submitted to it, make sure stats
+# string is non-empty and filled with zeros.
+#
 test_expect_success 'job stats lists jobs in correct state in each queue' '
 	defaultq=`flux job stats | jq ".queues[] | select( .name == \"defaultqueue\" )"` &&
 	updateq=`flux job stats | jq ".queues[] | select( .name == \"updatequeue\" )"` &&
+	nosubmitq=`flux job stats | jq ".queues[] | select( .name == \"nosubmitqueue\" )"` &&
+	test -n "$nosubmitq" &&
 	echo $defaultq | jq -e ".job_states.depend == 0" &&
 	echo $defaultq | jq -e ".job_states.priority == 0" &&
 	echo $defaultq | jq -e ".job_states.sched == 0" &&
@@ -2199,7 +2209,19 @@ test_expect_success 'job stats lists jobs in correct state in each queue' '
 	echo $updateq | jq -e ".failed == 0" &&
 	echo $updateq | jq -e ".canceled == 0" &&
 	echo $updateq | jq -e ".timeout == 0" &&
-	echo $updateq | jq -e ".inactive_purged == 0"
+	echo $updateq | jq -e ".inactive_purged == 0" &&
+	echo $nosubmitq | jq -e ".job_states.depend == 0" &&
+	echo $nosubmitq | jq -e ".job_states.priority == 0" &&
+	echo $nosubmitq | jq -e ".job_states.sched == 0" &&
+	echo $nosubmitq | jq -e ".job_states.run == 0" &&
+	echo $nosubmitq | jq -e ".job_states.cleanup == 0" &&
+	echo $nosubmitq | jq -e ".job_states.inactive == 0" &&
+	echo $nosubmitq | jq -e ".job_states.total == 0" &&
+	echo $nosubmitq | jq -e ".successful == 0" &&
+	echo $nosubmitq | jq -e ".failed == 0" &&
+	echo $nosubmitq | jq -e ".canceled == 0" &&
+	echo $nosubmitq | jq -e ".timeout == 0" &&
+	echo $nosubmitq | jq -e ".inactive_purged == 0"
 '
 
 test_expect_success 'reload the job-list module' '
@@ -2215,12 +2237,26 @@ test_expect_success 'job-list returns expected jobspec changes after reload' '
 
 #
 # After reload, job-list will not have ever seen any jobs submitted to
-# "defaultqueue" therefore the stats object is empty.
+# "defaultqueue", need to check that stats string is non-empty and all
+# stats are 0.
 #
 test_expect_success 'job stats in each queue correct after reload' '
 	defaultq=`flux job stats | jq ".queues[] | select( .name == \"defaultqueue\" )"` &&
 	updateq=`flux job stats | jq ".queues[] | select( .name == \"updatequeue\" )"` &&
-	test -z "$defaultq" &&
+	test -n "$defaultq" &&
+	test -n "$nosubmitq" &&
+	echo $defaultq | jq -e ".job_states.depend == 0" &&
+	echo $defaultq | jq -e ".job_states.priority == 0" &&
+	echo $defaultq | jq -e ".job_states.sched == 0" &&
+	echo $defaultq | jq -e ".job_states.run == 0" &&
+	echo $defaultq | jq -e ".job_states.cleanup == 0" &&
+	echo $defaultq | jq -e ".job_states.inactive == 0" &&
+	echo $defaultq | jq -e ".job_states.total == 0" &&
+	echo $defaultq | jq -e ".successful == 0" &&
+	echo $defaultq | jq -e ".failed == 0" &&
+	echo $defaultq | jq -e ".canceled == 0" &&
+	echo $defaultq | jq -e ".timeout == 0" &&
+	echo $defaultq | jq -e ".inactive_purged == 0" &&
 	echo $updateq | jq -e ".job_states.depend == 0" &&
 	echo $updateq | jq -e ".job_states.priority == 0" &&
 	echo $updateq | jq -e ".job_states.sched == 0" &&
@@ -2232,7 +2268,19 @@ test_expect_success 'job stats in each queue correct after reload' '
 	echo $updateq | jq -e ".failed == 0" &&
 	echo $updateq | jq -e ".canceled == 0" &&
 	echo $updateq | jq -e ".timeout == 0" &&
-	echo $updateq | jq -e ".inactive_purged == 0"
+	echo $updateq | jq -e ".inactive_purged == 0" &&
+	echo $nosubmitq | jq -e ".job_states.depend == 0" &&
+	echo $nosubmitq | jq -e ".job_states.priority == 0" &&
+	echo $nosubmitq | jq -e ".job_states.sched == 0" &&
+	echo $nosubmitq | jq -e ".job_states.run == 0" &&
+	echo $nosubmitq | jq -e ".job_states.cleanup == 0" &&
+	echo $nosubmitq | jq -e ".job_states.inactive == 0" &&
+	echo $nosubmitq | jq -e ".job_states.total == 0" &&
+	echo $nosubmitq | jq -e ".successful == 0" &&
+	echo $nosubmitq | jq -e ".failed == 0" &&
+	echo $nosubmitq | jq -e ".canceled == 0" &&
+	echo $nosubmitq | jq -e ".timeout == 0" &&
+	echo $nosubmitq | jq -e ".inactive_purged == 0"
 '
 
 test_expect_success 'remove jobtap plugins and remove queue config' '