job-list: add job db

Problem: We would like to store inactive jobs on disk for retrieval
at a later time.

Solution: Add an sqlite job db to the job-list module.  If the
job-list module is configured with a database path or if the flux
broker is configured with a statedir, store all inactive job data
to the db.

Author: chu11

src/modules/Makefile.am

@@ -190,7 +190,8 @@ job_list_la_SOURCES =
 job_list_la_CPPFLAGS = \
 	$(AM_CPPFLAGS) \
 	$(FLUX_SECURITY_CFLAGS) \
-	$(HWLOC_CFLAGS)
+	$(HWLOC_CFLAGS) \
+	$(SQLITE_CFLAGS)
 job_list_la_LIBADD = \
 	$(builddir)/job-list/libjob-list.la \
 	$(top_builddir)/src/common/libjob/libjob.la \
@@ -199,7 +200,8 @@ job_list_la_LIBADD = \
 	$(top_builddir)/src/common/libflux-optparse.la \
 	$(top_builddir)/src/common/librlist/librlist.la \
 	$(JANSSON_LIBS) \
-	$(HWLOC_LIBS)
+	$(HWLOC_LIBS) \
+	$(SQLITE_LIBS)
 job_list_la_LDFLAGS = $(fluxmod_ldflags) -module
 
 job_ingest_la_SOURCES =

src/modules/job-list/Makefile.am

@@ -22,6 +22,8 @@ libjob_list_la_SOURCES = \
 	job_state.c \
 	job_data.h \
 	job_data.c \
+	job_db.h \
+	job_db.c \
 	list.h \
 	list.c \
 	job_util.h \
@@ -35,7 +37,9 @@ libjob_list_la_SOURCES = \
 	state_match.h \
 	state_match.c \
 	match_util.h \
-	match_util.c
+	match_util.c \
+	util.h \
+	util.c
 
 TESTS = \
 	test_job_data.t \

src/modules/job-list/config.c

@@ -0,0 +1,105 @@
+static void config_reload_cb (flux_t *h,
+                              flux_msg_handler_t *mh,
+                              const flux_msg_t *msg,
+                              void *arg)
+{
+    struct conf *conf = arg;
+    const flux_conf_t *instance_conf;
+    struct conf_callback *ccb;
+    flux_error_t error;
+    const char *errstr = NULL;
+
+    if (flux_conf_reload_decode (msg, &instance_conf) < 0)
+        goto error;
+    if (policy_validate (instance_conf, &error) < 0) {
+        errstr = error.text;
+        goto error;
+    }
+    ccb = zlistx_first (conf->callbacks);
+    while (ccb) {
+        if (ccb->cb (instance_conf, &error, ccb->arg) < 0) {
+            errstr = error.text;
+            errno = EINVAL;
+            goto error;
+        }
+        ccb = zlistx_next (conf->callbacks);
+    }
+    if (flux_set_conf (h, flux_conf_incref (instance_conf)) < 0) {
+        errstr = "error updating cached configuration";
+        flux_conf_decref (instance_conf);
+        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[] = {
+    { FLUX_MSGTYPE_REQUEST, "job-manager.config-reload", config_reload_cb, 0 },
+    FLUX_MSGHANDLER_TABLE_END,
+};
+
+
+
+
+
+static int process_config (struct kvs_ctx *ctx)
+{
+    flux_error_t error;
+    if (kvs_checkpoint_config_parse (ctx->kcp,
+                                     flux_get_conf (ctx->h),
+                                     &error) < 0) {
+        flux_log (ctx->h, LOG_ERR, "%s", error.text);
+        return -1;
+    }
+    return 0;
+}
+
+
+
+static int checkpoint_period_parse (const flux_conf_t *conf,
+                                    flux_error_t *errp,
+                                    double *checkpoint_period)
+{
+    flux_error_t error;
+    const char *str = NULL;
+
+    if (flux_conf_unpack (conf,
+                          &error,
+                          "{s?{s?s}}",
+                          "kvs",
+                          "checkpoint-period", &str) < 0) {
+        errprintf (errp,
+                   "error reading config for kvs: %s",
+                   error.text);
+        return -1;
+    }
+
+    if (str) {
+        if (fsd_parse_duration (str, checkpoint_period) < 0) {
+            errprintf (errp,
+                       "invalid checkpoint-period config: %s",
+                       str);
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+int kvs_checkpoint_config_parse (kvs_checkpoint_t *kcp,
+                                 const flux_conf_t *conf,
+                                 flux_error_t *errp)
+{
+    if (kcp) {
+        double checkpoint_period = kcp->checkpoint_period;
+        if (checkpoint_period_parse (conf, errp, &checkpoint_period) < 0)
+            return -1;
+        kcp->checkpoint_period = checkpoint_period;
+    }
+    return 0;
+}
+

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

@@ -210,6 +210,8 @@ static void list_ctx_destroy (struct list_ctx *ctx)
         flux_msglist_destroy (ctx->deferred_requests);
         if (ctx->jsctx)
             job_state_destroy (ctx->jsctx);
+        if (ctx->dbctx)
+            job_db_ctx_destroy (ctx->dbctx);
         if (ctx->isctx)
             idsync_ctx_destroy (ctx->isctx);
         if (ctx->mctx)
@@ -219,7 +221,7 @@ static void list_ctx_destroy (struct list_ctx *ctx)
     }
 }
 
-static struct list_ctx *list_ctx_create (flux_t *h)
+static struct list_ctx *list_ctx_create (flux_t *h, int argc, char **argv)
 {
     struct list_ctx *ctx = calloc (1, sizeof (*ctx));
     if (!ctx)
@@ -231,6 +233,12 @@ static struct list_ctx *list_ctx_create (flux_t *h)
         goto error;
     if (!(ctx->isctx = idsync_ctx_create (ctx->h)))
         goto error;
+    /* job_db_setup() performs a job_db_ctx_create() and some
+     * initialization */
+    if (!(ctx->dbctx = job_db_setup (h, argc, argv))) {
+        if (errno != ENOTBLK)
+            goto error;
+    }
     if (!(ctx->jsctx = job_state_create (ctx)))
         goto error;
     if (!(ctx->deferred_requests = flux_msglist_create ()))
@@ -248,7 +256,7 @@ int mod_main (flux_t *h, int argc, char **argv)
     struct list_ctx *ctx;
     int rc = -1;
 
-    if (!(ctx = list_ctx_create (h))) {
+    if (!(ctx = list_ctx_create (h, argc, argv))) {
         flux_log_error (h, "initialization error");
         goto done;
     }

src/modules/job-list/job-list.h

@@ -16,13 +16,15 @@
 #include "src/common/libczmqcontainers/czmq_containers.h"
 
 #include "job_state.h"
+#include "job_db.h"
 #include "idsync.h"
 #include "match.h"
 
 struct list_ctx {
     flux_t *h;
     flux_msg_handler_t **handlers;
     struct job_state_ctx *jsctx;
+    struct job_db_ctx *dbctx;
     struct idsync_ctx *isctx;
     struct flux_msglist *deferred_requests;
     struct match_ctx *mctx;