Re-write of SSL support in Netdata; restoration of SIGCHLD; detection of stale plugins; streaming improvements (#15113)

* add information about streaming connections to /api/v2/nodes; reset defer time when sender or receivers connect or disconnect

* make each streaming destination respect its SSL settings

* to not send SSL traffic over non-SSL connection

* keep track of outgoing streaming connection attempts

* retry SSL reads when SSL_read() returns SSL_ERROR_WANT_READ

* Revert "retry SSL reads when SSL_read() returns SSL_ERROR_WANT_READ"

This reverts commit 14c858677c6f2d3b08c94f298e2f45ecdb74c801.

* cleanup SSL connections properly

* initialize SSL in rpt before takeover

* sender should free SSL when talking to a non-SSL destination

* do not shutdown SSL when receiver exits

* restore operation of SIGCHLD when the reaper is not enabled

* create an fgets function that checks for data and times out

* work on error handling of plugins exiting

* remove newlines from logs

* global call to waitid(), caching the result for netdata_pclose() to process

* receiver tid

* parser timeouts in 2 minutes instead of 10

* fix crash when UUID is NULL in SQLite

* abstract sqlite3 parsing for uuid and text

* write proper ssl errors on read and write

* fix for SSL_ERROR_WANT_RETRY_VERIFY

* SSL WANT per function

* unified SSL error logging

* fix compilation warning

* additional logging about parser cleanup

* streaming parser should call the pluginsd parser cleanup

* SSL error handling work

* SSL initialization unification

* check for pending data when receiving SSL response with timeout

* macro to check if an SSL connection has been established

* remove SSL_pending()

* check for SSL macros

* use SSL_peek() to find if there is a response

* SSL renames

* more SSL renames & cleanup

* rrdpush ssl connection function

* abstract all SSL functions into security.c

* keep track of SSL connections and always attempt to use SSL read/write when on SSL connection

* signal openssl to skip certificate validation when configured to do so

* better SSL error handling and logging

* SSL code cleanup

* SSL retry on SSL_connect and SSL_accept

* SSL provide default return value for old compilers

* SSL read/write functions emulate system read/write functions

* fix receive/send timeout and switch from SSL_peek() to SSL_pending()

* remove SSL_pending()

* removed sender auto-retry and debug info for initial recevier response

* ssl skip certificate verification config for web server

* ssl errors log ip and port of the peer

* keep ssl with web_client for its whole lifetime

* thread safe socket peers to text

* use error_limit() for common ssl errors

* cleanup

* more cleanup

* coverity fixes

* ssl error logs include both local and remote ip/port info

* remove obsolete code

4 files changed, 274 insertions, 200 deletions

diff --git a/streaming/receiver.c b/streaming/receiver.c
index ff7a956290..2b72486fd9 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -31,10 +31,14 @@ void receiver_state_free(struct receiver_state *rpt) {
     freez(rpt->program_version);
 
 #ifdef ENABLE_HTTPS
-    if(rpt->ssl.conn)
-        SSL_free(rpt->ssl.conn);
+    netdata_ssl_close(&rpt->ssl);
 #endif
 
+    if(rpt->fd != -1) {
+        internal_error(true, "closing socket...");
+        close(rpt->fd);
+    }
+
 #ifdef ENABLE_COMPRESSION
     if (rpt->decompressor)
         rpt->decompressor->destroy(&rpt->decompressor);
@@ -100,13 +104,18 @@ static int read_stream(struct receiver_state *r, char* buffer, size_t size) {
         return 0;
     }
 
+    ssize_t bytes_read;
+
 #ifdef ENABLE_HTTPS
-    if (r->ssl.conn && r->ssl.flags == NETDATA_SSL_HANDSHAKE_COMPLETE)
-        return (int)netdata_ssl_read(r->ssl.conn, buffer, size);
+    if (SSL_connection(&r->ssl))
+        bytes_read = netdata_ssl_read(&r->ssl, buffer, size);
+    else
+        bytes_read = read(r->fd, buffer, size);
+#else
+    bytes_read = read(r->fd, buffer, size);
 #endif
 
-    ssize_t bytes_read = read(r->fd, buffer, size);
-    if(bytes_read == 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS)) {
+    if((bytes_read == 0 || bytes_read == -1) && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS)) {
         error("STREAM: %s(): timeout while waiting for data on socket!", __FUNCTION__);
         bytes_read = -3;
     }
@@ -119,23 +128,6 @@ static int read_stream(struct receiver_state *r, char* buffer, size_t size) {
         bytes_read = -2;
     }
 
-//    do {
-//        bytes_read = (int) fread(buffer, 1, size, fp);
-//        if (unlikely(bytes_read <= 0)) {
-//            if(feof(fp)) {
-//                internal_error(true, "%s(): fread() failed with EOF", __FUNCTION__);
-//                bytes_read = -2;
-//            }
-//            else if(ferror(fp)) {
-//                internal_error(true, "%s(): fread() failed with ERROR", __FUNCTION__);
-//                bytes_read = -3;
-//            }
-//            else bytes_read = 0;
-//        }
-//        else
-//            worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, bytes_read);
-//    } while(bytes_read == 0);
-
     return (int)bytes_read;
 }
 
