3 files changed, 462 insertions, 13 deletions

diff --git a/src/journald/LMI_JournalMessageLogProvider.c b/src/journald/LMI_JournalMessageLogProvider.c
index c4d9457..cde9a33 100644
--- a/src/journald/LMI_JournalMessageLogProvider.c
+++ b/src/journald/LMI_JournalMessageLogProvider.c
@@ -21,12 +21,19 @@
 #include <konkret/konkret.h>
 #include "LMI_JournalMessageLog.h"
 
+#include <glib.h>
 #include <systemd/sd-journal.h>
 
 #include "globals.h"
 #include "journal.h"
+#include "instutil.h"
 
 
+/* As defined in CIM_MessageLog schema */
+#define CIM_MESSAGELOG_ITERATOR_RESULT_SUCCESS 0
+#define CIM_MESSAGELOG_ITERATOR_RESULT_NOT_SUPPORTED 1
+#define CIM_MESSAGELOG_ITERATOR_RESULT_FAILED 2
+
 static const CMPIBroker* _cb = NULL;
 
 static void LMI_JournalMessageLogInitialize()
@@ -204,6 +211,7 @@ KUint32 LMI_JournalMessageLog_ClearLog(
     KUint32 result = KUINT32_INIT;
 
     KSetStatus(status, ERR_NOT_SUPPORTED);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_NOT_SUPPORTED);
     return result;
 }
 
