format_influxdb: Add support for `METRIC_TYPE_FPCOUNTER`.

diff --git a/src/utils/format_influxdb/format_influxdb.c b/src/utils/format_influxdb/format_influxdb.c
index 2e9588bbfdb41baa92f43274e713c6e06e0dc292..bef8fa5181f5fe518e217417a15d272c347b5afa 100644 (file)
--- a/src/utils/format_influxdb/format_influxdb.c
+++ b/src/utils/format_influxdb/format_influxdb.c
@@ -31,73 +31,82 @@
 
 #include "utils/format_influxdb/format_influxdb.h"
 
-int format_influxdb_point(strbuf_t *sb, metric_t metric, bool store_rates) {
-  bool have_values = false;
+#define NEED_ESCAPE "\\ ,=\""
+#define ESCAPE_CHAR '\\'
 
-#define BUFFER_ADD_ESCAPE(...)                                                 \
+#define ERR_COMBINE(err, f)                                                    \
   do {                                                                         \
-    int status = strbuf_print_escaped(sb, __VA_ARGS__, "\\ ,=\"", '\\');       \
-    if (status != 0)                                                           \
-      return status;                                                           \
+    err = err ? err : f;                                                       \
   } while (0)
 
-#define BUFFER_ADD(...)                                                        \
-  do {                                                                         \
-    int status = strbuf_printf(sb, __VA_ARGS__);                               \
-    if (status != 0)                                                           \
-      return status;                                                           \
-  } while (0)
+static int format_metric_identity(strbuf_t *sb, metric_t const *m) {
+  int err = strbuf_print_escaped(sb, m->family->name, NEED_ESCAPE, ESCAPE_CHAR);
+  for (size_t j = 0; j < m->label.num; j++) {
+    label_pair_t label = m->label.ptr[j];
+    ERR_COMBINE(err, strbuf_printf(sb, ","));
+    ERR_COMBINE(err,
+                strbuf_print_escaped(sb, label.name, NEED_ESCAPE, ESCAPE_CHAR));
+    ERR_COMBINE(err, strbuf_printf(sb, "="));
+    ERR_COMBINE(err,
+                strbuf_print_escaped(sb, label.value, "\\ ,=\"", ESCAPE_CHAR));
+  }
+  ERR_COMBINE(err, strbuf_printf(sb, " "));
+  return err;
+}
 
-  have_values = false;
-  BUFFER_ADD_ESCAPE(metric.family->name);
-  for (size_t j = 0; j < metric.label.num; j++) {
-    label_pair_t label = metric.label.ptr[j];
-    BUFFER_ADD(",");
-    BUFFER_ADD_ESCAPE(label.name);
-    BUFFER_ADD("=");
-    BUFFER_ADD_ESCAPE(label.value);
+static int format_metric_rate(strbuf_t *sb, metric_t const *m) {
+  gauge_t rate = 0;
+  if (uc_get_rate(m, &rate) != 0) {
+    WARNING("write_influxdb_udp plugin: uc_get_rate failed.");
+    return EINVAL;
+  }
+  if (isnan(rate)) {
+    return EAGAIN;
   }
-  BUFFER_ADD(" ");
 
-  if (store_rates && (metric.family->type == METRIC_TYPE_COUNTER)) {
-    gauge_t rate;
-    if (uc_get_rate(&metric, &rate) != 0) {
-      WARNING("write_influxdb_udp plugin: "
-              "uc_get_rate failed.");
-      return EINVAL;
-    }
-    if (!isnan(rate)) {
-      BUFFER_ADD("value=" GAUGE_FORMAT, rate);
-      have_values = true;
-    }
-  } else {
-    switch (metric.family->type) {
-    case METRIC_TYPE_GAUGE:
-    case METRIC_TYPE_UNTYPED:
-      if (!isnan(metric.value.gauge)) {
-        BUFFER_ADD("value=" GAUGE_FORMAT, metric.value.gauge);
-        have_values = true;
-      }
-      break;
-    case METRIC_TYPE_COUNTER:
-      BUFFER_ADD("value=%" PRIi64 "i", metric.value.counter);
-      have_values = true;
-      break;
-    default:
-      WARNING("write_influxdb_udp plugin: "
-              "unknown family type.");
-      return EINVAL;
-      break;
+  return strbuf_printf(sb, "value=" GAUGE_FORMAT, rate);
+}
+
+static int format_metric_value(strbuf_t *sb, metric_t const *m,
+                               bool store_rate) {
+  if (store_rate) {
+    return format_metric_rate(sb, m);
+  }
+
+  switch (m->family->type) {
+  case METRIC_TYPE_GAUGE:
+    if (isnan(m->value.gauge)) {
+      return EAGAIN;
     }
+    return strbuf_printf(sb, "value=" GAUGE_FORMAT, m->value.gauge);
+  case METRIC_TYPE_COUNTER:
+    return strbuf_printf(sb, "value=%" PRIu64 "i", m->value.counter);
+  case METRIC_TYPE_FPCOUNTER:
+    return strbuf_printf(sb, "value=" GAUGE_FORMAT, m->value.fpcounter);
+  case METRIC_TYPE_UNTYPED:
   }
 
-  if (!have_values)
-    return 0;
+  ERROR("format_influxdb plugin: invalid metric type: %d", m->family->type);
+  return EINVAL;
+}
 
-  BUFFER_ADD(" %" PRIu64 "\n", CDTIME_T_TO_MS(metric.time));
+static int format_metric_time(strbuf_t *sb, metric_t const *m) {
+  return strbuf_printf(sb, " %" PRIu64 "\n", CDTIME_T_TO_MS(m->time));
+}
+
+int format_influxdb_point(strbuf_t *sb, metric_t const *m, bool store_rate) {
+  int err = format_metric_identity(sb, m);
+  if (err != 0) {
+    return err;
+  }
 
-#undef BUFFER_ADD_ESCAPE
-#undef BUFFER_ADD
+  err = format_metric_value(sb, m, store_rate);
+  if (err == EAGAIN) {
+    return 0;
+  }
+  if (err != 0) {
+    return err;
+  }
 
-  return 0;
+  return format_metric_time(sb, m);
 } /* int write_influxdb_point */

diff --git a/src/utils/format_influxdb/format_influxdb.h b/src/utils/format_influxdb/format_influxdb.h
index bd6b9746b97395f3c6121037f697697471a540d8..b9c5e5257fd3501dd0301fd02eaffa87b8dc2d01 100644 (file)
--- a/src/utils/format_influxdb/format_influxdb.h
+++ b/src/utils/format_influxdb/format_influxdb.h
@@ -30,6 +30,6 @@
 
 #include "plugin.h"
 
-int format_influxdb_point(strbuf_t *sb, metric_t metric, bool store_rates);
+int format_influxdb_point(strbuf_t *sb, metric_t const *m, bool store_rate);
 
 #endif /* UTILS_FORMAT_INFLUXDB_H */

diff --git a/src/write_http.c b/src/write_http.c
index 28eb3d69b07f5c0c1bbd43edf10356809a50e721..59a79af8653f9bcd4e2ab846dfd3a2c94b7e0de7 100644 (file)
--- a/src/write_http.c
+++ b/src/write_http.c
@@ -473,9 +473,8 @@ static int wh_write_influxdb(metric_family_t const *fam, wh_callback_t *cb) {
   pthread_mutex_lock(&cb->send_buffer_lock);
 
   for (size_t i = 0; i < fam->metric.num; i++) {
-    metric_t metric = fam->metric.ptr[i];
-    int status =
-        format_influxdb_point(&cb->send_buffer, metric, cb->store_rates);
+    metric_t const *m = fam->metric.ptr + i;
+    int status = format_influxdb_point(&cb->send_buffer, m, cb->store_rates);
     if (status != 0) {
       pthread_mutex_unlock(&cb->send_buffer_lock);
       ERROR("write_http plugin: format_influxdb_point failed: %s",

diff --git a/src/write_influxdb_udp.c b/src/write_influxdb_udp.c
index 4f9c9c4cd92d4f7b4ba45515ada47ec1f6a3d690..0f727c7f987ee3bdd8f3eb8e5b39012aa26742d3 100644 (file)
--- a/src/write_influxdb_udp.c
+++ b/src/write_influxdb_udp.c
@@ -356,9 +356,9 @@ static int write_influxdb_udp_write(metric_family_t const *fam,
   strbuf_t sb = STRBUF_CREATE_FIXED(buffer, buffer_len);
 
   for (size_t i = 0; i < fam->metric.num;) {
-    metric_t metric = fam->metric.ptr[i];
+    metric_t const *m = fam->metric.ptr + i;
     const size_t pos = sb.pos;
-    int status = format_influxdb_point(&sb, metric, wifxudp_config_store_rates);
+    int status = format_influxdb_point(&sb, m, wifxudp_config_store_rates);
     if (status == ENOSPC) {
       fill_send_buffer(sb.ptr, pos);
       strbuf_reset(&sb);