@@ -323,12 +315,6 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b
     return NULL;
 }
 
-static void streaming_parser_thread_cleanup(void *ptr) {
-    PARSER *parser = (PARSER *)ptr;
-    rrd_collector_finished();
-    parser_destroy(parser);
-}
-
 bool plugin_is_enabled(struct plugind *cd);
 
 static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
@@ -352,7 +338,7 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
 
     // 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);
+    netdata_thread_cleanup_push(pluginsd_process_thread_cleanup, parser);
 
     parser_add_keyword(parser, "CLAIMED_ID", streaming_claimed_id);
 
@@ -474,6 +460,8 @@ static bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
         rrdhost_flag_clear(rpt->host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
         aclk_queue_node_info(rpt->host, true);
 
+        rrdpush_reset_destinations_postpone_time(host);
+
         set_this = true;
     }
 
@@ -506,16 +494,16 @@ static void rrdhost_clear_receiver(struct receiver_state *rpt) {
             signal_rrdcontext = true;
             rrdpush_receiver_replication_reset(host);
 
-            if (host->receiver == rpt)
-                host->receiver = NULL;
-
             rrdhost_flag_set(host, RRDHOST_FLAG_ORPHAN);
+            host->receiver = NULL;
         }
 
         netdata_mutex_unlock(&host->receiver_lock);
 
         if(signal_rrdcontext)
             rrdcontext_host_child_disconnected(host);
+
+        rrdpush_reset_destinations_postpone_time(host);
     }
 }
 
@@ -549,7 +537,7 @@ bool stop_streaming_receiver(RRDHOST *host, const char *reason) {
               "thread %d takes too long to stop, giving up..."
         , rrdhost_hostname(host)
         , host->receiver->client_ip, host->receiver->client_port
-        , gettid());
+        , host->receiver->tid);
     else
         ret = true;
 
@@ -585,7 +573,7 @@ static void rrdhost_reset_destinations(RRDHOST *host) {
         d->postpone_reconnection_until = 0;
 }
 
-static int rrdpush_receive(struct receiver_state *rpt)
+static void rrdpush_receive(struct receiver_state *rpt)
 {
     rpt->config.mode = default_rrd_memory_mode;
     rpt->config.history = default_rrd_history_entries;
@@ -689,14 +677,12 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
         if(!host) {
             rrdpush_receive_log_status(rpt, "failed to find/create host structure", "INTERNAL ERROR DROPPING CONNECTION");
-            close(rpt->fd);
-            return 1;
+            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");
-            close(rpt->fd);
-            return 1;
+            goto cleanup;
         }
 
         // system_info has been consumed by the host structure
@@ -704,8 +690,7 @@ static int 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");
-            close(rpt->fd);
-            return 1;
+            goto cleanup;
         }
     }
 
@@ -776,15 +761,16 @@ static int rrdpush_receive(struct receiver_state *rpt)
         }
 
         debug(D_STREAM, "Initial response to %s: %s", rpt->client_ip, initial_response);
-        if(send_timeout(
+        ssize_t bytes_sent = send_timeout(
 #ifdef ENABLE_HTTPS
                 &rpt->ssl,
 #endif
-                rpt->fd, initial_response, strlen(initial_response), 0, 60) != (ssize_t)strlen(initial_response)) {
+                rpt->fd, initial_response, strlen(initial_response), 0, 60);
 
+        if(bytes_sent != (ssize_t)strlen(initial_response)) {
+            internal_error(true, "Cannot send response, got %zd bytes, expecting %zu bytes", bytes_sent, strlen(initial_response));
             rrdpush_receive_log_status(rpt, "cannot reply back", "CANT REPLY DROPPING CONNECTION");
-            close(rpt->fd);
-            return 0;
+            goto cleanup;
         }
     }
 
