api v2 nodes for streaming statuses (#15162)

* api v2 nodes for streaming statuses

* remove test

* move parts of the output

* in api/v2/data return 5 values per point when aggregation=percentage and raw option is given; return final values when aggregation=percentage is not the final grouping

5 files changed, 285 insertions, 31 deletions

diff --git a/streaming/receiver.c b/streaming/receiver.c
index 2b72486fd9..a89fdf04a3 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -423,6 +423,58 @@ static void rrdpush_receiver_replication_reset(RRDHOST *host) {
     rrdhost_receiver_replicating_charts_zero(host);
 }
 
+void rrdhost_receiver_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused) {
+    size_t receiver_hops = host->system_info ? host->system_info->hops : (host == localhost) ? 0 : 1;
+
+    netdata_mutex_lock(&host->receiver_lock);
+
+    buffer_json_member_add_object(wb, key);
+    buffer_json_member_add_uint64(wb, "hops", receiver_hops);
+
+    bool online = host == localhost || !rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN | RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
+    buffer_json_member_add_boolean(wb, "online", online);
+
+    if(host->child_connect_time || host->child_disconnected_time) {
+        time_t since = MAX(host->child_connect_time, host->child_disconnected_time);
+        buffer_json_member_add_time_t(wb, "since", since);
+        buffer_json_member_add_time_t(wb, "age", now - since);
+    }
+
+    if(!online && host->rrdpush_last_receiver_exit_reason)
+        buffer_json_member_add_string(wb, "reason", host->rrdpush_last_receiver_exit_reason);
+
+    if(host != localhost && host->receiver) {
+        buffer_json_member_add_object(wb, "replication");
+        {
+            size_t instances = rrdhost_receiver_replicating_charts(host);
+            buffer_json_member_add_boolean(wb, "in_progress", instances);
+            buffer_json_member_add_double(wb, "completion", host->rrdpush_receiver_replication_percent);
+            buffer_json_member_add_uint64(wb, "instances", instances);
+        }
+        buffer_json_object_close(wb); // replication
+
+        buffer_json_member_add_object(wb, "source");
+        {
+
+            char buf[1024 + 1];
+            SOCKET_PEERS peers = socket_peers(host->receiver->fd);
+            bool ssl = SSL_connection(&host->receiver->ssl);
+
+            snprintfz(buf, 1024, "[%s]:%d%s", peers.local.ip, peers.local.port, ssl ? ":SSL" : "");
+            buffer_json_member_add_string(wb, "local", buf);
+
+            snprintfz(buf, 1024, "[%s]:%d%s", peers.peer.ip, peers.peer.port, ssl ? ":SSL" : "");
+            buffer_json_member_add_string(wb, "remote", buf);
+
+            stream_capabilities_to_json_array(wb, host->receiver->capabilities, "capabilities");
+        }
+        buffer_json_object_close(wb); // source
+    }
+    buffer_json_object_close(wb); // collection
+
+    netdata_mutex_unlock(&host->receiver_lock);
+}
+
 static bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
     bool signal_rrdcontext = false;
     bool set_this = false;
@@ -496,6 +548,7 @@ static void rrdhost_clear_receiver(struct receiver_state *rpt) {
 
             rrdhost_flag_set(host, RRDHOST_FLAG_ORPHAN);
             host->receiver = NULL;
+            host->rrdpush_last_receiver_exit_reason = rpt->exit.reason;
         }
 
         netdata_mutex_unlock(&host->receiver_lock);
@@ -546,6 +599,18 @@ bool stop_streaming_receiver(RRDHOST *host, const char *reason) {
     return ret;
 }
 