@@ -216,8 +224,19 @@ KUint32 LMI_JournalMessageLog_PositionToFirstRecord(
     CMPIStatus* status)
 {
     KUint32 result = KUINT32_INIT;
+    gchar *iter_id;
+
+    KSetStatus(status, OK);
+
+    if ((iter_id = journal_iter_new(NULL, 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);
 
-    KSetStatus(status, ERR_NOT_SUPPORTED);
     return result;
 }
 
@@ -232,8 +251,39 @@ KUint32 LMI_JournalMessageLog_PositionAtRecord(
     CMPIStatus* status)
 {
     KUint32 result = KUINT32_INIT;
+    gchar *iter_id;
+    sd_journal *journal;
 
-    KSetStatus(status, ERR_NOT_SUPPORTED);
+    if (IterationIdentifier->null || ! IterationIdentifier->exists ||
+        MoveAbsolute->null || ! MoveAbsolute->exists ||
+        RecordNumber->null || ! RecordNumber->exists) {
+        KSetStatus(status, ERR_INVALID_PARAMETER);
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    if (MoveAbsolute->value != 0) {
+        KSetStatus2(_cb, status, ERR_INVALID_PARAMETER, "Absolute movement not supported\n");
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    iter_id = g_strdup(IterationIdentifier->chars);
+    if (! journal_iter_validate_id(&iter_id, &journal, NULL, _cb, status)) {
+        g_free(iter_id);
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    if (! journal_iter_seek(&iter_id, journal, RecordNumber->value)) {
+        g_free(iter_id);
+        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;
 }
 
@@ -249,8 +299,44 @@ KUint32 LMI_JournalMessageLog_GetRecord(
     CMPIStatus* status)
 {
     KUint32 result = KUINT32_INIT;
+    gchar *iter_id;
+    sd_journal *journal;
+    char *data;
+    guint i;
+
+    if (IterationIdentifier->null || ! IterationIdentifier->exists ||
+        PositionToNext->null || ! PositionToNext->exists) {
+        KSetStatus(status, ERR_INVALID_PARAMETER);
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
 
-    KSetStatus(status, ERR_NOT_SUPPORTED);
+    iter_id = g_strdup(IterationIdentifier->chars);
+    if (! journal_iter_validate_id(&iter_id, &journal, NULL, _cb, status)) {
+        g_free(iter_id);
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    data = journal_iter_get_data(&iter_id, journal, PositionToNext->value == 1);
+    if (! data) {
+        g_free(iter_id);
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    KUint8A_Init(RecordData, _cb, strlen(data));
+    /* FIXME: This will feed the array with unsigned chars, NOT terminated by zero.
+     *        Ideally we would like to pass string instead but the model is given. */
+    /* TODO: do the real signed vs. unsigned char conversion? */
+    for (i = 0; i < strlen(data); i++)
+        KUint8A_Set(RecordData, i, (unsigned char) data[i]);
+
+    g_free(data);
+
+    KString_Set(IterationIdentifier, _cb, iter_id);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_SUCCESS);
+    g_free(iter_id);
     return result;
 }
 
@@ -268,6 +354,7 @@ KUint32 LMI_JournalMessageLog_DeleteRecord(
     KUint32 result = KUINT32_INIT;
 
     KSetStatus(status, ERR_NOT_SUPPORTED);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_NOT_SUPPORTED);
     return result;
 }
 
@@ -285,6 +372,7 @@ KUint32 LMI_JournalMessageLog_WriteRecord(
     KUint32 result = KUINT32_INIT;
 
     KSetStatus(status, ERR_NOT_SUPPORTED);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_NOT_SUPPORTED);
     return result;
 }
 
@@ -297,8 +385,27 @@ KUint32 LMI_JournalMessageLog_CancelIteration(
     CMPIStatus* status)
 {
     KUint32 result = KUINT32_INIT;
+    gchar *iter_id_short;
 
-    KSetStatus(status, ERR_NOT_SUPPORTED);
+    if (IterationIdentifier->null || ! IterationIdentifier->exists) {
+        KSetStatus(status, ERR_INVALID_PARAMETER);
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    if (! journal_iter_parse_iterator_string(IterationIdentifier->chars, &iter_id_short, NULL, NULL)) {
+        KSetStatus2(_cb, status, ERR_INVALID_PARAMETER, "Malformed IterationIdentifier argument");
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+        return result;
+    }
+
+    KSetStatus(status, OK);
+    if (journal_iter_cancel(iter_id_short))
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_SUCCESS);
+    else
+        KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_FAILED);
+
+    g_free(iter_id_short);
     return result;
 }
 
@@ -313,6 +420,7 @@ KUint32 LMI_JournalMessageLog_FreezeLog(
     KUint32 result = KUINT32_INIT;
 
     KSetStatus(status, ERR_NOT_SUPPORTED);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_NOT_SUPPORTED);
     return result;
 }
 
@@ -329,6 +437,7 @@ KUint32 LMI_JournalMessageLog_FlagRecordForOverwrite(
     KUint32 result = KUINT32_INIT;
 
     KSetStatus(status, ERR_NOT_SUPPORTED);
+    KUint32_Set(&result, CIM_MESSAGELOG_ITERATOR_RESULT_NOT_SUPPORTED);
     return result;
 }
 
diff --git a/src/journald/instutil.c b/src/journald/instutil.c
index ac9e768..a38202a 100644
--- a/src/journald/instutil.c
+++ b/src/journald/instutil.c
@@ -22,6 +22,9 @@
 #include <syslog.h>
 #include <glib.h>
 
+#include <konkret/konkret.h>
+#include <cmpi/cmpimacs.h>
+
 #include "instutil.h"
 #include "globals.h"
 #include "journal.h"
@@ -33,6 +36,14 @@
 /* Assuming no thread safety */
 static sd_journal *ind_journal = NULL;
 
+/* LMI_JournalMessageLog iterators */
+static GHashTable *cmpi_iters = NULL;
+G_LOCK_DEFINE_STATIC(cmpi_iters);
+static sig_atomic_t cmpi_iters_count = 0;
+
+#define JOURNAL_ITER_PREFIX "LMI_JournalMessageLog_CMPI_Iter_"
+#define JOURNAL_ITER_SEPARATOR "#"
+
 
 int create_LMI_JournalLogRecordRef(sd_journal *j,
                                    LMI_JournalLogRecordRef *ref,
@@ -84,32 +95,60 @@ static int dup_journal_data(
     return 0;
 }
 
-int create_LMI_JournalLogRecord(sd_journal *j,
-                                LMI_JournalLogRecord *rec,
-                                const CMPIBroker *_cb)
+static int get_record_message(sd_journal *j, gboolean full_format, gchar **out)
 {
     int r;
-    uint64_t usec;
-    CMPIDateTime *date;
     gchar *d;
     gchar *syslog_identifier = NULL;
     gchar *comm = NULL;
     gchar *pid = NULL;
     gchar *fake_pid = NULL;
+    gchar *realtime = NULL;
+    gchar *hostname = NULL;
     GString *str;
 
-    /* Construct the message */
+    str = g_string_new(NULL);
+    if (! str)
+        return -ENOMEM;
+
     /* Message format is inspired by journalctl short output, containing the identifier (process name),
      * PID and the message. In other words, it's the traditional syslog record format.
      * Keep in sync with systemd:src/shared/logs-show.c:output_short() */
     r = dup_journal_data(j, "MESSAGE", &d);
     if (r < 0)
         return r;
+
     dup_journal_data(j, "SYSLOG_IDENTIFIER", &syslog_identifier);
     dup_journal_data(j, "_COMM", &comm);
     dup_journal_data(j, "_PID", &pid);
     dup_journal_data(j, "SYSLOG_PID", &fake_pid);
-    str = g_string_new(NULL);
+
+    if (full_format) {
+        /* Include timestamp and hostname */
+        char buf[64];
+        guint64 x = 0;
+        time_t t;
+        struct tm tm;
+        gchar *endptr = NULL;
+
+        dup_journal_data(j, "_SOURCE_REALTIME_TIMESTAMP", &realtime);
+        dup_journal_data(j, "_HOSTNAME", &hostname);
+        if (realtime) {
+            g_ascii_strtoull(realtime, &endptr, 10);
+            if (endptr != NULL && *endptr != '\0')
+                x = 0;
+        }
+        if (x == 0)
+            r = sd_journal_get_realtime_usec(j, &x);
+        if (r >= 0) {
+            t = (time_t) (x / 1000000ULL);
+            r = strftime(buf, sizeof(buf), "%b %d %H:%M:%S", localtime_r(&t, &tm));
+            if (r > 0)
+                g_string_append_printf(str, "%s ", buf);
+        }
+        if (hostname)
+            g_string_append_printf(str, "%s ", hostname);
+    }
     if (syslog_identifier || comm)
         g_string_append(str, syslog_identifier ? syslog_identifier : comm);
     if (pid || fake_pid)
@@ -117,13 +156,33 @@ int create_LMI_JournalLogRecord(sd_journal *j,
     if (str->len > 0)
         g_string_append(str, ": ");
     g_string_append(str, d);
-    LMI_JournalLogRecord_Set_DataFormat(rec, str->str);
-    g_string_free(str, TRUE);
+    *out = g_string_free(str, FALSE);
     g_free(d);
     g_free(syslog_identifier);
     g_free(comm);
     g_free(pid);
     g_free(fake_pid);
+    g_free(realtime);
+    g_free(hostname);
+
+    return 0;
+}
+
+int create_LMI_JournalLogRecord(sd_journal *j,
+                                LMI_JournalLogRecord *rec,
+                                const CMPIBroker *_cb)
+{
+    int r;
+    uint64_t usec;
+    CMPIDateTime *date;
+    gchar *d;
+
+    /* Construct the message */
+    r = get_record_message(j, FALSE, &d);
+    if (r < 0)
+        return r;
+    LMI_JournalLogRecord_Set_DataFormat(rec, d);
+    g_free(d);
 
     /* Set timestamp */
     r = sd_journal_get_realtime_usec(j, &usec);
@@ -319,3 +378,273 @@ bool ind_filter_cb(const CMPISelectExp *filter)
     if (strcasecmp(rhs_str, LMI_JournalLogRecord_ClassName) == 0) return true;
     return false;
 }
+
+/* --------------------------------------------------------------------------- */
+
+/* TODO: count references to the journal struct -- someone may cancel the iteration
+ *       while others are still using the journal struct (racy)  */
+
+#define set_cmpi_status_fmt(CB, STATUS, CODE, MSG, ...) \
+           { gchar *errs; \
+             errs = g_strdup_printf(MSG, ##__VA_ARGS__); \
+             KSetStatus2(CB, STATUS, CODE, errs); \
+             g_free(errs);\
+           }
+
+static gchar *
+make_iterator_string(sd_journal *journal, const char *cursor, gchar *iter_id_short)
+{
+    return g_strdup_printf("%s%s%p%s%s", iter_id_short, JOURNAL_ITER_SEPARATOR, (void *)journal, JOURNAL_ITER_SEPARATOR, cursor);
+}
+
+bool
+journal_iter_parse_iterator_string(const char *iter_id, gchar **out_iter_id_short, gpointer *out_iter_ptr, gchar **out_iter_cursor)
+{
+    gchar **s;
+    bool res;
+    gpointer valid_p;
+
+    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;
+    valid_p = NULL;
+    res = res && (sscanf(s[1], "%p", &valid_p) == 1);
+    if (res && out_iter_id_short)
+        res = res && ((*out_iter_id_short = g_strdup(s[0])) != NULL);
+    if (res && out_iter_ptr)
+        *out_iter_ptr = valid_p;
+    if (res && out_iter_cursor)
+        res = res && ((*out_iter_cursor = g_strdup(s[2])) != NULL);
+    g_strfreev(s);
+
+    return res;
+}
+
+gchar *
+journal_iter_new(const gchar *req_cursor, sd_journal **journal_out)
+{
+    gchar *iter_id = NULL;
+    gchar *iter_id_full = NULL;
+    char *cursor;
+    sd_journal *journal;
+    int r;
+
+    if (journal_out)
+        *journal_out = NULL;
+
+    r = sd_journal_open(&journal, 0);
+    if (r < 0) {
+        error("Error opening journal: %s\n", strerror(-r));
+        return NULL;
+    }
+
+    if (req_cursor)
+        r = sd_journal_seek_cursor(journal, req_cursor);
+    else
+        r = sd_journal_seek_head(journal);
+
+    if (r < 0) {
+        error("Error seeking to the requested journal position: %s\n", strerror(-r));
+        sd_journal_close(journal);
+        return NULL;
+    }
+
+    r = sd_journal_next(journal);
+    if (r < 0) {
+        error("Error stepping next in the journal: %s\n", strerror(-r));
+        sd_journal_close(journal);
+        return NULL;
+    }
+
+    r = sd_journal_get_cursor(journal, &cursor);
+    if (r < 0) {
+        error("Error getting current cursor: %s\n", strerror(-r));
+        sd_journal_close(journal);
+        return NULL;
+    }
+
+    G_LOCK(cmpi_iters);
+    if (cmpi_iters == NULL)
+        cmpi_iters = g_hash_table_new_full(g_str_hash, g_str_equal, (GDestroyNotify) g_free, (GDestroyNotify) sd_journal_close);
+    if (cmpi_iters == NULL) {
+        error("Memory allocation failure\n");
+        sd_journal_close(journal);
+        G_UNLOCK(cmpi_iters);
+        return NULL;
+    }
+    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);
+    if (iter_id == NULL || iter_id_full == NULL) {
+        error("Memory allocation failure\n");
+        sd_journal_close(journal);
+    } else {
+        g_hash_table_insert(cmpi_iters, iter_id, journal);
+    }
+    G_UNLOCK(cmpi_iters);
+
+    if (iter_id_full && journal_out)
+        *journal_out = journal;
+
+    return iter_id_full;
+}
+
+bool
+journal_iter_validate_id(gchar **iter_id, sd_journal **journal_out, gchar **prefix_out, const CMPIBroker *_cb, CMPIStatus *status)
+{
+    gboolean res;
+    gchar *iter_id_short, *iter_cursor;
+    gpointer iter_ptr;
+
+    res = TRUE;
+    if (journal_out)
+        *journal_out = NULL;
+    if (prefix_out)
+        *prefix_out = NULL;
+
+    if (! journal_iter_parse_iterator_string(*iter_id, &iter_id_short, &iter_ptr, &iter_cursor)) {
+        set_cmpi_status_fmt(_cb, status, ERR_INVALID_PARAMETER, "Malformed IterationIdentifier argument: \'%s\'\n", *iter_id);
+        return false;
+    }
+    KSetStatus(status, OK);
+
+    if (journal_out) {
+        G_LOCK(cmpi_iters);
+        if (cmpi_iters)
+            *journal_out = g_hash_table_lookup(cmpi_iters, iter_id_short);
+        G_UNLOCK(cmpi_iters);
+        if (*journal_out == NULL || *journal_out != iter_ptr) {
+            /* 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);
+            if (*iter_id == NULL) {
+                error("The IterationIdentifier is not valid anymore: \'%s\'\n", *iter_id);
+                res = FALSE;
+            }
+        }
+    }
+
+    if (res && prefix_out) {
+        /* No need to check prefix validity as it's supposed to be used in journal_iter_cancel() only */
+        res = res && ((*prefix_out = g_strdup(iter_id_short)) != NULL);
+    }
+
+    g_free(iter_id_short);
+    g_free(iter_cursor);
+    return res;
+}
+
+
+bool
+journal_iter_cancel(const char *iter_id)
+{
+    gboolean b;
+
+    g_return_val_if_fail(iter_id != NULL, false);
+
+    G_LOCK(cmpi_iters);
+    b = cmpi_iters && g_hash_table_remove(cmpi_iters, iter_id);
+    G_UNLOCK(cmpi_iters);
+    if (! b) {
+        error("IterationIdentifier \'%s\' not registered\n", iter_id);
+        return false;
+    }
+
+    return true;
+}
+
+static bool
+update_iter(gchar **iter_id, sd_journal *journal)
+{
+    gchar *iter_id_short;
+    char *cursor;
+    int r;
+
+    r = sd_journal_get_cursor(journal, &cursor);
+    if (r < 0) {
+        error("Error getting current cursor: %s\n", strerror(-r));
+        return false;
+    }
+
+    if (! journal_iter_parse_iterator_string(*iter_id, &iter_id_short, NULL, NULL))
+        return false;
+    *iter_id = make_iterator_string(journal, cursor, iter_id_short);
+    return *iter_id != NULL;
+}
+
+
+bool
+journal_iter_seek(gchar **iter_id, sd_journal *journal, gint64 position)
+{
+    int r;
+
+    g_return_val_if_fail(journal != NULL, false);
+
+    if (position == 0) {  /* NOP */
+        warn("journal_iter_seek(): Spurious seek request to relative position 0\n");
+        return true;
+    }
+
+    if (position > 0)
+        r = sd_journal_next_skip(journal, position);
+    else
+        r = sd_journal_previous_skip(journal, -position);
+
+    if (r < 0) {
+        error("Error seeking to the requested position: %s\n", strerror(-r));
+        return false;
+    }
+
+    if (! update_iter(iter_id, journal)) {
+        error("Error seeking to the requested position\n");
+        return false;
+    }
+
+    return true;
+}
+
+gchar *
+journal_iter_get_data(gchar **iter_id, sd_journal *journal, gboolean step_next)
+{
+    gchar *d;
+    int r;
+
+    g_return_val_if_fail(journal != NULL, false);
+
+    /* Construct the message */
+    r = get_record_message(journal, TRUE, &d);
+    if (r < 0) {
+        error("Error getting record message: %s\n", strerror(-r));
+        return NULL;
+    }
+
+    if (step_next) {
+        r = sd_journal_next(journal);
+        if (r < 0) {
+            error("Error advancing to the next record: %s\n", strerror(-r));
+            g_free(d);
+            return NULL;
+        }
+    }
+
+    if (! update_iter(iter_id, journal)) {
+        error("Error getting record message\n");
+        return NULL;
+    }
+
+    return d;
+}
+
+/* FIXME: unused for the moment as the hash table is global, shared across instances */
+void
+journal_iters_destroy()
+{
+    G_LOCK(cmpi_iters);
+    if (cmpi_iters != NULL) {
+        g_hash_table_destroy(cmpi_iters);
+        cmpi_iters = NULL;
+    }
+    G_UNLOCK(cmpi_iters);
+}
diff --git a/src/journald/instutil.h b/src/journald/instutil.h
index ee823c5..0ba7ff1 100644
--- a/src/journald/instutil.h
+++ b/src/journald/instutil.h
@@ -21,6 +21,8 @@
 #ifndef INSTUTIL_H_
 #define INSTUTIL_H_
 
+#include <glib.h>
+#include <konkret/konkret.h>
 #include <systemd/sd-journal.h>
 
 #include <ind_manager.h>
@@ -35,4 +37,13 @@ bool ind_filter_cb(const CMPISelectExp *filter);
 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);
+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);
+gchar * journal_iter_get_data(gchar **iter_id, sd_journal *journal, gboolean step_next);
+void    journal_iters_destroy();
+
+
 #endif /* INSTUTIL_H_ */