Adds more info to aclk-state API call (#12231)

4 files changed, 153 insertions, 1 deletions

diff --git a/database/sqlite/sqlite_aclk_alert.c b/database/sqlite/sqlite_aclk_alert.c
index d6773b3dc7..f57111408f 100644
--- a/database/sqlite/sqlite_aclk_alert.c
+++ b/database/sqlite/sqlite_aclk_alert.c
@@ -908,3 +908,44 @@ void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
     UNUSED(host);
 #endif
 }
+
+int get_proto_alert_status(RRDHOST *host, struct proto_alert_status *proto_alert_status)
+{
+    int rc;
+    struct aclk_database_worker_config *wc  = NULL;
+    wc = (struct aclk_database_worker_config *)host->dbsync_worker;
+    if (!wc)
+        return 1;
+
+    proto_alert_status->alert_updates = wc->alert_updates;
+    proto_alert_status->alerts_batch_id = wc->alerts_batch_id;
+
+    BUFFER *sql = buffer_create(1024);
+    sqlite3_stmt *res = NULL;
+
+    buffer_sprintf(sql, "SELECT MIN(sequence_id), MAX(sequence_id), " \
+                   "(select MAX(sequence_id) from aclk_alert_%s where date_cloud_ack is not NULL), " \
+                   "(select MAX(sequence_id) from aclk_alert_%s where date_submitted is not NULL) " \
+                   "FROM aclk_alert_%s where date_submitted is null;", wc->uuid_str, wc->uuid_str, wc->uuid_str);
+
+    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
+    if (rc != SQLITE_OK) {
+        error_report("Failed to prepare statement to get alert log status from the database.");
+        buffer_free(sql);
+        return 1;
+    }
+
+    while (sqlite3_step(res) == SQLITE_ROW) {
+        proto_alert_status->pending_min_sequence_id = sqlite3_column_bytes(res, 0) > 0 ? (uint64_t) sqlite3_column_int64(res, 0) : 0;
+        proto_alert_status->pending_max_sequence_id = sqlite3_column_bytes(res, 1) > 0 ? (uint64_t) sqlite3_column_int64(res, 1) : 0;
+        proto_alert_status->last_acked_sequence_id = sqlite3_column_bytes(res, 2) > 0 ? (uint64_t) sqlite3_column_int64(res, 2) : 0;
+        proto_alert_status->last_submitted_sequence_id = sqlite3_column_bytes(res, 3) > 0 ? (uint64_t) sqlite3_column_int64(res, 3) : 0;
+    }
+
+    rc = sqlite3_finalize(res);
+    if (unlikely(rc != SQLITE_OK))
+        error_report("Failed to finalize statement to get alert log status from the database, rc = %d", rc);
+
+    buffer_free(sql);
+    return 0;
+}
diff --git a/database/sqlite/sqlite_aclk_alert.h b/database/sqlite/sqlite_aclk_alert.h
index 1aaaa5d23a..957cb94ac8 100644
--- a/database/sqlite/sqlite_aclk_alert.h
+++ b/database/sqlite/sqlite_aclk_alert.h
@@ -5,6 +5,15 @@
 
 extern sqlite3 *db_meta;
 
+struct proto_alert_status {
+    int alert_updates;
+    uint64_t alerts_batch_id;
+    uint64_t last_acked_sequence_id;
+    uint64_t pending_min_sequence_id;
+    uint64_t pending_max_sequence_id;
+    uint64_t last_submitted_sequence_id;
+};
+
 int aclk_add_alert_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
 void aclk_push_alert_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
 void aclk_send_alarm_health_log(char *node_id);
@@ -16,5 +25,6 @@ void sql_queue_removed_alerts_to_aclk(RRDHOST *host);
 void sql_process_queue_removed_alerts_to_aclk(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
 void aclk_push_alert_snapshot_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
 void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id, uint64_t snapshot_id, uint64_t sequence_id);
+int get_proto_alert_status(RRDHOST *host, struct proto_alert_status *proto_alert_status);
 
 #endif //NETDATA_SQLITE_ACLK_ALERT_H
diff --git a/database/sqlite/sqlite_aclk_chart.c b/database/sqlite/sqlite_aclk_chart.c
index f4e8aea23a..f6f3ebeb32 100644
--- a/database/sqlite/sqlite_aclk_chart.c
+++ b/database/sqlite/sqlite_aclk_chart.c
@@ -1014,6 +1014,91 @@ void aclk_send_dimension_update(RRDDIM *rd)
     return;
 }
 
