in_head: a simple plugin to read a file.

CMakeLists.txt

@@ -36,6 +36,7 @@ option(WITH_IN_LIB      "Enable library mode input plugin"   Yes)
 option(WITH_IN_SERIAL   "Enable Serial input plugin"         Yes)
 option(WITH_IN_STDIN    "Enable Standard input plugin"       Yes)
 option(WITH_IN_MQTT     "Enable MQTT Broker input plugin"     No)
+option(WITH_IN_HEAD     "Enable Head input plugin"           Yes)
 option(WITH_OUT_ES      "Enable Elasticsearch output plugin" Yes)
 option(WITH_OUT_FLUENTD "Enable Fluentd output plugin"       Yes)
 option(WITH_OUT_NATS    "Enable NATS output plugin"           No)
@@ -56,6 +57,7 @@ if(WITH_ALL)
   set(WITH_IN_MQTT     1)
   set(WITH_IN_SERIAL   1)
   set(WITH_IN_STDIN    1)
+  set(WITH_IN_HEAD     1)
   set(WITH_OUT_FLUENTD 1)
   set(WITH_OUT_NATS    1)
   set(WITH_OUT_TD      1)

conf/in_head.conf

@@ -0,0 +1,26 @@
+# Head Input
+# ==========
+[HEAD]
+    # File
+    # ====
+    # File path. e.g. /proc/uptime
+    #
+    File    /path/to/file
+
+    # Buf_Size
+    # ====
+    # Buffer size to read file. Default 256
+    Buf_Size 256
+
+    # Total Interval 
+    #     = Interval Sec + ( Interval Nsec * 1000 * 1000 * 1000 )
+    #
+    # Interval Sec
+    # ====
+    # Read interval (sec) Default 1
+    Interval_Sec 1
+
+    # Interval NSec
+    # ====
+    # Read interval (nsec) Default 0
+    Interval_NSec 0

plugins/CMakeLists.txt

@@ -75,6 +75,7 @@ REGISTER_IN_PLUGIN("in_serial")
 REGISTER_IN_PLUGIN("in_stdin")
 REGISTER_IN_PLUGIN("in_mqtt")
 REGISTER_IN_PLUGIN("in_lib")
+REGISTER_IN_PLUGIN("in_head")
 REGISTER_OUT_PLUGIN("out_es")
 REGISTER_OUT_PLUGIN("out_fluentd")
 REGISTER_OUT_PLUGIN("out_td")

plugins/in_head/CMakeLists.txt

@@ -0,0 +1,4 @@
+set(src
+  in_head.c)
+
+FLB_PLUGIN(in_head "${src}" "")

plugins/in_head/in_head.c

