/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

11 files changed, 1494 insertions, 114 deletions

diff --git a/database/contexts/api_v2.c b/database/contexts/api_v2.c
index d83a9e9e3e..ec7db44507 100644
--- a/database/contexts/api_v2.c
+++ b/database/contexts/api_v2.c
@@ -88,6 +88,8 @@ static inline bool full_text_search_char(FTS_INDEX *fts, SIMPLE_PATTERN *q, char
 }
 
 struct rrdcontext_to_json_v2_data {
+    time_t now;
+
     BUFFER *wb;
     struct api_v2_contexts_request *request;
     DICTIONARY *ctx;
@@ -113,6 +115,13 @@ struct rrdcontext_to_json_v2_data {
         FTS_INDEX fts;
     } q;
 
+    struct {
+        bool enabled;
+        bool relative;
+        time_t after;
+        time_t before;
+    } window;
+
     struct query_timings timings;
 };
 
@@ -132,6 +141,9 @@ static FTS_MATCH rrdcontext_to_json_v2_full_text_search(struct rrdcontext_to_jso
     dfe_start_read(rc->rrdinstances, ri) {
         if(matched) break;
 
+        if(ctl->window.enabled && !query_matches_retention(ctl->window.after, ctl->window.before, ri->first_time_s, ri->last_time_s, 0))
+            continue;
+
         if(unlikely(full_text_search_string(&ctl->q.fts, q, ri->id)) ||
            (ri->name != ri->id && full_text_search_string(&ctl->q.fts, q, ri->name))) {
             matched = FTS_MATCHED_INSTANCE;
@@ -140,6 +152,9 @@ static FTS_MATCH rrdcontext_to_json_v2_full_text_search(struct rrdcontext_to_jso
 
         RRDMETRIC *rm;
         dfe_start_read(ri->rrdmetrics, rm) {
+            if(ctl->window.enabled && !query_matches_retention(ctl->window.after, ctl->window.before, rm->first_time_s, rm->last_time_s, 0))
+                continue;
+
             if(unlikely(full_text_search_string(&ctl->q.fts, q, rm->id)) ||
                (rm->name != rm->id && full_text_search_string(&ctl->q.fts, q, rm->name))) {
                 matched = FTS_MATCHED_DIMENSION;
@@ -185,16 +200,19 @@ static ssize_t rrdcontext_to_json_v2_add_context(void *data, RRDCONTEXT_ACQUIRED
 
     RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
 
+    if(ctl->window.enabled && !query_matches_retention(ctl->window.after, ctl->window.before, rc->first_time_s, rc->last_time_s, 0))
+        return 0; // continue to next context
+
     FTS_MATCH match = ctl->q.host_match;
     if((ctl->options & CONTEXTS_V2_SEARCH) && ctl->q.pattern) {
         match = rrdcontext_to_json_v2_full_text_search(ctl, rc, ctl->q.pattern);
 
         if(match == FTS_MATCHED_NONE)
-            return 0;
+            return 0; // continue to next context
     }
 
     struct rrdcontext_to_json_v2_entry t = {
-            .count = 0,
+            .count = 1,
             .id = rc->id,
             .family = string_dup(rc->family),
             .priority = rc->priority,
@@ -202,65 +220,197 @@ static ssize_t rrdcontext_to_json_v2_add_context(void *data, RRDCONTEXT_ACQUIRED
             .last_time_s = rc->last_time_s,
             .flags = rc->flags,
             .match = match,
-    }, *z = dictionary_set(ctl->ctx, string2str(rc->id), &t, sizeof(t));
-
-    if(!z->count) {
-        // we just added this
-        z->count = 1;
-    }
-    else {
-        // it is already in there
-        z->count++;
-        z->flags |= rc->flags;
-
-        if(z->priority > rc->priority)
-            z->priority = rc->priority;
+    };
 
-        if(z->first_time_s > rc->first_time_s)
-            z->first_time_s = rc->first_time_s;
+    dictionary_set(ctl->ctx, string2str(rc->id), &t, sizeof(t));
 
-        if(z->last_time_s < rc->last_time_s)
-            z->last_time_s = rc->last_time_s;
+    return 1;
+}
 
-        if(z->family != rc->family) {
-            z->family = string_2way_merge(z->family, rc->family);
-        }
+void buffer_json_agent_status_id(BUFFER *wb, size_t ai, usec_t duration_ut) {
+    buffer_json_member_add_object(wb, "st");
+    {
+        buffer_json_member_add_uint64(wb, "ai", ai);
+        buffer_json_member_add_uint64(wb, "code", 200);
+        buffer_json_member_add_string(wb, "msg", "");
+        if (duration_ut)
+            buffer_json_member_add_double(wb, "ms", (NETDATA_DOUBLE) duration_ut / 1000.0);
     }
-
-    return 1;
+    buffer_json_object_close(wb);
 }
 
-void buffer_json_node_add_v2(BUFFER *wb, RRDHOST *host, size_t ni, usec_t duration_ut) {
+void buffer_json_node_add_v2(BUFFER *wb, RRDHOST *host, size_t ni, usec_t duration_ut, bool status) {
     buffer_json_member_add_string(wb, "mg", host->machine_guid);
+
     if(host->node_id)
         buffer_json_member_add_uuid(wb, "nd", host->node_id);
     buffer_json_member_add_string(wb, "nm", rrdhost_hostname(host));
     buffer_json_member_add_uint64(wb, "ni", ni);
-    buffer_json_member_add_object(wb, "st");
-    buffer_json_member_add_uint64(wb, "ai", 0);
-    buffer_json_member_add_uint64(wb, "code", 200);
-    buffer_json_member_add_string(wb, "msg", "");
-    if(duration_ut)
-        buffer_json_member_add_double(wb, "ms", (NETDATA_DOUBLE)duration_ut / 1000.0);
-    buffer_json_object_close(wb);
+
+    if(status)
+        buffer_json_agent_status_id(wb, 0, duration_ut);
+}
+
+static void rrdhost_receiver_to_json(BUFFER *wb, RRDHOST_STATUS *s, const char *key) {
+    buffer_json_member_add_object(wb, key);
+    {
+        buffer_json_member_add_uint64(wb, "id", s->ingest.id);
+        buffer_json_member_add_uint64(wb, "hops", s->ingest.hops);
+        buffer_json_member_add_string(wb, "type", rrdhost_ingest_type_to_string(s->ingest.type));
+        buffer_json_member_add_string(wb, "status", rrdhost_ingest_status_to_string(s->ingest.status));
+        buffer_json_member_add_time_t(wb, "since", s->ingest.since);
+        buffer_json_member_add_time_t(wb, "age", s->now - s->ingest.since);
+
+        if(s->ingest.type == RRDHOST_INGEST_TYPE_CHILD) {
+            if(s->ingest.status == RRDHOST_INGEST_STATUS_OFFLINE)
+                buffer_json_member_add_string(wb, "reason", s->ingest.reason);
+
+            if(s->ingest.status == RRDHOST_INGEST_STATUS_REPLICATING) {
+                buffer_json_member_add_object(wb, "replication");
+                {
+                    buffer_json_member_add_boolean(wb, "in_progress", s->ingest.replication.in_progress);
+                    buffer_json_member_add_double(wb, "completion", s->ingest.replication.completion);
+                    buffer_json_member_add_uint64(wb, "instances", s->ingest.replication.instances);
+                }
+                buffer_json_object_close(wb); // replication
+            }
+
+            if(s->ingest.status == RRDHOST_INGEST_STATUS_REPLICATING || s->ingest.status == RRDHOST_INGEST_STATUS_ONLINE) {
+                buffer_json_member_add_object(wb, "source");
+                {
+                    char buf[1024 + 1];
+                    snprintfz(buf, 1024, "[%s]:%d%s", s->ingest.peers.local.ip, s->ingest.peers.local.port, s->ingest.ssl ? ":SSL" : "");
+                    buffer_json_member_add_string(wb, "local", buf);
+
+                    snprintfz(buf, 1024, "[%s]:%d%s", s->ingest.peers.peer.ip, s->ingest.peers.peer.port, s->ingest.ssl ? ":SSL" : "");
+                    buffer_json_member_add_string(wb, "remote", buf);
+
+                    stream_capabilities_to_json_array(wb, s->ingest.capabilities, "capabilities");
+                }
+                buffer_json_object_close(wb); // source
+            }
+        }
+    }
+    buffer_json_object_close(wb); // collection
+}
+
+static void rrdhost_sender_to_json(BUFFER *wb, RRDHOST_STATUS *s, const char *key) {
+    if(s->stream.status == RRDHOST_STREAM_STATUS_DISABLED)
+        return;
+
+    buffer_json_member_add_object(wb, key);
+
+    buffer_json_member_add_uint64(wb, "id", s->stream.id);
+    buffer_json_member_add_uint64(wb, "hops", s->stream.hops);
+    buffer_json_member_add_string(wb, "status", rrdhost_streaming_status_to_string(s->stream.status));
+    buffer_json_member_add_time_t(wb, "since", s->stream.since);
+    buffer_json_member_add_time_t(wb, "age", s->now - s->stream.since);
+
+    if(s->stream.status == RRDHOST_STREAM_STATUS_OFFLINE)
+        buffer_json_member_add_string(wb, "reason", s->stream.reason);
+
+    if(s->stream.status == RRDHOST_STREAM_STATUS_REPLICATING || s->stream.status == RRDHOST_STREAM_STATUS_ONLINE) {
+
+        if(s->stream.status == RRDHOST_STREAM_STATUS_REPLICATING) {
+            buffer_json_member_add_object(wb, "replication");
+            {
+                buffer_json_member_add_boolean(wb, "in_progress", s->stream.replication.in_progress);
+                buffer_json_member_add_double(wb, "completion", s->stream.replication.completion);
+                buffer_json_member_add_uint64(wb, "instances", s->stream.replication.instances);
+            }
+            buffer_json_object_close(wb);
+        }
+
+        buffer_json_member_add_object(wb, "destination");
+        {
+            char buf[1024 + 1];
+            snprintfz(buf, 1024, "[%s]:%d%s", s->stream.peers.local.ip, s->stream.peers.local.port, s->stream.ssl ? ":SSL" : "");
+            buffer_json_member_add_string(wb, "local", buf);
+
+            snprintfz(buf, 1024, "[%s]:%d%s", s->stream.peers.peer.ip, s->stream.peers.peer.port, s->stream.ssl ? ":SSL" : "");
+            buffer_json_member_add_string(wb, "remote", buf);
+
+            stream_capabilities_to_json_array(wb, s->stream.capabilities, "capabilities");
+
+            buffer_json_member_add_object(wb, "traffic");
+            {
+                buffer_json_member_add_boolean(wb, "compression", s->stream.compression);
+                buffer_json_member_add_uint64(wb, "data", s->stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_DATA]);
+                buffer_json_member_add_uint64(wb, "metadata", s->stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_METADATA]);
+                buffer_json_member_add_uint64(wb, "functions", s->stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_FUNCTIONS]);
+                buffer_json_member_add_uint64(wb, "replication", s->stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_REPLICATION]);
+            }
+            buffer_json_object_close(wb); // traffic
+
+            buffer_json_member_add_array(wb, "candidates");
+            struct rrdpush_destinations *d;
+            for (d = s->host->destinations; d; d = d->next) {
+                buffer_json_add_array_item_object(wb);
+                {
+
+                    if (d->ssl) {
+                        snprintfz(buf, 1024, "%s:SSL", string2str(d->destination));
+                        buffer_json_member_add_string(wb, "destination", buf);
+                    }
+                    else
+                        buffer_json_member_add_string(wb, "destination", string2str(d->destination));
+
+                    buffer_json_member_add_time_t(wb, "last_check", d->last_attempt);
+                    buffer_json_member_add_time_t(wb, "age", s->now - d->last_attempt);
+                    buffer_json_member_add_string(wb, "last_error", d->last_error);
+                    buffer_json_member_add_string(wb, "last_handshake",
+                                                  stream_handshake_error_to_string(d->last_handshake));
+                    buffer_json_member_add_time_t(wb, "next_check", d->postpone_reconnection_until);
+                    buffer_json_member_add_time_t(wb, "next_in",
+                                                  (d->postpone_reconnection_until > s->now) ?
+                                                  d->postpone_reconnection_until - s->now : 0);
+                }
+                buffer_json_object_close(wb); // each candidate
+            }
+            buffer_json_array_close(wb); // candidates
+        }
+        buffer_json_object_close(wb); // destination
+    }
+    buffer_json_object_close(wb); // streaming
+}
+
+static void agent_capabilities_to_json(BUFFER *wb, RRDHOST *host, const char *key) {
+    buffer_json_member_add_array(wb, key);
+
+    struct capability *capas = aclk_get_node_instance_capas(host);
+    for(struct capability *capa = capas; capa->name ;capa++) {
+        buffer_json_add_array_item_object(wb);
+        {
+            buffer_json_member_add_string(wb, "name", capa->name);
+            buffer_json_member_add_uint64(wb, "version", capa->version);
+            buffer_json_member_add_boolean(wb, "enabled", capa->enabled);
+        }
+        buffer_json_object_close(wb);
+    }
+    buffer_json_array_close(wb);
+    freez(capas);
 }
 
 static ssize_t rrdcontext_to_json_v2_add_host(void *data, RRDHOST *host, bool queryable_host) {
     if(!queryable_host || !host->rrdctx.contexts)
         // the host matches the 'scope_host' but does not match the 'host' patterns
         // or the host does not have any contexts
-        return 0;
+        return 0; // continue to next host
 
     struct rrdcontext_to_json_v2_data *ctl = data;
     BUFFER *wb = ctl->wb;
 
+    if(ctl->window.enabled && !rrdhost_matches_window(host, ctl->window.after, ctl->window.before, ctl->now))
+        // the host does not have data in the requested window
+        return 0; // continue to next host
+
     if(ctl->request->timeout_ms && now_monotonic_usec() > ctl->timings.received_ut + ctl->request->timeout_ms * USEC_PER_MS)
         // timed out
-        return -2;
+        return -2; // stop the query
 
     if(ctl->request->interrupt_callback && ctl->request->interrupt_callback(ctl->request->interrupt_callback_data))
         // interrupted
-        return -1;
+        return -1; // stop the query
 
     bool host_matched = (ctl->options & CONTEXTS_V2_NODES);
     bool do_contexts = (ctl->options & (CONTEXTS_V2_CONTEXTS | CONTEXTS_V2_SEARCH));
@@ -293,87 +443,157 @@ static ssize_t rrdcontext_to_json_v2_add_host(void *data, RRDHOST *host, bool qu
         // restore it
         ctl->q.pattern = old_q;
 
-        if(added == -1)
-            return -1;
+        if(unlikely(added < 0))
+            return -1; // stop the query
 
         if(added)
             host_matched = true;
     }
 
-    if(host_matched && (ctl->options & (CONTEXTS_V2_NODES | CONTEXTS_V2_NODES_DETAILED | CONTEXTS_V2_DEBUG))) {
-        buffer_json_add_array_item_object(wb);
-        buffer_json_node_add_v2(wb, host, ctl->nodes.ni++, 0);
-
-        if(ctl->options & CONTEXTS_V2_NODES_DETAILED) {
-            buffer_json_member_add_string(wb, "version", rrdhost_program_version(host));
-            buffer_json_member_add_uint64(wb, "hops", host->system_info ? host->system_info->hops : (host == localhost) ? 0 : 1);
-            buffer_json_member_add_string(wb, "state", (host == localhost || !rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN)) ? "reachable" : "stale");
-            buffer_json_member_add_boolean(wb, "isDeleted", false);
-
-            buffer_json_member_add_array(wb, "services");
-            buffer_json_array_close(wb);
+    if(!host_matched)
+        return 0;
 
-            buffer_json_member_add_array(wb, "nodeInstanceCapabilities");
+    if(ctl->options & CONTEXTS_V2_NODES) {
+        buffer_json_add_array_item_object(wb); // this node
+        buffer_json_node_add_v2(wb, host, ctl->nodes.ni++, 0,
+                                (ctl->options & CONTEXTS_V2_AGENTS) && !(ctl->options & CONTEXTS_V2_NODES_INSTANCES));
+
+        if(ctl->options & (CONTEXTS_V2_NODES_INFO)) {
+            buffer_json_member_add_string(wb, "v", rrdhost_program_version(host));
+
+            host_labels2json(host, wb, "labels");
+
+            if (host->system_info) {
+                buffer_json_member_add_object(wb, "hw");
+                {
+                    buffer_json_member_add_string_or_empty(wb, "architecture", host->system_info->architecture);
+                    buffer_json_member_add_string_or_empty(wb, "cpuFrequency", host->system_info->host_cpu_freq);
+                    buffer_json_member_add_string_or_empty(wb, "cpus", host->system_info->host_cores);
+                    buffer_json_member_add_string_or_empty(wb, "memory", host->system_info->host_ram_total);
+                    buffer_json_member_add_string_or_empty(wb, "diskSpace", host->system_info->host_disk_space);
+                    buffer_json_member_add_string_or_empty(wb, "virtualization", host->system_info->virtualization);
+                    buffer_json_member_add_string_or_empty(wb, "container", host->system_info->container);
+                }
+                buffer_json_object_close(wb);
 
-            struct capability *capas = aclk_get_node_instance_capas(host);
-            struct capability *capa = capas;
-            while(capa->name != NULL) {
-                buffer_json_add_array_item_object(wb);
-                buffer_json_member_add_string(wb, "name", capa->name);
-                buffer_json_member_add_uint64(wb, "version", capa->version);
-                buffer_json_member_add_boolean(wb, "enabled", capa->enabled);
+                buffer_json_member_add_object(wb, "os");
+                {
+                    buffer_json_member_add_string_or_empty(wb, "id", host->system_info->host_os_id);
+                    buffer_json_member_add_string_or_empty(wb, "nm", host->system_info->host_os_name);
+                    buffer_json_member_add_string_or_empty(wb, "v", host->system_info->host_os_version);
+                    buffer_json_member_add_object(wb, "kernel");
+                    buffer_json_member_add_string_or_empty(wb, "nm", host->system_info->kernel_name);
+                    buffer_json_member_add_string_or_empty(wb, "v", host->system_info->kernel_version);
+                    buffer_json_object_close(wb);
+                }
                 buffer_json_object_close(wb);
-                capa++;
             }
-            buffer_json_array_close(wb);
-            freez(capas);
 
-            web_client_api_request_v1_info_summary_alarm_statuses(host, wb, "alarmCounters");
+            // created      - the node is created but never connected to cloud
+            // unreachable  - not currently connected
+            // stale        - connected but not having live data
+            // reachable    - connected with live data
+            // pruned       - not connected for some time and has been removed
+            buffer_json_member_add_string(wb, "state", rrdhost_state_cloud_emulation(host)?"reachable":"stale");
 
-            host_labels2json(host, wb, "hostLabels");
+            agent_capabilities_to_json(wb, host, "capabilities");
+        }
 
-            buffer_json_member_add_object(wb, "mlInfo");
-            buffer_json_member_add_boolean(wb, "mlCapable", ml_capable(host));
-            buffer_json_member_add_boolean(wb, "mlEnabled", ml_enabled(host));
-            buffer_json_object_close(wb);
+        if(ctl->options & (CONTEXTS_V2_NODES_INSTANCES)) {
+            buffer_json_member_add_array(wb, "instances");
+            buffer_json_add_array_item_object(wb); // this instance
+            {
+                buffer_json_agent_status_id(wb, 0, 0);
+
+                time_t now = now_realtime_sec();
+                RRDHOST_STATUS s;
+                rrdhost_status(host, now, &s);
+                buffer_json_member_add_object(wb, "db");
+                {
+                    buffer_json_member_add_string(wb, "status", rrdhost_db_status_to_string(s.db.status));
+                    buffer_json_member_add_string(wb, "liveness", rrdhost_db_liveness_to_string(s.db.liveness));
+                    buffer_json_member_add_string(wb, "mode", rrd_memory_mode_name(s.db.mode));
+                    buffer_json_member_add_time_t(wb, "first_time", s.db.first_time_s);
+                    buffer_json_member_add_time_t(wb, "last_time", s.db.last_time_s);
+                    buffer_json_member_add_uint64(wb, "metrics", s.db.metrics);
+                    buffer_json_member_add_uint64(wb, "instances", s.db.instances);
+                    buffer_json_member_add_uint64(wb, "contexts", s.db.contexts);
+                }
+                buffer_json_object_close(wb);
 
-            if(host->system_info) {
-                buffer_json_member_add_string_or_empty(wb, "architecture", host->system_info->architecture);
-                buffer_json_member_add_string_or_empty(wb, "kernelName", host->system_info->kernel_name);
-                buffer_json_member_add_string_or_empty(wb, "kernelVersion", host->system_info->kernel_version);
-                buffer_json_member_add_string_or_empty(wb, "cpuFrequency", host->system_info->host_cpu_freq);
-                buffer_json_member_add_string_or_empty(wb, "cpus", host->system_info->host_cores);
-                buffer_json_member_add_string_or_empty(wb, "memory", host->system_info->host_ram_total);
-                buffer_json_member_add_string_or_empty(wb, "diskSpace", host->system_info->host_disk_space);
-                buffer_json_member_add_string_or_empty(wb, "container", host->system_info->container);
-                buffer_json_member_add_string_or_empty(wb, "virtualization", host->system_info->virtualization);
-                buffer_json_member_add_string_or_empty(wb, "os", host->system_info->host_os_id);
-                buffer_json_member_add_string_or_empty(wb, "osName", host->system_info->host_os_name);
-                buffer_json_member_add_string_or_empty(wb, "osVersion", host->system_info->host_os_version);
-            }
+                rrdhost_receiver_to_json(wb, &s, "ingest");
+                rrdhost_sender_to_json(wb, &s, "stream");
+
+                buffer_json_member_add_object(wb, "ml");
+                buffer_json_member_add_string(wb, "status", rrdhost_ml_status_to_string(s.ml.status));
+                buffer_json_member_add_string(wb, "type", rrdhost_ml_type_to_string(s.ml.type));
+                if (s.ml.status == RRDHOST_ML_STATUS_RUNNING) {
+                    buffer_json_member_add_object(wb, "metrics");
+                    {
+                        buffer_json_member_add_uint64(wb, "anomalous", s.ml.metrics.anomalous);
+                        buffer_json_member_add_uint64(wb, "normal", s.ml.metrics.normal);
+                        buffer_json_member_add_uint64(wb, "trained", s.ml.metrics.trained);
+                        buffer_json_member_add_uint64(wb, "pending", s.ml.metrics.pending);
+                        buffer_json_member_add_uint64(wb, "silenced", s.ml.metrics.silenced);
+                    }
+                    buffer_json_object_close(wb); // metrics
+                }
+                buffer_json_object_close(wb); // ml
+
+                buffer_json_member_add_object(wb, "health");
+                {
+                    buffer_json_member_add_string(wb, "status", rrdhost_health_status_to_string(s.health.status));
+                    if (s.health.status == RRDHOST_HEALTH_STATUS_RUNNING) {
+                        buffer_json_member_add_object(wb, "alerts");
+                        {
+                            buffer_json_member_add_uint64(wb, "critical", s.health.alerts.critical);
+                            buffer_json_member_add_uint64(wb, "warning", s.health.alerts.warning);
+                            buffer_json_member_add_uint64(wb, "clear", s.health.alerts.clear);
+                            buffer_json_member_add_uint64(wb, "undefined", s.health.alerts.undefined);
+                            buffer_json_member_add_uint64(wb, "uninitialized", s.health.alerts.uninitialized);
+                        }
+                        buffer_json_object_close(wb); // alerts
+                    }
+                }
+                buffer_json_object_close(wb); // health
 
-            time_t now = now_realtime_sec();
-            buffer_json_member_add_object(wb, "status");
-            {
-                rrdhost_receiver_to_json(wb, host, "collection", now);
-                rrdhost_sender_to_json(wb, host, "streaming", now);
+                host_functions2json(host, wb); // functions
+                agent_capabilities_to_json(wb, host, "capabilities");
             }
-            buffer_json_object_close(wb); // status
+            buffer_json_object_close(wb); // this instance
+            buffer_json_array_close(wb); // instances
         }
-
-        buffer_json_object_close(wb);
+        buffer_json_object_close(wb); // this node
     }
 
-    return host_matched ? 1 : 0;
+    return 1;
 }
 
 static void buffer_json_contexts_v2_options_to_array(BUFFER *wb, CONTEXTS_V2_OPTIONS options) {
     if(options & CONTEXTS_V2_DEBUG)
         buffer_json_add_array_item_string(wb, "debug");
 
-    if(options & (CONTEXTS_V2_NODES | CONTEXTS_V2_NODES_DETAILED))
+    if(options & CONTEXTS_V2_MINIFY)
+        buffer_json_add_array_item_string(wb, "minify");
+
+    if(options & CONTEXTS_V2_VERSIONS)
+        buffer_json_add_array_item_string(wb, "versions");
+
+    if(options & CONTEXTS_V2_AGENTS)
+        buffer_json_add_array_item_string(wb, "agents");
+
+    if(options & CONTEXTS_V2_AGENTS_INFO)
+        buffer_json_add_array_item_string(wb, "agents-info");
+
+    if(options & CONTEXTS_V2_NODES)
         buffer_json_add_array_item_string(wb, "nodes");
 
+    if(options & CONTEXTS_V2_NODES_INFO)
+        buffer_json_add_array_item_string(wb, "nodes-info");
+
+    if(options & CONTEXTS_V2_NODES_INSTANCES)
+        buffer_json_add_array_item_string(wb, "nodes-instances");
+
     if(options & CONTEXTS_V2_CONTEXTS)
         buffer_json_add_array_item_string(wb, "contexts");
 
@@ -396,7 +616,7 @@ void buffer_json_query_timings(BUFFER *wb, const char *key, struct query_timings
     buffer_json_object_close(wb);
 }
 
-void buffer_json_agents_array_v2(BUFFER *wb, struct query_timings *timings, time_t now_s) {
+void buffer_json_agents_array_v2(BUFFER *wb, struct query_timings *timings, time_t now_s, bool info) {
     if(!now_s)
         now_s = now_realtime_sec();
 
@@ -408,6 +628,56 @@ void buffer_json_agents_array_v2(BUFFER *wb, struct query_timings *timings, time
     buffer_json_member_add_time_t(wb, "now", now_s);
     buffer_json_member_add_uint64(wb, "ai", 0);
 
+    if(info) {
+        buffer_json_member_add_string(wb, "v", string2str(localhost->program_version));
+
+        buffer_json_member_add_object(wb, "cloud");
+        {
+            size_t id = cloud_connection_id();
+            CLOUD_STATUS status = cloud_status();
+            time_t last_change = cloud_last_change();
+            time_t next_connect = cloud_next_connection_attempt();
+            buffer_json_member_add_uint64(wb, "id", id);
+            buffer_json_member_add_string(wb, "status", cloud_status_to_string(status));
+            buffer_json_member_add_time_t(wb, "since", last_change);
+            buffer_json_member_add_time_t(wb, "age", now_s - last_change);
+
+            if (status != CLOUD_STATUS_ONLINE)
+                buffer_json_member_add_string(wb, "reason", cloud_offline_reason());
+
+            if (status == CLOUD_STATUS_OFFLINE && next_connect > now_s) {
+                buffer_json_member_add_time_t(wb, "next_check", next_connect);
+                buffer_json_member_add_time_t(wb, "next_in", next_connect - now_s);
+            }
+
+            if (status != CLOUD_STATUS_DISABLED && cloud_base_url())
+                buffer_json_member_add_string(wb, "url", cloud_base_url());
+        }
+        buffer_json_object_close(wb); // cloud
+
+        buffer_json_member_add_array(wb, "db_size");
+        for (size_t tier = 0; tier < storage_tiers; tier++) {
+            STORAGE_ENGINE *eng = localhost->db[tier].eng;
+            if (!eng) continue;
+
+            size_t max = storage_engine_disk_space_max(eng->backend, localhost->db[tier].instance);
+            size_t used = storage_engine_disk_space_used(eng->backend, localhost->db[tier].instance);
+            NETDATA_DOUBLE percent;
+            if (used && max)
+                percent = (NETDATA_DOUBLE) used * 100.0 / (NETDATA_DOUBLE) max;
+            else
+                percent = 0.0;
+
+            buffer_json_add_array_item_object(wb);
+            buffer_json_member_add_uint64(wb, "tier", tier);
+            buffer_json_member_add_uint64(wb, "disk_used", used);
+            buffer_json_member_add_uint64(wb, "disk_max", max);
+            buffer_json_member_add_double(wb, "disk_percent", percent);
+            buffer_json_object_close(wb);
+        }
+        buffer_json_array_close(wb); // db_size
+    }
+
     if(timings)
         buffer_json_query_timings(wb, "timings", timings);
 
@@ -416,6 +686,9 @@ void buffer_json_agents_array_v2(BUFFER *wb, struct query_timings *timings, time
 }
 
 void buffer_json_cloud_timings(BUFFER *wb, const char *key, struct query_timings *timings) {
+    if(!timings->finished_ut)
+        timings->finished_ut = now_monotonic_usec();
+
     buffer_json_member_add_object(wb, key);
     buffer_json_member_add_double(wb, "routing_ms", 0.0);
     buffer_json_member_add_double(wb, "node_max_ms", 0.0);
@@ -423,6 +696,48 @@ void buffer_json_cloud_timings(BUFFER *wb, const char *key, struct query_timings
     buffer_json_object_close(wb);
 }
 
+bool contexts_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *data __maybe_unused) {
+    struct rrdcontext_to_json_v2_entry *o = old_value;
+    struct rrdcontext_to_json_v2_entry *n = new_value;
+
+    o->count++;
+
+    if(o->family != n->family) {
+        STRING *m = string_2way_merge(o->family, n->family);
+        string_freez(o->family);
+        o->family = m;
+    }
+
+    if(o->priority != n->priority) {
+        if((o->flags & RRD_FLAG_COLLECTED) && !(n->flags & RRD_FLAG_COLLECTED))
+            // keep o
+            ;
+        else if(!(o->flags & RRD_FLAG_COLLECTED) && (n->flags & RRD_FLAG_COLLECTED))
+            // keep n
+            o->priority = n->priority;
+        else
+            // keep the min
+            o->priority = MIN(o->priority, n->priority);
+    }
+
+    if(o->first_time_s && n->first_time_s)
+        o->first_time_s = MIN(o->first_time_s, n->first_time_s);
+    else if(!o->first_time_s)
+        o->first_time_s = n->first_time_s;
+
+    if(o->last_time_s && n->last_time_s)
+        o->last_time_s = MAX(o->last_time_s, n->last_time_s);
+    else if(!o->last_time_s)
+        o->last_time_s = n->last_time_s;
+
+    o->flags |= n->flags;
+    o->match = MIN(o->match, n->match);
+
+    string_freez(n->family);
+
+    return true;
+}
+
 void contexts_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) {
     struct rrdcontext_to_json_v2_entry *z = value;
     string_freez(z->family);
@@ -434,6 +749,12 @@ int rrdcontext_to_json_v2(BUFFER *wb, struct api_v2_contexts_request *req, CONTE
     if(options & CONTEXTS_V2_SEARCH)
         options |= CONTEXTS_V2_CONTEXTS;
 
+    if(options & (CONTEXTS_V2_NODES_INFO | CONTEXTS_V2_NODES_INSTANCES))
+        options |= CONTEXTS_V2_NODES;
+
+    if(options & (CONTEXTS_V2_AGENTS_INFO))
+        options |= CONTEXTS_V2_AGENTS;
+
     struct rrdcontext_to_json_v2_data ctl = {
             .wb = wb,
             .request = req,
@@ -445,6 +766,12 @@ int rrdcontext_to_json_v2(BUFFER *wb, struct api_v2_contexts_request *req, CONTE
             .contexts.pattern = string_to_simple_pattern(req->contexts),
             .contexts.scope_pattern = string_to_simple_pattern(req->scope_contexts),
             .q.pattern = string_to_simple_pattern_nocase(req->q),
+            .window = {
+                    .enabled = false,
+                    .relative = false,
+                    .after = req->after,
+                    .before = req->before,
+            },
             .timings = {
                     .received_ut = now_monotonic_usec(),
             }
@@ -455,11 +782,20 @@ int rrdcontext_to_json_v2(BUFFER *wb, struct api_v2_contexts_request *req, CONTE
                 DICT_OPTION_SINGLE_THREADED | DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE, NULL,
                 sizeof(struct rrdcontext_to_json_v2_entry));
 
+        dictionary_register_conflict_callback(ctl.ctx, contexts_conflict_callback, NULL);
         dictionary_register_delete_callback(ctl.ctx, contexts_delete_callback, NULL);
     }
 
-    time_t now_s = now_realtime_sec();
-    buffer_json_initialize(wb, "\"", "\"", 0, true, false);
+    if(req->after || req->before) {
+        ctl.window.relative = rrdr_relative_window_to_absolute(&ctl.window.after, &ctl.window.before, &ctl.now);
+        ctl.window.enabled = true;
+    }
+    else
+        ctl.now = now_realtime_sec();
+
+    buffer_json_initialize(wb, "\"", "\"", 0,
+                           true, (options & CONTEXTS_V2_MINIFY) && !(options & CONTEXTS_V2_DEBUG));
+
     buffer_json_member_add_uint64(wb, "api", 2);
 
     if(options & CONTEXTS_V2_DEBUG) {
@@ -475,7 +811,12 @@ int rrdcontext_to_json_v2(BUFFER *wb, struct api_v2_contexts_request *req, CONTE
         buffer_json_member_add_string(wb, "contexts", req->contexts);
         buffer_json_object_close(wb);
 
+        buffer_json_member_add_object(wb, "filters");
         buffer_json_member_add_string(wb, "q", req->q);
+        buffer_json_member_add_time_t(wb, "after", req->after);
+        buffer_json_member_add_time_t(wb, "before", req->before);
+        buffer_json_object_close(wb);
+
         buffer_json_member_add_arr