Initial JSON UDP Plugin Implementation

This is a simple plugin that will format captured
records as JSON objects and then deliver them
to a listening UDP server.

configure.ac

@@ -747,6 +747,26 @@ AC_ARG_ENABLE(kafka,
 )
 dnl finish: Kafka handling
 
+dnl start: JSON UDP handling
+AC_MSG_CHECKING(whether to enable JSON UDP support)
+AC_ARG_ENABLE(jsonudp,
+  [  --enable-jsonudp             Enable JSON UDP support (default: no)],
+  [ case "$enableval" in
+  yes)
+    AC_MSG_RESULT(yes)
+    USING_JSONUDP="yes"
+    AC_DEFINE(WITH_JSONUDP, 1)
+    ;;
+  no)
+    AC_MSG_RESULT(no)
+    ;;
+  esac ],
+  [
+    AC_MSG_RESULT(no)
+  ]
+)
+dnl finish JSON UDP handling
+
 dnl start: geoip handling
 AC_MSG_CHECKING(whether to enable GeoIP support)
 AC_ARG_ENABLE(geoip,
@@ -1185,6 +1205,7 @@ AM_CONDITIONAL([WITH_SQLITE3], [test x"$USING_SQLITE3" = x"yes"])
 AM_CONDITIONAL([WITH_RABBITMQ], [test x"$USING_RABBITMQ" = x"yes"])
 AM_CONDITIONAL([WITH_ZMQ], [test x"$USING_ZMQ" = x"yes"])
 AM_CONDITIONAL([WITH_KAFKA], [test x"$USING_KAFKA" = x"yes"])
+AM_CONDITIONAL([WITH_JSONUDP], [test x"$USING_JSONUDP" = x"yes"])
 AM_CONDITIONAL([USING_SQL], [test x"$USING_SQL" = x"yes"])
 AM_CONDITIONAL([WITH_AVRO], [test x"$USING_AVRO" = x"yes"])
 AM_CONDITIONAL([WITH_SERDES], [test x"$USING_SERDES" = x"yes"])

src/Makefile.am

@@ -88,6 +88,9 @@ libdaemons_la_SOURCES += kafka_common.c kafka_common.h kafka_plugin.c kafka_plug
 libdaemons_la_LIBADD  += @KAFKA_LIBS@
 libdaemons_la_CFLAGS  += @KAFKA_CFLAGS@
 endif
+if WITH_JSONUDP
+libdaemons_la_SOURCES += jsonudp_plugin.c jsonudp_plugin.h
+endif
 if USING_SQL
 libdaemons_la_SOURCES += sql_common.c sql_handlers.c sql_common.h
 libdaemons_la_LIBADD  += -lm -lz

src/cfg.c

@@ -458,6 +458,7 @@ int create_plugin(char *filename, char *name, char *type)
 
   /* searching for a valid known plugin type */
   while(strcmp(plugin_types_list[index].string, "")) {
+    Log(LOG_ERR, "ERROR: Comparing %s with type: %s.\n", plugin_types_list[index].string, type);
     if (!strcmp(type, plugin_types_list[index].string)) ptype = &plugin_types_list[index];
     index++;
   }

src/cfg.h

@@ -177,6 +177,7 @@ struct configuration {
   char *print_output_separator;
   char *print_output_file;
   char *print_latest_file;
+  char *jsonudp_server;
   int nfacctd_port;
   char *nfacctd_ip;
   char *nfacctd_kafka_broker_host;

src/cfg_handlers.c

@@ -2829,6 +2829,25 @@ int cfg_key_print_output_separator(char *filename, char *name, char *value_ptr)
   return changes;
 }
 
+int cfg_key_jsonudp_server(char *filename, char *name, char *value_ptr)
+{
+  struct plugins_list_entry *list = plugins_list;
+  int changes = 0;
+  char *endptr;
+
+  if (!name) for (; list; list = list->next, changes++) list->cfg.jsonudp_server= value_ptr;
+  else {
+    for (; list; list = list->next) {
+      if (!strcmp(name, list->name)) {
+        list->cfg.jsonudp_server = value_ptr;
+        changes++;
+        break;
+      }
+    }
+  }
+  return changes;
+}
+
 int cfg_key_num_protos(char *filename, char *name, char *value_ptr)
 {
   struct plugins_list_entry *list = plugins_list;

src/cfg_handlers.h

@@ -141,6 +141,7 @@ EXT int cfg_key_print_output_file_append(char *, char *, char *);
 EXT int cfg_key_print_output_lock_file(char *, char *, char *);
 EXT int cfg_key_print_output_separator(char *, char *, char *);
 EXT int cfg_key_print_latest_file(char *, char *, char *);
+EXT int cfg_key_jsonudp_server(char *, char *, char *);
 EXT int cfg_key_nfacctd_port(char *, char *, char *);
 EXT int cfg_key_nfacctd_ip(char *, char *, char *);
 EXT int cfg_key_nfacctd_allow_file(char *, char *, char *);