Merging upstream version 1.35.0.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

5 files changed, 298 insertions, 49 deletions

diff --git a/streaming/receiver.c b/streaming/receiver.c
index b083766d..d20658e6 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -30,6 +30,8 @@ void destroy_receiver_state(struct receiver_state *rpt) {
 }
 
 static void rrdpush_receiver_thread_cleanup(void *ptr) {
+    worker_unregister();
+
     static __thread int executed = 0;
     if(!executed) {
         executed = 1;
@@ -338,26 +340,31 @@ static char *receiver_next_line(struct receiver_state *r, int *pos) {
     return NULL;
 }
 
+static void streaming_parser_thread_cleanup(void *ptr) {
+    PARSER *parser = (PARSER *)ptr;
+    parser_destroy(parser);
+}
+
 size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp) {
     size_t result;
-    PARSER_USER_OBJECT *user = callocz(1, sizeof(*user));
-    user->enabled = cd->enabled;
-    user->host = rpt->host;
-    user->opaque = rpt;
-    user->cd = cd;
-    user->trust_durations = 0;
-
-    PARSER *parser = parser_init(rpt->host, user, fp, PARSER_INPUT_SPLIT);
+
+    PARSER_USER_OBJECT user = {
+        .enabled = cd->enabled,
+        .host = rpt->host,
+        .opaque = rpt,
+        .cd = cd,
+        .trust_durations = 1
+    };
+
+    PARSER *parser = parser_init(rpt->host, &user, fp, PARSER_INPUT_SPLIT);
+
+    // this keeps the parser with its current value
+    // so, parser needs to be allocated before pushing it
+    netdata_thread_cleanup_push(streaming_parser_thread_cleanup, parser);
+
     parser_add_keyword(parser, "TIMESTAMP", streaming_timestamp);
     parser_add_keyword(parser, "CLAIMED_ID", streaming_claimed_id);
 
-    if (unlikely(!parser)) {
-        error("Failed to initialize parser");
-        cd->serial_failures++;
-        freez(user);
-        return 0;
-    }
-
     parser->plugins_action->begin_action     = &pluginsd_begin_action;
     parser->plugins_action->flush_action     = &pluginsd_flush_action;
     parser->plugins_action->end_action       = &pluginsd_end_action;
@@ -371,12 +378,13 @@ size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp
     parser->plugins_action->clabel_commit_action  = &pluginsd_clabel_commit_action;
     parser->plugins_action->clabel_action    = &pluginsd_clabel_action;
 
-    user->parser = parser;
+    user.parser = parser;
 
 #ifdef ENABLE_COMPRESSION
     if (rpt->decompressor)
         rpt->decompressor->reset(rpt->decompressor);
 #endif
+
     do{
         if (receiver_read(rpt, fp))
             break;
@@ -389,10 +397,13 @@ size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp
         rpt->last_msg_t = now_realtime_sec();
     }
     while(!netdata_exit);
+
 done:
-    result= user->count;
-    freez(user);
-    parser_destroy(parser);
+    result = user.count;
+
+    // free parser with the pop function
+    netdata_thread_cleanup_pop(1);
+
     return result;
 }
 
@@ -455,9 +466,23 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
     if (strcmp(rpt->machine_guid, localhost->machine_guid) == 0) {
         log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->machine_guid, rpt->hostname, "DENIED - ATTEMPT TO RECEIVE METRICS FROM MACHINE_GUID IDENTICAL TO PARENT");
-        error("STREAM %s [receive from %s:%s]: denied to receive metrics, machine GUID [%s] is my own. Did you copy the parent/proxy machine GUID to a child?", rpt->hostname, rpt->client_ip, rpt->client_port, rpt->machine_guid);
+        error("STREAM %s [receive from %s:%s]: denied to receive metrics, machine GUID [%s] is my own. Did you copy the parent/proxy machine GUID to a child, or is this an inter-agent loop?", rpt->hostname, rpt->client_ip, rpt->client_port, rpt->machine_guid);
+        char initial_response[HTTP_HEADER_SIZE + 1];
+        snprintfz(initial_response, HTTP_HEADER_SIZE, "%s", START_STREAMING_ERROR_SAME_LOCALHOST);
+#ifdef ENABLE_HTTPS
+        rpt->host->stream_ssl.conn = rpt->ssl.conn;
+        rpt->host->stream_ssl.flags = rpt->ssl.flags;
+        if(send_timeout(&rpt->ssl, rpt->fd, initial_response, strlen(initial_response), 0, 60) != (ssize_t)strlen(initial_response)) {
+#else
+        if(send_timeout(rpt->fd, initial_response, strlen(initial_response), 0, 60) != strlen(initial_response)) {
+#endif
+            log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rpt->host->hostname, "FAILED - CANNOT REPLY");
+            error("STREAM %s [receive from [%s]:%s]: cannot send command.", rpt->host->hostname, rpt->client_ip, rpt->client_port);
+            close(rpt->fd);
+            return 0;
+        }
         close(rpt->fd);
-        return 1;
+        return 0;
     }
 
     if (rpt->host==NULL) {
@@ -609,6 +634,12 @@ static int rrdpush_receive(struct receiver_state *rpt)
     if(sock_delnonblock(rpt->fd) < 0)
         error("STREAM %s [receive from [%s]:%s]: cannot remove the non-blocking flag from socket %d", rpt->host->hostname, rpt->client_ip, rpt->client_port, rpt->fd);
 
+    struct timeval timeout;
+    timeout.tv_sec = 120;
+    timeout.tv_usec = 0;
+    if (unlikely(setsockopt(rpt->fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout) != 0))
+        error("STREAM %s [receive from [%s]:%s]: cannot set timeout for socket %d", rpt->host->hostname, rpt->client_ip, rpt->client_port, rpt->fd);
+
     // convert the socket to a FILE *
     FILE *fp = fdopen(rpt->fd, "r");
     if(!fp) {
@@ -640,6 +671,9 @@ static int rrdpush_receive(struct receiver_state *rpt)
                 rpt->host->hostname);
         }
     }