@@ -850,9 +836,8 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
     rrdhost_set_is_parent_label(--localhost->connected_children_count);
 
-    // cleanup
-    close(rpt->fd);
-    return (int)count;
+cleanup:
+    ;
 }
 
 static void rrdpush_receiver_thread_cleanup(void *ptr) {
@@ -879,7 +864,8 @@ void *rrdpush_receiver_thread(void *ptr) {
     worker_register_job_custom_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, "replication completion", "%", WORKER_METRIC_ABSOLUTE);
 
     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());
+    rpt->tid = gettid();
+    info("STREAM %s [%s]:%s: receive thread created (task id %d)", rpt->hostname, rpt->client_ip, rpt->client_port, rpt->tid);
 
     rrdpush_receive(rpt);
 
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index 62b537f0c5..a113fbae05 100644
--- a/streaming/rrdpush.c
+++ b/streaming/rrdpush.c
@@ -49,7 +49,6 @@ bool default_rrdpush_enable_replication = true;
 time_t default_rrdpush_seconds_to_replicate = 86400;
 time_t default_rrdpush_replication_step = 600;
 #ifdef ENABLE_HTTPS
-int netdata_use_ssl_on_stream = NETDATA_SSL_OPTIONAL;
 char *netdata_ssl_ca_path = NULL;
 char *netdata_ssl_ca_file = NULL;
 #endif
@@ -137,24 +136,10 @@ int rrdpush_init() {
     }
 
 #ifdef ENABLE_HTTPS
-    if (netdata_use_ssl_on_stream == NETDATA_SSL_OPTIONAL) {
-        if (default_rrdpush_destination){
-            char *test = strstr(default_rrdpush_destination,":SSL");
-            if(test){
-                *test = 0X00;
-                netdata_use_ssl_on_stream = NETDATA_SSL_FORCE;
-            }
-        }
-    }
+    netdata_ssl_validate_certificate_sender = !appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "ssl skip certificate verification", !netdata_ssl_validate_certificate);
 
-    bool invalid_certificate = appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "ssl skip certificate verification", CONFIG_BOOLEAN_NO);
-
-    if(invalid_certificate == CONFIG_BOOLEAN_YES){
-        if(netdata_ssl_validate_server == NETDATA_SSL_VALID_CERTIFICATE){
-            info("Netdata is configured to accept invalid SSL certificate.");
-            netdata_ssl_validate_server = NETDATA_SSL_INVALID_CERTIFICATE;
-        }
-    }
+    if(!netdata_ssl_validate_certificate_sender)
+        info("SSL: streaming senders will skip SSL certificates verification.");
 
     netdata_ssl_ca_path = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "CApath", NULL);
     netdata_ssl_ca_file = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "CAfile", NULL);
