X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fprocevent.c;h=850f7742bc956b31d2027b820a7b2c84b7f08755;hb=34714af37d83d62189cda12ff336acbaf7f6a812;hp=e05fa0c8e79bba93a9a3e5c3606a404238cc3ecb;hpb=614f026c29b712c3a492b5cc82e48814d1609a99;p=collectd.git

diff --git a/src/procevent.c b/src/procevent.c
index e05fa0c8..850f7742 100644
--- a/src/procevent.c
+++ b/src/procevent.c
@@ -29,10 +29,10 @@
 #include "common.h"
 #include "plugin.h"
 #include "utils_complain.h"
+#include "utils_ignorelist.h"
 
 #include <errno.h>
 #include <pthread.h>
-#include <regex.h>
 #include <stdio.h>
 #include <string.h>
 #include <sys/socket.h>
@@ -48,12 +48,55 @@
 #include <stdlib.h>
 #include <string.h>
 
+#include <yajl/yajl_common.h>
+#include <yajl/yajl_gen.h>
+#if HAVE_YAJL_YAJL_VERSION_H
+#include <yajl/yajl_version.h>
+#endif
+#if defined(YAJL_MAJOR) && (YAJL_MAJOR > 1)
+#define HAVE_YAJL_V2 1
+#endif
+
 #define PROCEVENT_EXITED 0
 #define PROCEVENT_STARTED 1
-#define PROCEVENT_FIELDS 3 // pid, status, extra
+#define PROCEVENT_FIELDS 4 // pid, status, extra, timestamp
 #define BUFSIZE 512
 #define PROCDIR "/proc"
-#define PROCEVENT_REGEX_MATCHES 1
+
+#define PROCEVENT_DOMAIN_FIELD "domain"
+#define PROCEVENT_DOMAIN_VALUE "fault"
+#define PROCEVENT_EVENT_ID_FIELD "eventId"
+#define PROCEVENT_EVENT_NAME_FIELD "eventName"
+#define PROCEVENT_EVENT_NAME_DOWN_VALUE "down"
+#define PROCEVENT_EVENT_NAME_UP_VALUE "up"
+#define PROCEVENT_LAST_EPOCH_MICROSEC_FIELD "lastEpochMicrosec"
+#define PROCEVENT_PRIORITY_FIELD "priority"
+#define PROCEVENT_PRIORITY_VALUE "high"
+#define PROCEVENT_REPORTING_ENTITY_NAME_FIELD "reportingEntityName"
+#define PROCEVENT_REPORTING_ENTITY_NAME_VALUE "collectd procevent plugin"
+#define PROCEVENT_SEQUENCE_FIELD "sequence"
+#define PROCEVENT_SEQUENCE_VALUE "0"
+#define PROCEVENT_SOURCE_NAME_FIELD "sourceName"
+#define PROCEVENT_START_EPOCH_MICROSEC_FIELD "startEpochMicrosec"
+#define PROCEVENT_VERSION_FIELD "version"
+#define PROCEVENT_VERSION_VALUE "1.0"
+
+#define PROCEVENT_ALARM_CONDITION_FIELD "alarmCondition"
+#define PROCEVENT_ALARM_INTERFACE_A_FIELD "alarmInterfaceA"
+#define PROCEVENT_EVENT_SEVERITY_FIELD "eventSeverity"
+#define PROCEVENT_EVENT_SEVERITY_CRITICAL_VALUE "CRITICAL"
+#define PROCEVENT_EVENT_SEVERITY_NORMAL_VALUE "NORMAL"
+#define PROCEVENT_EVENT_SOURCE_TYPE_FIELD "eventSourceType"
+#define PROCEVENT_EVENT_SOURCE_TYPE_VALUE "process"
+#define PROCEVENT_FAULT_FIELDS_FIELD "faultFields"
+#define PROCEVENT_FAULT_FIELDS_VERSION_FIELD "faultFieldsVersion"
+#define PROCEVENT_FAULT_FIELDS_VERSION_VALUE "1.0"
+#define PROCEVENT_SPECIFIC_PROBLEM_FIELD "specificProblem"
+#define PROCEVENT_SPECIFIC_PROBLEM_DOWN_VALUE "down"
+#define PROCEVENT_SPECIFIC_PROBLEM_UP_VALUE "up"
+#define PROCEVENT_VF_STATUS_FIELD "vfStatus"
+#define PROCEVENT_VF_STATUS_CRITICAL_VALUE "Ready to terminate"
+#define PROCEVENT_VF_STATUS_NORMAL_VALUE "Active"
 
 /*
  * Private data types
@@ -63,17 +106,14 @@ typedef struct {
   int head;
   int tail;
   int maxLen;
-  int **buffer;
+  long long unsigned int **buffer;
 } circbuf_t;
 
 struct processlist_s {
   char *process;
-  char *process_regex;
-
-  regex_t process_regex_obj;
 
-  uint32_t is_regex;
-  uint32_t pid;
+  long pid;
+  int32_t last_status;
 
   struct processlist_s *next;
 };
@@ -82,106 +122,411 @@ typedef struct processlist_s processlist_t;
 /*
  * Private variables
  */
-
-static int procevent_thread_loop = 0;
-static int procevent_thread_error = 0;
-static pthread_t procevent_thread_id;
-static pthread_mutex_t procevent_lock = PTHREAD_MUTEX_INITIALIZER;
+static ignorelist_t *ignorelist = NULL;
+
+static int procevent_netlink_thread_loop = 0;
+static int procevent_netlink_thread_error = 0;
+static pthread_t procevent_netlink_thread_id;
+static int procevent_dequeue_thread_loop = 0;
+static pthread_t procevent_dequeue_thread_id;
+static pthread_mutex_t procevent_thread_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t procevent_data_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t procevent_cond = PTHREAD_COND_INITIALIZER;
-static pthread_mutex_t procevent_list_lock = PTHREAD_MUTEX_INITIALIZER;
 static int nl_sock = -1;
 static int buffer_length;
 static circbuf_t ring;
 static processlist_t *processlist_head = NULL;
+static int event_id = 0;
 
-static const char *config_keys[] = {"BufferLength", "Process", "RegexProcess"};
+static const char *config_keys[] = {"BufferLength", "Process", "ProcessRegex"};
 static int config_keys_num = STATIC_ARRAY_SIZE(config_keys);
 
 /*
+ * Prototypes
+ */
+
+static void procevent_dispatch_notification(long pid, const char *type,
+                                            gauge_t value, char *process,
+                                            long long unsigned int timestamp);
+
+/*
  * Private functions
  */
 
