/api/v2/nodes and streaming function (#15168)

* dummy streaming function

* expose global functions upstream

* separate function for pushing global functions

* add missing conditions

* allow streaming function to run async

* started internal API for functions

* cache host retention and expose it to /api/v2/nodes

* internal API for function table fields; more progress on streaming status

* abstracted and unified rrdhost status

* port old coverity warning fix - although it is not needed

* add ML information to rrdhost status

* add ML capability to streaming to signal the transmission of ML information; added ML information to host status

* protect host->receiver

* count metrics and instances per host

* exposed all inbound and outbound streaming

* fix for ML status and dependency of DATA_WITH_ML to INTERPOLATED, not IEEE754

* update ML dummy

* added all fields

* added streaming group by and cleaned up accepted values by cloud

* removed type

* Revert "removed type"

This reverts commit faae4177e603d4f85b7433f33f92ef3ccd23976e.

* added context to db summary

* new /api/v2/nodes schema

* added ML type

* change default function charts

* log to trace new capa

* add more debug

* removed debugging code

* retry on receive interrupted read; respect sender reconnect delay in all cases

* set disconnected host flag and manipulate localhost child count atomically, inside set/clear receiver

* fix infinite loop

* send_to_plugin() now has a spinlock to ensure that only 1 thread is writing to the plugin/child at the same time

* global cloud_status() call

* cloud should be a section, since it will contain error information

* put cloud capabilities into cloud

* aclk status in /api/v2 agents sections

* keep aclk_connection_counter

* updates on /api/v2/nodes

* final /api/v2/nodes and addition of /api/v2/nodes_instances

* parametrize all /api/v2/xxx output to control which info is outputed per endpoint

* always accept nodes selector

* st needs to be per instance, not per node

* fix merging of contexts; fix cups plugin priorities

* add after and before parameters to /api/v2/contexts/nodes/nodes_instances/q

* give each libuv worker a unique id

* aclk http_api_v2 version 4

4 files changed, 202 insertions, 2 deletions

diff --git a/libnetdata/buffer/buffer.h b/libnetdata/buffer/buffer.h
index 22686a5a16..0021c38ff6 100644
--- a/libnetdata/buffer/buffer.h
+++ b/libnetdata/buffer/buffer.h
@@ -921,4 +921,199 @@ static inline void buffer_json_array_close(BUFFER *wb) {
     _buffer_json_depth_pop(wb);
 }
 
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_OPTS_NONE         = 0,
+    RRDF_FIELD_OPTS_UNIQUE_KEY   = (1 << 0), // the field is the unique key of the row
+    RRDF_FIELD_OPTS_VISIBLE      = (1 << 1), // the field should be visible by default
+    RRDF_FIELD_OPTS_STICKY       = (1 << 2), // the field should be sticky
+} RRDF_FIELD_OPTIONS;
+
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_TYPE_INTEGER,
+    RRDF_FIELD_TYPE_STRING,
+    RRDF_FIELD_TYPE_DETAIL_STRING,
+    RRDF_FIELD_TYPE_BAR_WITH_INTEGER,
+    RRDF_FIELD_TYPE_DURATION,
+    RRDF_FIELD_TYPE_TIMESTAMP,
+    RRDF_FIELD_TYPE_ARRAY,
+} RRDF_FIELD_TYPE;
+
+static inline const char *rrdf_field_type_to_string(RRDF_FIELD_TYPE type) {
+    switch(type) {
+        default:
+        case RRDF_FIELD_TYPE_INTEGER:
+            return "integer";
+
+        case RRDF_FIELD_TYPE_STRING:
+            return "string";
+
+        case RRDF_FIELD_TYPE_DETAIL_STRING:
+            return "detail-string";
+
+        case RRDF_FIELD_TYPE_BAR_WITH_INTEGER:
+            return "bar-with-integer";
+
+        case RRDF_FIELD_TYPE_DURATION:
+            return "duration";
+
+        case RRDF_FIELD_TYPE_TIMESTAMP:
+            return "timestamp";
+
+        case RRDF_FIELD_TYPE_ARRAY:
+            return "array";
+    }
+}
+
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_VISUAL_VALUE,    // show the value, possibly applying a transformation
+    RRDF_FIELD_VISUAL_BAR,      // show the value and a bar, respecting the max field to fill the bar at 100%
+    RRDF_FIELD_VISUAL_PILL,     // array of values (transformation is respected)
+} RRDF_FIELD_VISUAL;
+
+static inline const char *rrdf_field_visual_to_string(RRDF_FIELD_VISUAL visual) {
+    switch(visual) {
+        default:
+        case RRDF_FIELD_VISUAL_VALUE:
+            return "value";
+
+        case RRDF_FIELD_VISUAL_BAR:
+            return "bar";
+
+        case RRDF_FIELD_VISUAL_PILL:
+            return "pill";
+    }
+}
+
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_TRANSFORM_NONE,      // show the value as-is
+    RRDF_FIELD_TRANSFORM_NUMBER,    // show the value repsecting the decimal_points
+    RRDF_FIELD_TRANSFORM_DURATION,  // transform as duration in second to a human readable duration
+    RRDF_FIELD_TRANSFORM_DATETIME,  // UNIX epoch timestamp in ms
+} RRDF_FIELD_TRANSFORM;
+
+static inline const char *rrdf_field_transform_to_string(RRDF_FIELD_TRANSFORM transform) {
+    switch(transform) {
+        default:
+        case RRDF_FIELD_TRANSFORM_NONE:
+            return "none";
+
+        case RRDF_FIELD_TRANSFORM_NUMBER:
+            return "number";
+
+        case RRDF_FIELD_TRANSFORM_DURATION:
+            return "duration";
+
+        case RRDF_FIELD_TRANSFORM_DATETIME:
+            return "datetime";
+    }
+}
+
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_SORT_ASCENDING  = (1 << 0),
+    RRDF_FIELD_SORT_DESCENDING = (1 << 1),
+
+    RRDF_FIELD_SORT_FIXED      = (1 << 7),
+} RRDF_FIELD_SORT;
+
+static inline const char *rrdf_field_sort_to_string(RRDF_FIELD_SORT sort) {
+    if(sort & RRDF_FIELD_SORT_DESCENDING)
+        return "descending";
+
+    else
+        return "ascending";
+}
+
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_SUMMARY_UNIQUECOUNT,     // Finds the number of unique values of a group of rows
+    RRDF_FIELD_SUMMARY_SUM,             // Sums the values of a group of rows
+    RRDF_FIELD_SUMMARY_MIN,             // Finds the minimum value of a group of rows
+    RRDF_FIELD_SUMMARY_MAX,             // Finds the maximum value of a group of rows
+    // RRDF_FIELD_SUMMARY_EXTENT,          // Finds the minimum and maximum values of a group of rows
+    RRDF_FIELD_SUMMARY_MEAN,            // Finds the mean/average value of a group of rows
+    RRDF_FIELD_SUMMARY_MEDIAN,          // Finds the median value of a group of rows
+    // RRDF_FIELD_SUMMARY_UNIQUE,         // Finds the unique values of a group of rows
+    RRDF_FIELD_SUMMARY_COUNT,           // Calculates the number of rows in a group
+} RRDF_FIELD_SUMMARY;
+
+static inline const char *rrdf_field_summary_to_string(RRDF_FIELD_SUMMARY summary) {
+    switch(summary) {
+        default:
+        case RRDF_FIELD_SUMMARY_COUNT:
+            return "count";
+
+        case RRDF_FIELD_SUMMARY_UNIQUECOUNT:
+            return "uniqueCount";
+
+        case RRDF_FIELD_SUMMARY_SUM:
+            return "sum";
+
+        case RRDF_FIELD_SUMMARY_MIN:
+            return "min";
+
+        case RRDF_FIELD_SUMMARY_MEAN:
+            return "mean";
+
+        case RRDF_FIELD_SUMMARY_MEDIAN:
+            return "median";
+
+        case RRDF_FIELD_SUMMARY_MAX:
+            return "max";
+    }
+}
+
+typedef enum __attribute__((packed)) {
+    RRDF_FIELD_FILTER_RANGE,
+    RRDF_FIELD_FILTER_MULTISELECT,
+} RRDF_FIELD_FILTER;
+
+static inline const char *rrdf_field_filter_to_string(RRDF_FIELD_FILTER filter) {
+    switch(filter) {
+        default:
+        case RRDF_FIELD_FILTER_RANGE:
+            return "range";
+
+        case RRDF_FIELD_FILTER_MULTISELECT:
+            return "multiselect";
+    }
+}
+
+static inline void
+buffer_rrdf_table_add_field(BUFFER *wb, size_t field_id, const char *key, const char *name, RRDF_FIELD_TYPE type,
+                            RRDF_FIELD_VISUAL visual, RRDF_FIELD_TRANSFORM transform, size_t decimal_points,
+                            const char *units, NETDATA_DOUBLE max, RRDF_FIELD_SORT sort, const char *pointer_to,
+                            RRDF_FIELD_SUMMARY summary, RRDF_FIELD_FILTER filter, RRDF_FIELD_OPTIONS options,
+                            const char *default_value) {
+
+    buffer_json_member_add_object(wb, key);
+    {
+        buffer_json_member_add_uint64(wb, "index", field_id);
+        buffer_json_member_add_boolean(wb, "unique_key", options & RRDF_FIELD_OPTS_UNIQUE_KEY);
+        buffer_json_member_add_string(wb, "name", name);
+        buffer_json_member_add_boolean(wb, "visible", options & RRDF_FIELD_OPTS_VISIBLE);
+        buffer_json_member_add_string(wb, "type", rrdf_field_type_to_string(type));
+        buffer_json_member_add_string_or_omit(wb, "units", units);
+        buffer_json_member_add_string(wb, "visualization", rrdf_field_visual_to_string(visual));
+
+        buffer_json_member_add_object(wb, "value_options");
+        {
+            buffer_json_member_add_string_or_omit(wb, "units", units);
+            buffer_json_member_add_string(wb, "transform", rrdf_field_transform_to_string(transform));
+            buffer_json_member_add_uint64(wb, "decimal_points", decimal_points);
+            buffer_json_member_add_string(wb, "default_value", default_value);
+        }
+        buffer_json_object_close(wb);
+
+        if (!isnan((NETDATA_DOUBLE) (max)))
+            buffer_json_member_add_double(wb, "max", (NETDATA_DOUBLE) (max));
+
+        buffer_json_member_add_string_or_omit(wb, "pointer_to", pointer_to);
+        buffer_json_member_add_string(wb, "sort", rrdf_field_sort_to_string(sort));
+        buffer_json_member_add_boolean(wb, "sortable", !(sort & RRDF_FIELD_SORT_FIXED));
+        buffer_json_member_add_boolean(wb, "sticky", options & RRDF_FIELD_OPTS_STICKY);
+        buffer_json_member_add_string(wb, "summary", rrdf_field_summary_to_string(summary));
+        buffer_json_member_add_string(wb, "filter", rrdf_field_filter_to_string(filter));
+    }
+    buffer_json_object_close(wb);
+}
+
 #endif /* NETDATA_WEB_BUFFER_H */
