@cryptotaxi247 / netdata-1 / commits / d8aba23d0

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

Timotej S committed Mar 9, 2022 at 14:08 UTC d8aba23d0f471ccb3f0107c4aac119cf35753f94
7 files changed +427 -36
aclk/aclk.c
+257 -15
@@ -24,6 +24,8 @@
24 #define ACLK_STABLE_TIMEOUT 3 // Minimum delay to mark AGENT as stable
25
26 int aclk_pubacks_per_conn = 0; // How many PubAcks we got since MQTT conn est.
27 +int aclk_rcvd_cloud_msgs = 0;
28 +int aclk_connection_counter = 0;
29 int disconnect_req = 0;
30
31 int aclk_alert_reloaded = 1; //1 on startup, and again on health_reload
@@ -43,8 +45,6 @@ struct aclk_shared_state aclk_shared_state = {
45 .mqtt_shutdown_msg_rcvd = 0
46 };
47
46 -//ENDTODO
47 -
48 static RSA *aclk_private_key = NULL;
49 static int load_private_key()
50 {
@@ -266,6 +266,7 @@ static void msg_callback_new_protocol(const char *topic, const void *msg, size_t
266 }
267
268 static inline void msg_callback(const char *topic, const void *msg, size_t msglen, int qos) {
269 + aclk_rcvd_cloud_msgs++;
270 if (aclk_use_new_cloud_arch)
271 msg_callback_new_protocol(topic, msg, msglen, qos);
272 else
@@ -402,6 +403,8 @@ static inline void mqtt_connected_actions(mqtt_wss_client client)
403 aclk_stats_upd_online(1);
404 aclk_connected = 1;
405 aclk_pubacks_per_conn = 0;
406 + aclk_rcvd_cloud_msgs = 0;
407 + aclk_connection_counter++;
408
409 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
410 if (!aclk_use_new_cloud_arch) {
@@ -1117,6 +1120,64 @@ void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, con
1120 aclk_send_bin_message_subtopic_pid(mqttwss_client, msg, msg_len, subtopic, msgname);
1121 }
1122
1123 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1124 +static void fill_alert_status_for_host(BUFFER *wb, RRDHOST *host)
1125 +{
1126 + struct proto_alert_status status;
1127 + memset(&status, 0, sizeof(status));
1128 + if (get_proto_alert_status(host, &status)) {
1129 + buffer_strcat(wb, "\nFailed to get alert streaming status for this host");
1130 + return;
1131 + }
1132 + buffer_sprintf(wb,
1133 + "\n\t\tUpdates: %d"
1134 + "\n\t\tBatch ID: %"PRIu64
1135 + "\n\t\tLast Acked Seq ID: %"PRIu64
1136 + "\n\t\tPending Min Seq ID: %"PRIu64
1137 + "\n\t\tPending Max Seq ID: %"PRIu64
1138 + "\n\t\tLast Submitted Seq ID: %"PRIu64,
1139 + status.alert_updates,
1140 + status.alerts_batch_id,
1141 + status.last_acked_sequence_id,
1142 + status.pending_min_sequence_id,
1143 + status.pending_max_sequence_id,
1144 + status.last_submitted_sequence_id
1145 + );
1146 +}
1147 +
1148 +static void fill_chart_status_for_host(BUFFER *wb, RRDHOST *host)
1149 +{
1150 + struct aclk_chart_sync_stats *stats = aclk_get_chart_sync_stats(host);
1151 + if (!stats) {
1152 + buffer_strcat(wb, "\n\t\tFailed to get alert streaming status for this host");
1153 + return;
1154 + }
1155 + buffer_sprintf(wb,
1156 + "\n\t\tUpdates: %d"
1157 + "\n\t\tBatch ID: %"PRIu64
1158 + "\n\t\tMin Seq ID: %"PRIu64
1159 + "\n\t\tMax Seq ID: %"PRIu64
1160 + "\n\t\tPending Min Seq ID: %"PRIu64
1161 + "\n\t\tPending Max Seq ID: %"PRIu64
1162 + "\n\t\tSent Min Seq ID: %"PRIu64
1163 + "\n\t\tSent Max Seq ID: %"PRIu64
1164 + "\n\t\tAcked Min Seq ID: %"PRIu64
1165 + "\n\t\tAcked Max Seq ID: %"PRIu64,
1166 + stats->updates,
1167 + stats->batch_id,
1168 + stats->min_seqid,
1169 + stats->max_seqid,
1170 + stats->min_seqid_pend,
1171 + stats->max_seqid_pend,
1172 + stats->min_seqid_sent,
1173 + stats->max_seqid_sent,
1174 + stats->min_seqid_ack,
1175 + stats->max_seqid_ack
1176 + );
1177 + freez(stats);
1178 +}
1179 +#endif
1180 +
1181 char *ng_aclk_state(void)
1182 {
1183 BUFFER *wb = buffer_create(1024);
@@ -1124,46 +1185,160 @@ char *ng_aclk_state(void)
1185
1186 buffer_strcat(wb,
1187 "ACLK Available: Yes\n"
1127 - "ACLK Implementation: Next Generation\n"
1188 + "ACLK Version: 2\n"
1189 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
1129 - "New Cloud Protocol Support: Yes\n"
1190 + "Protocols Supported: Legacy, Protobuf\n"
1191 #else
1131 - "New Cloud Protocol Support: No\n"
1192 + "Protocols Supported: Legacy\n"
1193 #endif
1133 - "Claimed: "
1194 );
1195 + buffer_sprintf(wb, "Protocol Used: %s\nClaimed: ", aclk_use_new_cloud_arch ? "Protobuf" : "Legacy");
1196
1197 char *agent_id = is_agent_claimed();
1198 if (agent_id == NULL)
1199 buffer_strcat(wb, "No\n");
1200 else {
1140 - buffer_sprintf(wb, "Yes\nClaimed Id: %s\n", agent_id);
1201 + char *cloud_base_url = appconfig_get(&cloud_config, CONFIG_SECTION_GLOBAL, "cloud base url", NULL);
1202 + buffer_sprintf(wb, "Yes\nClaimed Id: %s\nCloud URL: %s\n", agent_id, cloud_base_url ? cloud_base_url : "null");
1203 freez(agent_id);
1204 }
1205
1144 - buffer_sprintf(wb, "Online: %s\nUsed Cloud Protocol: %s", aclk_connected ? "Yes" : "No", aclk_use_new_cloud_arch ? "New" : "Legacy");
1206 + buffer_sprintf(wb, "Online: %s\nReconnect count: %d\n", aclk_connected ? "Yes" : "No", aclk_connection_counter > 0 ? (aclk_connection_counter - 1) : 0);
1207 +
1208 + if (aclk_connected) {
1209 + buffer_sprintf(wb, "Received Cloud MQTT Messages: %d\nMQTT Messages Confirmed by Remote Broker (PUBACKs): %d", aclk_rcvd_cloud_msgs, aclk_pubacks_per_conn);
1210 +
1211 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1212 + RRDHOST *host;
1213 + rrd_rdlock();
1214 + rrdhost_foreach_read(host) {
1215 + buffer_sprintf(wb, "\n\n> Node Instance for mGUID: \"%s\" hostname \"%s\"\n", host->machine_guid, host->hostname);
1216 +
1217 + buffer_strcat(wb, "\tClaimed ID: ");
1218 + rrdhost_aclk_state_lock(host);
1219 + if (host->aclk_state.claimed_id)
1220 + buffer_strcat(wb, host->aclk_state.claimed_id);
1221 + else
1222 + buffer_strcat(wb, "null");
1223 + rrdhost_aclk_state_unlock(host);
1224 +
1225 +
1226 + if (host->node_id == NULL || uuid_is_null(*host->node_id)) {
1227 + buffer_strcat(wb, "\n\tNode ID: null\n");
1228 + } else {
1229 + char node_id[GUID_LEN + 1];
1230 + uuid_unparse_lower(*host->node_id, node_id);
1231 + buffer_sprintf(wb, "\n\tNode ID: %s\n", node_id);
1232 + }
1233 +
1234 + buffer_sprintf(wb, "\tStreaming Hops: %d\n\tRelationship: %s", host->system_info->hops, host == localhost ? "self" : "child");
1235 +
1236 + if (host != localhost)
1237 + buffer_sprintf(wb, "\n\tStreaming Connection Live: %s", host->receiver ? "true" : "false");
1238 +
1239 + buffer_strcat(wb, "\n\tAlert Streaming Status:");
1240 + fill_alert_status_for_host(wb, host);
1241 +
1242 + buffer_strcat(wb, "\n\tChart Streaming Status:");
1243 + fill_chart_status_for_host(wb, host);
1244 + }
1245 + rrd_unlock();
1246 +#endif
1247 + }
1248
1249 ret = strdupz(buffer_tostring(wb));
1250 buffer_free(wb);
1251 return ret;
1252 }
1253
1254 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1255 +static void fill_alert_status_for_host_json(json_object *obj, RRDHOST *host)
1256 +{
1257 + struct proto_alert_status status;
1258 + memset(&status, 0, sizeof(status));
1259 + if (get_proto_alert_status(host, &status))
1260 + return;
1261 +
1262 + json_object *tmp = json_object_new_int(status.alert_updates);
1263 + json_object_object_add(obj, "updates", tmp);
1264 +
1265 + tmp = json_object_new_int(status.alerts_batch_id);
1266 + json_object_object_add(obj, "batch-id", tmp);
1267 +
1268 + tmp = json_object_new_int(status.last_acked_sequence_id);
1269 + json_object_object_add(obj, "last-acked-seq-id", tmp);
1270 +
1271 + tmp = json_object_new_int(status.pending_min_sequence_id);
1272 + json_object_object_add(obj, "pending-min-seq-id", tmp);
1273 +
1274 + tmp = json_object_new_int(status.pending_max_sequence_id);
1275 + json_object_object_add(obj, "pending-max-seq-id", tmp);
1276 +
1277 + tmp = json_object_new_int(status.last_submitted_sequence_id);
1278 + json_object_object_add(obj, "last-submitted-seq-id", tmp);
1279 +}
1280 +
1281 +static void fill_chart_status_for_host_json(json_object *obj, RRDHOST *host)
1282 +{
1283 + struct aclk_chart_sync_stats *stats = aclk_get_chart_sync_stats(host);
1284 + if (!stats)
1285 + return;
1286 +
1287 + json_object *tmp = json_object_new_int(stats->updates);
1288 + json_object_object_add(obj, "updates", tmp);
1289 +
1290 + tmp = json_object_new_int(stats->batch_id);
1291 + json_object_object_add(obj, "batch-id", tmp);
1292 +
1293 + tmp = json_object_new_int(stats->min_seqid);
1294 + json_object_object_add(obj, "min-seq-id", tmp);
1295 +
1296 + tmp = json_object_new_int(stats->max_seqid);
1297 + json_object_object_add(obj, "max-seq-id", tmp);
1298 +
1299 + tmp = json_object_new_int(stats->min_seqid_pend);
1300 + json_object_object_add(obj, "pending-min-seq-id", tmp);
1301 +
1302 + tmp = json_object_new_int(stats->max_seqid_pend);
1303 + json_object_object_add(obj, "pending-max-seq-id", tmp);
1304 +
1305 + tmp = json_object_new_int(stats->min_seqid_sent);
1306 + json_object_object_add(obj, "sent-min-seq-id", tmp);
1307 +
1308 + tmp = json_object_new_int(stats->max_seqid_sent);
1309 + json_object_object_add(obj, "sent-max-seq-id", tmp);
1310 +
1311 + tmp = json_object_new_int(stats->min_seqid_ack);
1312 + json_object_object_add(obj, "acked-min-seq-id", tmp);
1313 +
1314 + tmp = json_object_new_int(stats->max_seqid_ack);
1315 + json_object_object_add(obj, "acked-max-seq-id", tmp);
1316 +
1317 + freez(stats);
1318 +}
1319 +#endif
1320 +
1321 char *ng_aclk_state_json(void)
1322 {
1153 - json_object *tmp, *msg = json_object_new_object();
1323 + json_object *tmp, *grp, *msg = json_object_new_object();
1324
1325 tmp = json_object_new_boolean(1);
1326 json_object_object_add(msg, "aclk-available", tmp);
1327
1158 - tmp = json_object_new_string("Next Generation");
1159 - json_object_object_add(msg, "aclk-implementation", tmp);
1328 + tmp = json_object_new_int(2);
1329 + json_object_object_add(msg, "aclk-version", tmp);
1330
1331 + grp = json_object_new_array();
1332 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
1162 - tmp = json_object_new_boolean(1);
1333 + tmp = json_object_new_string("Legacy");
1334 + json_object_array_add(grp, tmp);
1335 + tmp = json_object_new_string("Protobuf");
1336 + json_object_array_add(grp, tmp);
1337 #else
1164 - tmp = json_object_new_boolean(0);
1338 + tmp = json_object_new_string("Legacy");
1339 + json_object_array_add(grp, tmp);
1340 #endif
1166 - json_object_object_add(msg, "new-cloud-protocol-supported", tmp);
1341 + json_object_object_add(msg, "protocols-supported", grp);
1342
1343 char *agent_id = is_agent_claimed();
1344 tmp = json_object_new_boolean(agent_id != NULL);
@@ -1176,12 +1351,79 @@ char *ng_aclk_state_json(void)
1351 tmp = NULL;
1352 json_object_object_add(msg, "claimed-id", tmp);
1353
1354 + char *cloud_base_url = appconfig_get(&cloud_config, CONFIG_SECTION_GLOBAL, "cloud base url", NULL);
1355 + tmp = cloud_base_url ? json_object_new_string(cloud_base_url) : NULL;
1356 + json_object_object_add(msg, "cloud-url", tmp);
1357 +
1358 tmp = json_object_new_boolean(aclk_connected);
1359 json_object_object_add(msg, "online", tmp);
1360
1182 - tmp = json_object_new_string(aclk_use_new_cloud_arch ? "New" : "Legacy");
1361 + tmp = json_object_new_string(aclk_use_new_cloud_arch ? "Protobuf" : "Legacy");
1362 json_object_object_add(msg, "used-cloud-protocol", tmp);
1363
1364 + tmp = json_object_new_int(aclk_rcvd_cloud_msgs);
1365 + json_object_object_add(msg, "received-app-layer-msgs", tmp);
1366 +
1367 + tmp = json_object_new_int(aclk_pubacks_per_conn);
1368 + json_object_object_add(msg, "received-mqtt-pubacks", tmp);
1369 +
1370 + tmp = json_object_new_int(aclk_connection_counter > 0 ? (aclk_connection_counter - 1) : 0);
1371 + json_object_object_add(msg, "reconnect-count", tmp);
1372 +
1373 +#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1374 + grp = json_object_new_array();
1375 +
1376 + RRDHOST *host;
1377 + rrd_rdlock();
1378 + rrdhost_foreach_read(host) {
1379 + json_object *nodeinstance = json_object_new_object();
1380 +
1381 + tmp = json_object_new_string(host->hostname);
1382 + json_object_object_add(nodeinstance, "hostname", tmp);
1383 +
1384 + tmp = json_object_new_string(host->machine_guid);
1385 + json_object_object_add(nodeinstance, "mguid", tmp);
1386 +
1387 + rrdhost_aclk_state_lock(host);
1388 + if (host->aclk_state.claimed_id) {
1389 + tmp = json_object_new_string(host->aclk_state.claimed_id);
1390 + json_object_object_add(nodeinstance, "claimed_id", tmp);
1391 + } else
1392 + json_object_object_add(nodeinstance, "claimed_id", NULL);
1393 + rrdhost_aclk_state_unlock(host);
1394 +
1395 + if (host->node_id == NULL || uuid_is_null(*host->node_id)) {
1396 + json_object_object_add(nodeinstance, "node-id", NULL);
1397 + } else {
1398 + char node_id[GUID_LEN + 1];
1399 + uuid_unparse_lower(*host->node_id, node_id);
1400 + tmp = json_object_new_string(node_id);
1401 + json_object_object_add(nodeinstance, "node-id", tmp);
1402 + }
1403 +
1404 + tmp = json_object_new_int(host->system_info->hops);
1405 + json_object_object_add(nodeinstance, "streaming-hops", tmp);
1406 +
1407 + tmp = json_object_new_string(host == localhost ? "self" : "child");
1408 + json_object_object_add(nodeinstance, "relationship", tmp);
1409 +
1410 + tmp = json_object_new_boolean((host->receiver || host == localhost));
1411 + json_object_object_add(nodeinstance, "streaming-online", tmp);
1412 +
1413 + tmp = json_object_new_object();
1414 + fill_alert_status_for_host_json(tmp, host);
1415 + json_object_object_add(nodeinstance, "alert-sync-status", tmp);
1416 +
1417 + tmp = json_object_new_object();
1418 + fill_chart_status_for_host_json(tmp, host);
1419 + json_object_object_add(nodeinstance, "chart-sync-status", tmp);
1420 +
1421 + json_object_array_add(grp, nodeinstance);
1422 + }
1423 + rrd_unlock();
1424 + json_object_object_add(msg, "node-instances", grp);
1425 +#endif
1426 +
1427 char *str = strdupz(json_object_to_json_string_ext(msg, JSON_C_TO_STRING_PLAIN));
1428 json_object_put(msg);
1429 return str;
database/sqlite/sqlite_aclk_alert.c
+41
@@ -908,3 +908,44 @@ void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
908 UNUSED(host);
909 #endif
910 }
911 +
912 +int get_proto_alert_status(RRDHOST *host, struct proto_alert_status *proto_alert_status)
913 +{
914 + int rc;
915 + struct aclk_database_worker_config *wc = NULL;
916 + wc = (struct aclk_database_worker_config *)host->dbsync_worker;
917 + if (!wc)
918 + return 1;
919 +
920 + proto_alert_status->alert_updates = wc->alert_updates;
921 + proto_alert_status->alerts_batch_id = wc->alerts_batch_id;
922 +
923 + BUFFER *sql = buffer_create(1024);
924 + sqlite3_stmt *res = NULL;
925 +
926 + buffer_sprintf(sql, "SELECT MIN(sequence_id), MAX(sequence_id), " \
927 + "(select MAX(sequence_id) from aclk_alert_%s where date_cloud_ack is not NULL), " \
928 + "(select MAX(sequence_id) from aclk_alert_%s where date_submitted is not NULL) " \
929 + "FROM aclk_alert_%s where date_submitted is null;", wc->uuid_str, wc->uuid_str, wc->uuid_str);
930 +
931 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
932 + if (rc != SQLITE_OK) {
933 + error_report("Failed to prepare statement to get alert log status from the database.");
934 + buffer_free(sql);
935 + return 1;
936 + }
937 +
938 + while (sqlite3_step(res) == SQLITE_ROW) {
939 + proto_alert_status->pending_min_sequence_id = sqlite3_column_bytes(res, 0) > 0 ? (uint64_t) sqlite3_column_int64(res, 0) : 0;
940 + proto_alert_status->pending_max_sequence_id = sqlite3_column_bytes(res, 1) > 0 ? (uint64_t) sqlite3_column_int64(res, 1) : 0;
941 + proto_alert_status->last_acked_sequence_id = sqlite3_column_bytes(res, 2) > 0 ? (uint64_t) sqlite3_column_int64(res, 2) : 0;
942 + proto_alert_status->last_submitted_sequence_id = sqlite3_column_bytes(res, 3) > 0 ? (uint64_t) sqlite3_column_int64(res, 3) : 0;
943 + }
944 +
945 + rc = sqlite3_finalize(res);
946 + if (unlikely(rc != SQLITE_OK))
947 + error_report("Failed to finalize statement to get alert log status from the database, rc = %d", rc);
948 +
949 + buffer_free(sql);
950 + return 0;
951 +}
database/sqlite/sqlite_aclk_alert.h
+10
@@ -5,6 +5,15 @@
5
6 extern sqlite3 *db_meta;
7
8 +struct proto_alert_status {
9 + int alert_updates;
10 + uint64_t alerts_batch_id;
11 + uint64_t last_acked_sequence_id;
12 + uint64_t pending_min_sequence_id;
13 + uint64_t pending_max_sequence_id;
14 + uint64_t last_submitted_sequence_id;
15 +};
16 +
17 int aclk_add_alert_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
18 void aclk_push_alert_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
19 void aclk_send_alarm_health_log(char *node_id);
@@ -16,5 +25,6 @@ void sql_queue_removed_alerts_to_aclk(RRDHOST *host);
25 void sql_process_queue_removed_alerts_to_aclk(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
26 void aclk_push_alert_snapshot_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
27 void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id, uint64_t snapshot_id, uint64_t sequence_id);
28 +int get_proto_alert_status(RRDHOST *host, struct proto_alert_status *proto_alert_status);
29
30 #endif //NETDATA_SQLITE_ACLK_ALERT_H
database/sqlite/sqlite_aclk_chart.c
+85 -1
@@ -1014,6 +1014,91 @@ void aclk_send_dimension_update(RRDDIM *rd)
1014 return;
1015 }
1016
1017 +#define SQL_SEQ_NULL(result, n) sqlite3_column_type(result, n) == SQLITE_NULL ? 0 : sqlite3_column_int64(result, n)
1018 +
1019 +struct aclk_chart_sync_stats *aclk_get_chart_sync_stats(RRDHOST *host)
1020 +{
1021 + struct aclk_chart_sync_stats *aclk_statistics = NULL;
1022 +
1023 + struct aclk_database_worker_config *wc = NULL;
1024 + wc = (struct aclk_database_worker_config *)host->dbsync_worker;
1025 + if (!wc)
1026 + return NULL;
1027 +
1028 + aclk_statistics = callocz(1, sizeof(struct aclk_chart_sync_stats));
1029 +
1030 + aclk_statistics->updates = wc->chart_updates;
1031 + aclk_statistics->batch_id = wc->batch_id;
1032 +
1033 + char host_uuid_fixed[GUID_LEN + 1];
1034 +
1035 + strncpy(host_uuid_fixed, host->machine_guid, GUID_LEN);
1036 + host_uuid_fixed[GUID_LEN] = 0;
1037 +
1038 + host_uuid_fixed[8] = '_';
1039 + host_uuid_fixed[13] = '_';
1040 + host_uuid_fixed[18] = '_';
1041 + host_uuid_fixed[23] = '_';
1042 +
1043 + sqlite3_stmt *res = NULL;
1044 + BUFFER *sql = buffer_create(1024);
1045 + buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s;", host_uuid_fixed);
1046 + buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_submitted IS NULL;", host_uuid_fixed);
1047 + buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_submitted IS NOT NULL;", host_uuid_fixed);
1048 + buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_updated IS NOT NULL;", host_uuid_fixed);
1049 + buffer_sprintf(sql, "SELECT max(date_created), max(date_submitted), max(date_updated), 0 FROM aclk_chart_%s;", host_uuid_fixed);
1050 +
1051 + int rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
1052 + if (rc != SQLITE_OK) {
1053 + buffer_free(sql);
1054 + freez(aclk_statistics);
1055 + return NULL;
1056 + }
1057 +
1058 + rc = sqlite3_step(res);
1059 + if (rc == SQLITE_ROW) {
1060 + aclk_statistics->min_seqid = SQL_SEQ_NULL(res, 0);
1061 + aclk_statistics->max_seqid = SQL_SEQ_NULL(res, 1);
1062 + }
1063 +
1064 + rc = sqlite3_step(res);
1065 + if (rc == SQLITE_ROW) {
1066 + aclk_statistics->min_seqid_pend = SQL_SEQ_NULL(res, 0);
1067 + aclk_statistics->max_seqid_pend = SQL_SEQ_NULL(res, 1);
1068 + }
1069 +
1070 + rc = sqlite3_step(res);
1071 + if (rc == SQLITE_ROW) {
1072 + aclk_statistics->min_seqid_sent = SQL_SEQ_NULL(res, 0);
1073 + aclk_statistics->max_seqid_sent = SQL_SEQ_NULL(res, 1);
1074 + }
1075 +
1076 + rc = sqlite3_step(res);
1077 + if (rc == SQLITE_ROW) {
1078 + aclk_statistics->min_seqid_ack = SQL_SEQ_NULL(res, 0);
1079 + aclk_statistics->max_seqid_ack = SQL_SEQ_NULL(res, 1);
1080 + }
1081 +
1082 + rc = sqlite3_step(res);
1083 + if (rc == SQLITE_ROW) {
1084 + aclk_statistics->min_seqid_ack = SQL_SEQ_NULL(res, 0);
1085 + aclk_statistics->max_seqid_ack = SQL_SEQ_NULL(res, 1);
1086 + }
1087 +
1088 + rc = sqlite3_step(res);
1089 + if (rc == SQLITE_ROW) {
1090 + aclk_statistics->max_date_created = (time_t) SQL_SEQ_NULL(res, 0);
1091 + aclk_statistics->max_date_submitted = (time_t) SQL_SEQ_NULL(res, 1);
1092 + aclk_statistics->max_date_ack = (time_t) SQL_SEQ_NULL(res, 2);
1093 + }
1094 +
1095 + rc = sqlite3_finalize(res);
1096 + if (unlikely(rc != SQLITE_OK))
1097 + error_report("Failed to finalize statement when fetching aclk sync statistics, rc = %d", rc);
1098 +
1099 + buffer_free(sql);
1100 + return aclk_statistics;
1101 +}
1102 #endif //ENABLE_NEW_CLOUD_PROTOCOL
1103
1104 // ST is read locked
@@ -1035,4 +1120,3 @@ int queue_chart_to_aclk(RRDSET *st)
1120 st, ACLK_DATABASE_ADD_CHART);
1121 #endif
1122 }
1038 -
database/sqlite/sqlite_aclk_chart.h
+17
@@ -16,6 +16,22 @@ extern sqlite3 *db_meta;
16 #define RRDSET_MINIMUM_LIVE_MULTIPLIER (1.5)
17 #endif
18
19 +struct aclk_chart_sync_stats {
20 + int updates;
21 + uint64_t batch_id;
22 + uint64_t min_seqid;
23 + uint64_t max_seqid;
24 + uint64_t min_seqid_pend;
25 + uint64_t max_seqid_pend;
26 + uint64_t min_seqid_sent;
27 + uint64_t max_seqid_sent;
28 + uint64_t min_seqid_ack;
29 + uint64_t max_seqid_ack;
30 + time_t max_date_created;
31 + time_t max_date_submitted;
32 + time_t max_date_ack;
33 +};
34 +
35 extern int queue_chart_to_aclk(RRDSET *st);
36 extern int queue_dimension_to_aclk(RRDDIM *rd);
37 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_
51 void aclk_process_dimension_deletion(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
52 uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc);
53 void aclk_send_dimension_update(RRDDIM *rd);
54 +struct aclk_chart_sync_stats *aclk_get_chart_sync_stats(RRDHOST *host);
55 #endif //NETDATA_SQLITE_ACLK_CHART_H
web/api/netdata-swagger.json
+9 -10
@@ -2085,17 +2085,16 @@
2085 "type": "boolean",
2086 "description": "Describes whether this agent is capable of connection to the Cloud. False means agent has been built without ACLK component either on purpose (user choice) or due to missing dependency."
2087 },
2088 - "aclk-implementation": {
2089 - "type": "string",
2090 - "description": "Describes which ACLK implementation is currently used.",
2091 - "enum": [
2092 - "Next Generation",
2093 - "Legacy"
2094 - ]
2088 + "aclk-version": {
2089 + "type": "integer",
2090 + "description": "Describes which ACLK version is currently used."
2091 },
2096 - "new-cloud-protocol-supported": {
2097 - "type": "boolean",
2098 - "description": "Informs about new protobuf based Cloud/Agent protocol support of this agent. If false agent has to be compiled with protobuf and protoc available."
2092 + "protocols-supported": {
2093 + "type": "array",
2094 + "description": "List of supported protocols for communication with Cloud.",
2095 + "items": {
2096 + "type": "string"
2097 + }
2098 },
2099 "agent-claimed": {
2100 "type": "boolean",
web/api/netdata-swagger.yaml
+8 -10
@@ -1628,16 +1628,14 @@ components:
1628 type: string
1629 description: Describes whether this agent is capable of connection to the Cloud.
1630 False means agent has been built without ACLK component either on purpose (user choice) or due to missing dependency.
1631 - aclk-implementation:
1632 - type: string
1633 - description: Describes which ACLK implementation is currently used.
1634 - enum:
1635 - - Next Generation
1636 - - Legacy
1637 - new-cloud-protocol-supported:
1638 - type: boolean
1639 - description: Informs about new protobuf based Cloud/Agent protocol support of this agent.
1640 - If false agent has to be compiled with protobuf and protoc available.
1631 + aclk-version:
1632 + type: integer
1633 + description: Describes which ACLK version is currently used.
1634 + protocols-supported:
1635 + type: array
1636 + description: List of supported protocols for communication with Cloud.
1637 + items:
1638 + type: string
1639 agent-claimed:
1640 type: boolean
1641 description: Informs whether this agent has been added to a space in the cloud (User has to perform claiming).