in_node_exporter_metrics: add scrape_on_start option

plugins/in_node_exporter_metrics/ne.c

@@ -69,6 +69,18 @@ static int cb_ne_collect(struct flb_input_instance *ins,
     return 0;
 }
 
+static int cb_ne_collect_start(struct flb_input_instance *ins,
+                               struct flb_config *config, void *in_context)
+{
+    struct flb_ne *ctx = in_context;
+
+    /* one-shot: pause the collector and forget it so resume skips it */
+    flb_input_collector_pause(ctx->coll_fd_start, ctx->ins);
+    ctx->coll_fd_start = -1;
+
+    return cb_ne_collect(ins, config, ctx);
+}
+
 static int collectors_common_init(struct flb_ne_collector *coll)
 {
     if (coll == NULL) {
@@ -216,6 +228,7 @@ static int in_ne_init(struct flb_input_instance *in,
 
     /* Initialize fds */
     ctx->coll_fd = -1;
+    ctx->coll_fd_start = -1;
 
     /* Associate context with the instance */
     flb_input_set_context(in, ctx);
@@ -249,6 +262,21 @@ static int in_ne_init(struct flb_input_instance *in,
         return -1;
     }
 
+    /* Scrape metrics on start / collector */
+    if (ctx->scrape_interval > 5 && ctx->scrape_on_start) {
+        ret = flb_input_set_collector_time(in,
+                                           cb_ne_collect_start,
+                                           5, 0,
+                                           config);
+        if (ret == -1) {
+            flb_plg_error(ctx->ins,
+                          "could not set collector on start for Node Exporter "
+                          "Metrics plugin");
+            return -1;
+        }
+        ctx->coll_fd_start = ret;
+    }
+
     /* Create the collector */
     ret = flb_input_set_collector_time(in,
                                        cb_ne_collect,
@@ -295,6 +323,9 @@ static void in_ne_pause(void *data, struct flb_config *config)
     struct mk_list *head;
 
     flb_input_collector_pause(ctx->coll_fd, ctx->ins);
+    if (ctx->coll_fd_start != -1) {
+        flb_input_collector_pause(ctx->coll_fd_start, ctx->ins);
+    }
 
     mk_list_foreach(head, &ctx->collectors) {
         coll = mk_list_entry(head, struct flb_ne_collector, _head);
@@ -312,6 +343,9 @@ static void in_ne_resume(void *data, struct flb_config *config)
     struct mk_list *head;
 
     flb_input_collector_resume(ctx->coll_fd, ctx->ins);
+    if (ctx->coll_fd_start != -1) {
+        flb_input_collector_resume(ctx->coll_fd_start, ctx->ins);
+    }
     mk_list_foreach(head, &ctx->collectors) {
         coll = mk_list_entry(head, struct flb_ne_collector, _head);
         if (coll->activated == FLB_FALSE) {
@@ -329,6 +363,13 @@ static struct flb_config_map config_map[] = {
      "scrape interval to collect metrics from the node."
     },
 
+    {
+     FLB_CONFIG_MAP_BOOL, "scrape_on_start", "false",
+     0, FLB_TRUE, offsetof(struct flb_ne, scrape_on_start),
+     "scrape metrics upon start, useful to avoid waiting for 'scrape_interval' "
+     "for the first round of metrics."
+    },
+
     {
      FLB_CONFIG_MAP_TIME, "collector.cpu.scrape_interval", "0",
      0, FLB_FALSE, 0,

plugins/in_node_exporter_metrics/ne.h

@@ -53,8 +53,10 @@ struct flb_ne {
     flb_sds_t path_sysfs;
     flb_sds_t path_textfile;
     int scrape_interval;
+    int scrape_on_start;
 
     int coll_fd;                                      /* collector fd     */
+    int coll_fd_start;                                /* start collector  */
     struct cmt *cmt;                                  /* cmetrics context */
     struct flb_input_instance *ins;                   /* input instance   */
     struct mk_list *metrics;                          /* enabled metrics */

tests/runtime/in_node_exporter_metrics.c

@@ -125,6 +125,78 @@ static void test_diskstats(void)
 
 #ifdef FLB_SYSTEM_LINUX
 
+/*
+ * Run with a scrape_interval far longer than the test, so any record that
+ * reaches the output within the wait can only come from scrape_on_start.
+ */
+static int count_first_scrape(char *scrape_on_start)
+{
+    int ret;
+    int input_fd;
+    int output_fd;
+    int count;
+    flb_ctx_t *ctx;
+    struct flb_lib_out_cb callback;
+
+    output_count = 0;
+    callback.cb = cb_count_output;
+    callback.data = NULL;
+
+    ctx = flb_create();
+    TEST_CHECK(ctx != NULL);
+
+    ret = flb_service_set(ctx,
+                          "Flush", "0.2",
+                          "Grace", "1",
+                          "Log_Level", "error",
+                          NULL);
+    TEST_CHECK(ret == 0);
+
+    input_fd = flb_input(ctx, (char *) "node_exporter_metrics", NULL);
+    TEST_CHECK(input_fd >= 0);
+
+    ret = flb_input_set(ctx, input_fd,
+                        "metrics", "loadavg",
+                        "scrape_interval", "3600",
+                        "scrape_on_start", scrape_on_start,
+                        NULL);
+    TEST_CHECK(ret == 0);
+
+    output_fd = flb_output(ctx, (char *) "lib", &callback);
+    TEST_CHECK(output_fd >= 0);
+
+    ret = flb_output_set(ctx, output_fd,
+                         "format", "json",
+                         NULL);
+    TEST_CHECK(ret == 0);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* the start collector fires 5 seconds after startup */
+    flb_time_msleep(8000);
+    count = get_output_count();
+
+    flb_stop(ctx);
+    flb_destroy(ctx);
+
+    return count;
+}
+
+static void test_scrape_on_start(void)
+{
+    TEST_CHECK(count_first_scrape("on") > 0);
+}
+
+static void test_no_scrape_on_start(void)
+{
+    TEST_CHECK(count_first_scrape("off") == 0);
+}
+
+#endif
+
+#ifdef FLB_SYSTEM_LINUX
+
 #define NETDEV_HEADER                                                        \
     "Inter-|   Receive                                                |"     \
     "  Transmit\n"                                                        \
@@ -312,6 +384,8 @@ TEST_LIST = {
     {"diskstats", test_diskstats},
 #ifdef FLB_SYSTEM_LINUX
     {"netdev_removes_stale_devices", test_netdev_removes_stale_devices},
+    {"scrape_on_start", test_scrape_on_start},
+    {"no_scrape_on_start", test_no_scrape_on_start},
 #endif
     {NULL, NULL}
 };