3 files changed, 25 insertions, 17 deletions

diff --git a/collectors/plugins.d/plugins_d.c b/collectors/plugins.d/plugins_d.c
index dc13cd2ee..da5226a5c 100644
--- a/collectors/plugins.d/plugins_d.c
+++ b/collectors/plugins.d/plugins_d.c
@@ -72,7 +72,7 @@ static void pluginsd_worker_thread_cleanup(void *arg)
             info("PLUGINSD: 'host:%s', waiting for data collection child process pid %d to exit...",
                  rrdhost_hostname(cd->host), pid);
 
-            waitid(P_PID, (id_t)pid, &info, WEXITED);
+            netdata_waitid(P_PID, (id_t)pid, &info, WEXITED);
         }
     }
 }
diff --git a/collectors/plugins.d/plugins_d.h b/collectors/plugins.d/plugins_d.h
index 68ed4940f..fe43a19f5 100644
--- a/collectors/plugins.d/plugins_d.h
+++ b/collectors/plugins.d/plugins_d.h
@@ -87,6 +87,7 @@ struct plugind {
 extern struct plugind *pluginsd_root;
 
 size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugin_input, FILE *fp_plugin_output, int trust_durations);
+void pluginsd_process_thread_cleanup(void *ptr);
 
 size_t pluginsd_initialize_plugin_directories();
 
diff --git a/collectors/plugins.d/pluginsd_parser.c b/collectors/plugins.d/pluginsd_parser.c
index 28fc0bd49..097e5ea60 100644
--- a/collectors/plugins.d/pluginsd_parser.c
+++ b/collectors/plugins.d/pluginsd_parser.c
@@ -11,10 +11,10 @@ static int send_to_plugin(const char *txt, void *data) {
         return 0;
 
 #ifdef ENABLE_HTTPS
-    struct netdata_ssl *ssl = parser->ssl_output;
+    NETDATA_SSL *ssl = parser->ssl_output;
     if(ssl) {
-        if(ssl->conn && ssl->flags == NETDATA_SSL_HANDSHAKE_COMPLETE)
-            return (int)netdata_ssl_write(ssl->conn, (void *)txt, strlen(txt));
+        if(SSL_connection(ssl))
+            return (int)netdata_ssl_write(ssl, (void *)txt, strlen(txt));
 
         error("PLUGINSD: cannot send command (SSL)");
         return -1;
@@ -108,11 +108,12 @@ void pluginsd_rrdset_cleanup(RRDSET *st) {
     st->pluginsd.pos = 0;
 }
 
-static inline void pluginsd_set_chart_from_parent(void *user, RRDSET *st, const char *keyword) {
+static inline void pluginsd_unlock_previous_chart(void *user, const char *keyword, bool stale) {
     PARSER_USER_OBJECT *u = (PARSER_USER_OBJECT *) user;
 
     if(unlikely(pluginsd_unlock_rrdset_data_collection(user))) {
-        error("PLUGINSD: 'host:%s/chart:%s/' stale data collection lock found during %s; it has been unlocked",
+        if(stale)
+            error("PLUGINSD: 'host:%s/chart:%s/' stale data collection lock found during %s; it has been unlocked",
               rrdhost_hostname(u->st->rrdhost), rrdset_id(u->st), keyword);
     }
 
@@ -120,9 +121,16 @@ static inline void pluginsd_set_chart_from_parent(void *user, RRDSET *st, const
         ml_chart_update_end(u->st);
         u->v2.ml_locked = false;
 
-        error("PLUGINSD: 'host:%s/chart:%s/' stale ML lock found during %s, it has been unlocked",
+        if(stale)
+            error("PLUGINSD: 'host:%s/chart:%s/' stale ML lock found during %s, it has been unlocked",
               rrdhost_hostname(u->st->rrdhost), rrdset_id(u->st), keyword);
     }
+}
+
+static inline void pluginsd_set_chart_from_parent(void *user, RRDSET *st, const char *keyword) {
+    PARSER_USER_OBJECT *u = (PARSER_USER_OBJECT *) user;
+
+    pluginsd_unlock_previous_chart(user, keyword, true);
 
     if(st) {
         size_t dims = dictionary_entries(st->rrddim_root_index);
@@ -1459,9 +1467,11 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
         time_t started = st->rrdhost->receiver->replication_first_time_t;
         time_t current = ((PARSER_USER_OBJECT *) user)->replay.end_time;
 
-        if(started && current > started)
+        if(started && current > started) {
+            host->rrdpush_receiver_replication_percent = (NETDATA_DOUBLE) (current - started) * 100.0 / (NETDATA_DOUBLE) (now - started);
             worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION,
-                          (NETDATA_DOUBLE)(current - started) * 100.0 / (NETDATA_DOUBLE)(now - started));
+                              host->rrdpush_receiver_replication_percent);
+        }
     }
 
     ((PARSER_USER_OBJECT *) user)->replay.start_time = 0;
@@ -1501,7 +1511,8 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
 
         pluginsd_set_chart_from_parent(user, NULL, PLUGINSD_KEYWORD_REPLAY_END);
 
-        worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, 100.0);
+        host->rrdpush_receiver_replication_percent = 100.0;
+        worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, host->rrdpush_receiver_replication_percent);
 
         return PARSER_RC_OK;
     }
@@ -1783,12 +1794,7 @@ PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_words __maybe_
     // ------------------------------------------------------------------------
     // unblock data collection
 
-    ml_chart_update_end(st);
-    u->v2.ml_locked = false;
-
-    timing_step(TIMING_STEP_END2_ML);
-
-    pluginsd_unlock_rrdset_data_collection(user);
+    pluginsd_unlock_previous_chart(user, PLUGINSD_KEYWORD_END_V2, false);
     rrdcontext_collected_rrdset(st);
     store_metric_collection_completed();
 
@@ -1823,13 +1829,14 @@ PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_words __maybe_
     return PARSER_RC_OK;
 }
 
-static void pluginsd_process_thread_cleanup(void *ptr) {
+void pluginsd_process_thread_cleanup(void *ptr) {
     PARSER *parser = (PARSER *)ptr;
 
     pluginsd_cleanup_v2(parser->user);
     pluginsd_host_define_cleanup(parser->user);
 
     rrd_collector_finished();
+
     parser_destroy(parser);
 }