@@ -0,0 +1,255 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2016 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#include <msgpack.h>
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_error.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_stats.h>
+
+#include "in_head.h"
+
+/* cb_collect callback */
+static int in_head_collect(struct flb_config *config, void *in_context)
+{
+    struct flb_in_head_config *head_config = in_context;
+    int     fd        = -1;
+    ssize_t read_size = 0;
+    int     ret = -1;
+
+    /* open at every collect callback */
+    fd = open(head_config->filepath, O_RDONLY);
+    if ( fd < 0) {
+        perror("open");
+        goto collect_fin;
+    }
+
+    head_config->buf_len = read(fd, head_config->buf,
+                                head_config->buf_size);
+    flb_debug("%s read_len=%d buf_size=%d",__FUNCTION__,
+              head_config->buf_len, head_config->buf_size);
+
+    if (head_config->buf_len < 0){
+        perror("read");
+        goto collect_fin;
+    }
+
+    msgpack_pack_array(&head_config->mp_pck,2);
+    msgpack_pack_uint64(&head_config->mp_pck, time(NULL));
+    msgpack_pack_map(&head_config->mp_pck,1);
+
+    msgpack_pack_bin(&head_config->mp_pck, 4);
+    msgpack_pack_bin_body(&head_config->mp_pck, "head", 4);
+    msgpack_pack_bin(&head_config->mp_pck, head_config->buf_len);
+    msgpack_pack_bin_body(&head_config->mp_pck,
+                          head_config->buf, head_config->buf_len);
+
+    ret = 0;
+    head_config->idx++;
+    flb_stats_update(in_head_plugin.stats_fd, 0, 1);
+ collect_fin:
+    if ( fd > 0 ) {
+        close(fd);
+    }
+    return ret;
+}
+
+/* read config file and*/
+static int in_head_config_read(struct flb_in_head_config *head_config,
+                               struct mk_rconf *config )
+{
+    struct mk_rconf_section *section = NULL;
+    char     *filepath  = NULL;
+    char     *pval = NULL;
+    size_t   buf_size = -1;
+
+    section = mk_rconf_section_get(config, "head");
+    if ( section == NULL ) {
+        return -1;
+    }
+
+    /* filepath setting */
+    filepath = mk_rconf_section_get_key(section, "file", MK_RCONF_STR);
+    if ( filepath == NULL ) {
+        return -1;
+    }
+    head_config->filepath = filepath;
+
+    /* buffer size setting */
+    pval = mk_rconf_section_get_key(section, "buf_size", MK_RCONF_STR);
+    if ( pval != NULL && atoi(pval) > 0 ) {
+        head_config->buf_size = atoi(pval);
+    } else {
+        head_config->buf_size = DEFAULT_BUF_SIZE;
+    }
+
+    /* interval settings */
+    pval = mk_rconf_section_get_key(section, "interval_sec", MK_RCONF_STR);
+    if ( pval != NULL && atoi(pval) > 0 ) {
+        head_config->interval_sec = atoi(pval);
+    } else {
+        head_config->interval_sec = DEFAULT_INTERVAL_SEC;
+    }
+    pval = mk_rconf_section_get_key(section, "interval_nsec", MK_RCONF_STR);
+    if ( pval != NULL && atoi(pval) > 0 ) {
+        head_config->interval_nsec = atoi(pval);
+    } else {
+        head_config->interval_nsec = DEFAULT_INTERVAL_NSEC;
+    }
+
+    flb_debug("Head config: buf_size=%d path=%s",
+              head_config->buf_size, head_config->filepath);
+    flb_debug("Head config: interval_sec=%d interval_nsec=%d",
+              head_config->interval_sec, head_config->interval_nsec);
+
+    return 0;
+}
+
+static void delete_head_config(struct flb_in_head_config *head_config)
+{
+    /* release buffer */
+    if ( head_config->buf != NULL ){
+        free(head_config->buf);
+    }
+    if ( head_config != NULL ) {
+        free(head_config);
+    }
+}
+
+/* Initialize plugin */
+/* cb_init callback  */
+static int in_head_init(struct flb_config *config, void *data)
+{
+    struct flb_in_head_config *head_config = NULL;
+    int ret      = -1;
+
+    /* Initialize head config */
+    if ( config->file == NULL ) {
+        flb_utils_error_c("config file not found");
+        return -1;
+    }
+
+    /* Allocate space for the configuration */
+    head_config = malloc( sizeof( struct flb_in_head_config) );
+    if ( head_config == NULL ){
+        return -1;
+    }
+    head_config->buf     = NULL;
+    head_config->buf_len = 0;
+    head_config->idx     = 0;
+
+    /* Initialize head config */
+    ret = in_head_config_read(head_config, config->file);
+    if ( ret < 0 ) {
+        goto init_error;
+    }
+
+    head_config->buf      = malloc( head_config->buf_size );
+    if ( head_config->buf == NULL ) {
+        flb_utils_error_c("could not allocate head buffer");
+        goto init_error;
+    }
+
+    flb_debug("%s read_len=%d buf_size=%d",__FUNCTION__,
+              head_config->buf_len, sizeof(head_config->buf));
+
+    ret = flb_input_set_context("head", head_config, config);
+    if ( ret < 0 ){
+        flb_utils_error_c("could not set context for head plugin");
+        goto init_error;
+    }
+
+    ret = flb_input_set_collector_time("head",
+				       in_head_collect,
+				       head_config->interval_sec,
+				       head_config->interval_nsec,
+				       config);
+
+    /* Initialize msgpack buffer */
+    msgpack_sbuffer_init(&head_config->mp_sbuf);
+    msgpack_packer_init(&head_config->mp_pck,
+                        &head_config->mp_sbuf,
+                        msgpack_sbuffer_write);
+
+    if ( ret < 0 ){
+        flb_utils_error_c("could not set collector for head input plugin");
+        goto init_error;
+    }
+
+    return 0;
+
+ init_error:
+    delete_head_config(head_config);
+
+    return -1;
+}
+
+/* cb_flush callback */
+static void *in_head_flush(void* in_context, int* size)
+{
+    char *buf = NULL;
+    struct flb_in_head_config *head_config = in_context;
+
+    if ( head_config->idx == 0 ) {
+        head_config = 0;
+        return NULL;
+    }
+    buf = malloc(head_config->mp_sbuf.size);
+    if (!buf) {
+        return NULL;
+    }
+
+    memcpy(buf, head_config->mp_sbuf.data, head_config->mp_sbuf.size);
+    *size = head_config->mp_sbuf.size;
+    msgpack_sbuffer_destroy(&head_config->mp_sbuf);
+    msgpack_sbuffer_init(&head_config->mp_sbuf);
+    msgpack_packer_init(&head_config->mp_pck,
+                        &head_config->mp_sbuf, msgpack_sbuffer_write);
+    head_config->idx = 0;
+
+    return buf;
+}
+
+int in_head_exit(void *data, struct flb_config *config)
+{
+    (void) *config;
+    struct flb_in_head_config *head_config = data;
+
+    delete_head_config(head_config);
+
+    return 0;
+}
+
+
+struct flb_input_plugin in_head_plugin = {
+    .name         = "head",
+    .description  = "Head Input",
+    .cb_init      = in_head_init,
+    .cb_pre_run   = NULL,
+    .cb_collect   = in_head_collect,
+    .cb_flush_buf = in_head_flush,
+    .cb_exit      = in_head_exit
+};

plugins/in_head/in_head.h

@@ -0,0 +1,49 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2016 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+#ifndef FLB_IN_HEAD_H
+#define FLB_IN_HEAD_H
+
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_input.h>
+#include <fluent-bit/flb_utils.h>
+
+#include <msgpack.h>
+
+#define DEFAULT_BUF_SIZE      256
+#define DEFAULT_INTERVAL_SEC  1
+#define DEFAULT_INTERVAL_NSEC 0
+
+struct flb_in_head_config {
+    int       idx;
+    size_t    buf_size; /* size of buf */
+    size_t    buf_len;  /* read size */
+    char     *buf;      /* read buf */
+
+    char     *filepath; /* to read */
+
+    int      interval_sec;
+    int      interval_nsec;
+
+    msgpack_packer   mp_pck;
+    msgpack_sbuffer  mp_sbuf;
+};
+
+extern struct flb_input_plugin in_head_plugin;
+
+#endif /* FLB_IN_HEAD_H */