http_server: storage: new api/v1/storage endpoint for storage metrics

Signed-off-by: Eduardo Silva <[email protected]>

Author: edsiper

src/http_server/api/v1/CMakeLists.txt

@@ -2,6 +2,7 @@
 set(src
   uptime.c
   metrics.c
+  storage.c
   plugins.c
   register.c
   )

src/http_server/api/v1/metrics.c

@@ -2,7 +2,7 @@
 
 /*  Fluent Bit
  *  ==========
- *  Copyright (C) 2019      The Fluent Bit Authors
+ *  Copyright (C) 2019-2020 The Fluent Bit Authors
  *  Copyright (C) 2015-2018 Treasure Data Inc.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
@@ -60,7 +60,7 @@ static struct flb_hs_buf *metrics_get_latest()
 }
 
 /* Delete unused metrics, note that we only care about the latest node */
-int cleanup_metrics()
+static int cleanup_metrics()
 {
     int c = 0;
     struct mk_list *tmp;
@@ -126,6 +126,7 @@ static void cb_mq_metrics(mk_mq_t *queue, void *data, size_t size)
     buf = flb_malloc(sizeof(struct flb_hs_buf));
     if (!buf) {
         flb_errno();
+        flb_sds_destroy(out_data);
         return;
     }
     buf->users = 0;
@@ -141,13 +142,15 @@ static void cb_mq_metrics(mk_mq_t *queue, void *data, size_t size)
 }
 
 int string_cmp(const void* a_arg, const void* b_arg) {
-  char* a = *(char **)a_arg;
-  char* b = *(char **)b_arg;
+  char *a = *(char **)a_arg;
+  char *b = *(char **)b_arg;
+
   return strcmp(a, b);
 }
 
 size_t extract_metric_name_end_position(char *s) {
     int i;
+
     for (i = 0; i < flb_sds_len(s); i++) {
         if (s[i] == '{') {
           return i;
@@ -157,41 +160,51 @@ size_t extract_metric_name_end_position(char *s) {
 }
 
 int is_same_metric(char *s1, char *s2) {
-  int i;
-  int p1 = extract_metric_name_end_position(s1);
-  int p2 = extract_metric_name_end_position(s2);
-  if (p1 != p2) {
-    return 0;
-  }
-  for (i = 0; i < p1; i++) {
-    if (s1[i] != s2[i]) {
-      return 0;
+    int i;
+    int p1 = extract_metric_name_end_position(s1);
+    int p2 = extract_metric_name_end_position(s2);
+
+    if (p1 != p2) {
+        return 0;
+    }
+
+    for (i = 0; i < p1; i++) {
+        if (s1[i] != s2[i]) {
+            return 0;
+        }
     }
-  }
-  return 1;
+    return 1;
 }
 
 /* derive HELP text from metricname */
 /* if help text length > 128, increase init memory for metric_helptxt */
 flb_sds_t metrics_help_txt(char *metric_name, flb_sds_t *metric_helptxt)
 {
-   if (strstr(metric_name, "input_bytes")) {
+    if (strstr(metric_name, "input_bytes")) {
         return flb_sds_cat(*metric_helptxt, " Number of input bytes.\n", 24);
-    } else if (strstr(metric_name, "input_records")) {
+    }
+    else if (strstr(metric_name, "input_records")) {
         return flb_sds_cat(*metric_helptxt, " Number of input records.\n", 26);
-    } else if (strstr(metric_name, "output_bytes")) {
+    }
+    else if (strstr(metric_name, "output_bytes")) {
         return flb_sds_cat(*metric_helptxt, " Number of output bytes.\n", 25);
-    } else if (strstr(metric_name, "output_records")) {
+    }
+    else if (strstr(metric_name, "output_records")) {
         return flb_sds_cat(*metric_helptxt, " Number of output records.\n", 27);
-    } else if (strstr(metric_name, "output_errors")) {
+    }
+    else if (strstr(metric_name, "output_errors")) {
         return flb_sds_cat(*metric_helptxt, " Number of output errors.\n", 26);
-    } else if (strstr(metric_name, "output_retries_failed")) {
+    }
+    else if (strstr(metric_name, "output_retries_failed")) {
         return flb_sds_cat(*metric_helptxt, " Number of output retries failed.\n", 34);
-    } else if (strstr(metric_name, "output_retries")) {
+    }
+    else if (strstr(metric_name, "output_retries")) {
         return flb_sds_cat(*metric_helptxt, " Number of output retries.\n", 27);
-    } else if (strstr(metric_name, "output_proc_records")) {
+    }
+    else if (strstr(metric_name, "output_proc_records")) {
         return flb_sds_cat(*metric_helptxt, " Number of processed output records.\n", 37);
-    } else if (strstr(metric_name, "output_proc_bytes")) {
+    }
+    else if (strstr(metric_name, "output_proc_bytes")) {
         return flb_sds_cat(*metric_helptxt, " Number of processed output bytes.\n", 35);
     }
     else {
@@ -473,7 +486,8 @@ int api_v1_metrics(struct flb_hs *hs)
     pthread_key_create(&hs_metrics_key, NULL);
 
     /* Create a message queue */
-    hs->qid = mk_mq_create(hs->ctx, "/metrics", cb_mq_metrics, NULL);
+    hs->qid_metrics = mk_mq_create(hs->ctx, "/metrics",
+                                   cb_mq_metrics, NULL);
 
     /* HTTP end-points */
     mk_vhost_handler(hs->ctx, hs->vid, "/api/v1/metrics/prometheus",

src/http_server/api/v1/metrics.h

@@ -26,7 +26,6 @@
 #include <fluent-bit/flb_sds.h>
 
 int api_v1_metrics(struct flb_hs *hs);
-
 flb_sds_t metrics_help_txt(char *metric_name, flb_sds_t *metric_helptxt);
 
 #endif

src/http_server/api/v1/register.c

@@ -2,7 +2,7 @@
 
 /*  Fluent Bit
  *  ==========
- *  Copyright (C) 2019      The Fluent Bit Authors
+ *  Copyright (C) 2019-2020 The Fluent Bit Authors
  *  Copyright (C) 2015-2018 Treasure Data Inc.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
@@ -23,6 +23,7 @@
 
 #include "uptime.h"
 #include "metrics.h"
+#include "storage.h"
 #include "plugins.h"
 
 int api_v1_registration(struct flb_hs *hs)
@@ -31,5 +32,9 @@ int api_v1_registration(struct flb_hs *hs)
     api_v1_metrics(hs);
     api_v1_plugins(hs);
 
+    if (hs->config->storage_metrics == FLB_TRUE) {
+        api_v1_storage_metrics(hs);
+    }
+
     return 0;
 }

src/http_server/api/v1/storage.c

@@ -0,0 +1,172 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2020 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 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 <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_sds.h>
+#include "storage.h"
+
+#include <fluent-bit/flb_http_server.h>
+#include <msgpack.h>
+
+pthread_key_t hs_storage_metrics_key;
+
+/* Return the newest storage metrics buffer */
+static struct flb_hs_buf *storage_metrics_get_latest()
+{
+    struct flb_hs_buf *buf;
+    struct mk_list *metrics_list;
+
+    metrics_list = pthread_getspecific(hs_storage_metrics_key);
+    if (!metrics_list) {
+        return NULL;
+    }
+
+    if (mk_list_size(metrics_list) == 0) {
+        return NULL;
+    }
+
+    buf = mk_list_entry_last(metrics_list, struct flb_hs_buf, _head);
+    return buf;
+}
+
+/* Delete unused metrics, note that we only care about the latest node */
+static int cleanup_metrics()
+{
+    int c = 0;
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct mk_list *metrics_list;
+    struct flb_hs_buf *last;
+    struct flb_hs_buf *entry;
+
+    metrics_list = pthread_getspecific(hs_storage_metrics_key);
+    if (!metrics_list) {
+        return -1;
+    }
+
+    last = storage_metrics_get_latest();
+    if (!last) {
+        return -1;
+    }
+
+    mk_list_foreach_safe(head, tmp, metrics_list) {
+        entry = mk_list_entry(head, struct flb_hs_buf, _head);
+        if (entry != last && entry->users == 0) {
+            mk_list_del(&entry->_head);
+            flb_sds_destroy(entry->data);
+            flb_free(entry->raw_data);
+            flb_free(entry);
+            c++;
+        }
+    }
+
+    return c;
+}
+
+/*
+ * Callback invoked every time some storage metrics are received through a
+ * message queue channel. This function runs in a Monkey HTTP thread
+ * worker and it purpose is to take the metrics data and store it
+ * somewhere so then it can be available by the end-points upon
+ * HTTP client requests.
+ */
+static void cb_mq_storage_metrics(mk_mq_t *queue, void *data, size_t size)
+{
+    flb_sds_t out_data;
+    struct flb_hs_buf *buf;
+    struct mk_list *metrics_list = NULL;
+
+    metrics_list = pthread_getspecific(hs_storage_metrics_key);
+    if (!metrics_list) {
+        metrics_list = flb_malloc(sizeof(struct mk_list));
+        if (!metrics_list) {
+            flb_errno();
+            return;
+        }
+        mk_list_init(metrics_list);
+        pthread_setspecific(hs_storage_metrics_key, metrics_list);
+    }
+
+    /* Convert msgpack to JSON */
+    out_data = flb_msgpack_raw_to_json_sds(data, size);
+    if (!out_data) {
+        return;
+    }
+
+    buf = flb_malloc(sizeof(struct flb_hs_buf));
+    if (!buf) {
+        flb_errno();
+        flb_sds_destroy(out_data);
+        return;
+    }
+    buf->users = 0;
+    buf->data = out_data;
+
+    buf->raw_data = flb_malloc(size);
+    memcpy(buf->raw_data, data, size);
+    buf->raw_size = size;
+
+    mk_list_add(&buf->_head, metrics_list);
+
+    cleanup_metrics();
+}
+
+static void cb_mq_storage_metrics_exit(mk_mq_t *queue, void *data)
+{
+
+}
+
+/* API: expose built-in storage metrics /api/v1/storage */
+static void cb_storage(mk_request_t *request, void *data)
+{
+    struct flb_hs_buf *buf;
+
+    buf = storage_metrics_get_latest();
+    if (!buf) {
+        mk_http_status(request, 404);
+        mk_http_done(request);
+        return;
+    }
+
+    buf->users++;
+
+    mk_http_status(request, 200);
+    mk_http_send(request, buf->data, flb_sds_len(buf->data), NULL);
+    mk_http_done(request);
+
+    buf->users--;
+}
+
+/* Perform registration */
+int api_v1_storage_metrics(struct flb_hs *hs)
+{
+    pthread_key_create(&hs_storage_metrics_key, NULL);
+
+    /* Create a message queue */
+    hs->qid_storage = mk_mq_create(hs->ctx, "/storage",
+                                   cb_mq_storage_metrics,
+                                   NULL);
+
+    /* HTTP end-point */
+    mk_vhost_handler(hs->ctx, hs->vid, "/api/v1/storage", cb_storage, hs);
+
+    return 0;
+}

src/http_server/api/v1/storage.h

@@ -0,0 +1,29 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019      The Fluent Bit Authors
+ *  Copyright (C) 2015-2017 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_HS_API_V1_STORAGE_METRICS_H
+#define FLB_HS_API_V1_STORAGE_METRICS_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_http_server.h>
+
+int api_v1_storage_metrics(struct flb_hs *hs);
+
+#endif

src/http_server/flb_hs.c

@@ -36,10 +36,16 @@ static void cb_root(mk_request_t *request, void *data)
     mk_http_done(request);
 }
 
-/* Ingest metrics into the web service context */
-int flb_hs_push_metrics(struct flb_hs *hs, void *data, size_t size)
+/* Ingest pipeline metrics into the web service context */
+int flb_hs_push_pipeline_metrics(struct flb_hs *hs, void *data, size_t size)
 {
-    return mk_mq_send(hs->ctx, hs->qid, data, size);
+    return mk_mq_send(hs->ctx, hs->qid_metrics, data, size);
+}
+
+/* Ingest storage metrics into the web service context */
+int flb_hs_push_storage_metrics(struct flb_hs *hs, void *data, size_t size)
+{
+    return mk_mq_send(hs->ctx, hs->qid_storage, data, size);
 }
 
 /* Create ROOT endpoints */