journald: Properly handle journal EOF

The logic of handling EOF is incompatible between journald API and
CIM_MessageLog model we use. The CIM_MessageLog.GetRecord() method
allows stepping to the next record after data retrieval but there's
no specific return code for the EOF case. And returning failure code
makes the client application think the whole data retrieval has failed
and thus throwing away the last record.

To work around this case, a new flag has been introduced to the
IterationIdentifier string, which is considered opaque to the client.
That way we return success from the GetRecord() method call and putting
a flag to the IterationIdentifier string so that any successive GetRecord()
method call properly fails without any data loss. Care is taken of the
case when a new message is logged between the GetRecord() calls.

This commit also adds the CIM_MessageLog.PositionToLastRecord() method
for convenient access to the end of the journal.

3 files changed, 77 insertions, 21 deletions

diff --git a/src/journald/LMI_JournalMessageLogProvider.c b/src/journald/LMI_JournalMessageLogProvider.c
index 2332f83..297a775 100644
--- a/src/journald/LMI_JournalMessageLogProvider.c
+++ b/src/journald/LMI_JournalMessageLogProvider.c
@@ -228,7 +228,32 @@ KUint32 LMI_JournalMessageLog_PositionToFirstRecord(
 
     KSetStatus(status, OK);
 
-    if ((iter_id = journal_iter_new(NULL, NULL)) == NULL) {
+    if ((iter_id = journal_iter_new(NULL, FALSE, NULL)) == NULL) {
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    KString_Set(IterationIdentifier, _cb, iter_id);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_SUCCESS);
+    g_free(iter_id);
+
+    return result;
+}
+
+KUint32 LMI_JournalMessageLog_PositionToLastRecord(
+    const CMPIBroker* cb,
+    CMPIMethodMI* mi,
+    const CMPIContext* context,
+    const LMI_JournalMessageLogRef* self,
+    KString* IterationIdentifier,
+    CMPIStatus* status)
+{
+    KUint32 result = KUINT32_INIT;
+    gchar *iter_id;
+
+    KSetStatus(status, OK);
+
+    if ((iter_id = journal_iter_new(NULL, TRUE, NULL)) == NULL) {
         KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
         return result;
     }
@@ -393,7 +418,7 @@ KUint32 LMI_JournalMessageLog_CancelIteration(
         return result;
     }
 
-    if (! journal_iter_parse_iterator_string(IterationIdentifier->chars, &iter_id_short, NULL, NULL)) {
+    if (! journal_iter_parse_iterator_string(IterationIdentifier->chars, &iter_id_short, NULL, NULL, NULL)) {
         KSetStatus2(_cb, status, ERR_INVALID_PARAMETER, "Malformed IterationIdentifier argument");
         KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
         return result;
diff --git a/src/journald/instutil.c b/src/journald/instutil.c
index be11610..413a8b1 100644
--- a/src/journald/instutil.c
+++ b/src/journald/instutil.c
@@ -44,6 +44,7 @@ static sig_atomic_t cmpi_iters_count = 0;
 
 #define JOURNAL_ITER_PREFIX "LMI_JournalMessageLog_CMPI_Iter_"
 #define JOURNAL_ITER_SEPARATOR "#"
+#define JOURNAL_ITER_EOF "<EOF>"
 
 
 int create_LMI_JournalLogRecordRef(sd_journal *j,
@@ -398,22 +399,26 @@ bool ind_gather(const IMManager *manager, CMPIInstance **old, CMPIInstance **new
            }
 
 static gchar *
-make_iterator_string(sd_journal *journal, const char *cursor, gchar *iter_id_short)
+make_iterator_string(sd_journal *journal, const char *cursor, gchar *iter_id_short, gboolean eof_set)
 {
-    return g_strdup_printf("%s%s%p%s%s", iter_id_short, JOURNAL_ITER_SEPARATOR, (void *)journal, JOURNAL_ITER_SEPARATOR, cursor);
+    return g_strdup_printf("%s%s%p%s%s%s%s", iter_id_short, JOURNAL_ITER_SEPARATOR, (void *)journal, JOURNAL_ITER_SEPARATOR, cursor,
+                                             eof_set ? JOURNAL_ITER_SEPARATOR : "", eof_set ? JOURNAL_ITER_EOF : "");
 }
 
 bool
-journal_iter_parse_iterator_string(const char *iter_id, gchar **out_iter_id_short, gpointer *out_iter_ptr, gchar **out_iter_cursor)
+journal_iter_parse_iterator_string(const char *iter_id, gchar **out_iter_id_short, gpointer *out_iter_ptr, gchar **out_iter_cursor, gboolean *out_eof_set)
 {
     gchar **s = NULL;
     bool res;
     gpointer valid_p;
 
+    if (out_eof_set)
+        *out_eof_set = FALSE;
+
     res = (iter_id && strlen(iter_id) > 0);
     if (res)
-        s = g_strsplit(iter_id, JOURNAL_ITER_SEPARATOR, 3);
-    res = res && s && g_strv_length(s) == 3 && strlen(s[0]) > 0 && strlen(s[1]) > 0 && strlen(s[2]) > 0;
+        s = g_strsplit(iter_id, JOURNAL_ITER_SEPARATOR, 4);
+    res = res && s && (g_strv_length(s) == 3 || g_strv_length(s) == 4) && strlen(s[0]) > 0 && strlen(s[1]) > 0 && strlen(s[2]) > 0 && (g_strv_length(s) == 3 || strlen(s[3]) > 0);
     valid_p = NULL;
     res = res && (sscanf(s[1], "%p", &valid_p) == 1);
     if (res && out_iter_id_short)
@@ -422,13 +427,18 @@ journal_iter_parse_iterator_string(const char *iter_id, gchar **out_iter_id_shor
         *out_iter_ptr = valid_p;
     if (res && out_iter_cursor)
         res = res && ((*out_iter_cursor = g_strdup(s[2])) != NULL);
+    if (res && out_eof_set && g_strv_length(s) == 4) {
+        *out_eof_set = g_strcmp0(s[3], JOURNAL_ITER_EOF) == 0;
+        /* the fourth part should only be present with valid string */
+        res = res && *out_eof_set;
+    }
     g_strfreev(s);
 
     return res;
 }
 
 gchar *
-journal_iter_new(const gchar *req_cursor, sd_journal **journal_out)
+journal_iter_new(const gchar *req_cursor, gboolean seek_tail, sd_journal **journal_out)
 {
     gchar *iter_id = NULL;
     gchar *iter_id_full = NULL;
@@ -436,6 +446,7 @@ journal_iter_new(const gchar *req_cursor, sd_journal **journal_out)
     sd_journal *journal;
     int r;
     char errbuf[STRERROR_BUF_LEN];
+    gboolean eof_set;
 
     if (journal_out)
         *journal_out = NULL;
@@ -448,8 +459,12 @@ journal_iter_new(const gchar *req_cursor, sd_journal **journal_out)
 
     if (req_cursor)
         r = sd_journal_seek_cursor(journal, req_cursor);
-    else
-        r = sd_journal_seek_head(journal);
+    else {
+        if (seek_tail)
+            r = sd_journal_seek_tail(journal);
+        else
+            r = sd_journal_seek_head(journal);
+    }
 
     if (r < 0) {
         error("Error seeking to the requested journal position: %s\n", strerror_r(-r, errbuf, sizeof(errbuf)));
@@ -457,12 +472,16 @@ journal_iter_new(const gchar *req_cursor, sd_journal **journal_out)
         return NULL;
     }
 
-    r = sd_journal_next(journal);
+    if (seek_tail)
+        r = sd_journal_previous(journal);
+    else
+        r = sd_journal_next(journal);
     if (r < 0) {
         error("Error stepping next in the journal: %s\n", strerror_r(-r, errbuf, sizeof(errbuf)));
         sd_journal_close(journal);
         return NULL;
     }
+    eof_set = r == 0;
 
     r = sd_journal_get_cursor(journal, &cursor);
     if (r < 0) {
@@ -482,7 +501,7 @@ journal_iter_new(const gchar *req_cursor, sd_journal **journal_out)
     }
     iter_id = g_strdup_printf("%s%d", JOURNAL_ITER_PREFIX, cmpi_iters_count++);
     if (iter_id)
-        iter_id_full = make_iterator_string(journal, cursor, iter_id);
+        iter_id_full = make_iterator_string(journal, cursor, iter_id, eof_set);
     if (iter_id == NULL || iter_id_full == NULL) {
         error("Memory allocation failure\n");
         sd_journal_close(journal);
@@ -503,6 +522,7 @@ journal_iter_validate_id(gchar **iter_id, sd_journal **journal_out, gchar **pref
     gboolean res;
     gchar *iter_id_short, *iter_cursor;
     gpointer iter_ptr;
+    gboolean eof_set;
 
     res = TRUE;
     if (journal_out)
@@ -510,7 +530,7 @@ journal_iter_validate_id(gchar **iter_id, sd_journal **journal_out, gchar **pref
     if (prefix_out)
         *prefix_out = NULL;
 
-    if (! journal_iter_parse_iterator_string(*iter_id, &iter_id_short, &iter_ptr, &iter_cursor)) {
+    if (! journal_iter_parse_iterator_string(*iter_id, &iter_id_short, &iter_ptr, &iter_cursor, &eof_set)) {
         set_cmpi_status_fmt(_cb, status, ERR_INVALID_PARAMETER, "Malformed IterationIdentifier argument: \'%s\'\n", *iter_id);
         return false;
     }
@@ -525,7 +545,7 @@ journal_iter_validate_id(gchar **iter_id, sd_journal **journal_out, gchar **pref
             /* Assume stale iterator ID, reopen journal and try to find the position by the cursor */
             warn("journal_iter_validate_id(): iterator pointer %p doesn't match with hashtable %p, reopening journal...\n", iter_ptr, *journal_out);
             g_free(*iter_id);
-            *iter_id = journal_iter_new(iter_cursor, journal_out);
+            *iter_id = journal_iter_new(iter_cursor, FALSE, journal_out);
             if (*iter_id == NULL) {
                 error("The IterationIdentifier is not valid anymore: \'%s\'\n", *iter_id);
                 res = FALSE;
@@ -563,12 +583,13 @@ journal_iter_cancel(const char *iter_id)
 }
 
 static bool
-update_iter(gchar **iter_id, sd_journal *journal)
+update_iter(gchar **iter_id, gboolean explicit_eof, sd_journal *journal)
 {
     gchar *iter_id_short;
     char *cursor;
     int r;
     char errbuf[STRERROR_BUF_LEN];
+    gboolean eof_set;
 
     r = sd_journal_get_cursor(journal, &cursor);
     if (r < 0) {
@@ -576,9 +597,9 @@ update_iter(gchar **iter_id, sd_journal *journal)
         return false;
     }
 
-    if (! journal_iter_parse_iterator_string(*iter_id, &iter_id_short, NULL, NULL))
+    if (! journal_iter_parse_iterator_string(*iter_id, &iter_id_short, NULL, NULL, &eof_set))
         return false;
-    *iter_id = make_iterator_string(journal, cursor, iter_id_short);
+    *iter_id = make_iterator_string(journal, cursor, iter_id_short, explicit_eof);
     return *iter_id != NULL;
 }
 
@@ -606,7 +627,7 @@ journal_iter_seek(gchar **iter_id, sd_journal *journal, gint64 position)
         return false;
     }
 
-    if (! update_iter(iter_id, journal)) {
+    if (! update_iter(iter_id, r == 0, journal)) {
         error("Error seeking to the requested position\n");
         return false;
     }
@@ -620,9 +641,19 @@ journal_iter_get_data(gchar **iter_id, sd_journal *journal, gboolean step_next)
     gchar *d;
     int r;
     char errbuf[STRERROR_BUF_LEN];
+    gboolean eof_set;
 
     g_return_val_if_fail(journal != NULL, false);
 
+    /* In case of EOF was reached previously, try to seek again to see if new record arrived */
+    if (! journal_iter_parse_iterator_string(*iter_id, FALSE, NULL, NULL, &eof_set))
+        return false;
+    if (eof_set) {
+        r = sd_journal_next(journal);
+        if (r <= 0)
+            return false;
+    }
+
     /* Construct the message */
     r = get_record_message(journal, TRUE, &d);
     if (r < 0) {
@@ -639,7 +670,7 @@ journal_iter_get_data(gchar **iter_id, sd_journal *journal, gboolean step_next)
         }
     }
 
-    if (! update_iter(iter_id, journal)) {
+    if (! update_iter(iter_id, r == 0, journal)) {
         error("Error getting record message\n");
         return NULL;
     }
diff --git a/src/journald/instutil.h b/src/journald/instutil.h
index 2e4ad0b..88572c5 100644
--- a/src/journald/instutil.h
+++ b/src/journald/instutil.h
@@ -42,8 +42,8 @@ bool ind_watcher(void **data);
 bool ind_gather(const IMManager *manager, CMPIInstance **old, CMPIInstance **new, void *data);
 void ind_destroy();
 
-gchar * journal_iter_new(const gchar *req_cursor, sd_journal **journal_out);
-bool    journal_iter_parse_iterator_string(const char *iter_id, gchar **out_iter_id_short, gpointer *out_iter_ptr, gchar **out_iter_cursor);
+gchar * journal_iter_new(const gchar *req_cursor, gboolean seek_tail, sd_journal **journal_out);
+bool    journal_iter_parse_iterator_string(const char *iter_id, gchar **out_iter_id_short, gpointer *out_iter_ptr, gchar **out_iter_cursor, gboolean *out_eof_set);
 bool    journal_iter_validate_id(gchar **iter_id, sd_journal **journal_out, gchar **prefix_out, const CMPIBroker *_cb, CMPIStatus *status);
 bool    journal_iter_cancel(const gchar *iter_id);
 bool    journal_iter_seek(gchar **iter_id, sd_journal *journal, gint64 position);