Improve cleaning up of orphan hosts (#12201)

* Move the rrdhost_cleanup_orphan_hosts_nolock to the service that processes obsolete charts

* Add OPCODE to mark a host as orphan

* Queue cmd to mark a host as orphan

3 files changed, 28 insertions, 7 deletions

diff --git a/database/rrdhost.c b/database/rrdhost.c
index 87d8c73259..dcdde6388d 100644
--- a/database/rrdhost.c
+++ b/database/rrdhost.c
@@ -642,8 +642,6 @@ RRDHOST *rrdhost_find_or_create(
         rrdhost_unlock(host);
     }
 
-    rrdhost_cleanup_orphan_hosts_nolock(host);
-
     rrd_unlock();
 
     return host;
@@ -652,7 +650,7 @@ inline int rrdhost_should_be_removed(RRDHOST *host, RRDHOST *protected_host, tim
     if(host != protected_host
        && host != localhost
        && rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN)
-       && host->receiver
+       && !host->receiver
        && host->senders_disconnected_time
        && host->senders_disconnected_time + rrdhost_free_orphan_time < now)
         return 1;
@@ -880,7 +878,20 @@ void rrdhost_free(RRDHOST *host) {
 
 
     rrdhost_wrlock(host);   // lock this RRDHOST
-
+#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
+    struct aclk_database_worker_config *wc =  host->dbsync_worker;
+    if (wc && !netdata_exit) {
+        struct aclk_database_cmd cmd;
+        memset(&cmd, 0, sizeof(cmd));
+        cmd.opcode = ACLK_DATABASE_ORPHAN_HOST;
+        struct aclk_completion compl ;
+        init_aclk_completion(&compl );
+        cmd.completion = &compl ;
+        aclk_database_enq_cmd(wc, &cmd);
+        wait_for_aclk_completion(&compl );
+        destroy_aclk_completion(&compl );
+    }
+#endif
     // ------------------------------------------------------------------------
     // release its children resources
 
@@ -983,7 +994,10 @@ void rrdhost_free(RRDHOST *host) {
     freez(host->node_id);
 
     freez(host);
-
+#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
+    if (wc)
+        wc->is_orphan = 0;
+#endif
     rrd_hosts_available--;
 }
 
diff --git a/database/sqlite/sqlite_aclk.c b/database/sqlite/sqlite_aclk.c
index 63196a81ea..ab02c6ce4e 100644
--- a/database/sqlite/sqlite_aclk.c
+++ b/database/sqlite/sqlite_aclk.c
@@ -508,11 +508,16 @@ void aclk_database_worker(void *arg)
                     aclk_update_retention(wc, cmd);
                     aclk_process_dimension_deletion(wc, cmd);
                     break;
-#endif
 
 // NODE_INSTANCE DETECTION
+                case ACLK_DATABASE_ORPHAN_HOST:
+                    wc->host = NULL;
+                    wc->is_orphan = 1;
+                    aclk_add_worker_thread(wc);
+                    break;
+#endif
                 case ACLK_DATABASE_TIMER:
-                    if (unlikely(localhost && !wc->host)) {
+                    if (unlikely(localhost && !wc->host && !wc->is_orphan)) {
                         if (claimed()) {
                             wc->host = rrdhost_find_by_guid(wc->host_guid, 0);
                             if (wc->host) {
diff --git a/database/sqlite/sqlite_aclk.h b/database/sqlite/sqlite_aclk.h
index 4249497409..894d934897 100644
--- a/database/sqlite/sqlite_aclk.h
+++ b/database/sqlite/sqlite_aclk.h
@@ -125,6 +125,7 @@ enum aclk_database_opcode {
     ACLK_DATABASE_CHART_ACK,
     ACLK_DATABASE_UPD_RETENTION,
     ACLK_DATABASE_DIM_DELETION,
+    ACLK_DATABASE_ORPHAN_HOST,
 #endif
     ACLK_DATABASE_ALARM_HEALTH_LOG,
     ACLK_DATABASE_CLEANUP,
@@ -194,6 +195,7 @@ struct aclk_database_worker_config {
     int chart_pending;
     int chart_reset_count;
     volatile unsigned is_shutting_down;
+    volatile unsigned is_orphan;
     struct aclk_database_worker_config  *next;
 };