+#define SQL_SEQ_NULL(result, n)  sqlite3_column_type(result, n) == SQLITE_NULL ? 0 : sqlite3_column_int64(result, n)
+
+struct aclk_chart_sync_stats *aclk_get_chart_sync_stats(RRDHOST *host)
+{
+    struct aclk_chart_sync_stats *aclk_statistics = NULL;
+
+    struct aclk_database_worker_config *wc  = NULL;
+    wc = (struct aclk_database_worker_config *)host->dbsync_worker;
+    if (!wc)
+        return NULL;
+
+    aclk_statistics = callocz(1, sizeof(struct aclk_chart_sync_stats));
+
+    aclk_statistics->updates = wc->chart_updates;
+    aclk_statistics->batch_id = wc->batch_id;
+
+    char host_uuid_fixed[GUID_LEN + 1];
+
+    strncpy(host_uuid_fixed, host->machine_guid, GUID_LEN);
+    host_uuid_fixed[GUID_LEN] = 0;
+
+    host_uuid_fixed[8] = '_';
+    host_uuid_fixed[13] = '_';
+    host_uuid_fixed[18] = '_';
+    host_uuid_fixed[23] = '_';
+
+    sqlite3_stmt *res = NULL;
+    BUFFER *sql = buffer_create(1024);
+    buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s;", host_uuid_fixed);
+    buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_submitted IS NULL;", host_uuid_fixed);
+    buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_submitted IS NOT NULL;", host_uuid_fixed);
+    buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_updated IS NOT NULL;", host_uuid_fixed);
+    buffer_sprintf(sql, "SELECT max(date_created), max(date_submitted), max(date_updated), 0 FROM aclk_chart_%s;", host_uuid_fixed);
+
+    int rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
+    if (rc != SQLITE_OK) {
+        buffer_free(sql);
+        freez(aclk_statistics);
+        return NULL;
+    }
+
+    rc = sqlite3_step(res);
+    if (rc == SQLITE_ROW) {
+        aclk_statistics->min_seqid = SQL_SEQ_NULL(res, 0);
+        aclk_statistics->max_seqid = SQL_SEQ_NULL(res, 1);
+    }
+
+    rc = sqlite3_step(res);
+    if (rc == SQLITE_ROW) {
+        aclk_statistics->min_seqid_pend = SQL_SEQ_NULL(res, 0);
+        aclk_statistics->max_seqid_pend = SQL_SEQ_NULL(res, 1);
+    }
+
+    rc = sqlite3_step(res);
+    if (rc == SQLITE_ROW) {
+        aclk_statistics->min_seqid_sent = SQL_SEQ_NULL(res, 0);
+        aclk_statistics->max_seqid_sent = SQL_SEQ_NULL(res, 1);
+    }
+
+    rc = sqlite3_step(res);
+    if (rc == SQLITE_ROW) {
+        aclk_statistics->min_seqid_ack = SQL_SEQ_NULL(res, 0);
+        aclk_statistics->max_seqid_ack = SQL_SEQ_NULL(res, 1);
+    }
+
+    rc = sqlite3_step(res);
+    if (rc == SQLITE_ROW) {
+        aclk_statistics->min_seqid_ack = SQL_SEQ_NULL(res, 0);
+        aclk_statistics->max_seqid_ack = SQL_SEQ_NULL(res, 1);
+    }
+
+    rc = sqlite3_step(res);
+    if (rc == SQLITE_ROW) {
+        aclk_statistics->max_date_created = (time_t) SQL_SEQ_NULL(res, 0);
+        aclk_statistics->max_date_submitted = (time_t) SQL_SEQ_NULL(res, 1);
+        aclk_statistics->max_date_ack = (time_t) SQL_SEQ_NULL(res, 2);
+    }
+
+    rc = sqlite3_finalize(res);
+    if (unlikely(rc != SQLITE_OK))
+        error_report("Failed to finalize statement when fetching aclk sync statistics, rc = %d", rc);
+
+    buffer_free(sql);
+    return aclk_statistics;
+}
 #endif //ENABLE_NEW_CLOUD_PROTOCOL
 
 // ST is read locked
@@ -1035,4 +1120,3 @@ int queue_chart_to_aclk(RRDSET *st)
                                        st, ACLK_DATABASE_ADD_CHART);
 #endif
 }
-
diff --git a/database/sqlite/sqlite_aclk_chart.h b/database/sqlite/sqlite_aclk_chart.h
index 370af4729a..fee5ecca2f 100644
--- a/database/sqlite/sqlite_aclk_chart.h
+++ b/database/sqlite/sqlite_aclk_chart.h
@@ -16,6 +16,22 @@ extern sqlite3 *db_meta;
 #define RRDSET_MINIMUM_LIVE_MULTIPLIER   (1.5)
 #endif
 
+struct aclk_chart_sync_stats {
+    int        updates;
+    uint64_t   batch_id;
+    uint64_t   min_seqid;
+    uint64_t   max_seqid;
+    uint64_t   min_seqid_pend;
+    uint64_t   max_seqid_pend;
+    uint64_t   min_seqid_sent;
+    uint64_t   max_seqid_sent;
+    uint64_t   min_seqid_ack;
+    uint64_t   max_seqid_ack;
+    time_t     max_date_created;
+    time_t     max_date_submitted;
+    time_t     max_date_ack;
+};
+
 extern int queue_chart_to_aclk(RRDSET *st);
 extern int queue_dimension_to_aclk(RRDDIM *rd);
 extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
@@ -35,4 +51,5 @@ void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_
 void aclk_process_dimension_deletion(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
 uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc);
 void aclk_send_dimension_update(RRDDIM *rd);
+struct aclk_chart_sync_stats *aclk_get_chart_sync_stats(RRDHOST *host);
 #endif //NETDATA_SQLITE_ACLK_CHART_H