job-list: rebuild and store eventlog

Problem: In the near future we will need access to the job's
eventlog when a job goes inactive.

Solution: Rebuild the job eventlog from the events journal and
store it internally in struct job.

Author: chu11

src/modules/job-list/job_data.c

@@ -36,6 +36,7 @@ void job_destroy (void *data)
         grudgeset_destroy (job->dependencies);
         json_decref (job->jobspec);
         json_decref (job->R);
+        free (job->eventlog);
         json_decref (job->exception_context);
         zlist_destroy (&job->next_states);
         free (job);

src/modules/job-list/job_data.h

@@ -66,6 +66,8 @@ struct job {
     /* cache of job information */
     json_t *jobspec;
     json_t *R;
+    char *eventlog;
+    size_t eventlog_len;
     json_t *exception_context;
 
     /* Track which states we have seen and have completed transition

src/modules/job-list/job_state.c

@@ -541,6 +541,42 @@ void job_state_unpause_cb (flux_t *h, flux_msg_handler_t *mh,
         flux_log_error (h, "error responding to unpause request");
 }
 
+static int store_eventlog_entry (struct job_state_ctx *jsctx,
+                                 struct job *job,
+                                 json_t *entry)
+{
+    char *s = json_dumps (entry, 0);
+    int rv = -1;
+
+    /* entry should have been verified via eventlog_entry_parse()
+     * earlier */
+    assert (s);
+
+    if (!job->eventlog) {
+        job->eventlog_len = strlen (s) + 2; /* +2 for \n and \0 */
+        if (!(job->eventlog = calloc (1, job->eventlog_len))) {
+            flux_log_error (jsctx->h, "calloc");
+            goto error;
+
+        }
+        strcpy (job->eventlog, s);
+        strcat (job->eventlog, "\n");
+    }
+    else {
+        job->eventlog_len += strlen (s) + 1; /* +1 for \n */
+        if (!(job->eventlog = realloc (job->eventlog, job->eventlog_len))) {
+            flux_log_error (jsctx->h, "realloc");
+            goto error;
+        }
+        strcat (job->eventlog, s);
+        strcat (job->eventlog, "\n");
+    }
+    rv = 0;
+error:
+    free (s);
+    return rv;
+}
+
 static struct job *eventlog_restart_parse (struct job_state_ctx *jsctx,
                                            const char *eventlog,
                                            flux_jobid_t id)
@@ -571,6 +607,9 @@ static struct job *eventlog_restart_parse (struct job_state_ctx *jsctx,
             goto error;
         }
 
+        if (store_eventlog_entry (jsctx, job, value) < 0)
+            goto error;
+
         job->eventlog_seq++;
         if (streq (name, "submit")) {
             if (submit_context_parse (jsctx->h, job, context) < 0)
@@ -927,11 +966,16 @@ static int journal_submit_event (struct job_state_ctx *jsctx,
                                  flux_jobid_t id,
                                  int eventlog_seq,
                                  double timestamp,
+                                 json_t *entry,
                                  json_t *context)
 {
     if (!job) {
         if (!(job = job_create (jsctx->h, id)))
             return -1;
+        if (store_eventlog_entry (jsctx, job, entry) < 0) {
+            job_destroy (job);
+            return -1;
+        }
         if (zhashx_insert (jsctx->index, &job->id, job) < 0) {
             job_destroy (job);
             errno = ENOMEM;
@@ -1303,12 +1347,18 @@ static int journal_process_event (struct job_state_ctx *jsctx, json_t *event)
         return 0;
     }
 
+    if (job && job->eventlog) {
+        if (store_eventlog_entry (jsctx, job, entry) < 0)
+            return -1;
+    }
+
     if (streq (name, "submit")) {
         if (journal_submit_event (jsctx,
                                   job,
                                   id,
                                   eventlog_seq,
                                   timestamp,
+                                  entry,
                                   context) < 0)
             return -1;
     }