From: Krzysztof Matczak <krzysztofx.matczak@intel.com>
Date: Sun, 8 Jan 2017 21:36:23 +0000 (+0000)
Subject: Utility for filtering and parsing log messages
X-Git-Tag: collectd-5.11.0~18^2~7
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=dd946578aec51a2dc87b7694ee531e56e7e19ce6;p=thirdparty%2Fcollectd.git

Utility for filtering and parsing log messages

This utility simplifies filtering and parsing specific messages
from given log file. It tracks log file using utils_tail_match.h API.
Main feature is automatic messages assembly based on set of user provided
regular expressions containing distinguished expressions for recognizing
start and the end of the message. That makes assembling single message
from multiple log lines quite easy. Utility supports multiple parsing jobs
and also user defined message part validation flags (mandatory or not).

Added fixes for memory leak in utils_match.c and for debug mode
segfault in src/utils_tail_match.c

Modification of utils_tail and utils_tail_match API with adding option
for reading file from the beginning. Minor modifications in this API
calls in tail and tail_csv plugins.

Change-Id: I6865833c8d5403294124187aa7f93cc957004a03
Signed-off-by: Krzysztof Matczak <krzysztofx.matczak@intel.com>
---

diff --git a/src/tail.c b/src/tail.c
index 8c9dfbe53..4cb7ea5e4 100644
--- a/src/tail.c
+++ b/src/tail.c
@@ -292,7 +292,7 @@ static int ctail_config(oconfig_item_t *ci) {
 static int ctail_read(user_data_t *ud) {
   int status;
 
-  status = tail_match_read((cu_tail_match_t *)ud->data);
+  status = tail_match_read((cu_tail_match_t *)ud->data, 0);
   if (status != 0) {
     ERROR("tail plugin: tail_match_read failed.");
     return -1;
diff --git a/src/tail_csv.c b/src/tail_csv.c
index ab8bf6d59..4e92eade3 100644
--- a/src/tail_csv.c
+++ b/src/tail_csv.c
@@ -221,7 +221,7 @@ static int tcsv_read(user_data_t *ud) {
     size_t buffer_len;
     int status;
 
-    status = cu_tail_readline(id->tail, buffer, (int)sizeof(buffer));
+    status = cu_tail_readline(id->tail, buffer, (int)sizeof(buffer), 0);
     if (status != 0) {
       ERROR("tail_csv plugin: File \"%s\": cu_tail_readline failed "
             "with status %i.",
diff --git a/src/utils/match/match.c b/src/utils/match/match.c
index ca6f1aaa7..c9bbd7b49 100644
--- a/src/utils/match/match.c
+++ b/src/utils/match/match.c
@@ -33,6 +33,7 @@
 
 #include <regex.h>
 
+#define UTILS_MATCH_FLAGS_REGEX 0x04
 #define UTILS_MATCH_FLAGS_EXCLUDE_REGEX 0x02
 #define UTILS_MATCH_FLAGS_REGEX 0x04
 
diff --git a/src/utils/tail/tail.c b/src/utils/tail/tail.c
index db34a72b1..d87254095 100644
--- a/src/utils/tail/tail.c
+++ b/src/utils/tail/tail.c
@@ -41,7 +41,7 @@ struct cu_tail_s {
   struct stat stat;
 };
 
-static int cu_tail_reopen(cu_tail_t *obj) {
+static int cu_tail_reopen(cu_tail_t *obj, _Bool force_rewind) {
   int seek_end = 0;
   struct stat stat_buf = {0};
 
@@ -68,10 +68,11 @@ static int cu_tail_reopen(cu_tail_t *obj) {
     return 1;
   }
 
-  /* Seek to the end if we re-open the same file again or the file opened
-   * is the first at all or the first after an error */
+  /* Unless flag for rewinding to the start is set, seek to the end
+   * if we re-open the same file again or the file opened is the first at all
+   * or the first after an error */
   if ((obj->stat.st_ino == 0) || (obj->stat.st_ino == stat_buf.st_ino))
-    seek_end = 1;
+    seek_end = force_rewind ? 0 : 1;
 
   FILE *fh = fopen(obj->file, "r");
   if (fh == NULL) {
@@ -123,7 +124,7 @@ int cu_tail_destroy(cu_tail_t *obj) {
   return 0;
 } /* int cu_tail_destroy */
 
-int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen) {
+int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen, _Bool force_rewind) {
   int status;
 
   if (buflen < 1) {
@@ -132,7 +133,7 @@ int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen) {
   }
 
   if (obj->fh == NULL) {
-    status = cu_tail_reopen(obj);
+    status = cu_tail_reopen(obj, force_rewind);
     if (status < 0)
       return status;
   }
@@ -155,7 +156,7 @@ int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen) {
   /* else: eof -> check if the file was moved away and reopen the new file if
    * so.. */
 
-  status = cu_tail_reopen(obj);
+  status = cu_tail_reopen(obj, force_rewind);
   /* error -> return with error */
   if (status < 0)
     return status;
@@ -186,13 +187,13 @@ int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen) {
 } /* int cu_tail_readline */
 
 int cu_tail_read(cu_tail_t *obj, char *buf, int buflen, tailfunc_t *callback,
-                 void *data) {
+                 void *data, _Bool force_rewind) {
   int status;
 
   while (42) {
     size_t len;
 
-    status = cu_tail_readline(obj, buf, buflen);
+    status = cu_tail_readline(obj, buf, buflen, force_rewind);
     if (status != 0) {
       ERROR("utils_tail: cu_tail_read: cu_tail_readline "
             "failed.");
diff --git a/src/utils/tail/tail.h b/src/utils/tail/tail.h
index 73a6de215..b3aa07231 100644
--- a/src/utils/tail/tail.h
+++ b/src/utils/tail/tail.h
@@ -73,7 +73,7 @@ int cu_tail_destroy(cu_tail_t *obj);
  *
  * Returns 0 when successful and non-zero otherwise.
  */
-int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen);
+int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen, _Bool force_rewind);
 
 /*
  * cu_tail_readline
@@ -83,6 +83,6 @@ int cu_tail_readline(cu_tail_t *obj, char *buf, int buflen);
  * Returns 0 when successful and non-zero otherwise.
  */
 int cu_tail_read(cu_tail_t *obj, char *buf, int buflen, tailfunc_t *callback,
-                 void *data);
+                 void *data, _Bool force_rewind);
 
 #endif /* UTILS_TAIL_H */
diff --git a/src/utils_message_parser.c b/src/utils_message_parser.c
new file mode 100644
index 000000000..29c64d688
--- /dev/null
+++ b/src/utils_message_parser.c
@@ -0,0 +1,320 @@
+/*
+ * collectd - src/utils_message_parser.c
+ * MIT License
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ *
+ * Authors:
+ *   Krzysztof Matczak <krzysztofx.matczak@intel.com>
+ */
+
+#include "collectd.h"
+
+#include "common.h"
+#include "plugin.h"
+
+#include "utils_message_parser.h"
+
+#define UTIL_NAME "utils_message_parser"
+
+#define MSG_STOR_INIT_LEN 64
+#define MSG_STOR_INC_STEP 10
+
+typedef struct checked_match_t {
+  parser_job_data *parser_job;
+  message_pattern msg_pattern;
+  int msg_pattern_idx;
+} checked_match;
+
+struct parser_job_data_t {
+  const char *filename;
+  unsigned int start_idx;
+  unsigned int stop_idx;
+  cu_tail_match_t *tm;
+  message *messages_storage;
+  size_t messages_max_len;
+  int message_idx;
+  unsigned int message_item_idx;
+  unsigned int messages_completed;
+  message_pattern *message_patterns;
+  size_t message_patterns_len;
+  int (*resize_message_buffer)(parser_job_data *self, size_t);
+  int (*start_message_assembly)(parser_job_data *self);
+  void (*end_message_assembly)(parser_job_data *self);
+  void (*message_item_assembly)(parser_job_data *self, checked_match *cm,
+                                char *const *matches);
+};
+
+static void message_item_assembly(parser_job_data *self, checked_match *cm,
+                                  char *const *matches) {
+  message_item *msg_it = &(self->messages_storage[self->message_idx]
+                               .message_items[self->message_item_idx]);
+  sstrncpy(msg_it->name, (cm->msg_pattern).name, sizeof(msg_it->name));
+  sstrncpy(msg_it->value, matches[cm->msg_pattern.submatch_idx],
+           sizeof(msg_it->value));
+  self->messages_storage[self->message_idx]
+      .matched_patterns_check[cm->msg_pattern_idx] = 1;
+  ++(self->message_item_idx);
+}
+
+static int start_message_assembly(parser_job_data *self) {
+  /* Remove previous message assembly if unfinished */
+  if (self->message_idx >= 0 &&
+      self->messages_storage[self->message_idx].started == 1 &&
+      self->messages_storage[self->message_idx].completed == 0) {
+    DEBUG(UTIL_NAME ": Removing unfinished assembly of previous message");
+    self->messages_storage[self->message_idx] = (message){0};
+    self->message_item_idx = 0;
+  } else
+    ++(self->message_idx);
+
+  /* Resize messages buffer if needed */
+  if (self->message_idx >= self->messages_max_len) {
+    INFO(UTIL_NAME ": Exceeded message buffer size: %lu", self->messages_max_len);
+    if (self->resize_message_buffer(self, self->messages_max_len +
+                                              MSG_STOR_INC_STEP) != 0) {
+      ERROR(UTIL_NAME ": Insufficient message buffer size: %lu. Remaining "
+                      "messages for this read will be skipped",
+            self->messages_max_len);
+      self->message_idx = self->messages_max_len;
+      return -1;
+    }
+  }
+  self->messages_storage[self->message_idx] = (message){0};
+  self->message_item_idx = 0;
+  self->messages_storage[self->message_idx].started = 1;
+  self->messages_storage[self->message_idx].completed = 0;
+  return 0;
+}
+
+static int resize_message_buffer(parser_job_data *self, size_t new_size) {
+  INFO(UTIL_NAME ": Resizing message buffer size to %lu", new_size);
+  void *new_storage = realloc(self->messages_storage,
+                              new_size * sizeof(*(self->messages_storage)));
+  if (new_storage == NULL) {
+    ERROR(UTIL_NAME ": Error while reallocating message buffer");
+    return -1;
+  }
+  self->messages_storage = new_storage;
+  self->messages_max_len = new_size;
+  /* Fill unused memory with 0 */
+  memset(self->messages_storage + self->message_idx, 0,
+         (self->messages_max_len - self->message_idx) *
+             sizeof(*(self->messages_storage)));
+  return 0;
+}
+
+static void end_message_assembly(parser_job_data *self) {
+  /* Checking mandatory items */
+  for (size_t i = 0; i < self->message_patterns_len; i++) {
+    if (self->message_patterns[i].is_mandatory &&
+        !(self->messages_storage[self->message_idx]
+              .matched_patterns_check[i])) {
+      WARNING(
+          UTIL_NAME
+          ": Mandatory message item pattern %s not found. Message discarded",
+          self->message_patterns[i].regex);
+      self->messages_storage[self->message_idx] = (message){0};
+      self->message_item_idx = 0;
+      if (self->message_idx > 0)
+        --(self->message_idx);
+      return;
+    }
+  }
+  self->messages_storage[self->message_idx].completed = 1;
+  self->message_item_idx = 0;
+}
+
+static int message_assembler(const char *row, char *const *matches,
+                             size_t matches_num, void *user_data) {
+  if (user_data == NULL) {
+    ERROR(UTIL_NAME ": Invalid user_data pointer");
+    return -1;
+  }
+  checked_match *cm = (checked_match *)user_data;
+  parser_job_data *parser_job = cm->parser_job;
+
+  if (cm->msg_pattern.submatch_idx < 0 ||
+      cm->msg_pattern.submatch_idx >= matches_num) {
+    ERROR(UTIL_NAME ": Invalid target submatch index: %d",
+          cm->msg_pattern.submatch_idx);
+    return -1;
+  }
+  if (parser_job->message_item_idx >=
+      STATIC_ARRAY_SIZE(parser_job->messages_storage[parser_job->message_idx]
+                            .message_items)) {
+    ERROR(UTIL_NAME ": Message items number exceeded. Forced message end.");
+    parser_job->end_message_assembly(parser_job);
+    return -1;
+  }
+
+  /* Every matched start pattern resets current message items and starts
+   * assembling new messages */
+  if (strcmp((cm->msg_pattern).regex,
+             parser_job->message_patterns[parser_job->start_idx].regex) == 0) {
+    DEBUG(UTIL_NAME ": Found beginning pattern");
+    if (parser_job->start_message_assembly(parser_job) != 0)
+      return -1;
+  }
+  /* Ignoring message items without corresponding start item */
+  if (parser_job->message_idx < 0 ||
+      parser_job->messages_storage[parser_job->message_idx].started == 0) {
+    DEBUG(UTIL_NAME ": Dropping item with no corresponding start element");
+    return 0;
+  }
+  /* Populate message items */
+  parser_job->message_item_assembly(parser_job, cm, matches);
+
+  /* Handle message ending */
+  if (strcmp((cm->msg_pattern).regex,
+             parser_job->message_patterns[parser_job->stop_idx].regex) == 0) {
+    DEBUG(UTIL_NAME ": Found ending pattern");
+    parser_job->end_message_assembly(parser_job);
+  }
+  return 0;
+}
+
+parser_job_data *message_parser_init(const char *filename,
+                                     unsigned int start_idx,
+                                     unsigned int stop_idx,
+                                     message_pattern message_patterns[],
+                                     size_t message_patterns_len) {
+  parser_job_data *parser_job = calloc(1, sizeof(*parser_job));
+  if (parser_job == NULL) {
+    ERROR(UTIL_NAME ": Error allocating parser_job");
+    return NULL;
+  }
+  parser_job->resize_message_buffer = resize_message_buffer;
+  parser_job->start_message_assembly = start_message_assembly;
+  parser_job->end_message_assembly = end_message_assembly;
+  parser_job->message_item_assembly = message_item_assembly;
+  parser_job->messages_max_len = MSG_STOR_INIT_LEN;
+  parser_job->filename = filename;
+  parser_job->start_idx = start_idx;
+  parser_job->stop_idx = stop_idx;
+  parser_job->message_idx = -1;
+  parser_job->message_patterns =
+      calloc(message_patterns_len, sizeof(*(parser_job->message_patterns)));
+  if (parser_job->message_patterns == NULL) {
+    ERROR(UTIL_NAME ": Error allocating message_patterns");
+    return NULL;
+  }
+  parser_job->messages_storage = calloc(
+      parser_job->messages_max_len, sizeof(*(parser_job->messages_storage)));
+  if (parser_job->messages_storage == NULL) {
+    ERROR(UTIL_NAME ": Error allocating messages_storage");
+    return NULL;
+  }
+  /* Crete own copy of regex patterns */
+  memcpy(parser_job->message_patterns, message_patterns,
+         sizeof(*(parser_job->message_patterns)) * message_patterns_len);
+  parser_job->message_patterns_len = message_patterns_len;
+  /* Init tail match */
+  parser_job->tm = tail_match_create(parser_job->filename);
+  if (parser_job->tm == NULL) {
+    ERROR(UTIL_NAME ": Error creating tail match");
+    return NULL;
+  }
+
+  for (size_t i = 0; i < message_patterns_len; i++) {
+    /* Create current_match container for passing regex info
+     * to callback function */
+    checked_match *current_match = calloc(1, sizeof(*current_match));
+    if (current_match == NULL) {
+      ERROR(UTIL_NAME ": Error allocating current_match");
+      return NULL;
+    }
+    current_match->parser_job = parser_job;
+    current_match->msg_pattern = message_patterns[i];
+    current_match->msg_pattern_idx = i;
+    /* Create callback */
+    cu_match_t *m = match_create_callback(
+        message_patterns[i].regex, message_patterns[i].excluderegex,
+        message_assembler, current_match, free);
+    if (m == NULL) {
+      ERROR(UTIL_NAME ": Error creating match callback");
+      return NULL;
+    }
+    if (tail_match_add_match(parser_job->tm, m, 0, 0, 0) != 0) {
+      ERROR(UTIL_NAME ": Error adding match callback");
+      return NULL;
+    }
+  }
+
+  return parser_job;
+}
+
+int message_parser_read(parser_job_data *parser_job, message **messages_storage,
+                        _Bool force_rewind) {
+  if (parser_job == NULL) {
+    ERROR(UTIL_NAME ": Invalid parser_job pointer");
+    return -1;
+  }
+
+  /* Finish incomplete message assembly in this read */
+  if (parser_job->message_idx >= 0 &&
+      parser_job->messages_storage[parser_job->message_idx].started &&
+      !(parser_job->messages_storage[parser_job->message_idx].completed)) {
+    INFO(UTIL_NAME ": Found incomplete message from previous read.");
+    message tmp_message = parser_job->messages_storage[parser_job->message_idx];
+    int tmp_message_item_idx = parser_job->message_item_idx;
+    memset(parser_job->messages_storage, 0,
+           parser_job->messages_max_len *
+               sizeof(*(parser_job->messages_storage)));
+    memcpy(parser_job->messages_storage, &tmp_message,
+           sizeof(*(parser_job->messages_storage)));
+    parser_job->message_item_idx = tmp_message_item_idx;
+    parser_job->message_idx = 0;
+  } else {
+    memset(parser_job->messages_storage, 0,
+           parser_job->messages_max_len *
+               sizeof(*(parser_job->messages_storage)));
+    parser_job->message_item_idx = 0;
+    parser_job->message_idx = -1;
+  }
+
+  int status = tail_match_read(parser_job->tm, force_rewind);
+  if (status != 0) {
+    ERROR(UTIL_NAME ": Error while parser read. Status: %d", status);
+    return -1;
+  }
+
+  /* Get no of messagess completed in this read */
+  int messages_completed = 0;
+  for (size_t i = 0; i < parser_job->messages_max_len; i++) {
+    if (parser_job->messages_storage[i].completed)
+      ++messages_completed;
+  }
+  *messages_storage = parser_job->messages_storage;
+  return messages_completed;
+}
+
+void message_parser_cleanup(parser_job_data *parser_job) {
+  if (parser_job == NULL) {
+    ERROR(UTIL_NAME ": Invalid parser_job pointer");
+    return;
+  }
+  sfree(parser_job->messages_storage);
+  sfree(parser_job->message_patterns);
+  if (parser_job->tm)
+    tail_match_destroy(parser_job->tm);
+  sfree(parser_job);
+}
diff --git a/src/utils_message_parser.h b/src/utils_message_parser.h
new file mode 100644
index 000000000..4d59e6a21
--- /dev/null
+++ b/src/utils_message_parser.h
@@ -0,0 +1,146 @@
+/*
+ * collectd - src/utils_message_parser.h
+ * MIT License
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ *
+ * Authors:
+ *   Krzysztof Matczak <krzysztofx.matczak@intel.com>
+ */
+
+#ifndef UTILS_MESSAGE_PARSER_H
+#define UTILS_MESSAGE_PARSER_H 1
+
+#include "utils_tail_match.h"
+
+typedef struct message_item_pattern_t {
+  /* User defined name for message item */
+  char *name;
+  /* Regular expression for finding out specific message item. The match result
+   * is taken only from submatch pointed by submatch_idx variable, so multiple
+   * submatches are supported. Example syntax:
+   * "(STRING1|STRING2)(.*)"
+   * For this example (STRING1|STRING2) is the first submatch and (.*) second
+   * so user can choose which one to store by setting submatch_idx argument.
+   */
+  char *regex;
+  /* Index of regex submatch that is stored as result. Value 0 takes whole
+   * match result */
+  int submatch_idx;
+  /* Regular expression that excludes string from further processing */
+  char *excluderegex;
+  /* Flag indicating if this item is mandatory for message validation */
+  _Bool is_mandatory;
+} message_pattern;
+
+typedef struct message_item_t {
+  char name[32];
+  char value[64];
+} message_item;
+
+typedef struct message_t {
+  message_item message_items[32];
+  int matched_patterns_check[32];
+  _Bool started;
+  _Bool completed;
+} message;
+
+typedef struct parser_job_data_t parser_job_data;
+
+/*
+ * NAME
+ *   message_parser_init
+ *
+ * DESCRIPTION
+ *   Configures unique message parser job. Performs necessary buffer
+ *   allocations. If there's need to create multiple parser jobs with different
+ *   parameters, this function may be called many times accordingly.
+ *
+ * PARAMETERS
+ *   `filename'   The name of the file to be parsed.
+ *   `start_idx'  Index of regular expression that indicates beginning of
+ *                message.
+ *                This index refers to 'message_patterns' array.
+ *   `stop_idx'   Index of the regular expression that indicates the end of
+ *                message. This index refers to 'message_patterns' array.
+ *   `message_patterns' Array of regular expression patterns for populating
+ *                      message items. Example:
+ *    message_pattern message_patterns[]= {
+ *      {.name = "START", .regex = "(Running trigger.*)", submatch_idx = 1,
+ *       .excluderegex = "kernel", .is_mandatory = 1},
+ *      {.name = "BANK", .regex = "BANK ([0-9]*)", submatch_idx = 1,
+ *       .excluderegex = "kernel", .is_mandatory = 1},
+ *      {.name = "TSC", .regex = "TSC ([a-z0-9]*)", submatch_idx = 1,
+ *       .excluderegex = "kernel", .is_mandatory = 1},
+ *      {.name = "MCA", .regex = "MCA: (.*)", submatch_idx = 1,
+ *       .excluderegex = "kernel", .is_mandatory = 0},
+ *      {.name = "END", .regex = "(CPUID Vendor.*)", submatch_idx = 1,
+ *       .excluderegex = "kernel", .is_mandatory = 1}
+ *    };
+ *
+ *   `message_patterns_len' Length of message_patterns array.
+ *
+ * RETURN VALUE
+ *   Returns NULL upon failure, pointer to new parser job otherwise.
+ */
+parser_job_data *message_parser_init(const char *filename,
+                                     unsigned int start_idx,
+                                     unsigned int stop_idx,
+                                     message_pattern message_patterns[],
+                                     size_t message_patterns_len);
+
+/*
+ * NAME
+ *   message_parser_read
+ *
+ * DESCRIPTION
+ *   Collects all new messages matching criteria set in 'message_parser_init'.
+ *   New messages means those reported or finished after previous call for this
+ *   function.
+ *
+ * PARAMETERS
+ *   `parser_job' Pointer to parser job to read.
+ *   `messages_storage' Pointer to be set to internally allocated messages
+ *                      storage.
+ *   `force_rewind' If value set to non zero, file will be parsed from
+ *                  beginning, otherwise tailing end of file.
+ *
+ * RETURN VALUE
+ *   Returns -1 upon failure, number of messages collected from last read
+ *   otherwise.
+ */
+int message_parser_read(parser_job_data *parser_job, message **messages_storage,
+                        _Bool force_rewind);
+
+/*
+ * NAME
+ *   message_parser_cleanup
+ *
+ * DESCRIPTION
+ *   Performs parser job resource deallocation and cleanup.
+ *
+ * PARAMETERS
+ *   `parser_job' Pointer to parser job to be cleaned up.
+ *
+ */
+void message_parser_cleanup(parser_job_data *parser_job);
+
+#endif /* UTILS_MESSAGE_PARSER_H */
diff --git a/src/utils_tail_match.c b/src/utils_tail_match.c
index 0e5a86114..a0220abdd 100644
--- a/src/utils_tail_match.c
+++ b/src/utils_tail_match.c
@@ -306,12 +306,12 @@ out:
   return status;
 } /* int tail_match_add_match_simple */
 
-int tail_match_read(cu_tail_match_t *obj) {
+int tail_match_read(cu_tail_match_t *obj, _Bool force_rewind) {
   char buffer[4096];
   int status;
 
   status = cu_tail_read(obj->tail, buffer, sizeof(buffer), tail_callback,
-                        (void *)obj);
+                        (void *)obj, force_rewind);
   if (status != 0) {
     ERROR("tail_match: cu_tail_read failed.");
     return status;
diff --git a/src/utils_tail_match.h b/src/utils_tail_match.h
index 771c24109..0f510294d 100644
--- a/src/utils_tail_match.h
+++ b/src/utils_tail_match.h
@@ -135,6 +135,6 @@ int tail_match_add_match_simple(cu_tail_match_t *obj, const char *regex,
  * RETURN VALUE
  *   Zero on success, nonzero on failure.
  */
-int tail_match_read(cu_tail_match_t *obj);
+int tail_match_read(cu_tail_match_t *obj, _Bool force_rewind);
 
 #endif /* UTILS_TAIL_MATCH_H */