@@ -579,6 +564,7 @@ int connect_to_one_of_destinations(
         if (reconnects_counter)
             *reconnects_counter += 1;
 
+        d->last_attempt = now;
         sock = connect_to_this(string2str(d->destination), default_port, timeout);
 
         if (sock != -1) {
@@ -610,6 +596,14 @@ bool destinations_init_add_one(char *entry, void *data) {
     struct destinations_init_tmp *t = data;
 
     struct rrdpush_destinations *d = callocz(1, sizeof(struct rrdpush_destinations));
+    char *colon_ssl = strstr(entry, ":SSL");
+    if(colon_ssl) {
+        *colon_ssl = '\0';
+        d->ssl = true;
+    }
+    else
+        d->ssl = false;
+
     d->destination = string_strdupz(entry);
 
     __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_senders, sizeof(struct rrdpush_destinations), __ATOMIC_RELAXED);
@@ -724,6 +718,31 @@ int rrdpush_receiver_too_busy_now(struct web_client *w) {
     return HTTP_RESP_SERVICE_UNAVAILABLE;
 }
 
+static void rrdpush_receiver_takeover_web_connection(struct web_client *w, struct receiver_state *rpt) {
+    rpt->fd                = w->ifd;
+
+#ifdef ENABLE_HTTPS
+    rpt->ssl.conn          = w->ssl.conn;
+    rpt->ssl.state         = w->ssl.state;
+
+    w->ssl = NETDATA_SSL_UNSET_CONNECTION;
+#endif
+
+    WEB_CLIENT_IS_DEAD(w);
+
+    if(web_server_mode == WEB_SERVER_MODE_STATIC_THREADED) {
+        web_client_flag_set(w, WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET);
+    }
+    else {
+        if(w->ifd == w->ofd)
+            w->ifd = w->ofd = -1;
+        else
+            w->ifd = -1;
+    }
+
+    buffer_flush(w->response.data);
+}
+
 void *rrdpush_receiver_thread(void *ptr);
 int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_string) {
 
@@ -741,20 +760,16 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
     rpt->system_info = callocz(1, sizeof(struct rrdhost_system_info));
     rpt->system_info->hops = rpt->hops;
 
-    rpt->fd                = w->ifd;
+    rpt->fd                = -1;
     rpt->client_ip         = strdupz(w->client_ip);
     rpt->client_port       = strdupz(w->client_port);
 
-    rpt->config.update_every = default_rrd_update_every;
-
 #ifdef ENABLE_HTTPS
-    rpt->ssl.conn          = w->ssl.conn;
-    rpt->ssl.flags         = w->ssl.flags;
-
-    w->ssl.conn = NULL;
-    w->ssl.flags = NETDATA_SSL_START;
+    rpt->ssl = NETDATA_SSL_UNSET_CONNECTION;
 #endif
 
+    rpt->config.update_every = default_rrd_update_every;
+
     // parse the parameters and fill rpt and rpt->system_info
 
     while(decoded_query_string) {
@@ -1011,6 +1026,8 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
 
     if (strcmp(rpt->machine_guid, localhost->machine_guid) == 0) {
 
+        rrdpush_receiver_takeover_web_connection(w, rpt);
+
         rrdpush_receive_log_status(
                 rpt,
                 "machine GUID is my own",
@@ -1032,9 +1049,8 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
             );
         }
 
-        close(rpt->fd);
         receiver_state_free(rpt);
-        return web_client_socket_is_now_used_for_streaming(w);
+        return HTTP_RESP_OK;
     }
 
     if(unlikely(web_client_streaming_rate_t > 0)) {
@@ -1138,8 +1154,11 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
 
     debug(D_SYSTEM, "starting STREAM receive thread.");
 
-    char tag[FILENAME_MAX + 1];
-    snprintfz(tag, FILENAME_MAX, THREAD_TAG_STREAM_RECEIVER "[%s,[%s]:%s]", rpt->hostname, w->client_ip, w->client_port);
+    rrdpush_receiver_takeover_web_connection(w, rpt);
+
+    char tag[NETDATA_THREAD_TAG_MAX + 1];
+    snprintfz(tag, NETDATA_THREAD_TAG_MAX, THREAD_TAG_STREAM_RECEIVER "[%s]", rpt->hostname);
+    tag[NETDATA_THREAD_TAG_MAX] = '\0';
 
     if(netdata_thread_create(&rpt->thread, tag, NETDATA_THREAD_OPTION_DEFAULT, rrdpush_receiver_thread, (void *)rpt)) {
         rrdpush_receive_log_status(
@@ -1154,23 +1173,83 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
     }
 
     // prevent the caller from closing the streaming socket
-    return web_client_socket_is_now_used_for_streaming(w);
+    return HTTP_RESP_OK;
+}
+
+void rrdpush_reset_destinations_postpone_time(RRDHOST *host) {
+    struct rrdpush_destinations *d;
+    for (d = host->destinations; d; d = d->next)
+        d->postpone_reconnection_until = 0;
 }
 
+static struct {
+    STREAM_HANDSHAKE err;
+    const char *str;
+} handshake_errors[] = {
+    { STREAM_HANDSHAKE_OK_V5, "OK_V5" },
+    { STREAM_HANDSHAKE_OK_V4, "OK_V4" },
+    { 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" },
+    { 0, NULL },
+};
+
+const char *stream_handshake_error_to_string(STREAM_HANDSHAKE handshake_error) {
+    for(size_t i = 0; handshake_errors[i].str ; i++) {
+        if(handshake_error == handshake_errors[i].err)
+            return handshake_errors[i].str;
+    }
+
+    return "";
+}
+
+static struct {
+    STREAM_CAPABILITIES cap;
+    const char *str;
+} capability_names[] = {
+    { STREAM_CAP_V1, "V1" },
+    { STREAM_CAP_V2, "V2" },
+    { STREAM_CAP_VN, "VN" },
+    { STREAM_CAP_VCAPS, "VCAPS" },
+    { STREAM_CAP_HLABELS, "HLABELS" },
+    { STREAM_CAP_CLAIM, "CLAIM" },
+    { STREAM_CAP_CLABELS, "CLABELS" },
+    { STREAM_CAP_COMPRESSION, "COMPRESSION" },
+    { STREAM_CAP_FUNCTIONS, "FUNCTIONS" },
+    { STREAM_CAP_REPLICATION, "REPLICATION" },
+    { STREAM_CAP_BINARY, "BINARY" },
+    { STREAM_CAP_INTERPOLATED, "INTERPOLATED" },
+    { STREAM_CAP_IEEE754, "IEEE754" },
+    { 0 , NULL },
+};
+
 static void stream_capabilities_to_string(BUFFER *wb, STREAM_CAPABILITIES caps) {
-    if(caps & STREAM_CAP_V1) buffer_strcat(wb, "V1 ");
-    if(caps & STREAM_CAP_V2) buffer_strcat(wb, "V2 ");
-    if(caps & STREAM_CAP_VN) buffer_strcat(wb, "VN ");
-    if(caps & STREAM_CAP_VCAPS) buffer_strcat(wb, "VCAPS ");
-    if(caps & STREAM_CAP_HLABELS) buffer_strcat(wb, "HLABELS ");
-    if(caps & STREAM_CAP_CLAIM) buffer_strcat(wb, "CLAIM ");
-    if(caps & STREAM_CAP_CLABELS) buffer_strcat(wb, "CLABELS ");
-    if(caps & STREAM_CAP_COMPRESSION) buffer_strcat(wb, "COMPRESSION ");
-    if(caps & STREAM_CAP_FUNCTIONS) buffer_strcat(wb, "FUNCTIONS ");
-    if(caps & STREAM_CAP_REPLICATION) buffer_strcat(wb, "REPLICATION ");
-    if(caps & STREAM_CAP_BINARY) buffer_strcat(wb, "BINARY ");
-    if(caps & STREAM_CAP_INTERPOLATED) buffer_strcat(wb, "INTERPOLATED ");
-    if(caps & STREAM_CAP_IEEE754) buffer_strcat(wb, "IEEE754 ");
+    for(size_t i = 0; capability_names[i].str ; i++) {
+        if(caps & capability_names[i].cap) {
+            buffer_strcat(wb, capability_names[i].str);
+            buffer_strcat(wb, " ");
+        }
+    }
+}
+
+void stream_capabilities_to_json_array(BUFFER *wb, STREAM_CAPABILITIES caps, const char *key) {
+    buffer_json_member_add_array(wb, key);
+
+    for(size_t i = 0; capability_names[i].str ; i++) {
+        if(caps & capability_names[i].cap)
+            buffer_json_add_array_item_string(wb, capability_names[i].str);
+    }
+
+    buffer_json_array_close(wb);
 }
 
 void log_receiver_capabilities(struct receiver_state *rpt) {
diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index ff89584403..cb33694cda 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -166,7 +166,7 @@ struct sender_state {
     struct compressor_state *compressor;
 #endif
 #ifdef ENABLE_HTTPS
-    struct netdata_ssl ssl;                     // structure used to encrypt the connection
+    NETDATA_SSL ssl;                     // structure used to encrypt the connection
 #endif
 
     struct {
@@ -221,6 +221,7 @@ struct sender_state {
 
 struct receiver_state {
     RRDHOST *host;
+    pid_t tid;
     netdata_thread_t thread;
     int fd;
     char *key;
@@ -266,7 +267,7 @@ struct receiver_state {
     } config;
 
 #ifdef ENABLE_HTTPS
-    struct netdata_ssl ssl;
+    NETDATA_SSL ssl;
 #endif
 #ifdef ENABLE_COMPRESSION
     unsigned int rrdpush_compression;
@@ -278,8 +279,10 @@ struct receiver_state {
 
 struct rrdpush_destinations {
     STRING *destination;
+    bool ssl;
 
     const char *last_error;
+    time_t last_attempt;
     time_t postpone_reconnection_until;
     STREAM_HANDSHAKE last_handshake;
 
@@ -351,7 +354,9 @@ void rrdpush_signal_sender_to_wake_up(struct sender_state *s);
 struct compressor_state *create_compressor();
 struct decompressor_state *create_decompressor();
 #endif
-
+void rrdpush_reset_destinations_postpone_time(RRDHOST *host);
+const char *stream_handshake_error_to_string(STREAM_HANDSHAKE handshake_error);
+void stream_capabilities_to_json_array(BUFFER *wb, STREAM_CAPABILITIES caps, const char *key);
 void rrdpush_receive_log_status(struct receiver_state *rpt, const char *msg, const char *status);
 void log_receiver_capabilities(struct receiver_state *rpt);
 void log_sender_capabilities(struct sender_state *s);
diff --git a/streaming/sender.c b/streaming/sender.c
index 179c2dc603..cedf58ce2d 100644
--- a/streaming/sender.c
+++ b/streaming/sender.c
@@ -29,7 +29,6 @@
 #endif
 
 extern struct config stream_config;
-extern int netdata_use_ssl_on_stream;
 extern char *netdata_ssl_ca_path;
 extern char *netdata_ssl_ca_file;
 
@@ -320,6 +319,10 @@ static void rrdpush_sender_after_connect(RRDHOST *host) {
 }
 
 static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
+#ifdef ENABLE_HTTPS
+    netdata_ssl_close(&host->sender->ssl);
+#endif
+
     if(host->sender->rrdpush_sender_socket != -1) {
         close(host->sender->rrdpush_sender_socket);
         host->sender->rrdpush_sender_socket = -1;
@@ -480,6 +483,53 @@ static inline bool rrdpush_sender_validate_response(RRDHOST *host, struct sender
     return false;
 }
 
+static bool rrdpush_sender_connect_ssl(struct sender_state *s) {
+#ifdef ENABLE_HTTPS
+    RRDHOST *host = s->host;
+    bool ssl_required = host->destination && host->destination->ssl;
+
+    netdata_ssl_close(&host->sender->ssl);
+
+    if(!ssl_required)
+        return true;
+
+    if (netdata_ssl_open(&host->sender->ssl, netdata_ssl_streaming_sender_ctx, s->rrdpush_sender_socket)) {
+        if(!netdata_ssl_connect(&host->sender->ssl)) {
+            // couldn't connect
+
+            worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
+            rrdpush_sender_thread_close_socket(host);
+            host->destination->last_error = "SSL error";
+            host->destination->last_handshake = STREAM_HANDSHAKE_ERROR_SSL_ERROR;
+            host->destination->postpone_reconnection_until = now_realtime_sec() + 5 * 60;
+            return false;
+        }
+
+        if (netdata_ssl_validate_certificate_sender &&
+            security_test_certificate(host->sender->ssl.conn)) {
+            // certificate is not valid
+
+            worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
+            error("SSL: closing the stream connection, because the server SSL certificate is not valid.");
+            rrdpush_sender_thread_close_socket(host);
+            host->destination->last_error = "invalid SSL certificate";
+            host->destination->last_handshake = STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE;
+            host->destination->postpone_reconnection_until = now_realtime_sec() + 5 * 60;
+            return false;
+        }
+
+        return true;
+    }
+
+    // failed to establish connection
+    return false;
+
+#else
+    // SSL is not enabled
+    return true;
+#endif
+}
+
 static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_port, int timeout, struct sender_state *s) {
 
     struct timeval tv = {
@@ -507,35 +557,6 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
 
     // info("STREAM %s [send to %s]: initializing communication...", rrdhost_hostname(host), s->connected_to);
 
-#ifdef ENABLE_HTTPS
-    if(netdata_ssl_client_ctx){
-        host->sender->ssl.flags = NETDATA_SSL_START;
-        if (!host->sender->ssl.conn){
-            host->sender->ssl.conn = SSL_new(netdata_ssl_client_ctx);
-            if(!host->sender->ssl.conn){
-                error("Failed to allocate SSL structure.");
-                host->sender->ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
-            }
-        }
-        else{
-            SSL_clear(host->sender->ssl.conn);
-        }
-
-        if (host->sender->ssl.conn)
-        {
-            if (SSL_set_fd(host->sender->ssl.conn, s->rrdpush_sender_socket) != 1) {
-                error("Failed to set the socket to the SSL on socket fd %d.", s->rrdpush_sender_socket);
-                host->sender->ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
-            } else{
-                host->sender->ssl.flags = NETDATA_SSL_HANDSHAKE_COMPLETE;
-            }
-        }
-    }
-    else {
-        host->sender->ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
-    }
-#endif
-
     // reset our capabilities to default
     s->capabilities = stream_our_capabilities();
 
@@ -651,43 +672,8 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
     http[eol] = 0x00;
     rrdpush_clean_encoded(&se);
 
-#ifdef ENABLE_HTTPS
-    if (!host->sender->ssl.flags) {
-        ERR_clear_error();
-        SSL_set_connect_state(host->sender->ssl.conn);
-        int err = SSL_connect(host->sender->ssl.conn);
-        if (err != 1){
-            err = SSL_get_error(host->sender->ssl.conn, err);
-            error("SSL cannot connect with the server:  %s ",ERR_error_string((long)SSL_get_error(host->sender->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);
-                host->destination->last_error = "SSL error";
-                host->destination->last_handshake = STREAM_HANDSHAKE_ERROR_SSL_ERROR;
-                host->destination->postpone_reconnection_until = now_realtime_sec() + 5 * 60;
-                return false;
-            }
-            else {
-                host->sender->ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
-            }
-        }
-        else {
-            if (netdata_use_ssl_on_stream == NETDATA_SSL_FORCE) {
-                if (netdata_ssl_validate_server == NETDATA_SSL_VALID_CERTIFICATE) {
-                    if ( security_test_certificate(host->sender->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);
-                        host->destination->last_error = "invalid SSL certificate";
-                        host->destination->last_handshake = STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE;
-                        host->destination->postpone_reconnection_until = now_realtime_sec() + 5 * 60;
-                        return false;
-                    }
-                }
-            }
-        }
-    }
-#endif
+    if(!rrdpush_sender_connect_ssl(s))
+        return false;
 
     ssize_t bytes;
 
@@ -733,6 +719,12 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
         return false;
     }
 
+    if(sock_setnonblock(s->rrdpush_sender_socket) < 0)
+        error("STREAM %s [send to %s]: cannot set non-blocking mode for socket.", rrdhost_hostname(host), s->connected_to);
+
+    if(sock_enlarge_out(s->rrdpush_sender_socket) < 0)
+        error("STREAM %s [send to %s]: cannot enlarge the socket buffer.", rrdhost_hostname(host), s->connected_to);
+
     http[bytes] = '\0';
     debug(D_STREAM, "Response to sender from far end: %s", http);
     if(!rrdpush_sender_validate_response(host, s, http, bytes))
@@ -749,12 +741,6 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
 
     log_sender_capabilities(s);
 
-    if(sock_setnonblock(s->rrdpush_sender_socket) < 0)
-        error("STREAM %s [send to %s]: cannot set non-blocking mode for socket.", rrdhost_hostname(host), s->connected_to);
-
-    if(sock_enlarge_out(s->rrdpush_sender_socket) < 0)
-        error("STREAM %s [send to %s]: cannot enlarge the socket buffer.", rrdhost_hostname(host), s->connected_to);
-
     debug(D_STREAM, "STREAM: Connected on fd %d...", s->rrdpush_sender_socket);
 
     return true;
@@ -819,9 +805,8 @@ static ssize_t attempt_to_send(struct sender_state *s) {
     debug(D_STREAM, "STREAM: Sending data. Buffer r=%zu w=%zu s=%zu, next chunk=%zu", cb->read, cb->write, cb->size, outstanding);
 
 #ifdef ENABLE_HTTPS
-    SSL *conn = s->ssl.conn ;
-    if(conn && s->ssl.flags == NETDATA_SSL_HANDSHAKE_COMPLETE)
-        ret = netdata_ssl_write(conn, chunk, outstanding);
+    if(SSL_connection(&s->ssl))
+        ret = netdata_ssl_write(&s->ssl, chunk, outstanding);
     else
         ret = send(s->rrdpush_sender_socket, chunk, outstanding, MSG_DONTWAIT);
 #else
@@ -852,25 +837,17 @@ static ssize_t attempt_to_send(struct sender_state *s) {
 }
 
 static ssize_t attempt_read(struct sender_state *s) {
-    ssize_t ret = 0;
+    ssize_t ret;
 
 #ifdef ENABLE_HTTPS
-    if (s->ssl.conn && s->ssl.flags == NETDATA_SSL_HANDSHAKE_COMPLETE) {
-        size_t desired = sizeof(s->read_buffer) - s->read_len - 1;
-        ret = netdata_ssl_read(s->ssl.conn, s->read_buffer, desired);
-        if (ret > 0 ) {
-            s->read_len += (int)ret;
-            return ret;
-        }
-
-        if (ret == -1) {
-            worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
-            rrdpush_sender_thread_close_socket(s->host);
-        }
-        return ret;
-    }
-#endif
+    if (SSL_connection(&s->ssl))
+        ret = netdata_ssl_read(&s->ssl, s->read_buffer + s->read_len, sizeof(s->read_buffer) - s->read_len - 1);
+    else
+        ret = recv(s->rrdpush_sender_socket, s->read_buffer + s->read_len, sizeof(s->read_buffer) - s->read_len - 1,MSG_DONTWAIT);
+#else
     ret = recv(s->rrdpush_sender_socket, s->read_buffer + s->read_len, sizeof(s->read_buffer) - s->read_len - 1,MSG_DONTWAIT);
+#endif
+
     if (ret > 0) {
         s->read_len += ret;
         return ret;
@@ -879,6 +856,12 @@ static ssize_t attempt_read(struct sender_state *s) {
     if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
         return ret;
 
+#ifdef ENABLE_HTTPS
+    if (SSL_connection(&s->ssl))
+        worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);
+    else
+#endif
+
     if (ret == 0 || errno == ECONNRESET) {
         worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_PARENT_CLOSED);
         error("STREAM %s [send to %s]: connection closed by far end.", rrdhost_hostname(s->host), s->connected_to);
@@ -887,6 +870,7 @@ static ssize_t attempt_read(struct sender_state *s) {
         worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR);
         error("STREAM %s [send to %s]: error during receive (%zd) - closing connection.", rrdhost_hostname(s->host), s->connected_to, ret);
     }
+
     rrdpush_sender_thread_close_socket(s->host);
 
     return ret;
@@ -1096,6 +1080,8 @@ static bool rrdhost_set_sender(RRDHOST *host) {
     }
     netdata_mutex_unlock(&host->sender->mutex);
 
+    rrdpush_reset_destinations_postpone_time(host);
+
     return ret;
 }
 
@@ -1108,6 +1094,8 @@ static void rrdhost_clear_sender___while_having_sender_mutex(RRDHOST *host) {
         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);
     }
+
+    rrdpush_reset_destinations_postpone_time(host);
 }
 
 static bool rrdhost_sender_should_exit(struct sender_state *s) {
@@ -1134,7 +1122,7 @@ static bool rrdhost_sender_should_exit(struct sender_state *s) {
 
     if(unlikely(rrdhost_flag_check(s->host, RRDHOST_FLAG_ORPHAN))) {
         if(!s->exit.reason)
-            s->exit.reason = "RECEIVER LEFT";
+            s->exit.reason = "RECEIVER LEFT (ORPHAN HOST)";
         return true;
     }
 
@@ -1162,6 +1150,32 @@ static void rrdpush_sender_thread_cleanup_callback(void *ptr) {
     freez(s);
 }
 
+void rrdpush_initialize_ssl_ctx(RRDHOST *host) {
+#ifdef ENABLE_HTTPS
+    static SPINLOCK sp = NETDATA_SPINLOCK_INITIALIZER;
+    netdata_spinlock_lock(&sp);
+
+    if(netdata_ssl_streaming_sender_ctx || !host) {
+        netdata_spinlock_unlock(&sp);
+        return;
+    }
+
+    for(struct rrdpush_destinations *d = host->destinations; d ; d = d->next) {
+        if (d->ssl) {
+            // we need to initialize SSL
+
+            netdata_ssl_initialize_ctx(NETDATA_SSL_STREAMING_SENDER_CTX);
+            ssl_security_location_for_context(netdata_ssl_streaming_sender_ctx, netdata_ssl_ca_file, netdata_ssl_ca_path);
+
+            // stop the loop
+            break;
+        }
+    }
+
+    netdata_spinlock_unlock(&sp);
+#endif
+}
+
 void *rrdpush_sender_thread(void *ptr) {
     worker_register("STREAMSND");
     worker_register_job_name(WORKER_SENDER_JOB_CONNECT, "connect");
@@ -1206,17 +1220,7 @@ void *rrdpush_sender_thread(void *ptr) {
         return NULL;
     }
 
-#ifdef ENABLE_HTTPS
-    if (netdata_use_ssl_on_stream & NETDATA_SSL_FORCE ) {
-        static SPINLOCK sp = NETDATA_SPINLOCK_INITIALIZER;
-        netdata_spinlock_lock(&sp);
-        if(!netdata_ssl_client_ctx) {
-            security_start_ssl(NETDATA_SSL_CONTEXT_STREAMING);
-            ssl_security_location_for_context(netdata_ssl_client_ctx, netdata_ssl_ca_file, netdata_ssl_ca_path);
-        }
-        netdata_spinlock_unlock(&sp);
-    }
-#endif
+    rrdpush_initialize_ssl_ctx(s->host);
 
     info("STREAM %s [send]: thread created (task id %d)", rrdhost_hostname(s->host), gettid());