+    rpt->host->senders_connect_time = now_realtime_sec();
+    rpt->host->senders_last_chart_command = 0;
+    rpt->host->trigger_chart_obsoletion_check = 1;
     rrdhost_unlock(rpt->host);
 
     // call the plugins.d processor to receive the metrics
@@ -648,9 +682,9 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
     cd.version = rpt->stream_version;
 
-#if defined(ENABLE_ACLK)
+#if defined(ENABLE_NEW_CLOUD_PROTOCOL)
     // in case we have cloud connection we inform cloud
-    // new slave connected
+    // new child connected
     if (netdata_cloud_setting)
         aclk_host_state_update(rpt->host, 1);
 #endif
@@ -662,9 +696,9 @@ static int rrdpush_receive(struct receiver_state *rpt)
     error("STREAM %s [receive from [%s]:%s]: disconnected (completed %zu updates).", rpt->hostname, rpt->client_ip,
           rpt->client_port, count);
 
-#if defined(ENABLE_ACLK)
+#if defined(ENABLE_NEW_CLOUD_PROTOCOL)
     // in case we have cloud connection we inform cloud
-    // new slave connected
+    // new child connected
     if (netdata_cloud_setting)
         aclk_host_state_update(rpt->host, 0);
 #endif
@@ -675,6 +709,8 @@ static int rrdpush_receive(struct receiver_state *rpt)
         rrdhost_wrlock(rpt->host);
         netdata_mutex_lock(&rpt->host->receiver_lock);
         if (rpt->host->receiver == rpt) {
+            rpt->host->senders_connect_time = 0;
+            rpt->host->trigger_chart_obsoletion_check = 0;
             rpt->host->senders_disconnected_time = now_realtime_sec();
             rrdhost_flag_set(rpt->host, RRDHOST_FLAG_ORPHAN);
             if(health_enabled == CONFIG_BOOLEAN_AUTO)
@@ -699,7 +735,9 @@ void *rrdpush_receiver_thread(void *ptr) {
     struct receiver_state *rpt = (struct receiver_state *)ptr;
     info("STREAM %s [%s]:%s: receive thread created (task id %d)", rpt->hostname, rpt->client_ip, rpt->client_port, gettid());
 
+    worker_register("STREAMRCV");
     rrdpush_receive(rpt);
+    worker_unregister();
 
     netdata_thread_cleanup_pop(1);
     return NULL;
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index 8829d1ee..77774d8d 100644
--- a/streaming/rrdpush.c
+++ b/streaming/rrdpush.c
@@ -136,11 +136,7 @@ static inline int should_send_chart_matching(RRDSET *st) {
     if (rrdset_flag_check(st, RRDSET_FLAG_ANOMALY_DETECTION))
         return ml_streaming_enabled();
 
-    if(unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED))) {
-        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_SEND);
-        rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_IGNORE);
-    }
-    else if(!rrdset_flag_check(st, RRDSET_FLAG_UPSTREAM_SEND|RRDSET_FLAG_UPSTREAM_IGNORE)) {
+    if(!rrdset_flag_check(st, RRDSET_FLAG_UPSTREAM_SEND|RRDSET_FLAG_UPSTREAM_IGNORE)) {
         RRDHOST *host = st->rrdhost;
 
         if(simple_pattern_matches(host->rrdpush_send_charts_matching, st->id) ||
@@ -421,6 +417,86 @@ void rrdpush_claimed_id(RRDHOST *host)
         error("STREAM %s [send]: cannot write to internal pipe", host->hostname);
 }
 
+int connect_to_one_of_destinations(
+    struct rrdpush_destinations *destinations,
+    int default_port,
+    struct timeval *timeout,
+    size_t *reconnects_counter,
+    char *connected_to,
+    size_t connected_to_size,
+    struct rrdpush_destinations **destination)
+{
+    int sock = -1;
+
+    for (struct rrdpush_destinations *d = destinations; d; d = d->next) {
+        if (d->disabled_no_proper_reply) {
+            d->disabled_no_proper_reply = 0;
+            continue;
+        } else if (d->disabled_because_of_localhost) {
+            continue;
+        } else if (d->disabled_already_streaming && (d->disabled_already_streaming + 30 > now_realtime_sec())) {
+            continue;
+        } else if (d->disabled_because_of_denied_access) {
+            d->disabled_because_of_denied_access = 0;
+            continue;
+        }
+
+        if (reconnects_counter)
+            *reconnects_counter += 1;
+        sock = connect_to_this(d->destination, default_port, timeout);
+        if (sock != -1) {
+            if (connected_to && connected_to_size) {
+                strncpy(connected_to, d->destination, connected_to_size);
+                connected_to[connected_to_size - 1] = '\0';
+            }
+            *destination = d;
+            break;
+        }
+    }
+
+    return sock;
+}
+
+struct rrdpush_destinations *destinations_init(const char *dests) {
+    const char *s = dests;
+    struct rrdpush_destinations *destinations = NULL, *prev = NULL;
+    while(*s) {
+        const char *e = s;
+
+        // skip path, moving both s(tart) and e(nd)
+        if(*e == '/')
+            while(!isspace(*e) && *e != ',') s = ++e;
+
+        // skip separators, moving both s(tart) and e(nd)
+        while(isspace(*e) || *e == ',') s = ++e;
+
+        // move e(nd) to the first separator
+        while(*e && !isspace(*e) && *e != ',' && *e != '/') e++;
+
+        // is there anything?
+        if(!*s || s == e) break;
+
+        char buf[e - s + 1];
+        strncpyz(buf, s, e - s);
+        struct rrdpush_destinations *d = callocz(1, sizeof(struct rrdpush_destinations));
+        strncpyz(d->destination, buf, sizeof(d->destination)-1);
+        d->disabled_no_proper_reply = 0;
+        d->disabled_because_of_localhost = 0;
+        d->disabled_already_streaming = 0;
+        d->disabled_because_of_denied_access = 0;
+        d->next = NULL;
+        if (!destinations) {
+            destinations = d;
+        } else {
+            prev->next = d;
+        }
+        prev = d;
+
+        s = e;
+    }
+    return destinations;
+}
+
 // ----------------------------------------------------------------------------
 // rrdpush sender thread
 
@@ -539,6 +615,8 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
             system_info->ml_capable = strtoul(value, NULL, 0);
         else if(!strcmp(name, "ml_enabled"))
             system_info->ml_enabled = strtoul(value, NULL, 0);
+        else if(!strcmp(name, "mc_version"))
+            system_info->mc_version = strtoul(value, NULL, 0);
         else if(!strcmp(name, "tags"))
             tags = value;
         else if(!strcmp(name, "ver"))
diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index 7eb2c6e5..6efe8cd6 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -26,6 +26,10 @@
 #define START_STREAMING_PROMPT_V2  "Hit me baby, push them over and bring the host labels..."
 #define START_STREAMING_PROMPT_VN "Hit me baby, push them over with the version="
 
+#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 HTTP_HEADER_SIZE 8192
 
 typedef enum {
@@ -138,6 +142,14 @@ struct receiver_state {
 #endif
 };
 
+struct rrdpush_destinations {
+    char destination[CONNECTED_TO_SIZE + 1];
+    int disabled_no_proper_reply;
+    int disabled_because_of_localhost;
+    time_t disabled_already_streaming;
+    int disabled_because_of_denied_access;
+    struct rrdpush_destinations *next;
+};
 
 extern unsigned int default_rrdpush_enabled;
 #ifdef ENABLE_COMPRESSION
@@ -149,6 +161,7 @@ extern char *default_rrdpush_send_charts_matching;
 extern unsigned int remote_clock_resync_iterations;
 
 extern void sender_init(struct sender_state *s, RRDHOST *parent);
+extern struct rrdpush_destinations *destinations_init(const char *destinations);
 void sender_start(struct sender_state *s);
 void sender_commit(struct sender_state *s);
 extern int rrdpush_init();
@@ -164,6 +177,14 @@ extern void rrdpush_sender_thread_stop(RRDHOST *host);
 
 extern void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, RRDVAR *rv);
 extern void log_stream_connection(const char *client_ip, const char *client_port, const char *api_key, const char *machine_guid, const char *host, const char *msg);
+extern int connect_to_one_of_destinations(
+    struct rrdpush_destinations *destinations,
+    int default_port,
+    struct timeval *timeout,
+    size_t *reconnects_counter,
+    char *connected_to,
+    size_t connected_to_size,
+    struct rrdpush_destinations **destination);
 
 #ifdef ENABLE_COMPRESSION
 struct compressor_state *create_compressor();
diff --git a/streaming/sender.c b/streaming/sender.c
index 72259c3a..a95cc867 100644
--- a/streaming/sender.c
+++ b/streaming/sender.c
@@ -2,6 +2,26 @@
 
 #include "rrdpush.h"
 
+#define WORKER_SENDER_JOB_CONNECT                    0
+#define WORKER_SENDER_JOB_PIPE_READ                  1
+#define WORKER_SENDER_JOB_SOCKET_RECEIVE             2
+#define WORKER_SENDER_JOB_EXECUTE                    3
+#define WORKER_SENDER_JOB_SOCKET_SEND                4
+#define WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE   5
+#define WORKER_SENDER_JOB_DISCONNECT_OVERFLOW        6
+#define WORKER_SENDER_JOB_DISCONNECT_TIMEOUT         7
+#define WORKER_SENDER_JOB_DISCONNECT_POLL_ERROR      8
+#define WORKER_SENDER_JOB_DISCONNECT_SOCKER_ERROR    9
+#define WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR      10
+#define WORKER_SENDER_JOB_DISCONNECT_PARENT_CLOSED  11
+#define WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR  12
+#define WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR     13
+#define WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION 14
+
+#if WORKER_UTILIZATION_MAX_JOB_TYPES < 15
+#error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 15
+#endif
+
 extern struct config stream_config;
 extern int netdata_use_ssl_on_stream;
 extern char *netdata_ssl_ca_path;
@@ -21,8 +41,8 @@ static inline void rrdpush_sender_thread_close_socket(RRDHOST *host);
 * Inform the user through the error log file and 
 * deactivate compression by downgrading the stream protocol.
 */
-static inline void deactivate_compression(struct sender_state *s)
-{
+static inline void deactivate_compression(struct sender_state *s) {
+    worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION);
     error("STREAM_COMPRESSION: Deactivating compression to avoid stream corruption");
     default_compression_enabled = 0;
     s->rrdpush_compression = 0;
@@ -183,6 +203,18 @@ void rrdpush_clean_encoded(stream_encoded_t *se)
         freez(se->kernel_version);
 }
 
+static inline long int parse_stream_version_for_errors(char *http)
+{
+    if (!memcmp(http, START_STREAMING_ERROR_SAME_LOCALHOST, sizeof(START_STREAMING_ERROR_SAME_LOCALHOST)))
+        return -2;
+    else if (!memcmp(http, START_STREAMING_ERROR_ALREADY_STREAMING, sizeof(START_STREAMING_ERROR_ALREADY_STREAMING)))
+        return -3;
+    else if (!memcmp(http, START_STREAMING_ERROR_NOT_PERMITTED, sizeof(START_STREAMING_ERROR_NOT_PERMITTED)))
+        return -4;
+    else
+        return -1;
+}
+
 static inline long int parse_stream_version(RRDHOST *host, char *http)
 {
     long int stream_version = -1;
@@ -207,6 +239,9 @@ static inline long int parse_stream_version(RRDHOST *host, char *http)
                 host->labels.labels_flag |= LABEL_FLAG_STOP_STREAM;
                 host->labels.labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
             }
+            else {
+                stream_version = parse_stream_version_for_errors(http);
+            }
         }
     }
     return stream_version;