+static void rrdpush_send_error_on_taken_over_connection(struct receiver_state *rpt, const char *msg) {
+    send_timeout(
+#ifdef ENABLE_HTTPS
+            &rpt->ssl,
+#endif
+            rpt->fd,
+            (char *)msg,
+            strlen(msg),
+            0,
+            5);
+}
+
 void rrdpush_receive_log_status(struct receiver_state *rpt, const char *msg, const char *status) {
 
     log_stream_connection(rpt->client_ip, rpt->client_port,
@@ -677,11 +742,13 @@ static void rrdpush_receive(struct receiver_state *rpt)
 
         if(!host) {
             rrdpush_receive_log_status(rpt, "failed to find/create host structure", "INTERNAL ERROR DROPPING CONNECTION");
+            rrdpush_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INTERNAL_ERROR);
             goto cleanup;
         }
 
         if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) {
             rrdpush_receive_log_status(rpt, "host is initializing", "INITIALIZATION IN PROGRESS RETRY LATER");
+            rrdpush_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INITIALIZATION);
             goto cleanup;
         }
 
@@ -690,6 +757,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
 
         if(!rrdhost_set_receiver(host, rpt)) {
             rrdpush_receive_log_status(rpt, "host is already served by another receiver", "DUPLICATE RECEIVER DROPPING CONNECTION");
+            rrdpush_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_ALREADY_STREAMING);
             goto cleanup;
         }
     }
diff --git a/streaming/replication.c b/streaming/replication.c
index a50913a1a4..c6fafc357b 100644
--- a/streaming/replication.c
+++ b/streaming/replication.c
@@ -274,6 +274,12 @@ static void replication_query_finalize(BUFFER *wb, struct replication_query *q,
         replication_queries.queries_finished += queries;
         replication_queries.points_read += q->points_read;
         replication_queries.points_generated += q->points_generated;
+
+        if(q->st && q->st->rrdhost->sender) {
+            struct sender_state *s = q->st->rrdhost->sender;
+            s->replication.latest_completed_before_t = q->query.before;
+        }
+
         netdata_spinlock_unlock(&replication_queries.spinlock);
     }
 