+static int gen_message_payload(int state, long pid, char *process,
+                               long long unsigned int timestamp, char **buf) {
+  const unsigned char *buf2;
+  yajl_gen g;
+  char json_str[DATA_MAX_NAME_LEN];
+
+#if !defined(HAVE_YAJL_V2)
+  yajl_gen_config conf = {};
+
+  conf.beautify = 0;
+#endif
+
+#if HAVE_YAJL_V2
+  size_t len;
+  g = yajl_gen_alloc(NULL);
+  yajl_gen_config(g, yajl_gen_beautify, 0);
+#else
+  unsigned int len;
+  g = yajl_gen_alloc(&conf, NULL);
+#endif
+
+  yajl_gen_clear(g);
+
+  // *** BEGIN common event header ***
+
+  if (yajl_gen_map_open(g) != yajl_gen_status_ok)
+    goto err;
+
+  // domain
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_DOMAIN_FIELD,
+                      strlen(PROCEVENT_DOMAIN_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_DOMAIN_VALUE,
+                      strlen(PROCEVENT_DOMAIN_VALUE)) != yajl_gen_status_ok)
+    goto err;
+
+  // eventId
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_EVENT_ID_FIELD,
+                      strlen(PROCEVENT_EVENT_ID_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  event_id = event_id + 1;
+  int event_id_len = sizeof(char) * sizeof(int) * 4 + 1;
+  memset(json_str, '\0', DATA_MAX_NAME_LEN);
+  snprintf(json_str, event_id_len, "%d", event_id);
+
+  if (yajl_gen_number(g, json_str, strlen(json_str)) != yajl_gen_status_ok) {
+    goto err;
+  }
+
+  // eventName
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_EVENT_NAME_FIELD,
+                      strlen(PROCEVENT_EVENT_NAME_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  int event_name_len = 0;
+  event_name_len = event_name_len + (sizeof(char) * sizeof(int) * 4); // pid
+  event_name_len = event_name_len + strlen(process);      // process name
+  event_name_len = event_name_len + (state == 0 ? 4 : 2); // "down" or "up"
+  event_name_len = event_name_len +
+                   13; // "process", 3 spaces, 2 parentheses and null-terminator
+  memset(json_str, '\0', DATA_MAX_NAME_LEN);
+  snprintf(json_str, event_name_len, "process %s (%ld) %s", process, pid,
+           (state == 0 ? PROCEVENT_EVENT_NAME_DOWN_VALUE
+                       : PROCEVENT_EVENT_NAME_UP_VALUE));
+
+  if (yajl_gen_string(g, (u_char *)json_str, strlen(json_str)) !=
+      yajl_gen_status_ok) {
+    goto err;
+  }
+
+  // lastEpochMicrosec
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_LAST_EPOCH_MICROSEC_FIELD,
+                      strlen(PROCEVENT_LAST_EPOCH_MICROSEC_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  int last_epoch_microsec_len =
+      sizeof(char) * sizeof(long long unsigned int) * 4 + 1;
+  memset(json_str, '\0', DATA_MAX_NAME_LEN);
+  snprintf(json_str, last_epoch_microsec_len, "%llu",
+           (long long unsigned int)CDTIME_T_TO_US(cdtime()));
+
+  if (yajl_gen_number(g, json_str, strlen(json_str)) != yajl_gen_status_ok) {
+    goto err;
+  }
+
+  // priority
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_PRIORITY_FIELD,
+                      strlen(PROCEVENT_PRIORITY_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_PRIORITY_VALUE,
+                      strlen(PROCEVENT_PRIORITY_VALUE)) != yajl_gen_status_ok)
+    goto err;
+
+  // reportingEntityName
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_REPORTING_ENTITY_NAME_FIELD,
+                      strlen(PROCEVENT_REPORTING_ENTITY_NAME_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_REPORTING_ENTITY_NAME_VALUE,
+                      strlen(PROCEVENT_REPORTING_ENTITY_NAME_VALUE)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  // sequence
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_SEQUENCE_FIELD,
+                      strlen(PROCEVENT_SEQUENCE_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_number(g, PROCEVENT_SEQUENCE_VALUE,
+                      strlen(PROCEVENT_SEQUENCE_VALUE)) != yajl_gen_status_ok)
+    goto err;
+
+  // sourceName
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_SOURCE_NAME_FIELD,
+                      strlen(PROCEVENT_SOURCE_NAME_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(g, (u_char *)process, strlen(process)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  // startEpochMicrosec
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_START_EPOCH_MICROSEC_FIELD,
+                      strlen(PROCEVENT_START_EPOCH_MICROSEC_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  int start_epoch_microsec_len =
+      sizeof(char) * sizeof(long long unsigned int) * 4 + 1;
+  memset(json_str, '\0', DATA_MAX_NAME_LEN);
+  snprintf(json_str, start_epoch_microsec_len, "%llu",
+           (long long unsigned int)timestamp);
+
+  if (yajl_gen_number(g, json_str, strlen(json_str)) != yajl_gen_status_ok) {
+    goto err;
+  }
+
+  // version
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_VERSION_FIELD,
+                      strlen(PROCEVENT_VERSION_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_number(g, PROCEVENT_VERSION_VALUE,
+                      strlen(PROCEVENT_VERSION_VALUE)) != yajl_gen_status_ok)
+    goto err;
+
+  // *** END common event header ***
+
+  // *** BEGIN fault fields ***
+
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_FAULT_FIELDS_FIELD,
+                      strlen(PROCEVENT_FAULT_FIELDS_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_map_open(g) != yajl_gen_status_ok)
+    goto err;
+
+  // alarmCondition
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_ALARM_CONDITION_FIELD,
+                      strlen(PROCEVENT_ALARM_CONDITION_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  int alarm_condition_len = 0;
+  alarm_condition_len =
+      alarm_condition_len + (sizeof(char) * sizeof(int) * 4);  // pid
+  alarm_condition_len = alarm_condition_len + strlen(process); // process name
+  alarm_condition_len =
+      alarm_condition_len + 25; // "process", "state", "change", 4 spaces, 2
+                                // parentheses and null-terminator
+  memset(json_str, '\0', DATA_MAX_NAME_LEN);
+  snprintf(json_str, alarm_condition_len, "process %s (%ld) state change",
+           process, pid);
+
+  if (yajl_gen_string(g, (u_char *)json_str, strlen(json_str)) !=
+      yajl_gen_status_ok) {
+    goto err;
+  }
+
+  // alarmInterfaceA
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_ALARM_INTERFACE_A_FIELD,
+                      strlen(PROCEVENT_ALARM_INTERFACE_A_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(g, (u_char *)process, strlen(process)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  // eventSeverity
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_EVENT_SEVERITY_FIELD,
+                      strlen(PROCEVENT_EVENT_SEVERITY_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(
+          g, (u_char *)(state == 0 ? PROCEVENT_EVENT_SEVERITY_CRITICAL_VALUE
+                                   : PROCEVENT_EVENT_SEVERITY_NORMAL_VALUE),
+          strlen((state == 0 ? PROCEVENT_EVENT_SEVERITY_CRITICAL_VALUE
+                             : PROCEVENT_EVENT_SEVERITY_NORMAL_VALUE))) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  // eventSourceType
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_EVENT_SOURCE_TYPE_FIELD,
+                      strlen(PROCEVENT_EVENT_SOURCE_TYPE_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_EVENT_SOURCE_TYPE_VALUE,
+                      strlen(PROCEVENT_EVENT_SOURCE_TYPE_VALUE)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  // faultFieldsVersion
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_FAULT_FIELDS_VERSION_FIELD,
+                      strlen(PROCEVENT_FAULT_FIELDS_VERSION_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_number(g, PROCEVENT_FAULT_FIELDS_VERSION_VALUE,
+                      strlen(PROCEVENT_FAULT_FIELDS_VERSION_VALUE)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  // specificProblem
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_SPECIFIC_PROBLEM_FIELD,
+                      strlen(PROCEVENT_SPECIFIC_PROBLEM_FIELD)) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  int specific_problem_len = 0;
+  specific_problem_len =
+      specific_problem_len + (sizeof(char) * sizeof(int) * 4);   // pid
+  specific_problem_len = specific_problem_len + strlen(process); // process name
+  specific_problem_len =
+      specific_problem_len + (state == 0 ? 4 : 2); // "down" or "up"
+  specific_problem_len =
+      specific_problem_len +
+      13; // "process", 3 spaces, 2 parentheses and null-terminator
+  memset(json_str, '\0', DATA_MAX_NAME_LEN);
+  snprintf(json_str, specific_problem_len, "process %s (%ld) %s", process, pid,
+           (state == 0 ? PROCEVENT_SPECIFIC_PROBLEM_DOWN_VALUE
+                       : PROCEVENT_SPECIFIC_PROBLEM_UP_VALUE));
+
+  if (yajl_gen_string(g, (u_char *)json_str, strlen(json_str)) !=
+      yajl_gen_status_ok) {
+    goto err;
+  }
+
+  // vfStatus
+  if (yajl_gen_string(g, (u_char *)PROCEVENT_VF_STATUS_FIELD,
+                      strlen(PROCEVENT_VF_STATUS_FIELD)) != yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_string(
+          g, (u_char *)(state == 0 ? PROCEVENT_VF_STATUS_CRITICAL_VALUE
+                                   : PROCEVENT_VF_STATUS_NORMAL_VALUE),
+          strlen((state == 0 ? PROCEVENT_VF_STATUS_CRITICAL_VALUE
+                             : PROCEVENT_VF_STATUS_NORMAL_VALUE))) !=
+      yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_map_close(g) != yajl_gen_status_ok)
+    goto err;
+
+  // *** END fault fields ***
+
+  if (yajl_gen_map_close(g) != yajl_gen_status_ok)
+    goto err;
+
+  if (yajl_gen_get_buf(g, &buf2, &len) != yajl_gen_status_ok)
+    goto err;
+
+  *buf = strdup((char *)buf2);
+
+  if (*buf == NULL) {
+    ERROR("procevent plugin: strdup failed during gen_message_payload: %s",
+          STRERRNO);
+    goto err;
+  }
+
+  yajl_gen_free(g);
+
+  return 0;
+
+err:
+  yajl_gen_free(g);
+  ERROR("procevent plugin: gen_message_payload failed to generate JSON");
+  return -1;
+}
+
 // Does /proc/<pid>/comm contain a process name we are interested in?
-static processlist_t *process_check(int pid) {
-  int len, is_match, status, retval;
+// NOTE: Caller MUST hold procevent_data_lock when calling this function
+static processlist_t *process_check(long pid) {
   char file[BUFSIZE];
-  FILE *fh;
-  char buffer[BUFSIZE];
-  regmatch_t matches[PROCEVENT_REGEX_MATCHES];
 
-  len = snprintf(file, sizeof(file), PROCDIR "/%d/comm", pid);
+  int len = snprintf(file, sizeof(file), PROCDIR "/%ld/comm", pid);
 
   if ((len < 0) || (len >= BUFSIZE)) {
     WARNING("procevent process_check: process name too large");
     return NULL;
   }
 
+  FILE *fh;
+
   if (NULL == (fh = fopen(file, "r"))) {
     // No /proc/<pid>/comm for this pid, just ignore
-    DEBUG("procevent plugin: no comm file available for pid %d", pid);
+    DEBUG("procevent plugin: no comm file available for pid %ld", pid);
     return NULL;
   }
 
-  retval = fscanf(fh, "%[^\n]", buffer);
+  char buffer[BUFSIZE];
+  int retval = fscanf(fh, "%[^\n]", buffer);
 
   if (retval < 0) {
-    WARNING("procevent process_check: unable to read comm file for pid %d",
+    WARNING("procevent process_check: unable to read comm file for pid %ld",
             pid);
+    fclose(fh);
     return NULL;
   }
 
+  // Now that we have the process name in the buffer, check if we are
+  // even interested in it
+  if (ignorelist_match(ignorelist, buffer) != 0) {
+    DEBUG("procevent process_check: ignoring process %s (%ld)", buffer, pid);
+    fclose(fh);
+    return NULL;
+  }
+
+  if (fh != NULL) {
+    fclose(fh);
+    fh = NULL;
+  }
+
   //
   // Go through the processlist linked list and look for the process name
   // in /proc/<pid>/comm.  If found:
-  // 1. If pl->pid is -1, then set pl->pid to <pid>
+  // 1. If pl->pid is -1, then set pl->pid to <pid> (and return that object)
   // 2. If pl->pid is not -1, then another <process name> process was already
   //    found.  If <pid> == pl->pid, this is an old match, so do nothing.
-  //    If the <pid> is different, however,  make a new processlist_t and
+  //    If the <pid> is different, however, make a new processlist_t and
   //    associate <pid> with it (with the same process name as the existing).
   //
 
-  pthread_mutex_lock(&procevent_list_lock);
-
-  processlist_t *pl;
   processlist_t *match = NULL;
 
-  for (pl = processlist_head; pl != NULL; pl = pl->next) {
-    if (pl->is_regex != 0) {
-      is_match = (regexec(&pl->process_regex_obj, buffer,
-                          PROCEVENT_REGEX_MATCHES, matches, 0) == 0
-                      ? 1
-                      : 0);
-    } else {
-      is_match = (strcmp(buffer, pl->process) == 0 ? 1 : 0);
-    }
+  for (processlist_t *pl = processlist_head; pl != NULL; pl = pl->next) {
+
+    int is_match = (strcmp(buffer, pl->process) == 0 ? 1 : 0);
 
     if (is_match == 1) {
-      DEBUG("procevent plugin: process %d name match (pattern: %s) for %s", pid,
-            (pl->is_regex == 0 ? pl->process : pl->process_regex), buffer);
-
-      if (pl->is_regex == 1) {
-        // If this is a regex name, copy the actual process name into the object
-        // for cleaner log reporting
-
-        if (pl->process != NULL)
-          sfree(pl->process);
-        pl->process = strdup(buffer);
-        if (pl->process == NULL) {
-          char errbuf[1024];
-          ERROR("procevent plugin: strdup failed during process_check: %s",
-                sstrerror(errno, errbuf, sizeof(errbuf)));
-          pthread_mutex_unlock(&procevent_list_lock);
-          return NULL;
-        }
-      }
+      DEBUG("procevent plugin: process %ld name match for %s", pid, buffer);
 
       if (pl->pid == pid) {
         // this is a match, and we've already stored the exact pid/name combo
+        DEBUG("procevent plugin: found exact match with name %s, PID %ld for "
+              "incoming PID %ld",
+              pl->process, pl->pid, pid);
         match = pl;
         break;
       } else if (pl->pid == -1) {
         // this is a match, and we've found a candidate processlist_t to store
         // this new pid/name combo
+        DEBUG("procevent plugin: reusing pl object with PID %ld for incoming "
+              "PID %ld",
+              pl->pid, pid);
         pl->pid = pid;
         match = pl;
         break;
@@ -189,67 +534,41 @@ static processlist_t *process_check(int pid) {
         // this is a match, but another instance of this process has already
         // claimed this pid/name combo,
         // so keep looking
+        DEBUG("procevent plugin: found pl object with matching name for "
+              "incoming PID %ld, but object is in use by PID %ld",
+              pid, pl->pid);
         match = pl;
         continue;
       }
     }
   }
 
-  if (match != NULL && match->pid != -1 && match->pid != pid) {
+  if (match == NULL ||
+      (match != NULL && match->pid != -1 && match->pid != pid)) {
+    // if there wasn't an existing match, OR
     // if there was a match but the associated processlist_t object already
     // contained a pid/name combo,
     // then make a new one and add it to the linked list
 
-    DEBUG(
-        "procevent plugin: allocating new processlist_t object for PID %d (%s)",
-        pid, match->process);
-
-    processlist_t *pl2;
-    char *process;
-    char *process_regex;
+    DEBUG("procevent plugin: allocating new processlist_t object for PID %ld "
+          "(%s)",
+          pid, buffer);
 
-    pl2 = malloc(sizeof(*pl2));
+    processlist_t *pl2 = calloc(1, sizeof(*pl2));
     if (pl2 == NULL) {
-      char errbuf[1024];
-      ERROR("procevent plugin: malloc failed during process_check: %s",
-            sstrerror(errno, errbuf, sizeof(errbuf)));
-      pthread_mutex_unlock(&procevent_list_lock);
+      ERROR("procevent plugin: calloc failed during process_check: %s",
+            STRERRNO);
       return NULL;
     }
 
-    process = strdup(match->process);
+    char *process = strdup(buffer);
     if (process == NULL) {
-      char errbuf[1024];
       sfree(pl2);
       ERROR("procevent plugin: strdup failed during process_check: %s",
-            sstrerror(errno, errbuf, sizeof(errbuf)));
-      pthread_mutex_unlock(&procevent_list_lock);
+            STRERRNO);
       return NULL;
     }
 
-    if (match->is_regex == 1) {
-      pl2->is_regex = 1;
-      status =
-          regcomp(&pl2->process_regex_obj, match->process_regex, REG_EXTENDED);
-
-      if (status != 0) {
-        ERROR("procevent plugin: invalid regular expression: %s",
-              match->process_regex);
-        return NULL;
-      }
-
-      process_regex = strdup(match->process_regex);
-      if (process_regex == NULL) {
-        char errbuf[1024];
-        sfree(pl);
-        ERROR("procevent plugin: strdup failed during process_check: %s",
-              sstrerror(errno, errbuf, sizeof(errbuf)));
-        return NULL;
-      }
-
-      pl2->process_regex = process_regex;
-    }
-
     pl2->process = process;
     pl2->pid = pid;
     pl2->next = processlist_head;
@@ -258,37 +577,29 @@ static processlist_t *process_check(int pid) {
     match = pl2;
   }
 
-  pthread_mutex_unlock(&procevent_list_lock);
-
-  if (fh != NULL) {
-    fclose(fh);
-    fh = NULL;
-  }
-
   return match;
 }
 
 // Does our map have this PID or name?
-static processlist_t *process_map_check(int pid, char *process) {
-  processlist_t *pl;
-
-  pthread_mutex_lock(&procevent_list_lock);
-
-  for (pl = processlist_head; pl != NULL; pl = pl->next) {
+// NOTE: Caller MUST hold procevent_data_lock when calling this function
+static processlist_t *process_map_check(long pid, char *process) {
+  for (processlist_t *pl = processlist_head; pl != NULL; pl = pl->next) {
     int match_pid = 0;
-    int match_process = 0;
-    int match = 0;
 
     if (pid > 0) {
       if (pl->pid == pid)
         match_pid = 1;
     }
 
+    int match_process = 0;
+
     if (process != NULL) {
       if (strcmp(pl->process, process) == 0)
         match_process = 1;
     }
 
+    int match = 0;
+
     if (pid > 0 && process == NULL && match_pid == 1)
       match = 1;
     else if (pid < 0 && process != NULL && match_process == 1)
@@ -297,47 +608,31 @@ static processlist_t *process_map_check(int pid, char *process) {
       match = 1;
 
     if (match == 1) {
-      pthread_mutex_unlock(&procevent_list_lock);
       return pl;
     }
   }
 
-  pthread_mutex_unlock(&procevent_list_lock);
-
   return NULL;
 }
 
 static int process_map_refresh(void) {
-  DIR *proc;
-
   errno = 0;
-  proc = opendir(PROCDIR);
+  DIR *proc = opendir(PROCDIR);
+
   if (proc == NULL) {
-    char errbuf[1024];
-    ERROR("procevent plugin: fopen (%s): %s", PROCDIR,
-          sstrerror(errno, errbuf, sizeof(errbuf)));
+    ERROR("procevent plugin: fopen (%s): %s", PROCDIR, STRERRNO);
     return -1;
   }
 
   while (42) {
-    struct dirent *dent;
-    int len;
-    char file[BUFSIZE];
-
-    struct stat statbuf;
-
-    int status;
-
     errno = 0;
-    dent = readdir(proc);
+    struct dirent *dent = readdir(proc);
     if (dent == NULL) {
-      char errbuf[4096];
-
       if (errno == 0) /* end of directory */
         break;
 
       ERROR("procevent plugin: failed to read directory %s: %s", PROCDIR,
-            sstrerror(errno, errbuf, sizeof(errbuf)));
+            STRERRNO);
       closedir(proc);
       return -1;
     }
@@ -345,15 +640,17 @@ static int process_map_refresh(void) {
     if (dent->d_name[0] == '.')
       continue;
 
-    len = snprintf(file, sizeof(file), PROCDIR "/%s", dent->d_name);
+    char file[BUFSIZE];
+
+    int len = snprintf(file, sizeof(file), PROCDIR "/%s", dent->d_name);
     if ((len < 0) || (len >= BUFSIZE))
       continue;
 
-    status = stat(file, &statbuf);
+    struct stat statbuf;
+
+    int status = stat(file, &statbuf);
     if (status != 0) {
-      char errbuf[4096];
-      WARNING("procevent plugin: stat (%s) failed: %s", file,
-              sstrerror(errno, errbuf, sizeof(errbuf)));
+      WARNING("procevent plugin: stat (%s) failed: %s", file, STRERRNO);
       continue;
     }
 
@@ -379,7 +676,9 @@ static int process_map_refresh(void) {
     // Check if we need to store this pid/name combo in our processlist_t linked
     // list
     int this_pid = atoi(dent->d_name);
+    pthread_mutex_lock(&procevent_data_lock);
     processlist_t *pl = process_check(this_pid);
+    pthread_mutex_unlock(&procevent_data_lock);
 
     if (pl != NULL)
       DEBUG("procevent plugin: process map refreshed for PID %d and name %s",
@@ -392,22 +691,19 @@ static int process_map_refresh(void) {
 }
 
 static int nl_connect() {
-  int rc;
-  struct sockaddr_nl sa_nl;
+  struct sockaddr_nl sa_nl = {
+      .nl_family = AF_NETLINK, .nl_groups = CN_IDX_PROC, .nl_pid = getpid(),
+  };
 
   nl_sock = socket(PF_NETLINK, SOCK_DGRAM, NETLINK_CONNECTOR);
   if (nl_sock == -1) {
-    ERROR("procevent plugin: socket open failed.");
+    ERROR("procevent plugin: socket open failed: %d", errno);
     return -1;
   }
 
-  sa_nl.nl_family = AF_NETLINK;
-  sa_nl.nl_groups = CN_IDX_PROC;
-  sa_nl.nl_pid = getpid();
-
-  rc = bind(nl_sock, (struct sockaddr *)&sa_nl, sizeof(sa_nl));
+  int rc = bind(nl_sock, (struct sockaddr *)&sa_nl, sizeof(sa_nl));
   if (rc == -1) {
-    ERROR("procevent plugin: socket bind failed.");
+    ERROR("procevent plugin: socket bind failed: %d", errno);
     close(nl_sock);
     return -1;
   }
@@ -416,7 +712,6 @@ static int nl_connect() {
 }
 
 static int set_proc_ev_listen(bool enable) {
-  int rc;
   struct __attribute__((aligned(NLMSG_ALIGNTO))) {
     struct nlmsghdr nl_hdr;
     struct __attribute__((__packed__)) {
@@ -436,21 +731,19 @@ static int set_proc_ev_listen(bool enable) {
 
   nlcn_msg.cn_mcast = enable ? PROC_CN_MCAST_LISTEN : PROC_CN_MCAST_IGNORE;
 
-  rc = send(nl_sock, &nlcn_msg, sizeof(nlcn_msg), 0);
+  int rc = send(nl_sock, &nlcn_msg, sizeof(nlcn_msg), 0);
   if (rc == -1) {
-    ERROR("procevent plugin: subscribing to netlink process events failed.");
+    ERROR("procevent plugin: subscribing to netlink process events failed: %d",
+          errno);
     return -1;
   }
 
   return 0;
 }
 
+// Read from netlink socket and write to ring buffer
 static int read_event() {
-  int status;
-  int ret = 0;
-  int proc_id = -1;
-  int proc_status = -1;
-  int proc_extra = -1;
+  int recv_flags = MSG_DONTWAIT;
   struct __attribute__((aligned(NLMSG_ALIGNTO))) {
     struct nlmsghdr nl_hdr;
     struct __attribute__((__packed__)) {
@@ -460,415 +753,580 @@ static int read_event() {
   } nlcn_msg;
 
   if (nl_sock == -1)
-    return ret;
+    return 0;
 
-  status = recv(nl_sock, &nlcn_msg, sizeof(nlcn_msg), 0);
+  while (42) {
+    pthread_mutex_lock(&procevent_thread_lock);
 
-  if (status == 0) {
-    return 0;
-  } else if (status == -1) {
-    if (errno != EINTR) {
-      ERROR("procevent plugin: socket receive error: %d", errno);
-      return -1;
+    if (procevent_netlink_thread_loop <= 0) {
+      pthread_mutex_unlock(&procevent_thread_lock);
+      return 0;
+    }
+
+    pthread_mutex_unlock(&procevent_thread_lock);
+
+    int status = recv(nl_sock, &nlcn_msg, sizeof(nlcn_msg), recv_flags);
+
+    if (status == 0) {
+      return 0;
+    } else if (status < 0) {
+      if (errno == EAGAIN || errno == EWOULDBLOCK) {
+        pthread_mutex_lock(&procevent_data_lock);
+
+        // There was nothing more to receive for now, so...
+        // If ring head does not equal ring tail, there is data
+        // in the ring buffer for the dequeue thread to read, so
+        // signal it
+        if (ring.head != ring.tail)
+          pthread_cond_signal(&procevent_cond);
+
+        pthread_mutex_unlock(&procevent_data_lock);
+
+        // Since there was nothing to receive, set recv to block and
+        // try again
+        recv_flags = 0;
+        continue;
+      } else if (errno != EINTR) {
+        ERROR("procevent plugin: socket receive error: %d", errno);
+        return -1;
+      } else {
+        // Interrupt, so just return
+        return 0;
+      }
+    }
+
+    // We successfully received a message, so don't block on the next
+    // read in case there are more (and if there aren't, it will be
+    // handled above in the error-checking)
+    recv_flags = MSG_DONTWAIT;
+
+    int proc_id = -1;
+    int proc_status = -1;
+    int proc_extra = -1;
+
+    switch (nlcn_msg.proc_ev.what) {
+    case PROC_EVENT_NONE:
+    case PROC_EVENT_FORK:
+    case PROC_EVENT_UID:
+    case PROC_EVENT_GID:
+      // Not of interest in current version
+      break;
+    case PROC_EVENT_EXEC:
+      proc_status = PROCEVENT_STARTED;
+      proc_id = nlcn_msg.proc_ev.event_data.exec.process_pid;
+      break;
+    case PROC_EVENT_EXIT:
+      proc_id = nlcn_msg.proc_ev.event_data.exit.process_pid;
+      proc_status = PROCEVENT_EXITED;
+      proc_extra = nlcn_msg.proc_ev.event_data.exit.exit_code;
+      break;
+    default:
+      break;
+    }
+
+    // If we're interested in this process status event, place the event
+    // in the ring buffer for consumption by the main polling thread.
+
+    if (proc_status != -1) {
+      pthread_mutex_lock(&procevent_data_lock);
+
+      int next = ring.head + 1;
+      if (next >= ring.maxLen)
+        next = 0;
+
+      if (next == ring.tail) {
+        // Buffer is full, signal the dequeue thread to process the buffer
+        // and clean it out, and then sleep
+        WARNING("procevent plugin: ring buffer full");
+
+        pthread_cond_signal(&procevent_cond);
+        pthread_mutex_unlock(&procevent_data_lock);
+
+        usleep(1000);
+        continue;
+      } else {
+        DEBUG("procevent plugin: Process %d status is now %s at %llu", proc_id,
+              (proc_status == PROCEVENT_EXITED ? "EXITED" : "STARTED"),
+              (long long unsigned int)CDTIME_T_TO_US(cdtime()));
+
+        if (proc_status == PROCEVENT_EXITED) {
+          ring.buffer[ring.head][0] = proc_id;
+          ring.buffer[ring.head][1] = proc_status;
+          ring.buffer[ring.head][2] = proc_extra;
+          ring.buffer[ring.head][3] =
+              (long long unsigned int)CDTIME_T_TO_US(cdtime());
+        } else {
+          ring.buffer[ring.head][0] = proc_id;
+          ring.buffer[ring.head][1] = proc_status;
+          ring.buffer[ring.head][2] = 0;
+          ring.buffer[ring.head][3] =
+              (long long unsigned int)CDTIME_T_TO_US(cdtime());
+        }
+
+        ring.head = next;
+      }
+
+      pthread_mutex_unlock(&procevent_data_lock);
     }
   }
 
-  switch (nlcn_msg.proc_ev.what) {
-  case PROC_EVENT_NONE:
-    // printf("set mcast listen ok\n");
-    break;
-  case PROC_EVENT_FORK:
-    // printf("fork: parent tid=%d pid=%d -> child tid=%d pid=%d\n",
-    //         nlcn_msg.proc_ev.event_data.fork.parent_pid,
-    //         nlcn_msg.proc_ev.event_data.fork.parent_tgid,
-    //         nlcn_msg.proc_ev.event_data.fork.child_pid,
-    //         nlcn_msg.proc_ev.event_data.fork.child_tgid);
-    // proc_status = PROCEVENT_STARTED;
-    // proc_id = nlcn_msg.proc_ev.event_data.fork.child_pid;
-    break;
-  case PROC_EVENT_EXEC:
-    // printf("exec: tid=%d pid=%d\n",
-    //         nlcn_msg.proc_ev.event_data.exec.process_pid,
-    //         nlcn_msg.proc_ev.event_data.exec.process_tgid);
-    proc_status = PROCEVENT_STARTED;
-    proc_id = nlcn_msg.proc_ev.event_data.exec.process_pid;
-    break;
-  case PROC_EVENT_UID:
-    // printf("uid change: tid=%d pid=%d from %d to %d\n",
-    //         nlcn_msg.proc_ev.event_data.id.process_pid,
-    //         nlcn_msg.proc_ev.event_data.id.process_tgid,
-    //         nlcn_msg.proc_ev.event_data.id.r.ruid,
-    //         nlcn_msg.proc_ev.event_data.id.e.euid);
-    break;
-  case PROC_EVENT_GID:
-    // printf("gid change: tid=%d pid=%d from %d to %d\n",
-    //         nlcn_msg.proc_ev.event_data.id.process_pid,
-    //         nlcn_msg.proc_ev.event_data.id.process_tgid,
-    //         nlcn_msg.proc_ev.event_data.id.r.rgid,
-    //         nlcn_msg.proc_ev.event_data.id.e.egid);
-    break;
-  case PROC_EVENT_EXIT:
-    proc_id = nlcn_msg.proc_ev.event_data.exit.process_pid;
-    proc_status = PROCEVENT_EXITED;
-    proc_extra = nlcn_msg.proc_ev.event_data.exit.exit_code;
-    break;
-  default:
-    break;
-  }
-
-  // If we're interested in this process status event, place the event
-  // in the ring buffer for consumption by the main polling thread.
-
-  if (proc_status != -1) {
-    pthread_mutex_unlock(&procevent_lock);
-
-    int next = ring.head + 1;
+  return 0;
+}
+
+// Read from ring buffer and dispatch to write plugins
+static void read_ring_buffer() {
+  pthread_mutex_lock(&procevent_data_lock);
+
+  // If there's currently nothing to read from the buffer,
+  // then wait
+  if (ring.head == ring.tail)
+    pthread_cond_wait(&procevent_cond, &procevent_data_lock);
+
+  while (ring.head != ring.tail) {
+    int next = ring.tail + 1;
+
     if (next >= ring.maxLen)
       next = 0;
 
-    if (next == ring.tail) {
-      WARNING("procevent plugin: ring buffer full");
-    } else {
-      DEBUG("procevent plugin: Process %d status is now %s", proc_id,
-            (proc_status == PROCEVENT_EXITED ? "EXITED" : "STARTED"));
+    if (ring.buffer[ring.tail][1] == PROCEVENT_EXITED) {
+      processlist_t *pl = process_map_check(ring.buffer[ring.tail][0], NULL);
 
-      if (proc_status == PROCEVENT_EXITED) {
-        ring.buffer[ring.head][0] = proc_id;
-        ring.buffer[ring.head][1] = proc_status;
-        ring.buffer[ring.head][2] = proc_extra;
-      } else {
-        ring.buffer[ring.head][0] = proc_id;
-        ring.buffer[ring.head][1] = proc_status;
-        ring.buffer[ring.head][2] = 0;
+      if (pl != NULL) {
+        // This process is of interest to us, so publish its EXITED status
+        procevent_dispatch_notification(ring.buffer[ring.tail][0], "gauge",
+                                        ring.buffer[ring.tail][1], pl->process,
+                                        ring.buffer[ring.tail][3]);
+        DEBUG(
+            "procevent plugin: PID %ld (%s) EXITED, removing PID from process "
+            "list",
+            pl->pid, pl->process);
+        pl->pid = -1;
+        pl->last_status = -1;
       }
+    } else if (ring.buffer[ring.tail][1] == PROCEVENT_STARTED) {
+      // a new process has started, so check if we should monitor it
+      processlist_t *pl = process_check(ring.buffer[ring.tail][0]);
+
+      // If we had already seen this process name and pid combo before,
+      // and the last message was a "process started" message, don't send
+      // the notfication again
+
+      if (pl != NULL && pl->last_status != PROCEVENT_STARTED) {
+        // This process is of interest to us, so publish its STARTED status
+        procevent_dispatch_notification(ring.buffer[ring.tail][0], "gauge",
+                                        ring.buffer[ring.tail][1], pl->process,
+                                        ring.buffer[ring.tail][3]);
+
+        pl->last_status = PROCEVENT_STARTED;
 
-      ring.head = next;
+        DEBUG("procevent plugin: PID %ld (%s) STARTED, adding PID to process "
+              "list",
+              pl->pid, pl->process);
+      }
     }
 
-    pthread_mutex_unlock(&procevent_lock);
+    ring.tail = next;
   }
 
-  return ret;
+  pthread_mutex_unlock(&procevent_data_lock);
 }
 
-static void *procevent_thread(void *arg) /* {{{ */
+// Entry point for thread responsible for listening
+// to netlink socket and writing data to ring buffer
+static void *procevent_netlink_thread(void *arg) /* {{{ */
 {
-  pthread_mutex_lock(&procevent_lock);
-
-  while (procevent_thread_loop > 0) {
-    int status;
-
-    pthread_mutex_unlock(&procevent_lock);
+  pthread_mutex_lock(&procevent_thread_lock);
 
-    usleep(1000);
+  while (procevent_netlink_thread_loop > 0) {
+    pthread_mutex_unlock(&procevent_thread_lock);
 
-    status = read_event();
+    int status = read_event();
 
-    pthread_mutex_lock(&procevent_lock);
+    pthread_mutex_lock(&procevent_thread_lock);
 
     if (status < 0) {
-      procevent_thread_error = 1;
+      procevent_netlink_thread_error = 1;
       break;
     }
+  } /* while (procevent_netlink_thread_loop > 0) */
 
-    if (procevent_thread_loop <= 0)
-      break;
-  } /* while (procevent_thread_loop > 0) */
-
-  pthread_mutex_unlock(&procevent_lock);
+  pthread_mutex_unlock(&procevent_thread_lock);
 
-  return ((void *)0);
-} /* }}} void *procevent_thread */
+  return (void *)0;
+} /* }}} void *procevent_netlink_thread */
 
-static int start_thread(void) /* {{{ */
+// Entry point for thread responsible for reading from
+// ring buffer and dispatching notifications
+static void *procevent_dequeue_thread(void *arg) /* {{{ */
 {
-  int status;
+  pthread_mutex_lock(&procevent_thread_lock);
+
+  while (procevent_dequeue_thread_loop > 0) {
+    pthread_mutex_unlock(&procevent_thread_lock);
+
+    read_ring_buffer();
+
+    pthread_mutex_lock(&procevent_thread_lock);
+  } /* while (procevent_dequeue_thread_loop > 0) */
+
+  pthread_mutex_unlock(&procevent_thread_lock);
 
-  pthread_mutex_lock(&procevent_lock);
+  return (void *)0;
+} /* }}} void *procevent_dequeue_thread */
 
-  if (procevent_thread_loop != 0) {
-    pthread_mutex_unlock(&procevent_lock);
-    return (0);
+static int start_netlink_thread(void) /* {{{ */
+{
+  pthread_mutex_lock(&procevent_thread_lock);
+
+  if (procevent_netlink_thread_loop != 0) {
+    pthread_mutex_unlock(&procevent_thread_lock);
+    return 0;
   }
 
+  int status;
+
   if (nl_sock == -1) {
     status = nl_connect();
 
-    if (status != 0)
+    if (status != 0) {
+      pthread_mutex_unlock(&procevent_thread_lock);
       return status;
+    }
 
     status = set_proc_ev_listen(true);
-    if (status == -1)
+    if (status == -1) {
+      pthread_mutex_unlock(&procevent_thread_lock);
       return status;
+    }
   }
 
   DEBUG("procevent plugin: socket created and bound");
 
-  procevent_thread_loop = 1;
-  procevent_thread_error = 0;
+  procevent_netlink_thread_loop = 1;
+  procevent_netlink_thread_error = 0;
 
-  status = plugin_thread_create(&procevent_thread_id, /* attr = */ NULL,
-                                procevent_thread,
+  status = plugin_thread_create(&procevent_netlink_thread_id, /* attr = */ NULL,
+                                procevent_netlink_thread,
                                 /* arg = */ (void *)0, "procevent");
   if (status != 0) {
-    procevent_thread_loop = 0;
-    ERROR("procevent plugin: Starting thread failed.");
-    pthread_mutex_unlock(&procevent_lock);
-    return (-1);
+    procevent_netlink_thread_loop = 0;
+    ERROR("procevent plugin: Starting netlink thread failed.");
+    pthread_mutex_unlock(&procevent_thread_lock);
+
+    int status2 = close(nl_sock);
+
+    if (status2 != 0) {
+      ERROR("procevent plugin: failed to close socket %d: %d (%s)", nl_sock,
+            status2, STRERRNO);
+    } else
+      nl_sock = -1;
+
+    return -1;
   }
 
-  pthread_mutex_unlock(&procevent_lock);
-  return (0);
-} /* }}} int start_thread */
+  pthread_mutex_unlock(&procevent_thread_lock);
 
-static int stop_thread(int shutdown) /* {{{ */
+  return status;
+} /* }}} int start_netlink_thread */
+
+static int start_dequeue_thread(void) /* {{{ */
 {
-  int status;
+  pthread_mutex_lock(&procevent_thread_lock);
+
+  if (procevent_dequeue_thread_loop != 0) {
+    pthread_mutex_unlock(&procevent_thread_lock);
+    return 0;
+  }
+
+  procevent_dequeue_thread_loop = 1;
+
+  int status = plugin_thread_create(&procevent_dequeue_thread_id,
+                                    /* attr = */ NULL, procevent_dequeue_thread,
+                                    /* arg = */ (void *)0, "procevent");
+  if (status != 0) {
+    procevent_dequeue_thread_loop = 0;
+    ERROR("procevent plugin: Starting dequeue thread failed.");
+    pthread_mutex_unlock(&procevent_thread_lock);
+    return -1;
+  }
+
+  pthread_mutex_unlock(&procevent_thread_lock);
+
+  return status;
+} /* }}} int start_dequeue_thread */
+
+static int start_threads(void) /* {{{ */
+{
+  int status = start_netlink_thread();
+  int status2 = start_dequeue_thread();
+
+  if (status != 0)
+    return status;
+  else
+    return status2;
+} /* }}} int start_threads */
+
+static int stop_netlink_thread(int shutdown) /* {{{ */
+{
+  int socket_status;
 
   if (nl_sock != -1) {
-    status = close(nl_sock);
-    if (status != 0) {
+    socket_status = close(nl_sock);
+    if (socket_status != 0) {
       ERROR("procevent plugin: failed to close socket %d: %d (%s)", nl_sock,
-            status, strerror(errno));
-      return (-1);
+            socket_status, strerror(errno));
+      return -1;
     } else
       nl_sock = -1;
+  } else
+    socket_status = 0;
+
+  pthread_mutex_lock(&procevent_thread_lock);
+
+  if (procevent_netlink_thread_loop == 0) {
+    pthread_mutex_unlock(&procevent_thread_lock);
+    return -1;
   }
 
-  pthread_mutex_lock(&procevent_lock);
+  // Set thread termination status
+  procevent_netlink_thread_loop = 0;
+  pthread_mutex_unlock(&procevent_thread_lock);
+
+  // Let threads waiting on access to the data know to move
+  // on such that they'll see the thread's termination status
+  pthread_cond_broadcast(&procevent_cond);
+
+  int thread_status;
+
+  if (shutdown == 1) {
+    // Calling pthread_cancel here in
+    // the case of a shutdown just assures that the thread is
+    // gone and that the process has been fully terminated.
+
+    DEBUG("procevent plugin: Canceling netlink thread for process shutdown");
 
-  if (procevent_thread_loop == 0) {
-    pthread_mutex_unlock(&procevent_lock);
-    return (-1);
+    thread_status = pthread_cancel(procevent_netlink_thread_id);
+
+    if (thread_status != 0 && thread_status != ESRCH) {
+      ERROR("procevent plugin: Unable to cancel netlink thread: %d",
+            thread_status);
+      thread_status = -1;
+    } else
+      thread_status = 0;
+  } else {
+    thread_status =
+        pthread_join(procevent_netlink_thread_id, /* return = */ NULL);
+    if (thread_status != 0 && thread_status != ESRCH) {
+      ERROR("procevent plugin: Stopping netlink thread failed.");
+      thread_status = -1;
+    } else
+      thread_status = 0;
   }
 
-  procevent_thread_loop = 0;
+  pthread_mutex_lock(&procevent_thread_lock);
+  memset(&procevent_netlink_thread_id, 0, sizeof(procevent_netlink_thread_id));
+  procevent_netlink_thread_error = 0;
+  pthread_mutex_unlock(&procevent_thread_lock);
+
+  DEBUG("procevent plugin: Finished requesting stop of netlink thread");
+
+  if (socket_status != 0)
+    return socket_status;
+  else
+    return thread_status;
+} /* }}} int stop_netlink_thread */
+
+static int stop_dequeue_thread(int shutdown) /* {{{ */
+{
+  int status;
+
+  pthread_mutex_lock(&procevent_thread_lock);
+
+  if (procevent_dequeue_thread_loop == 0) {
+    pthread_mutex_unlock(&procevent_thread_lock);
+    return -1;
+  }
+
+  procevent_dequeue_thread_loop = 0;
+  pthread_mutex_unlock(&procevent_thread_lock);
+
   pthread_cond_broadcast(&procevent_cond);
-  pthread_mutex_unlock(&procevent_lock);
 
   if (shutdown == 1) {
     // Calling pthread_cancel here in
     // the case of a shutdown just assures that the thread is
     // gone and that the process has been fully terminated.
 
-    DEBUG("procevent plugin: Canceling thread for process shutdown");
+    DEBUG("procevent plugin: Canceling dequeue thread for process shutdown");
 
-    status = pthread_cancel(procevent_thread_id);
+    status = pthread_cancel(procevent_dequeue_thread_id);
 
-    if (status != 0) {
-      ERROR("procevent plugin: Unable to cancel thread: %d", status);
+    if (status != 0 && status != ESRCH) {
+      ERROR("procevent plugin: Unable to cancel dequeue thread: %d", status);
       status = -1;
-    }
+    } else
+      status = 0;
   } else {
-    status = pthread_join(procevent_thread_id, /* return = */ NULL);
-    if (status != 0) {
-      ERROR("procevent plugin: Stopping thread failed.");
+    status = pthread_join(procevent_dequeue_thread_id, /* return = */ NULL);
+    if (status != 0 && status != ESRCH) {
+      ERROR("procevent plugin: Stopping dequeue thread failed.");
       status = -1;
-    }
+    } else
+      status = 0;
   }
 
-  pthread_mutex_lock(&procevent_lock);
-  memset(&procevent_thread_id, 0, sizeof(procevent_thread_id));
-  procevent_thread_error = 0;
-  pthread_mutex_unlock(&procevent_lock);
+  pthread_mutex_lock(&procevent_thread_lock);
+  memset(&procevent_dequeue_thread_id, 0, sizeof(procevent_dequeue_thread_id));
+  pthread_mutex_unlock(&procevent_thread_lock);
 
-  DEBUG("procevent plugin: Finished requesting stop of thread");
+  DEBUG("procevent plugin: Finished requesting stop of dequeue thread");
 
-  return (status);
-} /* }}} int stop_thread */
+  return status;
+} /* }}} int stop_dequeue_thread */
 
-static int procevent_init(void) /* {{{ */
+static int stop_threads(int shutdown) /* {{{ */
 {
-  int status;
+  int status = stop_netlink_thread(shutdown);
+  int status2 = stop_dequeue_thread(shutdown);
 
-  if (processlist_head == NULL) {
-    NOTICE("procevent plugin: No processes have been configured.");
-    return (-1);
-  }
+  if (status != 0)
+    return status;
+  else
+    return status2;
+} /* }}} int stop_threads */
 
+static int procevent_init(void) /* {{{ */
+{
   ring.head = 0;
   ring.tail = 0;
   ring.maxLen = buffer_length;
-  ring.buffer = (int **)malloc(buffer_length * sizeof(int *));
+  ring.buffer = (long long unsigned int **)calloc(
+      buffer_length, sizeof(long long unsigned int *));
 
   for (int i = 0; i < buffer_length; i++) {
-    ring.buffer[i] = (int *)malloc(PROCEVENT_FIELDS * sizeof(int));
+    ring.buffer[i] = (long long unsigned int *)calloc(
+        PROCEVENT_FIELDS, sizeof(long long unsigned int));
   }
 
-  status = process_map_refresh();
+  int status = process_map_refresh();
 
   if (status == -1) {
     ERROR("procevent plugin: Initial process mapping failed.");
-    return (-1);
+    return -1;
+  }
+
+  if (ignorelist == NULL) {
+    NOTICE("procevent plugin: No processes have been configured.");
+    return -1;
   }
 
-  return (start_thread());
+  return start_threads();
 } /* }}} int procevent_init */
 
 static int procevent_config(const char *key, const char *value) /* {{{ */
 {
-  int status;
+  if (ignorelist == NULL)
+    ignorelist = ignorelist_create(/* invert = */ 1);
 
   if (strcasecmp(key, "BufferLength") == 0) {
     buffer_length = atoi(value);
-  } else if (strcasecmp(key, "Process") == 0 ||
-             strcasecmp(key, "RegexProcess") == 0) {
-
-    processlist_t *pl;
-    char *process;
-    char *process_regex;
-
-    pl = malloc(sizeof(*pl));
-    if (pl == NULL) {
-      char errbuf[1024];
-      ERROR("procevent plugin: malloc failed during procevent_config: %s",
-            sstrerror(errno, errbuf, sizeof(errbuf)));
-      return (1);
-    }
-
-    process = strdup(value);
-    if (process == NULL) {
-      char errbuf[1024];
-      sfree(pl);
-      ERROR("procevent plugin: strdup failed during procevent_config: %s",
-            sstrerror(errno, errbuf, sizeof(errbuf)));
-      return (1);
-    }
-
-    if (strcasecmp(key, "RegexProcess") == 0) {
-      pl->is_regex = 1;
-      status = regcomp(&pl->process_regex_obj, value, REG_EXTENDED);
-
-      if (status != 0) {
-        ERROR("procevent plugin: invalid regular expression: %s", value);
-        return (1);
-      }
+  } else if (strcasecmp(key, "Process") == 0) {
+    ignorelist_add(ignorelist, value);
+  } else if (strcasecmp(key, "ProcessRegex") == 0) {
+#if HAVE_REGEX_H
+    int status = ignorelist_add(ignorelist, value);
 
-      process_regex = strdup(value);
-      if (process_regex == NULL) {
-        char errbuf[1024];
-        sfree(pl);
-        ERROR("procevent plugin: strdup failed during procevent_config: %s",
-              sstrerror(errno, errbuf, sizeof(errbuf)));
-        return (1);
-      }
-
-      pl->process_regex = process_regex;
-    } else {
-      pl->is_regex = 0;
+    if (status != 0) {
+      ERROR("procevent plugin: invalid regular expression: %s", value);
+      return 1;
     }
-
-    pl->process = process;
-    pl->pid = -1;
-    pl->next = processlist_head;
-    processlist_head = pl;
+#else
+    WARNING("procevent plugin: The plugin has been compiled without support "
+            "for the \"ProcessRegex\" option.");
+#endif
   } else {
-    return (-1);
+    return -1;
   }
 
-  return (0);
+  return 0;
 } /* }}} int procevent_config */
 
-static void submit(int pid, const char *type, /* {{{ */
-                   gauge_t value, const char *process) {
-  value_list_t vl = VALUE_LIST_INIT;
-  char hostname[1024];
-
-  vl.values = &(value_t){.gauge = value};
-  vl.values_len = 1;
-  sstrncpy(vl.plugin, "procevent", sizeof(vl.plugin));
-  sstrncpy(vl.plugin_instance, process, sizeof(vl.plugin_instance));
-  sstrncpy(vl.type, type, sizeof(vl.type));
-
-  DEBUG("procevent plugin: dispatching state %d for PID %d (%s)", (int)value,
-        pid, process);
+static void procevent_dispatch_notification(long pid,
+                                            const char *type, /* {{{ */
+                                            gauge_t value, char *process,
+                                            long long unsigned int timestamp) {
+
+  notification_t n = {(value == 1 ? NOTIF_OKAY : NOTIF_FAILURE),
+                      cdtime(),
+                      "",
+                      "",
+                      "procevent",
+                      "",
+                      "",
+                      "",
+                      NULL};
+  sstrncpy(n.host, hostname_g, sizeof(n.host));
+  sstrncpy(n.plugin_instance, process, sizeof(n.plugin_instance));
+  sstrncpy(n.type, "gauge", sizeof(n.type));
+  sstrncpy(n.type_instance, "process_status", sizeof(n.type_instance));
+
+  char *buf = NULL;
+  gen_message_payload(value, pid, process, timestamp, &buf);
+
+  notification_meta_t *m = calloc(1, sizeof(*m));
+
+  if (m == NULL) {
+    sfree(buf);
+    ERROR("procevent plugin: unable to allocate metadata: %s", STRERRNO);
+    return;
+  }
 
-  // Create metadata to store JSON key-values
-  meta_data_t *meta = meta_data_create();
+  sstrncpy(m->name, "ves", sizeof(m->name));
+  m->nm_value.nm_string = sstrdup(buf);
+  m->type = NM_TYPE_STRING;
+  n.meta = m;
 
-  vl.meta = meta;
+  DEBUG("procevent plugin: notification message: %s",
+        n.meta->nm_value.nm_string);
 
-  gethostname(hostname, sizeof(hostname));
+  DEBUG("procevent plugin: dispatching state %d for PID %ld (%s)", (int)value,
+        pid, process);
 
-  if (value == 1) {
-    meta_data_add_string(meta, "condition", "process_up");
-    meta_data_add_string(meta, "entity", process);
-    meta_data_add_string(meta, "source", hostname);
-    meta_data_add_string(meta, "dest", "process_down");
-  } else {
-    meta_data_add_string(meta, "condition", "process_down");
-    meta_data_add_string(meta, "entity", process);
-    meta_data_add_string(meta, "source", hostname);
-    meta_data_add_string(meta, "dest", "process_up");
-  }
+  plugin_dispatch_notification(&n);
+  plugin_notification_meta_free(n.meta);
 
-  plugin_dispatch_values(&vl);
-} /* }}} void interface_submit */
+  // strdup'd in gen_message_payload
+  if (buf != NULL)
+    sfree(buf);
+}
 
 static int procevent_read(void) /* {{{ */
 {
-  if (procevent_thread_error != 0) {
-    ERROR(
-        "procevent plugin: The interface thread had a problem. Restarting it.");
+  pthread_mutex_lock(&procevent_thread_lock);
 
-    stop_thread(0);
+  if (procevent_netlink_thread_error != 0) {
 
-    start_thread();
+    pthread_mutex_unlock(&procevent_thread_lock);
 
-    return (-1);
-  } /* if (procevent_thread_error != 0) */
+    ERROR("procevent plugin: The netlink thread had a problem. Restarting it.");
 
-  pthread_mutex_lock(&procevent_lock);
-
-  while (ring.head != ring.tail) {
-    int next = ring.tail + 1;
-
-    if (next >= ring.maxLen)
-      next = 0;
-
-    if (ring.buffer[ring.tail][1] == PROCEVENT_EXITED) {
-      processlist_t *pl = process_map_check(ring.buffer[ring.tail][0], NULL);
-
-      if (pl != NULL) {
-        // This process is of interest to us, so publish its EXITED status
-        submit(ring.buffer[ring.tail][0], "gauge", ring.buffer[ring.tail][1],
-               pl->process);
-        DEBUG("procevent plugin: PID %d (%s) EXITED, removing PID from process "
-              "list",
-              pl->pid, pl->process);
-        pl->pid = -1;
-      }
-    } else if (ring.buffer[ring.tail][1] == PROCEVENT_STARTED) {
-      // a new process has started, so check if we should monitor it
-      processlist_t *pl = process_check(ring.buffer[ring.tail][0]);
+    stop_netlink_thread(0);
 
-      if (pl != NULL) {
-        // This process is of interest to us, so publish its STARTED status
-        submit(ring.buffer[ring.tail][0], "gauge", ring.buffer[ring.tail][1],
-               pl->process);
-        DEBUG(
-            "procevent plugin: PID %d (%s) STARTED, adding PID to process list",
-            pl->pid, pl->process);
-      }
-    }
+    start_netlink_thread();
 
-    ring.tail = next;
-  }
+    return -1;
+  } /* if (procevent_netlink_thread_error != 0) */
 
-  pthread_mutex_unlock(&procevent_lock);
+  pthread_mutex_unlock(&procevent_thread_lock);
 
-  return (0);
+  return 0;
 } /* }}} int procevent_read */
 
 static int procevent_shutdown(void) /* {{{ */
 {
-  // int status = 0;
-  processlist_t *pl;
+  DEBUG("procevent plugin: Shutting down threads.");
 
-  DEBUG("procevent plugin: Shutting down thread.");
-
-  if (stop_thread(1) < 0)
-    return (-1);
+  int status = stop_threads(1);
 
   for (int i = 0; i < buffer_length; i++) {
     free(ring.buffer[i]);
@@ -876,24 +1334,21 @@ static int procevent_shutdown(void) /* {{{ */
 
   free(ring.buffer);
 
-  pl = processlist_head;
+  processlist_t *pl = processlist_head;
   while (pl != NULL) {
     processlist_t *pl_next;
 
     pl_next = pl->next;
 
-    if (pl->is_regex == 1) {
-      sfree(pl->process_regex);
-      regfree(&pl->process_regex_obj);
-    }
-
     sfree(pl->process);
     sfree(pl);
 
     pl = pl_next;
   }
 
-  return (0);
+  ignorelist_free(ignorelist);
+
+  return status;
 } /* }}} int procevent_shutdown */
 
 void module_register(void) {
@@ -902,4 +1357,4 @@ void module_register(void) {
   plugin_register_init("procevent", procevent_init);
   plugin_register_read("procevent", procevent_read);
   plugin_register_shutdown("procevent", procevent_shutdown);
-} /* void module_register */
\ No newline at end of file
+} /* void module_register */