Adding upstream version 1.16.0.upstream/1.16.0

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

1 files changed, 431 insertions, 91 deletions

diff --git a/backends/backends.c b/backends/backends.c
index 0e7918916..15a0cb41b 100644
--- a/backends/backends.c
+++ b/backends/backends.c
@@ -246,6 +246,194 @@ static void backends_main_cleanup(void *ptr) {
     static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
 }
 
+/**
+ * Set Kinesis variables
+ *
+ * Set the variables necessaries to work with this specific backend.
+ *
+ * @param default_port  the default port of the backend
+ * @param brc function called to check the result.
+ * @param brf function called to format the msessage to the backend
+ * @param type the backend string selector.
+ */
+void backend_set_kinesis_variables(int *default_port,
+                                   backend_response_checker_t brc,
+                                   backend_request_formatter_t brf)
+{
+    (void)default_port;
+#ifndef HAVE_KINESIS
+    (void)brc;
+    (void)brf;
+#endif
+
+#if HAVE_KINESIS
+    *brc = process_json_response;
+    if (BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
+        *brf = format_dimension_collected_json_plaintext;
+    else
+        *brf = format_dimension_stored_json_plaintext;
+#endif
+}
+
+/**
+ * Set Prometheus variables
+ *
+ * Set the variables necessaries to work with this specific backend.
+ *
+ * @param default_port  the default port of the backend
+ * @param brc function called to check the result.
+ * @param brf function called to format the msessage to the backend
+ * @param type the backend string selector.
+ */
+void backend_set_prometheus_variables(int *default_port,
+                                      backend_response_checker_t brc,
+                                      backend_request_formatter_t brf)
+{
+    (void)default_port;
+    (void)brf;
+#ifndef ENABLE_PROMETHEUS_REMOTE_WRITE
+    (void)brc;
+#endif
+
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    *brc = process_prometheus_remote_write_response;
+#endif /* ENABLE_PROMETHEUS_REMOTE_WRITE */
+}
+
+/**
+ * Set JSON variables
+ *
+ * Set the variables necessaries to work with this specific backend.
+ *
+ * @param default_port  the default port of the backend
+ * @param brc function called to check the result.
+ * @param brf function called to format the msessage to the backend
+ * @param type the backend string selector.
+ */
+void backend_set_json_variables(int *default_port,
+                                backend_response_checker_t brc,
+                                backend_request_formatter_t brf)
+{
+    *default_port = 5448;
+    *brc = process_json_response;
+
+    if (BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
+        *brf = format_dimension_collected_json_plaintext;
+    else
+        *brf = format_dimension_stored_json_plaintext;
+}
+
+/**
+ * Set OpenTSDB HTTP variables
+ *
+ * Set the variables necessaries to work with this specific backend.
+ *
+ * @param default_port  the default port of the backend
+ * @param brc function called to check the result.
+ * @param brf function called to format the msessage to the backend
+ * @param type the backend string selector.
+ */
+void backend_set_opentsdb_http_variables(int *default_port,
+                                         backend_response_checker_t brc,
+                                         backend_request_formatter_t brf)
+{
+    *default_port = 4242;
+    *brc = process_opentsdb_response;
+
+    if(BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
+        *brf = format_dimension_collected_opentsdb_http;
+    else
+        *brf = format_dimension_stored_opentsdb_http;
+
+}
+
+/**
+ * Set OpenTSDB Telnet variables
+ *
+ * Set the variables necessaries to work with this specific backend.
+ *
+ * @param default_port  the default port of the backend
+ * @param brc function called to check the result.
+ * @param brf function called to format the msessage to the backend
+ * @param type the backend string selector.
+ */
+void backend_set_opentsdb_telnet_variables(int *default_port,
+                                           backend_response_checker_t brc,
+                                           backend_request_formatter_t brf)
+{
+    *default_port = 4242;
+    *brc = process_opentsdb_response;
+
+    if(BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
+        *brf = format_dimension_collected_opentsdb_telnet;
+    else
+        *brf = format_dimension_stored_opentsdb_telnet;
+}
+
+/**
+ * Set Graphite variables
+ *
+ * Set the variables necessaries to work with this specific backend.
+ *
+ * @param default_port  the default port of the backend
+ * @param brc function called to check the result.
+ * @param brf function called to format the msessage to the backend
+ * @param type the backend string selector.
+ */
+void backend_set_graphite_variables(int *default_port,
+                                    backend_response_checker_t brc,
+                                    backend_request_formatter_t brf)
+{
+    *default_port = 2003;
+    *brc = process_graphite_response;
+
+    if(BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
+        *brf = format_dimension_collected_graphite_plaintext;
+    else
+        *brf = format_dimension_stored_graphite_plaintext;
+}
+
+/**
+ * Select Type
+ *
+ * Select the backedn type based in the user input
+ *
+ * @param type is the string that defines the backend type
+ *
+ * @return It returns the backend id.
+ */
+BACKEND_TYPE backend_select_type(const char *type) {
+    if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
+        return BACKEND_TYPE_GRAPHITE;
+    }
+    else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
+        return BACKEND_TYPE_OPENTSDB_USING_TELNET;
+    }
+    else if(!strcmp(type, "opentsdb:http") || !strcmp(type, "opentsdb:https")) {
+        return BACKEND_TYPE_OPENTSDB_USING_HTTP;
+    }
+    else if (!strcmp(type, "json") || !strcmp(type, "json:plaintext")) {
+        return BACKEND_TYPE_JSON;
+    }
+    else if (!strcmp(type, "prometheus_remote_write")) {
+        return  BACKEND_TYPE_PROMETEUS;
+    }
+    else if (!strcmp(type, "kinesis") || !strcmp(type, "kinesis:plaintext")) {
+        return BACKEND_TYPE_KINESIS;
+    }
+
+    return BACKEND_TYPE_UNKNOWN;
+}
+
+/**
+ * Backend main
+ *
+ * The main thread used to control the backedns.
+ *
+ * @param ptr a pointer to netdata_static_structure.
+ *
+ * @return It always return NULL.
+ */
 void *backends_main(void *ptr) {
     netdata_thread_cleanup_push(backends_main_cleanup, ptr);
 
@@ -260,6 +448,15 @@ void *backends_main(void *ptr) {
     char *kinesis_auth_key_id = NULL, *kinesis_secure_key = NULL, *kinesis_stream_name = NULL;
 #endif
 
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    int do_prometheus_remote_write = 0;
+    BUFFER *http_request_header = buffer_create(1);
+#endif
+
+#ifdef ENABLE_HTTPS
+            struct netdata_ssl opentsdb_ssl = {NULL , NETDATA_SSL_START};
+#endif
+
     // ------------------------------------------------------------------------
     // collect configuration options
 
@@ -285,6 +482,10 @@ void *backends_main(void *ptr) {
     charts_pattern = simple_pattern_create(config_get(CONFIG_SECTION_BACKEND, "send charts matching", "*"), NULL, SIMPLE_PATTERN_EXACT);
     hosts_pattern  = simple_pattern_create(config_get(CONFIG_SECTION_BACKEND, "send hosts matching", "localhost *"), NULL, SIMPLE_PATTERN_EXACT);
 
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    const char *remote_write_path = config_get(CONFIG_SECTION_BACKEND, "remote write URL path", "/receive");
+#endif
+
     // ------------------------------------------------------------------------
     // validate configuration options
     // and prepare for sending data to our backend
@@ -303,90 +504,95 @@ void *backends_main(void *ptr) {
 
     // ------------------------------------------------------------------------
     // select the backend type
-
-    if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
-
-        default_port = 2003;
-        backend_response_checker = process_graphite_response;
-
-        if(BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
-            backend_request_formatter = format_dimension_collected_graphite_plaintext;
-        else
-            backend_request_formatter = format_dimension_stored_graphite_plaintext;
-
-    }
-    else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
-
-        default_port = 4242;
-        backend_response_checker = process_opentsdb_response;
-
-        if(BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
-            backend_request_formatter = format_dimension_collected_opentsdb_telnet;
-        else
-            backend_request_formatter = format_dimension_stored_opentsdb_telnet;
-
+    BACKEND_TYPE work_type = backend_select_type(type);
+    if (work_type == BACKEND_TYPE_UNKNOWN) {
+        error("BACKEND: Unknown backend type '%s'", type);
+        goto cleanup;
     }
-    else if (!strcmp(type, "json") || !strcmp(type, "json:plaintext")) {
 
-        default_port = 5448;
-        backend_response_checker = process_json_response;
-
-        if (BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
-            backend_request_formatter = format_dimension_collected_json_plaintext;
-        else
-            backend_request_formatter = format_dimension_stored_json_plaintext;
+    switch (work_type) {
+        case BACKEND_TYPE_OPENTSDB_USING_HTTP: {
+#ifdef ENABLE_HTTPS
+            if (!strcmp(type, "opentsdb:https")) {
+                security_start_ssl(NETDATA_SSL_CONTEXT_OPENTSDB);
+            }
+#endif
+            backend_set_opentsdb_http_variables(&default_port,&backend_response_checker,&backend_request_formatter);
+            break;
+        }
+        case BACKEND_TYPE_PROMETEUS: {
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+            do_prometheus_remote_write = 1;
 
-    }
+            init_write_request();
+#else
+            error("BACKEND: Prometheus remote write support isn't compiled");
+#endif // ENABLE_PROMETHEUS_REMOTE_WRITE
+            backend_set_prometheus_variables(&default_port,&backend_response_checker,&backend_request_formatter);
+            break;
+        }
+        case BACKEND_TYPE_KINESIS: {
 #if HAVE_KINESIS
-    else if (!strcmp(type, "kinesis") || !strcmp(type, "kinesis:plaintext")) {
+            do_kinesis = 1;
 
-        do_kinesis = 1;
+            if(unlikely(read_kinesis_conf(netdata_configured_user_config_dir, &kinesis_auth_key_id, &kinesis_secure_key, &kinesis_stream_name))) {
+                error("BACKEND: kinesis backend type is set but cannot read its configuration from %s/aws_kinesis.conf", netdata_configured_user_config_dir);
+                goto cleanup;
+            }
 
-        if(unlikely(read_kinesis_conf(netdata_configured_user_config_dir, &kinesis_auth_key_id, &kinesis_secure_key, &kinesis_stream_name))) {
-            error("BACKEND: kinesis backend type is set but cannot read its configuration from %s/aws_kinesis.conf", netdata_configured_user_config_dir);
-            goto cleanup;
+            kinesis_init(destination, kinesis_auth_key_id, kinesis_secure_key, timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
+#else
+            error("BACKEND: AWS Kinesis support isn't compiled");
+#endif // HAVE_KINESIS
+            backend_set_kinesis_variables(&default_port,&backend_response_checker,&backend_request_formatter);
+            break;
+        }
+        case BACKEND_TYPE_GRAPHITE: {
+            backend_set_graphite_variables(&default_port,&backend_response_checker,&backend_request_formatter);
+            break;
+        }
+        case BACKEND_TYPE_OPENTSDB_USING_TELNET: {
+            backend_set_opentsdb_telnet_variables(&default_port,&backend_response_checker,&backend_request_formatter);
+            break;
+        }
+        case BACKEND_TYPE_JSON: {
+            backend_set_json_variables(&default_port,&backend_response_checker,&backend_request_formatter);
+            break;
+        }
+        case BACKEND_TYPE_UNKNOWN: {
+            break;
         }
-
-        kinesis_init(destination, kinesis_auth_key_id, kinesis_secure_key, timeout.tv_sec * 1000 + timeout.tv_usec / 1000);
-
-        backend_response_checker = process_json_response;
-        if (BACKEND_OPTIONS_DATA_SOURCE(global_backend_options) == BACKEND_SOURCE_DATA_AS_COLLECTED)
-            backend_request_formatter = format_dimension_collected_json_plaintext;
-        else
-            backend_request_formatter = format_dimension_stored_json_plaintext;
-
-    }
-#endif /* HAVE_KINESIS */
-    else {
-        error("BACKEND: Unknown backend type '%s'", type);
-        goto cleanup;
     }
 
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    if((backend_request_formatter == NULL && !do_prometheus_remote_write) || backend_response_checker == NULL) {
+#else
     if(backend_request_formatter == NULL || backend_response_checker == NULL) {
+#endif
         error("BACKEND: backend is misconfigured - disabling it.");
         goto cleanup;
     }
 
 
-    // ------------------------------------------------------------------------
-    // prepare the charts for monitoring the backend operation
+// ------------------------------------------------------------------------
+// prepare the charts for monitoring the backend operation
 
     struct rusage thread;
 
     collected_number
-            chart_buffered_metrics = 0,
-            chart_lost_metrics = 0,
-            chart_sent_metrics = 0,
-            chart_buffered_bytes = 0,
-            chart_received_bytes = 0,
-            chart_sent_bytes = 0,
-            chart_receptions = 0,
-            chart_transmission_successes = 0,
-            chart_transmission_failures = 0,
-            chart_data_lost_events = 0,
-            chart_lost_bytes = 0,
-            chart_backend_reconnects = 0;
-            // chart_backend_latency = 0;
+        chart_buffered_metrics = 0,
+        chart_lost_metrics = 0,
+        chart_sent_metrics = 0,
+        chart_buffered_bytes = 0,
+        chart_received_bytes = 0,
+        chart_sent_bytes = 0,
+        chart_receptions = 0,
+        chart_transmission_successes = 0,
+        chart_transmission_failures = 0,
+        chart_data_lost_events = 0,
+        chart_lost_bytes = 0,
+        chart_backend_reconnects = 0;
+        // chart_backend_latency = 0;
 
     RRDSET *chart_metrics = rrdset_create_localhost("netdata", "backend_metrics", NULL, "backend", NULL, "Netdata Buffered Metrics", "metrics", "backends", NULL, 130600, global_backend_update_every, RRDSET_TYPE_LINE);
     rrddim_add(chart_metrics, "buffered", NULL,  1, 1, RRD_ALGORITHM_ABSOLUTE);
@@ -407,12 +613,12 @@ void *backends_main(void *ptr) {
     rrddim_add(chart_ops, "read",      NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
 
     /*
-     * this is misleading - we can only measure the time we need to send data
-     * this time is not related to the time required for the data to travel to
-     * the backend database and the time that server needed to process them
-     *
-     * issue #1432 and https://www.softlab.ntua.gr/facilities/documentation/unix/unix-socket-faq/unix-socket-faq-2.html
-     *
+    * this is misleading - we can only measure the time we need to send data
+    * this time is not related to the time required for the data to travel to
+    * the backend database and the time that server needed to process them
+    *
+    * issue #1432 and https://www.softlab.ntua.gr/facilities/documentation/unix/unix-socket-faq/unix-socket-faq-2.html
+    *
     RRDSET *chart_latency = rrdset_create_localhost("netdata", "backend_latency", NULL, "backend", NULL, "Netdata Backend Latency", "ms", "backends", NULL, 130620, global_backend_update_every, RRDSET_TYPE_AREA);
     rrddim_add(chart_latency, "latency",   NULL,  1, 1000, RRD_ALGORITHM_ABSOLUTE);
     */
@@ -451,6 +657,9 @@ void *backends_main(void *ptr) {
         size_t count_charts_total = 0;
         size_t count_dims_total = 0;
 
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+        clear_write_request();
+#endif
         rrd_rdlock();
         RRDHOST *host;
         rrdhost_foreach_read(host) {
@@ -478,26 +687,45 @@ void *backends_main(void *ptr) {
 
             const char *__hostname = (host == localhost)?hostname:host->hostname;
 
-            RRDSET *st;
-            rrdset_foreach_read(st, host) {
-                if(likely(backends_can_send_rrdset(global_backend_options, st))) {
-                    rrdset_rdlock(st);
-
-                    count_charts++;
-
-                    RRDDIM *rd;
-                    rrddim_foreach_read(rd, st) {
-                        if (likely(rd->last_collected_time.tv_sec >= after)) {
-                            chart_buffered_metrics += backend_request_formatter(b, global_backend_prefix, host, __hostname, st, rd, after, before, global_backend_options);
-                            count_dims++;
-                        }
-                        else {
-                            debug(D_BACKEND, "BACKEND: not sending dimension '%s' of chart '%s' from host '%s', its last data collection (%lu) is not within our timeframe (%lu to %lu)", rd->id, st->id, __hostname, (unsigned long)rd->last_collected_time.tv_sec, (unsigned long)after, (unsigned long)before);
-                            count_dims_skipped++;
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+            if(do_prometheus_remote_write) {
+                rrd_stats_remote_write_allmetrics_prometheus(
+                    host
+                    , __hostname
+                    , global_backend_prefix
+                    , global_backend_options
+                    , after
+                    , before
+                    , &count_charts
+                    , &count_dims
+                    , &count_dims_skipped
+                );
+                chart_buffered_metrics += count_dims;
+            }
+            else
+#endif
+            {
+                RRDSET *st;
+                rrdset_foreach_read(st, host) {
+                    if(likely(backends_can_send_rrdset(global_backend_options, st))) {
+                        rrdset_rdlock(st);
+
+                        count_charts++;
+
+                        RRDDIM *rd;
+                        rrddim_foreach_read(rd, st) {
+                            if (likely(rd->last_collected_time.tv_sec >= after)) {
+                                chart_buffered_metrics += backend_request_formatter(b, global_backend_prefix, host, __hostname, st, rd, after, before, global_backend_options);
+                                count_dims++;
+                            }
+                            else {
+                                debug(D_BACKEND, "BACKEND: not sending dimension '%s' of chart '%s' from host '%s', its last data collection (%lu) is not within our timeframe (%lu to %lu)", rd->id, st->id, __hostname, (unsigned long)rd->last_collected_time.tv_sec, (unsigned long)after, (unsigned long)before);
+                                count_dims_skipped++;
+                            }
                         }
-                    }
 
-                    rrdset_unlock(st);
+                        rrdset_unlock(st);
+                    }
                 }
             }
 
@@ -621,7 +849,16 @@ void *backends_main(void *ptr) {
                 while(sock != -1 && errno != EWOULDBLOCK) {
                     buffer_need_bytes(response, 4096);
 
-                    ssize_t r = recv(sock, &response->buffer[response->len], response->size - response->len, MSG_DONTWAIT);
+                    ssize_t r;
+#ifdef ENABLE_HTTPS
+                    if(opentsdb_ssl.conn && !opentsdb_ssl.flags) {
+                        r = SSL_read(opentsdb_ssl.conn, &response->buffer[response->len], response->size - response->len);
+                    } else {
+                        r = recv(sock, &response->buffer[response->len], response->size - response->len, MSG_DONTWAIT);
+                    }
+#else
+                    r = recv(sock, &response->buffer[response->len], response->size - response->len, MSG_DONTWAIT);
+#endif
                     if(likely(r > 0)) {
                         // we received some data
                         response->len += r;
@@ -654,7 +891,37 @@ void *backends_main(void *ptr) {
                 size_t reconnects = 0;
 
                 sock = connect_to_one_of(destination, default_port, &timeout, &reconnects, NULL, 0);
+#ifdef ENABLE_HTTPS
+                if(sock != -1) {
+                    if(netdata_opentsdb_ctx) {
+                        if(!opentsdb_ssl.conn) {
+                            opentsdb_ssl.conn = SSL_new(netdata_opentsdb_ctx);
+                            if(!opentsdb_ssl.conn) {
+                                error("Failed to allocate SSL structure %d.", sock);
+                                opentsdb_ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
+                            }
+                        } else {
+                            SSL_clear(opentsdb_ssl.conn);
+                        }
+                    }
 
+                    if(opentsdb_ssl.conn) {
+                        if(SSL_set_fd(opentsdb_ssl.conn, sock) != 1) {
+                            error("Failed to set the socket to the SSL on socket fd %d.", host->rrdpush_sender_socket);
+                            opentsdb_ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
+                        } else {
+                            opentsdb_ssl.flags = NETDATA_SSL_HANDSHAKE_COMPLETE;
+                            SSL_set_connect_state(opentsdb_ssl.conn);
+                            int err = SSL_connect(opentsdb_ssl.conn);
+                            if (err != 1) {
+                                err = SSL_get_error(opentsdb_ssl.conn, err);
+                                error("SSL cannot connect with the server:  %s ", ERR_error_string((long)SSL_get_error(opentsdb_ssl.conn, err), NULL));
+                                opentsdb_ssl.flags = NETDATA_SSL_NO_HANDSHAKE;
+                            } //TODO: check certificate here
+                        }
+                    }
+                }
+#endif
                 chart_backend_reconnects += reconnects;
                 // chart_backend_latency += now_monotonic_usec() - start_ut;
             }
@@ -672,7 +939,54 @@ void *backends_main(void *ptr) {
                 flags += MSG_NOSIGNAL;
     #endif
 
-                ssize_t written = send(sock, buffer_tostring(b), len, flags);
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+                if(do_prometheus_remote_write) {
+                    size_t data_size = get_write_request_size();
+
+                    if(unlikely(!data_size)) {
+                        error("BACKEND: write request size is out of range");
+                        continue;
+                    }
+
+                    buffer_flush(b);
+                    buffer_need_bytes(b, data_size);
+                    if(unlikely(pack_write_request(b->buffer, &data_size))) {
+                        error("BACKEND: cannot pack write request");
+                        continue;
+                    }
+                    b->len = data_size;
+                    chart_buffered_bytes = (collected_number)buffer_strlen(b);
+
+                    buffer_flush(http_request_header);
+                    buffer_sprintf(http_request_header,
+                                    "POST %s HTTP/1.1\r\n"
+                                    "Host: %s\r\n"
+                                    "Accept: */*\r\n"
+                                    "Content-Length: %zu\r\n"
+                                    "Content-Type: application/x-www-form-urlencoded\r\n\r\n",
+                                    remote_write_path,
+                                    hostname,
+                                    data_size
+                    );
+
+                    len = buffer_strlen(http_request_header);
+                    send(sock, buffer_tostring(http_request_header), len, flags);
+
+                    len = data_size;
+                }
+#endif
+
+                ssize_t written;
+#ifdef ENABLE_HTTPS
+                if(opentsdb_ssl.conn && !opentsdb_ssl.flags) {
+                    written = SSL_write(opentsdb_ssl.conn, buffer_tostring(b), len);
+                } else {
+                    written = send(sock, buffer_tostring(b), len, flags);
+                }
+#else
+                written = send(sock, buffer_tostring(b), len, flags);
+#endif
+
                 // chart_backend_latency += now_monotonic_usec() - start_ut;
                 if(written != -1 && (size_t)written == len) {
                     // we sent the data successfully
@@ -711,6 +1025,16 @@ void *backends_main(void *ptr) {
             }
         }
 
+
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+        if(failures) {
+            (void) buffer_on_failures;
+            failures = 0;
+            chart_lost_bytes = chart_buffered_bytes = get_write_request_size(); // estimated write request size
+            chart_data_lost_events++;
+            chart_lost_metrics = chart_buffered_metrics;
+        }
+#else
         if(failures > buffer_on_failures) {
             // too bad! we are going to lose data
             chart_lost_bytes += buffer_strlen(b);
@@ -720,6 +1044,7 @@ void *backends_main(void *ptr) {
             chart_data_lost_events++;
             chart_lost_metrics = chart_buffered_metrics;
         }
+#endif /* ENABLE_PROMETHEUS_REMOTE_WRITE */
 
         if(unlikely(netdata_exit)) break;
 
@@ -775,12 +1100,27 @@ cleanup:
     }
 #endif
 
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    if(do_prometheus_remote_write) {
+        buffer_free(http_request_header);
+        protocol_buffers_shutdown();
+    }
+#endif
+
     if(sock != -1)
         close(sock);
 
     buffer_free(b);
     buffer_free(response);
 
+#ifdef ENABLE_HTTPS
+    if(netdata_opentsdb_ctx) {
+        if(opentsdb_ssl.conn) {
+            SSL_free(opentsdb_ssl.conn);
+        }
+    }
+#endif
+
     netdata_thread_cleanup_pop(1);
     return NULL;
 }