@@ -644,7 +650,7 @@ bool replication_response_execute_and_finalize(struct replication_query *q, size
     buffer_fast_strcat(wb, "\n", 1);
 
     worker_is_busy(WORKER_JOB_BUFFER_COMMIT);
-    sender_commit(host->sender, wb);
+    sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_REPLICATION);
     worker_is_busy(WORKER_JOB_CLEANUP);
 
     if(enable_streaming) {
@@ -1466,6 +1472,9 @@ void replication_add_request(struct sender_state *sender, const char *chart_id,
             .not_indexed_preprocessing = false,
     };
 
+    if(!sender->replication.oldest_request_after_t || rq.after < sender->replication.oldest_request_after_t)
+        sender->replication.oldest_request_after_t = rq.after;
+
     if(start_streaming && rrdpush_sender_get_buffer_used_percent(sender) <= STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)
         replication_execute_request(&rq, false);
 
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index a113fbae05..c481871cc7 100644
--- a/streaming/rrdpush.c
+++ b/streaming/rrdpush.c
@@ -375,7 +375,7 @@ bool rrdset_push_chart_definition_now(RRDSET *st) {
 
     BUFFER *wb = sender_start(host->sender);
     rrdpush_send_chart_definition(wb, st);
-    sender_commit(host->sender, wb);
+    sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
     sender_thread_buffer_free();
 
     return true;
@@ -443,7 +443,7 @@ void rrdset_push_metrics_finished(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
         buffer_fast_strcat(rsb->wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
     }
 
-    sender_commit(st->rrdhost->sender, rsb->wb);
+    sender_commit(st->rrdhost->sender, rsb->wb, STREAM_TRAFFIC_TYPE_DATA);
 
     *rsb = (RRDSET_STREAM_BUFFER){ .wb = NULL, };
 }
@@ -483,7 +483,7 @@ RRDSET_STREAM_BUFFER rrdset_push_metric_initialize(RRDSET *st, time_t wall_clock
     if(unlikely(!exposed_upstream)) {
         BUFFER *wb = sender_start(host->sender);
         replication_in_progress = rrdpush_send_chart_definition(wb, st);
-        sender_commit(host->sender, wb);
+        sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
     }
 
     if(replication_in_progress)
@@ -514,7 +514,7 @@ void rrdpush_send_host_labels(RRDHOST *host) {
     rrdlabels_walkthrough_read(host->rrdlabels, send_labels_callback, wb);
     buffer_sprintf(wb, "OVERWRITE %s\n", "labels");
 
-    sender_commit(host->sender, wb);
+    sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
 
     sender_thread_buffer_free();
 }
@@ -533,7 +533,7 @@ void rrdpush_claimed_id(RRDHOST *host)
     buffer_sprintf(wb, "CLAIMED_ID %s %s\n", host->machine_guid, (host->aclk_state.claimed_id ? host->aclk_state.claimed_id : "NULL") );
 
     rrdhost_aclk_state_unlock(host);
-    sender_commit(host->sender, wb);
+    sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
 
     sender_thread_buffer_free();
 }
@@ -706,7 +706,7 @@ int rrdpush_receiver_permission_denied(struct web_client *w) {
     // we always respond with the same message and error code
     // to prevent an attacker from gaining info about the error
     buffer_flush(w->response.data);
-    buffer_sprintf(w->response.data, "You are not permitted to access this. Check the logs for more info.");
+    buffer_strcat(w->response.data, START_STREAMING_ERROR_NOT_PERMITTED);
     return HTTP_RESP_UNAUTHORIZED;
 }
 
@@ -714,7 +714,7 @@ int rrdpush_receiver_too_busy_now(struct web_client *w) {
     // we always respond with the same message and error code
     // to prevent an attacker from gaining info about the error
     buffer_flush(w->response.data);
-    buffer_sprintf(w->response.data, "The server is too busy now to accept this request. Try later.");
+    buffer_strcat(w->response.data, START_STREAMING_ERROR_BUSY_TRY_LATER);
     return HTTP_RESP_SERVICE_UNAVAILABLE;
 }
 
@@ -1146,7 +1146,7 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
 
             // Have not set WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET - caller should clean up
             buffer_flush(w->response.data);
-            buffer_strcat(w->response.data, "This GUID is already streaming to this server");
+            buffer_strcat(w->response.data, START_STREAMING_ERROR_ALREADY_STREAMING);
             receiver_state_free(rpt);
             return HTTP_RESP_CONFLICT;
         }
@@ -1191,15 +1191,18 @@ static struct {
     { STREAM_HANDSHAKE_OK_V3, "OK_V3" },
     { STREAM_HANDSHAKE_OK_V2, "OK_V2" },
     { STREAM_HANDSHAKE_OK_V1, "OK_V1" },
-    { STREAM_HANDSHAKE_ERROR_BAD_HANDSHAKE, "ERROR_BAD_HANDSHAKE" },
-    { STREAM_HANDSHAKE_ERROR_LOCALHOST, "ERROR_LOCALHOST" },
-    { STREAM_HANDSHAKE_ERROR_ALREADY_CONNECTED, "ERROR_ALREADY_CONNECTED" },
-    { STREAM_HANDSHAKE_ERROR_DENIED, "ERROR_DENIED" },
-    { STREAM_HANDSHAKE_ERROR_SEND_TIMEOUT, "ERROR_SEND_TIMEOUT" },
-    { STREAM_HANDSHAKE_ERROR_RECEIVE_TIMEOUT, "ERROR_RECEIVE_TIMEOUT" },
-    { STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE, "ERROR_INVALID_CERTIFICATE" },
-    { STREAM_HANDSHAKE_ERROR_SSL_ERROR, "ERROR_SSL_ERROR" },
-    { STREAM_HANDSHAKE_ERROR_CANT_CONNECT, "ERROR_CANT_CONNECT" },
+    { STREAM_HANDSHAKE_ERROR_BAD_HANDSHAKE, "BAD HANDSHAKE" },
+    { STREAM_HANDSHAKE_ERROR_LOCALHOST, "LOCALHOST" },
+    { STREAM_HANDSHAKE_ERROR_ALREADY_CONNECTED, "ALREADY CONNECTED" },
+    { STREAM_HANDSHAKE_ERROR_DENIED, "DENIED" },
+    { STREAM_HANDSHAKE_ERROR_SEND_TIMEOUT, "SEND TIMEOUT" },
+    { STREAM_HANDSHAKE_ERROR_RECEIVE_TIMEOUT, "RECEIVE TIMEOUT" },
+    { STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE, "INVALID CERTIFICATE" },
+    { STREAM_HANDSHAKE_ERROR_SSL_ERROR, "SSL ERROR" },
+    { STREAM_HANDSHAKE_ERROR_CANT_CONNECT, "CANT CONNECT" },
+    { STREAM_HANDSHAKE_BUSY_TRY_LATER, "BUSY TRY LATER" },
+    { STREAM_HANDSHAKE_INTERNAL_ERROR, "INTERNAL ERROR" },
+    { STREAM_HANDSHAKE_INITIALIZATION, "INITIALIZING" },
     { 0, NULL },
 };
 
diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index cb33694cda..f97c8ddfb3 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -72,6 +72,9 @@ STREAM_CAPABILITIES stream_our_capabilities();
 #define START_STREAMING_ERROR_SAME_LOCALHOST "Don't hit me baby, you are trying to stream my localhost back"
 #define START_STREAMING_ERROR_ALREADY_STREAMING "This GUID is already streaming to this server"
 #define START_STREAMING_ERROR_NOT_PERMITTED "You are not permitted to access this. Check the logs for more info."
+#define START_STREAMING_ERROR_BUSY_TRY_LATER "The server is too busy now to accept this request. Try later."
+#define START_STREAMING_ERROR_INTERNAL_ERROR "The server encountered an internal error. Try later."
+#define START_STREAMING_ERROR_INITIALIZATION "The server is initializing. Try later."
 
 typedef enum {
     STREAM_HANDSHAKE_OK_V5 = 5, // COMPRESSION
@@ -87,12 +90,27 @@ typedef enum {
     STREAM_HANDSHAKE_ERROR_RECEIVE_TIMEOUT = -6,
     STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE = -7,
     STREAM_HANDSHAKE_ERROR_SSL_ERROR = -8,
-    STREAM_HANDSHAKE_ERROR_CANT_CONNECT = -9
+    STREAM_HANDSHAKE_ERROR_CANT_CONNECT = -9,
+    STREAM_HANDSHAKE_BUSY_TRY_LATER = -10,
+    STREAM_HANDSHAKE_INTERNAL_ERROR = -11,
+    STREAM_HANDSHAKE_INITIALIZATION = -12,
 } STREAM_HANDSHAKE;
 
 
 // ----------------------------------------------------------------------------
 
+typedef enum __attribute__((packed)) {
+    STREAM_TRAFFIC_TYPE_REPLICATION,
+    STREAM_TRAFFIC_TYPE_FUNCTIONS,
+    STREAM_TRAFFIC_TYPE_METADATA,
+    STREAM_TRAFFIC_TYPE_DATA,
+
+    // terminator
+    STREAM_TRAFFIC_TYPE_MAX,
+} STREAM_TRAFFIC_TYPE;
+
+// ----------------------------------------------------------------------------
+
 typedef struct {
     char *os_name;
     char *os_id;
@@ -148,6 +166,7 @@ struct sender_state {
     size_t sent_bytes_on_this_connection;
     size_t send_attempts;
     time_t last_traffic_seen_t;
+    time_t last_state_since_t;              // the timestamp of the last state (online/offline) change
     size_t not_connected_loops;
     // Metrics are collected asynchronously by collector threads calling rrdset_done_push(). This can also trigger
     // the lazy creation of the sender thread - both cases (buffer access and thread creation) are guarded here.
@@ -157,6 +176,8 @@ struct sender_state {
     int read_len;
     STREAM_CAPABILITIES capabilities;
 
+    size_t sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_MAX];
+
     int rrdpush_sender_pipe[2];                     // collector to sender thread signaling
     int rrdpush_sender_socket;
 
@@ -176,6 +197,8 @@ struct sender_state {
 
     struct {
         DICTIONARY *requests;                   // de-duplication of replication requests, per chart
+        time_t oldest_request_after_t;          // the timestamp of the oldest replication request
+        time_t latest_completed_before_t;       // the timestamp of the latest replication request
 
         struct {
             size_t pending_requests;            // the currently outstanding replication requests
@@ -306,7 +329,7 @@ void rrdpush_destinations_init(RRDHOST *host);
 void rrdpush_destinations_free(RRDHOST *host);
 
 BUFFER *sender_start(struct sender_state *s);
-void sender_commit(struct sender_state *s, BUFFER *wb);
+void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type);
 int rrdpush_init();
 bool rrdpush_receiver_needs_dbengine();
 int configured_as_parent();
@@ -368,6 +391,9 @@ bool stop_streaming_receiver(RRDHOST *host, const char *reason);
 
 void sender_thread_buffer_free(void);
 
+void rrdhost_receiver_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused);
+void rrdhost_sender_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused);
+
 #include "replication.h"
 
 #endif //NETDATA_RRDPUSH_H
diff --git a/streaming/sender.c b/streaming/sender.c
index cedf58ce2d..b207039b5e 100644
--- a/streaming/sender.c
+++ b/streaming/sender.c
@@ -84,7 +84,7 @@ static inline void deactivate_compression(struct sender_state *s) {
 #define SENDER_BUFFER_ADAPT_TO_TIMES_MAX_SIZE 3
 
 // Collector thread finishing a transmission
-void sender_commit(struct sender_state *s, BUFFER *wb) {
+void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type) {
 
     if(unlikely(wb != sender_thread_buffer))
         fatal("STREAMING: sender is trying to commit a buffer that is not this thread's buffer.");
@@ -163,6 +163,8 @@ void sender_commit(struct sender_state *s, BUFFER *wb) {
 
             if(cbuffer_add_unsafe(s->buffer, dst, dst_len))
                 s->flags |= SENDER_FLAG_OVERFLOW;
+            else
+                s->sent_bytes_on_this_connection_per_type[type] += dst_len;
 
             src = src + size_to_compress;
             src_len -= size_to_compress;
@@ -170,9 +172,13 @@ void sender_commit(struct sender_state *s, BUFFER *wb) {
     }
     else if(cbuffer_add_unsafe(s->buffer, src, src_len))
         s->flags |= SENDER_FLAG_OVERFLOW;
+    else
+        s->sent_bytes_on_this_connection_per_type[type] += src_len;
 #else
     if(cbuffer_add_unsafe(s->buffer, src, src_len))
         s->flags |= SENDER_FLAG_OVERFLOW;
+    else
+        s->sent_bytes_on_this_connection_per_type[type] += src_len;
 #endif
 
     replication_recalculate_buffer_used_ratio_unsafe(s);
@@ -204,7 +210,7 @@ void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, const RRDVAR_ACQU
     if(rrdhost_can_send_definitions_to_parent(host)) {
         BUFFER *wb = sender_start(host->sender);
         rrdpush_sender_add_host_variable_to_buffer(wb, rva);
-        sender_commit(host->sender, wb);
+        sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
         sender_thread_buffer_free();
     }
 }
@@ -233,7 +239,7 @@ static void rrdpush_sender_thread_send_custom_host_variables(RRDHOST *host) {
         };
         int ret = rrdvar_walkthrough_read(host->rrdvars, rrdpush_sender_thread_custom_host_variables_callback, &tmp);
         (void)ret;
-        sender_commit(host->sender, wb);
+        sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
         sender_thread_buffer_free();
 
         debug(D_STREAM, "RRDVAR sent %d VARIABLES", ret);
@@ -426,6 +432,33 @@ struct {
         .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE,
         .postpone_reconnect_seconds = 1 * 60, // 1 minute
     },
+    {
+        .response = START_STREAMING_ERROR_BUSY_TRY_LATER,
+        .length = sizeof(START_STREAMING_ERROR_BUSY_TRY_LATER) - 1,
+        .version = STREAM_HANDSHAKE_BUSY_TRY_LATER,
+        .dynamic = false,
+        .error = "remote server is currently busy, we should try later",
+        .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE,
+        .postpone_reconnect_seconds = 2 * 60, // 2 minutes
+    },
+    {
+        .response = START_STREAMING_ERROR_INTERNAL_ERROR,
+        .length = sizeof(START_STREAMING_ERROR_INTERNAL_ERROR) - 1,
+        .version = STREAM_HANDSHAKE_INTERNAL_ERROR,
+        .dynamic = false,
+        .error = "remote server is encountered an internal error, we should try later",
+        .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE,
+        .postpone_reconnect_seconds = 5 * 60, // 5 minutes
+    },
+    {
+        .response = START_STREAMING_ERROR_INITIALIZATION,
+        .length = sizeof(START_STREAMING_ERROR_INITIALIZATION) - 1,
+        .version = STREAM_HANDSHAKE_INITIALIZATION,
+        .dynamic = false,
+        .error = "remote server is initializing, we should try later",
+        .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE,
+        .postpone_reconnect_seconds = 2 * 60, // 2 minute
+    },
 
     // terminator
     {
@@ -750,6 +783,10 @@ static bool attempt_to_connect(struct sender_state *state)
 {
     state->send_attempts = 0;
 
+    // reset the bytes we have sent for this session
+    state->sent_bytes_on_this_connection = 0;
+    memset(state->sent_bytes_on_this_connection_per_type, 0, sizeof(state->sent_bytes_on_this_connection_per_type));
+
     if(rrdpush_sender_thread_connect_to_parent(state->host, state->default_port, state->timeout, state)) {
         // reset the buffer, to properly send charts and metrics
         rrdpush_sender_on_connect(state->host);
@@ -760,9 +797,6 @@ static bool attempt_to_connect(struct sender_state *state)
         // make sure the next reconnection will be immediate
         state->not_connected_loops = 0;
 
-        // reset the bytes we have sent for this session
-        state->sent_bytes_on_this_connection = 0;
-
         // let the data collection threads know we are ready
         rrdhost_flag_set(state->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
 
@@ -776,9 +810,6 @@ static bool attempt_to_connect(struct sender_state *state)
     // increase the failed connections counter
     state->not_connected_loops++;
 
-    // reset the number of bytes sent
-    state->sent_bytes_on_this_connection = 0;
-
     // slow re-connection on repeating errors
     usec_t now_ut = now_monotonic_usec();
     usec_t end_ut = now_ut + USEC_PER_SEC * state->reconnect_delay;
@@ -899,7 +930,7 @@ void stream_execute_function_callback(BUFFER *func_wb, int code, void *data) {
         buffer_fast_strcat(wb, buffer_tostring(func_wb), buffer_strlen(func_wb));
         pluginsd_function_result_end_to_buffer(wb);
 
-        sender_commit(s, wb);
+        sender_commit(s, wb, STREAM_TRAFFIC_TYPE_FUNCTIONS);
         sender_thread_buffer_free();
 
         internal_error(true, "STREAM %s [send to %s] FUNCTION transaction %s sending back response (%zu bytes, %llu usec).",
@@ -1067,6 +1098,120 @@ void rrdpush_signal_sender_to_wake_up(struct sender_state *s) {
     }
 }
 
+static NETDATA_DOUBLE rrdhost_sender_replication_completion(RRDHOST *host, time_t now, size_t *instances) {
+    size_t charts = rrdhost_sender_replicating_charts(host);
+    NETDATA_DOUBLE completion;
+    if(!charts || !host->sender->replication.oldest_request_after_t)
+        completion = 100.0;
+    else if(!host->sender->replication.latest_completed_before_t || host->sender->replication.latest_completed_before_t < host->sender->replication.oldest_request_after_t)
+        completion = 0.0;
+    else {
+        time_t total = now - host->sender->replication.oldest_request_after_t;
+        time_t current = host->sender->replication.latest_completed_before_t - host->sender->replication.oldest_request_after_t;
+        completion = (NETDATA_DOUBLE) current * 100.0 / (NETDATA_DOUBLE) total;
+    }
+
+    *instances = charts;
+
+    return completion;
+}
+
+void rrdhost_sender_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused) {
+    bool online = rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
+    buffer_json_member_add_object(wb, key);
+
+    if(host->sender)
+        buffer_json_member_add_uint64(wb, "hops", host->sender->hops);
+
+    buffer_json_member_add_boolean(wb, "online", online);
+
+    if(host->sender && host->sender->last_state_since_t) {
+        buffer_json_member_add_time_t(wb, "since", host->sender->last_state_since_t);
+        buffer_json_member_add_time_t(wb, "age", now - host->sender->last_state_since_t);
+    }
+
+    if(!online && host->sender && host->sender->exit.reason)
+        buffer_json_member_add_string(wb, "reason", host->sender->exit.reason);
+
+    buffer_json_member_add_object(wb, "replication");
+    {
+        size_t instances;
+        NETDATA_DOUBLE completion = rrdhost_sender_replication_completion(host, now, &instances);
+        buffer_json_member_add_boolean(wb, "in_progress", instances);
+        buffer_json_member_add_double(wb, "completion", completion);
+        buffer_json_member_add_uint64(wb, "instances", instances);
+    }
+    buffer_json_object_close(wb);
+
+    if(host->sender) {
+        netdata_mutex_lock(&host->sender->mutex);
+
+        buffer_json_member_add_object(wb, "destination");
+        {
+            char buf[1024 + 1];
+            if(online && host->sender->rrdpush_sender_socket != -1) {
+                SOCKET_PEERS peers = socket_peers(host->sender->rrdpush_sender_socket);
+                bool ssl = SSL_connection(&host->sender->ssl);
+
+                snprintfz(buf, 1024, "[%s]:%d%s", peers.local.ip, peers.local.port, ssl ? ":SSL" : "");
+                buffer_json_member_add_string(wb, "local", buf);
+
+                snprintfz(buf, 1024, "[%s]:%d%s", peers.peer.ip, peers.peer.port, ssl ? ":SSL" : "");
+                buffer_json_member_add_string(wb, "remote", buf);
+
+                stream_capabilities_to_json_array(wb, online ? host->sender->capabilities : 0,
+                                                  "capabilities");
+
+                buffer_json_member_add_object(wb, "traffic");
+                {
+                    bool compression = false;
+#ifdef ENABLE_COMPRESSION
+                    compression = (stream_has_capability(host->sender, STREAM_CAP_COMPRESSION) && host->sender->compressor);
+#endif
+                    buffer_json_member_add_boolean(wb, "compression", compression);
+                    buffer_json_member_add_uint64(wb, "data", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_DATA]);
+                    buffer_json_member_add_uint64(wb, "metadata", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_METADATA]);
+                    buffer_json_member_add_uint64(wb, "functions", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_FUNCTIONS]);
+                    buffer_json_member_add_uint64(wb, "replication", host->sender->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 = 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", 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 > now) ?
+                                                  d->postpone_reconnection_until - now : 0);
+                }
+                buffer_json_object_close(wb); // each candidate
+            }
+            buffer_json_array_close(wb); // candidates
+        }
+        buffer_json_object_close(wb); // destination
+
+        netdata_mutex_unlock(&host->sender->mutex);
+    }
+
+    buffer_json_object_close(wb); // streaming
+}
+
 static bool rrdhost_set_sender(RRDHOST *host) {
     if(unlikely(!host->sender)) return false;
 
@@ -1076,6 +1221,8 @@ static bool rrdhost_set_sender(RRDHOST *host) {
         rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED | RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
         rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN);
         host->sender->tid = gettid();
+        host->sender->last_state_since_t = now_realtime_sec();
+        host->sender->exit.reason = NULL;
         ret = true;
     }
     netdata_mutex_unlock(&host->sender->mutex);
@@ -1091,8 +1238,8 @@ static void rrdhost_clear_sender___while_having_sender_mutex(RRDHOST *host) {
     if(host->sender->tid == gettid()) {
         host->sender->tid = 0;
         host->sender->exit.shutdown = false;
-        host->sender->exit.reason = NULL;
         rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN | RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED | RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
+        host->sender->last_state_since_t = now_realtime_sec();
     }
 
     rrdpush_reset_destinations_postpone_time(host);
@@ -1291,6 +1438,7 @@ void *rrdpush_sender_thread(void *ptr) {
             now_s = s->last_traffic_seen_t = now_monotonic_sec();
             rrdpush_claimed_id(s->host);
             rrdpush_send_host_labels(s->host);
+            s->replication.oldest_request_after_t = 0;
 
             rrdhost_flag_set(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
             info("STREAM %s [send to %s]: enabling metrics streaming...", rrdhost_hostname(s->host), s->connected_to);