diff --git a/libnetdata/parser/parser.c b/libnetdata/parser/parser.c
index 80c9a2639a..d760233318 100644
--- a/libnetdata/parser/parser.c
+++ b/libnetdata/parser/parser.c
@@ -46,6 +46,7 @@ PARSER *parser_init(void *user, FILE *fp_input, FILE *fp_output, int fd,
     parser->flags = flags;
     parser->worker_job_next_id = WORKER_PARSER_FIRST_JOB;
 
+    netdata_spinlock_init(&parser->writer.spinlock);
     return parser;
 }
 
diff --git a/libnetdata/parser/parser.h b/libnetdata/parser/parser.h
index c21cbaf7e6..a40db90705 100644
--- a/libnetdata/parser/parser.h
+++ b/libnetdata/parser/parser.h
@@ -65,6 +65,10 @@ typedef struct parser {
         DICTIONARY *functions;
         usec_t smaller_timeout;
     } inflight;
+
+    struct {
+        SPINLOCK spinlock;
+    } writer;
 } PARSER;
 
 PARSER *parser_init(void *user, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl);
diff --git a/libnetdata/socket/socket.c b/libnetdata/socket/socket.c
index 220db7601f..daf1629a0e 100644
--- a/libnetdata/socket/socket.c
+++ b/libnetdata/socket/socket.c
@@ -15,10 +15,10 @@ SOCKET_PEERS socket_peers(int sock_fd) {
     SOCKET_PEERS peers;
 
     if(sock_fd < 0) {
-        strncpyz(peers.peer.ip, "unknown", sizeof(peers.peer.ip) - 1);
+        strncpyz(peers.peer.ip, "not connected", sizeof(peers.peer.ip) - 1);
         peers.peer.port = 0;
 
-        strncpyz(peers.local.ip, "unknown", sizeof(peers.local.ip) - 1);
+        strncpyz(peers.local.ip, "not connected", sizeof(peers.local.ip) - 1);
         peers.local.port = 0;
 
         return peers;