Merge pull request #1863 from ktsaou/master

[netdata.git] / src / rrdpush.c

diff --git a/src/rrdpush.c b/src/rrdpush.c
index 0e7f0aedb170fabade8dbe8db6a33bf7f3070c82..06695c6f207e8a36e9f01019e22e4feb60fe50e2 100644 (file)
--- a/src/rrdpush.c
+++ b/src/rrdpush.c
@@ -33,6 +33,7 @@ int rrdpush_init() {
     default_rrdpush_enabled     = appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "enabled", default_rrdpush_enabled);
     default_rrdpush_destination = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "destination", "");
     default_rrdpush_api_key     = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "api key", "");
+    rrdhost_free_orphan_time    = appconfig_get_number(&stream_config, CONFIG_SECTION_STREAM, "free orphan hosts after seconds", rrdhost_free_orphan_time);
 
     if(default_rrdpush_enabled && (!default_rrdpush_destination || !*default_rrdpush_destination || !default_rrdpush_api_key || !*default_rrdpush_api_key)) {
         error("STREAM [send]: cannot enable sending thread - information is missing.");
@@ -65,7 +66,7 @@ static unsigned int remote_clock_resync_iterations = 60;
 static inline int need_to_send_chart_definition(RRDSET *st) {
     RRDDIM *rd;
     rrddim_foreach_read(rd, st)
-        if(!rrddim_flag_check(rd, RRDDIM_FLAG_EXPOSED))
+        if(!rd->exposed)
             return 1;
 
     return 0;
@@ -96,7 +97,7 @@ static inline void send_chart_definition(RRDSET *st) {
                        , rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)?"hidden":""
                        , rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)?"noreset":""
         );
-        rrddim_flag_set(rd, RRDDIM_FLAG_EXPOSED);
+        rd->exposed = 1;
     }
 }
 
@@ -106,7 +107,7 @@ static inline void send_chart_metrics(RRDSET *st) {
 
     RRDDIM *rd;
     rrddim_foreach_read(rd, st) {
-        if(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED) && rrddim_flag_check(rd, RRDDIM_FLAG_EXPOSED))
+        if(rd->updated && rd->exposed)
             buffer_sprintf(st->rrdhost->rrdpush_buffer, "SET %s = " COLLECTED_NUMBER_FORMAT "\n"
                        , rd->id
                        , rd->collected_value
@@ -174,7 +175,7 @@ static void rrdpush_sender_thread_reset_all_charts(RRDHOST *host) {
 
         RRDDIM *rd;
         rrddim_foreach_read(rd, st)
-            rrddim_flag_clear(rd, RRDDIM_FLAG_EXPOSED);
+            rd->exposed = 0;
 
         rrdset_unlock(st);
     }
@@ -206,28 +207,45 @@ static inline void rrdpush_sender_thread_unlock(RRDHOST *host) {
     rrdpush_unlock(host);
 }
 
-void rrdpush_sender_thread_cleanup(RRDHOST *host) {
+static void rrdpush_sender_thread_cleanup(RRDHOST *host) {
     rrdpush_lock(host);
 
     host->rrdpush_connected = 0;
 
-    if(host->rrdpush_socket != -1) close(host->rrdpush_socket);
+    if(host->rrdpush_socket != -1) {
+        close(host->rrdpush_socket);
+        host->rrdpush_socket = -1;
+    }
 
     // close the pipe
-    if(host->rrdpush_pipe[PIPE_READ] != -1)  close(host->rrdpush_pipe[PIPE_READ]);
-    if(host->rrdpush_pipe[PIPE_WRITE] != -1) close(host->rrdpush_pipe[PIPE_WRITE]);
-    host->rrdpush_pipe[PIPE_READ] = -1;
-    host->rrdpush_pipe[PIPE_WRITE] = -1;
+    if(host->rrdpush_pipe[PIPE_READ] != -1) {
+        close(host->rrdpush_pipe[PIPE_READ]);
+        host->rrdpush_pipe[PIPE_READ] = -1;
+    }
+
+    if(host->rrdpush_pipe[PIPE_WRITE] != -1) {
+        close(host->rrdpush_pipe[PIPE_WRITE]);
+        host->rrdpush_pipe[PIPE_WRITE] = -1;
+    }
 
     buffer_free(host->rrdpush_buffer);
     host->rrdpush_buffer = NULL;
 
     host->rrdpush_spawn = 0;
-    host->rrdpush_enabled = 0;
 
     rrdpush_unlock(host);
 }
 
+void rrdpush_sender_thread_stop(RRDHOST *host) {
+    rrdhost_check_wrlock(host);
+
+    if(host->rrdpush_spawn) {
+        info("STREAM %s [send]: stopping sending thread...", host->hostname);
+        pthread_cancel(host->rrdpush_thread);
+        rrdpush_sender_thread_cleanup(host);
+    }
+}
+
 void *rrdpush_sender_thread(void *ptr) {
     RRDHOST *host = (RRDHOST *)ptr;
 
@@ -531,7 +549,7 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *m
     }
 
     rrdhost_wrlock(host);
-    host->use_counter++;
+    host->connected_senders++;
     if(health_enabled != CONFIG_BOOLEAN_NO)
         host->health_delay_up_to = now_realtime_sec() + alarms_delay;
     rrdhost_unlock(host);
@@ -542,15 +560,14 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *m
     error("STREAM %s [receive from [%s]:%s]: disconnected (completed updates %zu).", host->hostname, client_ip, client_port, count);
 
     rrdhost_wrlock(host);
-    host->use_counter--;
-    if(!host->use_counter) {
+    host->connected_senders--;
+    if(!host->connected_senders) {
         if(health_enabled == CONFIG_BOOLEAN_AUTO)
             host->health_enabled = 0;
 
-        if(host->rrdpush_enabled && host->rrdpush_spawn) {
-            pthread_cancel(host->rrdpush_thread);
-            rrdpush_sender_thread_cleanup(host);
-        }
+        host->senders_disconnected_time = now_realtime_sec();
+
+        rrdpush_sender_thread_stop(host);
     }
     rrdhost_unlock(host);
 
@@ -672,7 +689,7 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
         return 404;
     }
 
-    if(!appconfig_get_boolean(&stream_config, key, "enabled", 1)) {
+    if(!appconfig_get_boolean(&stream_config, key, "enabled", 0)) {
         error("STREAM [receive from [%s]:%s]: API key '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, machine_guid);
         buffer_flush(w->response.data);
         buffer_sprintf(w->response.data, "Your API key is not permitted access.");