@@ -226,13 +261,14 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po
     debug(D_STREAM, "STREAM: Attempting to connect...");
     info("STREAM %s [send to %s]: connecting...", host->hostname, host->rrdpush_send_destination);
 
-    host->rrdpush_sender_socket = connect_to_one_of(
-            host->rrdpush_send_destination
+    host->rrdpush_sender_socket = connect_to_one_of_destinations(
+            host->destinations
             , default_port
             , &tv
             , &s->reconnects_counter
             , s->connected_to
             , sizeof(s->connected_to)-1
+            , &host->destination
     );
 
     if(unlikely(host->rrdpush_sender_socket == -1)) {
@@ -299,6 +335,7 @@ if(!s->rrdpush_compression)
                  "&hops=%d"
                  "&ml_capable=%d"
                  "&ml_enabled=%d"
+                 "&mc_version=%d"
                  "&tags=%s"
                  "&ver=%d"
                  "&NETDATA_INSTANCE_CLOUD_TYPE=%s"
@@ -344,6 +381,7 @@ if(!s->rrdpush_compression)
                  , host->system_info->hops + 1
                  , host->system_info->ml_capable
                  , host->system_info->ml_enabled
+                 , host->system_info->mc_version
                  , (host->tags) ? host->tags : ""
                  , s->version
                  , (host->system_info->cloud_provider_type) ? host->system_info->cloud_provider_type : ""
@@ -389,7 +427,10 @@ if(!s->rrdpush_compression)
             err = SSL_get_error(host->ssl.conn, err);
             error("SSL cannot connect with the server:  %s ",ERR_error_string((long)SSL_get_error(host->ssl.conn,err),NULL));
             if (netdata_use_ssl_on_stream == NETDATA_SSL_FORCE) {
+                worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
                 rrdpush_sender_thread_close_socket(host);
+                if (host->destination->next)
+                    host->destination->disabled_no_proper_reply = 1;
                 return 0;
             }else {
                 host->ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
@@ -399,8 +440,11 @@ if(!s->rrdpush_compression)
             if (netdata_use_ssl_on_stream == NETDATA_SSL_FORCE) {
                 if (netdata_validate_server == NETDATA_SSL_VALID_CERTIFICATE) {
                     if ( security_test_certificate(host->ssl.conn)) {
+                        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
                         error("Closing the stream connection, because the server SSL certificate is not valid.");
                         rrdpush_sender_thread_close_socket(host);
+                        if (host->destination->next)
+                            host->destination->disabled_no_proper_reply = 1;
                         return 0;
                     }
                 }
@@ -411,6 +455,7 @@ if(!s->rrdpush_compression)
 #else
     if(send_timeout(host->rrdpush_sender_socket, http, strlen(http), 0, timeout) == -1) {
 #endif
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT);
         error("STREAM %s [send to %s]: failed to send HTTP header to remote netdata.", host->hostname, s->connected_to);
         rrdpush_sender_thread_close_socket(host);
         return 0;
@@ -426,6 +471,7 @@ if(!s->rrdpush_compression)
     received = recv_timeout(host->rrdpush_sender_socket, http, HTTP_HEADER_SIZE, 0, timeout);
     if(received == -1) {
 #endif
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT);
         error("STREAM %s [send to %s]: remote netdata does not respond.", host->hostname, s->connected_to);
         rrdpush_sender_thread_close_socket(host);
         return 0;
@@ -435,8 +481,31 @@ if(!s->rrdpush_compression)
     debug(D_STREAM, "Response to sender from far end: %s", http);
     int32_t version = (int32_t)parse_stream_version(host, http);
     if(version == -1) {
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE);
         error("STREAM %s [send to %s]: server is not replying properly (is it a netdata?).", host->hostname, s->connected_to);
         rrdpush_sender_thread_close_socket(host);
+        //catch other reject reasons and force to check other destinations
+        if (host->destination->next)
+            host->destination->disabled_no_proper_reply = 1;
+        return 0;
+    }
+    else if(version == -2) {
+        error("STREAM %s [send to %s]: remote server is the localhost for [%s].", host->hostname, s->connected_to, host->hostname);
+        rrdpush_sender_thread_close_socket(host);
+        host->destination->disabled_because_of_localhost = 1;
+        return 0;
+    }
+    else if(version == -3) {
+        error("STREAM %s [send to %s]: remote server already receives metrics for [%s].", host->hostname, s->connected_to, host->hostname);
+        rrdpush_sender_thread_close_socket(host);
+        host->destination->disabled_already_streaming = now_realtime_sec();
+        return 0;
+    }
+    else if(version == -4) {
+        error("STREAM %s [send to %s]: remote server denied access for [%s].", host->hostname, s->connected_to, host->hostname);
+        rrdpush_sender_thread_close_socket(host);
+        if (host->destination->next)
+            host->destination->disabled_because_of_denied_access = 1;
         return 0;
     }
     s->version = version;
@@ -541,9 +610,9 @@ void attempt_to_send(struct sender_state *s) {
         s->last_sent_t = now_monotonic_sec();
     }
     else if (ret == -1 && (errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK))
-        debug(D_STREAM, "STREAM %s [send to %s]: unavailable after polling POLLOUT", s->host->hostname,
-              s->connected_to);
+        debug(D_STREAM, "STREAM %s [send to %s]: unavailable after polling POLLOUT", s->host->hostname, s->connected_to);
     else if (ret == -1) {
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR);
         debug(D_STREAM, "STREAM: Send failed - closing socket...");
         error("STREAM %s [send to %s]: failed to send metrics - closing connection - we have sent %zu bytes on this connection.",  s->host->hostname, s->connected_to, s->sent_bytes_on_this_connection);
         rrdpush_sender_thread_close_socket(s->host);
@@ -570,6 +639,8 @@ int ret;
         int sslerrno = SSL_get_error(s->host->ssl.conn, desired);
         if (sslerrno == SSL_ERROR_WANT_READ || sslerrno == SSL_ERROR_WANT_WRITE)
             return;
+
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
         u_long err;
         char buf[256];
         while ((err = ERR_get_error()) != 0) {
@@ -581,20 +652,25 @@ int ret;
         return;
     }
 #endif
-    ret = recv(s->host->rrdpush_sender_socket, s->read_buffer + s->read_len, sizeof(s->read_buffer) - s->read_len - 1,
-               MSG_DONTWAIT);
+    ret = recv(s->host->rrdpush_sender_socket, s->read_buffer + s->read_len, sizeof(s->read_buffer) - s->read_len - 1,MSG_DONTWAIT);
     if (ret>0) {
         s->read_len += ret;
         return;
     }
+
     debug(D_STREAM, "Socket was POLLIN, but req %zu bytes gave %d", sizeof(s->read_buffer) - s->read_len - 1, ret);
+
     if (ret<0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
         return;
-    if (ret==0)
+
+    if (ret==0) {
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_PARENT_CLOSED);
         error("STREAM %s [send to %s]: connection closed by far end. Restarting connection", s->host->hostname, s->connected_to);
-    else
-        error("STREAM %s [send to %s]: error during read (%d). Restarting connection", s->host->hostname, s->connected_to,
-              ret);
+    }
+    else {
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR);
+        error("STREAM %s [send to %s]: error during receive (%d). Restarting connection", s->host->hostname, s->connected_to, ret);
+    }
     rrdpush_sender_thread_close_socket(s->host);
 }
 
@@ -615,6 +691,8 @@ void execute_commands(struct sender_state *s) {
 
 
 static void rrdpush_sender_thread_cleanup_callback(void *ptr) {
+    worker_unregister();
+
     RRDHOST *host = (RRDHOST *)ptr;
 
     netdata_mutex_lock(&host->sender->mutex);
@@ -683,7 +761,7 @@ void *rrdpush_sender_thread(void *ptr) {
     s->timeout = (int)appconfig_get_number(&stream_config, CONFIG_SECTION_STREAM, "timeout seconds", 60);
     s->default_port = (int)appconfig_get_number(&stream_config, CONFIG_SECTION_STREAM, "default port", 19999);
     s->buffer->max_size =
-        (size_t)appconfig_get_number(&stream_config, CONFIG_SECTION_STREAM, "buffer size bytes", 1024 * 1024);
+        (size_t)appconfig_get_number(&stream_config, CONFIG_SECTION_STREAM, "buffer size bytes", 1024 * 1024 * 10);
     s->reconnect_delay =
         (unsigned int)appconfig_get_number(&stream_config, CONFIG_SECTION_STREAM, "reconnect delay seconds", 5);
     remote_clock_resync_iterations = (unsigned int)appconfig_get_number(
@@ -707,6 +785,25 @@ void *rrdpush_sender_thread(void *ptr) {
     fds[Collector].fd = s->host->rrdpush_sender_pipe[PIPE_READ];
     fds[Collector].events = POLLIN;
 
+    worker_register("STREAMSND");
+    worker_register_job_name(WORKER_SENDER_JOB_CONNECT, "connect");
+    worker_register_job_name(WORKER_SENDER_JOB_PIPE_READ, "pipe read");
+    worker_register_job_name(WORKER_SENDER_JOB_SOCKET_RECEIVE, "receive");
+    worker_register_job_name(WORKER_SENDER_JOB_EXECUTE, "execute");
+    worker_register_job_name(WORKER_SENDER_JOB_SOCKET_SEND, "send");
+
+    // disconnection reasons
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT, "disconnect timeout");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_POLL_ERROR, "disconnect poll error");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_SOCKER_ERROR, "disconnect socket error");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_OVERFLOW, "disconnect overflow");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR, "disconnect ssl error");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_PARENT_CLOSED, "disconnect parent closed");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR, "disconnect receive error");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR, "disconnect send error");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION, "disconnect no compression");
+    worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE, "disconnect bad handshake");
+
     netdata_thread_cleanup_push(rrdpush_sender_thread_cleanup_callback, s->host);
     for(; s->host->rrdpush_send_enabled && !netdata_exit ;) {
         // check for outstanding cancellation requests
@@ -714,6 +811,7 @@ void *rrdpush_sender_thread(void *ptr) {
 
         // The connection attempt blocks (after which we use the socket in nonblocking)
         if(unlikely(s->host->rrdpush_sender_socket == -1)) {
+            worker_is_busy(WORKER_SENDER_JOB_CONNECT);
             s->overflow = 0;
             s->read_len = 0;
             s->buffer->read = 0;
@@ -731,11 +829,14 @@ void *rrdpush_sender_thread(void *ptr) {
 
         // If the TCP window never opened then something is wrong, restart connection
         if(unlikely(now_monotonic_sec() - s->last_sent_t > s->timeout)) {
+            worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT);
             error("STREAM %s [send to %s]: could not send metrics for %d seconds - closing connection - we have sent %zu bytes on this connection via %zu send attempts.", s->host->hostname, s->connected_to, s->timeout, s->sent_bytes_on_this_connection, s->send_attempts);
             rrdpush_sender_thread_close_socket(s->host);
             continue;
         }
 
+        worker_is_idle();
+
         // Wait until buffer opens in the socket or a rrdset_done_push wakes us
         fds[Collector].revents = 0;
         fds[Socket].revents = 0;
@@ -757,16 +858,18 @@ void *rrdpush_sender_thread(void *ptr) {
         int retval = poll(fds, 2, 1000);
         debug(D_STREAM, "STREAM: poll() finished collector=%d socket=%d (current chunk %zu bytes)...",
               fds[Collector].revents, fds[Socket].revents, outstanding);
+
         if(unlikely(netdata_exit)) break;
 
         // Spurious wake-ups without error - loop again
-        if (retval == 0 || ((retval == -1) && (errno == EAGAIN || errno == EINTR)))
-        {
+        if (retval == 0 || ((retval == -1) && (errno == EAGAIN || errno == EINTR))) {
             debug(D_STREAM, "Spurious wakeup");
             continue;
         }
+
         // Only errors from poll() are internal, but try restarting the connection
         if(unlikely(retval == -1)) {
+            worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_POLL_ERROR);
             error("STREAM %s [send to %s]: failed to poll(). Closing socket.", s->host->hostname, s->connected_to);
             rrdpush_sender_thread_close_socket(s->host);
             continue;
@@ -774,6 +877,7 @@ void *rrdpush_sender_thread(void *ptr) {
 
         // If the collector woke us up then empty the pipe to remove the signal
         if (fds[Collector].revents & POLLIN || fds[Collector].revents & POLLPRI) {
+            worker_is_busy(WORKER_SENDER_JOB_PIPE_READ);
             debug(D_STREAM, "STREAM: Data added to send buffer (current buffer chunk %zu bytes)...", outstanding);
 
             char buffer[1000 + 1];
@@ -782,13 +886,19 @@ void *rrdpush_sender_thread(void *ptr) {
         }
 
         // Read as much as possible to fill the buffer, split into full lines for execution.
-        if (fds[Socket].revents & POLLIN)
+        if (fds[Socket].revents & POLLIN) {
+            worker_is_busy(WORKER_SENDER_JOB_SOCKET_RECEIVE);
             attempt_read(s);
+        }
+
+        worker_is_busy(WORKER_SENDER_JOB_EXECUTE);
         execute_commands(s);
 
         // If we have data and have seen the TCP window open then try to close it by a transmission.
-        if (outstanding && fds[Socket].revents & POLLOUT)
+        if (outstanding && fds[Socket].revents & POLLOUT) {
+            worker_is_busy(WORKER_SENDER_JOB_SOCKET_SEND);
             attempt_to_send(s);
+        }
 
         // TODO-GAPS - why do we only check this on the socket, not the pipe?
         if (outstanding) {
@@ -800,6 +910,7 @@ void *rrdpush_sender_thread(void *ptr) {
             else if (unlikely(fds[Socket].revents & POLLNVAL))
                 error = "connection is invalid (POLLNVAL)";
             if(unlikely(error)) {
+                worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SOCKER_ERROR);
                 error("STREAM %s [send to %s]: restart stream because %s - %zu bytes transmitted.", s->host->hostname,
                       s->connected_to, error, s->sent_bytes_on_this_connection);
                 rrdpush_sender_thread_close_socket(s->host);
@@ -808,6 +919,7 @@ void *rrdpush_sender_thread(void *ptr) {
 
         // protection from overflow
         if (s->overflow) {
+            worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_OVERFLOW);
             errno = 0;
             error("STREAM %s [send to %s]: buffer full (%zu-bytes) after %zu bytes. Restarting connection",
                   s->host->hostname, s->connected_to, s->buffer->size, s->sent_bytes_on_this_connection);
diff --git a/streaming/stream.conf b/streaming/stream.conf
index e65e76fa..33172bbc 100644
--- a/streaming/stream.conf
+++ b/streaming/stream.conf
@@ -82,9 +82,9 @@
     send charts matching = *
 
     # The buffer to use for sending metrics.
-    # 1MB is good for 10-20 seconds of data, so increase this if you expect latencies.
+    # 10MB is good for 60 seconds of data, so increase this if you expect latencies.
     # The buffer is flushed on reconnects (this will not prevent gaps at the charts).
-    buffer size bytes = 1048576
+    buffer size bytes = 10485760
 
     # If the connection fails, or it disconnects,
     # retry after that many seconds.