sqlite_aclk_chart.c 45 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #include "sqlite_functions.h"
  3. #include "sqlite_aclk_chart.h"
  4. #if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
  5. #include "../../aclk/aclk_charts_api.h"
  6. #include "../../aclk/aclk.h"
  7. static inline int
  8. sql_queue_chart_payload(struct aclk_database_worker_config *wc, void *data, enum aclk_database_opcode opcode)
  9. {
  10. int rc;
  11. if (unlikely(!wc))
  12. return 1;
  13. struct aclk_database_cmd cmd;
  14. memset(&cmd, 0, sizeof(cmd));
  15. cmd.opcode = opcode;
  16. cmd.data = data;
  17. rc = aclk_database_enq_cmd_noblock(wc, &cmd);
  18. return rc;
  19. }
  20. static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size)
  21. {
  22. static __thread sqlite3_stmt *res = NULL;
  23. int rc;
  24. int send_status = 0;
  25. if (unlikely(!res)) {
  26. char sql[ACLK_SYNC_QUERY_SIZE];
  27. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "
  28. "WHERE acl.unique_id = acp.unique_id AND acl.uuid = @uuid AND acp.payload = @payload;",
  29. uuid_str, uuid_str);
  30. rc = prepare_statement(db_meta, sql, &res);
  31. if (rc != SQLITE_OK) {
  32. error_report("Failed to prepare statement to check payload data");
  33. return 0;
  34. }
  35. }
  36. rc = sqlite3_bind_blob(res, 1, uuid, sizeof(*uuid), SQLITE_STATIC);
  37. if (unlikely(rc != SQLITE_OK))
  38. goto bind_fail;
  39. rc = sqlite3_bind_blob(res, 2, payload, payload_size, SQLITE_STATIC);
  40. if (unlikely(rc != SQLITE_OK))
  41. goto bind_fail;
  42. while (sqlite3_step(res) == SQLITE_ROW) {
  43. send_status = sqlite3_column_int(res, 0);
  44. }
  45. bind_fail:
  46. if (unlikely(sqlite3_reset(res) != SQLITE_OK))
  47. error_report("Failed to reset statement in check payload, rc = %d", rc);
  48. return send_status;
  49. }
  50. static int aclk_add_chart_payload(struct aclk_database_worker_config *wc, uuid_t *uuid, char *claim_id,
  51. ACLK_PAYLOAD_TYPE payload_type, void *payload, size_t payload_size, int *send_status)
  52. {
  53. static __thread sqlite3_stmt *res_chart = NULL;
  54. int rc;
  55. rc = payload_sent(wc->uuid_str, uuid, payload, payload_size);
  56. if (send_status)
  57. *send_status = rc;
  58. if (rc == 1)
  59. return 0;
  60. if (unlikely(!res_chart)) {
  61. char sql[ACLK_SYNC_QUERY_SIZE];
  62. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,
  63. "INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \
  64. "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);
  65. rc = prepare_statement(db_meta, sql, &res_chart);
  66. if (rc != SQLITE_OK) {
  67. error_report("Failed to prepare statement to store chart payload data");
  68. return 1;
  69. }
  70. }
  71. uuid_t unique_uuid;
  72. uuid_generate(unique_uuid);
  73. uuid_t claim_uuid;
  74. if (uuid_parse(claim_id, claim_uuid))
  75. return 1;
  76. rc = sqlite3_bind_blob(res_chart, 1, &unique_uuid, sizeof(unique_uuid), SQLITE_STATIC);
  77. if (unlikely(rc != SQLITE_OK))
  78. goto bind_fail;
  79. rc = sqlite3_bind_blob(res_chart, 2, uuid, sizeof(*uuid), SQLITE_STATIC);
  80. if (unlikely(rc != SQLITE_OK))
  81. goto bind_fail;
  82. rc = sqlite3_bind_blob(res_chart, 3, &claim_uuid, sizeof(claim_uuid), SQLITE_STATIC);
  83. if (unlikely(rc != SQLITE_OK))
  84. goto bind_fail;
  85. rc = sqlite3_bind_int(res_chart, 4, payload_type);
  86. if (unlikely(rc != SQLITE_OK))
  87. goto bind_fail;
  88. rc = sqlite3_bind_blob(res_chart, 5, payload, payload_size, SQLITE_STATIC);
  89. if (unlikely(rc != SQLITE_OK))
  90. goto bind_fail;
  91. rc = execute_insert(res_chart);
  92. if (unlikely(rc != SQLITE_DONE))
  93. error_report("Failed store chart payload event, rc = %d", rc);
  94. else {
  95. wc->chart_payload_count++;
  96. time_t now = now_realtime_sec();
  97. if (wc->rotation_after > now && wc->rotation_after < now + ACLK_DATABASE_ROTATION_DELAY)
  98. wc->rotation_after = now + ACLK_DATABASE_ROTATION_DELAY;
  99. }
  100. bind_fail:
  101. if (unlikely(sqlite3_reset(res_chart) != SQLITE_OK))
  102. error_report("Failed to reset statement in store chart payload, rc = %d", rc);
  103. return (rc != SQLITE_DONE);
  104. }
  105. int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  106. {
  107. int rc = 0;
  108. CHECK_SQLITE_CONNECTION(db_meta);
  109. char *claim_id = is_agent_claimed();
  110. RRDSET *st = cmd.data;
  111. if (likely(claim_id)) {
  112. struct chart_instance_updated chart_payload;
  113. memset(&chart_payload, 0, sizeof(chart_payload));
  114. chart_payload.config_hash = get_str_from_uuid(&st->state->hash_id);
  115. chart_payload.update_every = st->update_every;
  116. chart_payload.memory_mode = st->rrd_memory_mode;
  117. chart_payload.name = (char *)st->name;
  118. chart_payload.node_id = wc->node_id;
  119. chart_payload.claim_id = claim_id;
  120. chart_payload.id = strdupz(st->id);
  121. struct label_index *labels = &st->state->labels;
  122. netdata_rwlock_wrlock(&labels->labels_rwlock);
  123. struct label *label_list = labels->head;
  124. struct label *chart_label = NULL;
  125. while (label_list) {
  126. chart_label = add_label_to_list(chart_label, label_list->key, label_list->value, label_list->label_source);
  127. label_list = label_list->next;
  128. }
  129. netdata_rwlock_unlock(&labels->labels_rwlock);
  130. chart_payload.label_head = chart_label;
  131. size_t size;
  132. char *payload = generate_chart_instance_updated(&size, &chart_payload);
  133. if (likely(payload))
  134. rc = aclk_add_chart_payload(wc, st->chart_uuid, claim_id, ACLK_PAYLOAD_CHART, (void *) payload, size, NULL);
  135. freez(payload);
  136. chart_instance_updated_destroy(&chart_payload);
  137. }
  138. return rc;
  139. }
  140. static inline int aclk_upd_dimension_event(struct aclk_database_worker_config *wc, char *claim_id, uuid_t *dim_uuid,
  141. const char *dim_id, const char *dim_name, const char *chart_type_id, time_t first_time, time_t last_time,
  142. int *send_status)
  143. {
  144. int rc = 0;
  145. size_t size;
  146. if (unlikely(!dim_uuid || !dim_id || !dim_name || !chart_type_id))
  147. return 0;
  148. struct chart_dimension_updated dim_payload;
  149. memset(&dim_payload, 0, sizeof(dim_payload));
  150. #ifdef NETDATA_INTERNAL_CHECKS
  151. if (!first_time)
  152. info("Host %s (node %s) deleting dimension id=[%s] name=[%s] chart=[%s]",
  153. wc->host_guid, wc->node_id, dim_id, dim_name, chart_type_id);
  154. if (last_time)
  155. info("Host %s (node %s) stopped collecting dimension id=[%s] name=[%s] chart=[%s] %ld seconds ago at %ld",
  156. wc->host_guid, wc->node_id, dim_id, dim_name, chart_type_id, now_realtime_sec() - last_time, last_time);
  157. #endif
  158. dim_payload.node_id = wc->node_id;
  159. dim_payload.claim_id = claim_id;
  160. dim_payload.name = dim_name;
  161. dim_payload.id = dim_id;
  162. dim_payload.chart_id = chart_type_id;
  163. dim_payload.created_at.tv_sec = first_time;
  164. dim_payload.last_timestamp.tv_sec = last_time;
  165. char *payload = generate_chart_dimension_updated(&size, &dim_payload);
  166. if (likely(payload))
  167. rc = aclk_add_chart_payload(wc, dim_uuid, claim_id, ACLK_PAYLOAD_DIMENSION, (void *)payload, size, send_status);
  168. freez(payload);
  169. return rc;
  170. }
  171. void aclk_process_dimension_deletion(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  172. {
  173. int rc = 0;
  174. sqlite3_stmt *res = NULL;
  175. if (!aclk_use_new_cloud_arch || !aclk_connected)
  176. return;
  177. if (unlikely(!db_meta))
  178. return;
  179. uuid_t host_id;
  180. if (uuid_parse(wc->host_guid, host_id))
  181. return;
  182. char *claim_id = is_agent_claimed();
  183. if (!claim_id)
  184. return;
  185. rc = sqlite3_prepare_v2(
  186. db_meta,
  187. "DELETE FROM dimension_delete where host_id = @host_id "
  188. "RETURNING dimension_id, dimension_name, chart_type_id, dim_id LIMIT 10;",
  189. -1,
  190. &res,
  191. 0);
  192. if (rc != SQLITE_OK) {
  193. error_report("Failed to prepare statement when trying to delete dimension deletes");
  194. freez(claim_id);
  195. return;
  196. }
  197. rc = sqlite3_bind_blob(res, 1, &host_id, sizeof(host_id), SQLITE_STATIC);
  198. if (unlikely(rc != SQLITE_OK))
  199. goto bind_fail;
  200. unsigned count = 0;
  201. while (sqlite3_step(res) == SQLITE_ROW) {
  202. (void) aclk_upd_dimension_event(
  203. wc,
  204. claim_id,
  205. (uuid_t *)sqlite3_column_text(res, 3),
  206. (const char *)sqlite3_column_text(res, 0),
  207. (const char *)sqlite3_column_text(res, 1),
  208. (const char *)sqlite3_column_text(res, 2),
  209. 0,
  210. 0,
  211. NULL);
  212. count++;
  213. }
  214. if (count) {
  215. memset(&cmd, 0, sizeof(cmd));
  216. cmd.opcode = ACLK_DATABASE_DIM_DELETION;
  217. if (aclk_database_enq_cmd_noblock(wc, &cmd))
  218. info("Failed to queue a dimension deletion message");
  219. }
  220. bind_fail:
  221. rc = sqlite3_finalize(res);
  222. if (unlikely(rc != SQLITE_OK))
  223. error_report("Failed to finalize statement when adding dimension deletion events, rc = %d", rc);
  224. freez(claim_id);
  225. return;
  226. }
  227. int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  228. {
  229. int rc = 0;
  230. CHECK_SQLITE_CONNECTION(db_meta);
  231. char *claim_id = is_agent_claimed();
  232. RRDDIM *rd = cmd.data;
  233. if (likely(claim_id)) {
  234. int send_status = 0;
  235. time_t now = now_realtime_sec();
  236. time_t first_t = rd->state->query_ops.oldest_time(rd);
  237. time_t last_t = rd->state->query_ops.latest_time(rd);
  238. int live = ((now - last_t) < (RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every));
  239. rc = aclk_upd_dimension_event(
  240. wc,
  241. claim_id,
  242. &rd->state->metric_uuid,
  243. rd->id,
  244. rd->name,
  245. rd->rrdset->id,
  246. first_t,
  247. live ? 0 : last_t,
  248. &send_status);
  249. if (!send_status)
  250. rd->state->aclk_live_status = live;
  251. freez(claim_id);
  252. }
  253. rrddim_flag_clear(rd, RRDDIM_FLAG_ACLK);
  254. return rc;
  255. }
  256. void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  257. {
  258. int rc;
  259. wc->chart_pending = 0;
  260. if (unlikely(!wc->chart_updates)) {
  261. log_access(
  262. "ACLK STA [%s (%s)]: Ignoring chart push event, updates have been turned off for this node.",
  263. wc->node_id,
  264. wc->host ? wc->host->hostname : "N/A");
  265. return;
  266. }
  267. char *claim_id = is_agent_claimed();
  268. if (unlikely(!claim_id))
  269. return;
  270. uuid_t claim_uuid;
  271. if (uuid_parse(claim_id, claim_uuid))
  272. return;
  273. int limit = cmd.count > 0 ? cmd.count : 1;
  274. uint64_t first_sequence;
  275. uint64_t last_sequence;
  276. time_t last_timestamp = 0;
  277. char sql[ACLK_SYNC_QUERY_SIZE];
  278. static __thread sqlite3_stmt *res = NULL;
  279. if (unlikely(!res)) {
  280. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid " \
  281. "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \
  282. "WHERE ac.date_submitted IS NULL AND ac.unique_id = acp.unique_id AND ac.update_count > 0 " \
  283. "AND acp.claim_id = @claim_id ORDER BY ac.sequence_id ASC LIMIT %d;", wc->uuid_str, wc->uuid_str, limit);
  284. rc = prepare_statement(db_meta, sql, &res);
  285. if (rc != SQLITE_OK) {
  286. error_report("Failed to prepare statement when trying to send a chart update via ACLK");
  287. freez(claim_id);
  288. return;
  289. }
  290. }
  291. rc = sqlite3_bind_blob(res, 1, claim_uuid, sizeof(claim_uuid), SQLITE_STATIC);
  292. if (unlikely(rc != SQLITE_OK))
  293. goto bind_fail;
  294. char **payload_list = callocz(limit + 1, sizeof(char *));
  295. size_t *payload_list_size = callocz(limit + 1, sizeof(size_t));
  296. size_t *payload_list_max_size = callocz(limit + 1, sizeof(size_t));
  297. struct aclk_message_position *position_list = callocz(limit + 1, sizeof(*position_list));
  298. int *is_dim = callocz(limit + 1, sizeof(*is_dim));
  299. int loop = cmd.param1;
  300. uint64_t start_sequence_id = wc->chart_sequence_id;
  301. while (loop > 0) {
  302. uint64_t previous_sequence_id = wc->chart_sequence_id;
  303. int count = 0;
  304. first_sequence = 0;
  305. last_sequence = 0;
  306. while (count < limit && sqlite3_step(res) == SQLITE_ROW) {
  307. size_t payload_size = sqlite3_column_bytes(res, 1);
  308. if (payload_list_max_size[count] < payload_size) {
  309. freez(payload_list[count]);
  310. payload_list_max_size[count] = payload_size;
  311. payload_list[count] = mallocz(payload_size);
  312. }
  313. payload_list_size[count] = payload_size;
  314. memcpy(payload_list[count], sqlite3_column_blob(res, 1), payload_size);
  315. position_list[count].sequence_id = (uint64_t)sqlite3_column_int64(res, 0);
  316. position_list[count].previous_sequence_id = previous_sequence_id;
  317. position_list[count].seq_id_creation_time.tv_sec = sqlite3_column_int64(res, 2);
  318. position_list[count].seq_id_creation_time.tv_usec = 0;
  319. if (!first_sequence)
  320. first_sequence = position_list[count].sequence_id;
  321. last_sequence = position_list[count].sequence_id;
  322. last_timestamp = position_list[count].seq_id_creation_time.tv_sec;
  323. previous_sequence_id = last_sequence;
  324. is_dim[count] = sqlite3_column_int(res, 3) > 0;
  325. count++;
  326. if (wc->chart_payload_count)
  327. wc->chart_payload_count--;
  328. }
  329. freez(payload_list[count]);
  330. payload_list_max_size[count] = 0;
  331. payload_list[count] = NULL;
  332. rc = sqlite3_reset(res);
  333. if (unlikely(rc != SQLITE_OK))
  334. error_report("Failed to reset statement when pushing chart events, rc = %d", rc);
  335. if (likely(first_sequence)) {
  336. db_lock();
  337. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=strftime('%%s','now') "
  338. "WHERE date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",
  339. wc->uuid_str, first_sequence, last_sequence);
  340. db_execute(sql);
  341. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "
  342. " SELECT uuid, unique_id, date_submitted FROM aclk_chart_%s s "
  343. " WHERE date_submitted IS NOT NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64
  344. " ;",
  345. wc->uuid_str, wc->uuid_str, first_sequence, last_sequence);
  346. db_execute(sql);
  347. db_unlock();
  348. aclk_chart_inst_and_dim_update(payload_list, payload_list_size, is_dim, position_list, wc->batch_id);
  349. log_access(
  350. "ACLK RES [%s (%s)]: CHARTS SENT from %" PRIu64 " to %" PRIu64 " batch=%" PRIu64,
  351. wc->node_id,
  352. wc->host ? wc->host->hostname : "N/A",
  353. first_sequence,
  354. last_sequence,
  355. wc->batch_id);
  356. wc->chart_sequence_id = last_sequence;
  357. wc->chart_timestamp = last_timestamp;
  358. } else
  359. break;
  360. --loop;
  361. }
  362. if (start_sequence_id != wc->chart_sequence_id) {
  363. time_t now = now_realtime_sec();
  364. if (wc->rotation_after > now && wc->rotation_after < now + ACLK_DATABASE_ROTATION_DELAY)
  365. wc->rotation_after = now + ACLK_DATABASE_ROTATION_DELAY;
  366. } else {
  367. wc->chart_payload_count = sql_get_pending_count(wc);
  368. if (!wc->chart_payload_count)
  369. log_access(
  370. "ACLK STA [%s (%s)]: Sync of charts and dimensions done in %ld seconds.",
  371. wc->node_id,
  372. wc->host ? wc->host->hostname : "N/A",
  373. now_realtime_sec() - wc->startup_time);
  374. }
  375. for (int i = 0; i <= limit; ++i)
  376. freez(payload_list[i]);
  377. freez(payload_list);
  378. freez(payload_list_size);
  379. freez(payload_list_max_size);
  380. freez(position_list);
  381. freez(is_dim);
  382. bind_fail:
  383. rc = sqlite3_reset(res);
  384. if (unlikely(rc != SQLITE_OK))
  385. error_report("Failed to reset statement when pushing chart events, rc = %d", rc);
  386. freez(claim_id);
  387. return;
  388. }
  389. // Push one chart config to the cloud
  390. int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  391. {
  392. UNUSED(wc);
  393. CHECK_SQLITE_CONNECTION(db_meta);
  394. sqlite3_stmt *res = NULL;
  395. int rc = 0;
  396. char *hash_id = (char *) cmd.data_param;
  397. uuid_t hash_uuid;
  398. rc = uuid_parse(hash_id, hash_uuid);
  399. if (unlikely(rc)) {
  400. freez((char *) cmd.data_param);
  401. return 1;
  402. }
  403. BUFFER *sql = buffer_create(1024);
  404. buffer_sprintf(sql, "SELECT type, family, context, title, priority, plugin, module, unit, chart_type " \
  405. "FROM chart_hash WHERE hash_id = @hash_id;");
  406. rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
  407. if (rc != SQLITE_OK) {
  408. error_report("Failed to prepare statement when trying to fetch a chart hash configuration");
  409. goto fail;
  410. }
  411. rc = sqlite3_bind_blob(res, 1, &hash_uuid , sizeof(hash_uuid), SQLITE_STATIC);
  412. if (unlikely(rc != SQLITE_OK))
  413. goto bind_fail;
  414. struct chart_config_updated chart_config;
  415. chart_config.config_hash = NULL;
  416. while (sqlite3_step(res) == SQLITE_ROW) {
  417. chart_config.type = strdupz((char *)sqlite3_column_text(res, 0));
  418. chart_config.family = strdupz((char *)sqlite3_column_text(res, 1));
  419. chart_config.context = strdupz((char *)sqlite3_column_text(res, 2));
  420. chart_config.title = strdupz((char *)sqlite3_column_text(res, 3));
  421. chart_config.priority = sqlite3_column_int64(res, 4);
  422. chart_config.plugin = strdupz((char *)sqlite3_column_text(res, 5));
  423. chart_config.module = sqlite3_column_bytes(res, 6) > 0 ? strdupz((char *)sqlite3_column_text(res, 6)) : NULL;
  424. chart_config.chart_type = (RRDSET_TYPE) sqlite3_column_int(res,8);
  425. chart_config.units = strdupz((char *)sqlite3_column_text(res, 7));
  426. chart_config.config_hash = strdupz(hash_id);
  427. }
  428. if (likely(chart_config.config_hash)) {
  429. log_access(
  430. "ACLK REQ [%s (%s)]: Sending chart config for %s.",
  431. wc->node_id,
  432. wc->host ? wc->host->hostname : "N/A",
  433. hash_id);
  434. aclk_chart_config_updated(&chart_config, 1);
  435. destroy_chart_config_updated(&chart_config);
  436. } else
  437. log_access(
  438. "ACLK STA [%s (%s)]: Chart config for %s not found.",
  439. wc->node_id,
  440. wc->host ? wc->host->hostname : "N/A",
  441. hash_id);
  442. bind_fail:
  443. rc = sqlite3_finalize(res);
  444. if (unlikely(rc != SQLITE_OK))
  445. error_report("Failed to reset statement when pushing chart config hash, rc = %d", rc);
  446. fail:
  447. freez((char *)cmd.data_param);
  448. buffer_free(sql);
  449. return rc;
  450. }
  451. void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  452. {
  453. int rc;
  454. sqlite3_stmt *res = NULL;
  455. char sql[ACLK_SYNC_QUERY_SIZE];
  456. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"UPDATE aclk_chart_%s SET date_updated=strftime('%%s','now') WHERE sequence_id <= @sequence_id "
  457. "AND date_submitted IS NOT NULL AND date_updated IS NULL;", wc->uuid_str);
  458. rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
  459. if (rc != SQLITE_OK) {
  460. error_report("Failed to prepare statement to ack chart sequence ids");
  461. return;
  462. }
  463. rc = sqlite3_bind_int64(res, 1, (uint64_t) cmd.param1);
  464. if (unlikely(rc != SQLITE_OK))
  465. goto bind_fail;
  466. rc = execute_insert(res);
  467. if (rc != SQLITE_DONE)
  468. error_report("Failed to ACK sequence id, rc = %d", rc);
  469. else
  470. log_access(
  471. "ACLK STA [%s (%s)]: CHARTS ACKNOWLEDGED in the database upto %" PRIu64,
  472. wc->node_id,
  473. wc->host ? wc->host->hostname : "N/A",
  474. cmd.param1);
  475. bind_fail:
  476. if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
  477. error_report("Failed to finalize statement to ACK older sequence ids, rc = %d", rc);
  478. return;
  479. }
  480. void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  481. {
  482. BUFFER *sql = buffer_create(1024);
  483. buffer_sprintf(
  484. sql,
  485. "UPDATE aclk_chart_%s SET status = NULL, date_submitted = NULL WHERE sequence_id >= %" PRIu64 ";",
  486. wc->uuid_str,
  487. cmd.param1);
  488. db_execute(buffer_tostring(sql));
  489. if (cmd.param1 == 1) {
  490. buffer_flush(sql);
  491. log_access("ACLK REQ [%s (%s)]: Received chart full resync.", wc->node_id, wc->host ? wc->host->hostname : "N/A");
  492. buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s; DELETE FROM aclk_chart_%s; " \
  493. "DELETE FROM aclk_chart_latest_%s;", wc->uuid_str, wc->uuid_str, wc->uuid_str);
  494. db_lock();
  495. db_execute("BEGIN TRANSACTION;");
  496. db_execute(buffer_tostring(sql));
  497. db_execute("COMMIT TRANSACTION;");
  498. db_unlock();
  499. wc->chart_sequence_id = 0;
  500. wc->chart_timestamp = 0;
  501. wc->chart_payload_count = 0;
  502. RRDHOST *host = wc->host;
  503. if (likely(host)) {
  504. rrdhost_rdlock(host);
  505. RRDSET *st;
  506. rrdset_foreach_read(st, host)
  507. {
  508. rrdset_rdlock(st);
  509. rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
  510. RRDDIM *rd;
  511. rrddim_foreach_read(rd, st)
  512. {
  513. rd->state->aclk_live_status = (rd->state->aclk_live_status == 0);
  514. }
  515. rrdset_unlock(st);
  516. }
  517. rrdhost_unlock(host);
  518. } else
  519. error_report("ACLK synchronization thread for %s is not linked to HOST", wc->host_guid);
  520. } else {
  521. log_access(
  522. "ACLK STA [%s (%s)]: Restarting chart sync from sequence %" PRIu64,
  523. wc->node_id,
  524. wc->host ? wc->host->hostname : "N/A",
  525. cmd.param1);
  526. wc->chart_payload_count = sql_get_pending_count(wc);
  527. sql_get_last_chart_sequence(wc);
  528. }
  529. buffer_free(sql);
  530. wc->chart_updates = 1;
  531. return;
  532. }
  533. //
  534. // Functions called directly from ACLK threads and will queue commands
  535. //
  536. void aclk_get_chart_config(char **hash_id)
  537. {
  538. struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *)localhost->dbsync_worker;
  539. if (unlikely(!wc || !hash_id))
  540. return;
  541. struct aclk_database_cmd cmd;
  542. memset(&cmd, 0, sizeof(cmd));
  543. cmd.opcode = ACLK_DATABASE_PUSH_CHART_CONFIG;
  544. for (int i = 0; hash_id[i]; ++i) {
  545. // TODO: Verify that we have a valid hash_id
  546. log_access(
  547. "ACLK REQ [%s (%s)]: Request %d for chart config with hash %s received.",
  548. wc->node_id,
  549. wc->host ? wc->host->hostname : "N/A",
  550. i,
  551. hash_id[i]);
  552. cmd.data_param = (void *)strdupz(hash_id[i]);
  553. aclk_database_enq_cmd(wc, &cmd);
  554. }
  555. return;
  556. }
  557. // Send a command to a node_id
  558. // Need to discover the thread that will handle the request
  559. // if thread not in active hosts, then try to find in the queue
  560. static void aclk_submit_param_command(char *node_id, enum aclk_database_opcode aclk_command, uint64_t param)
  561. {
  562. if (unlikely(!node_id))
  563. return;
  564. struct aclk_database_worker_config *wc = NULL;
  565. struct aclk_database_cmd cmd;
  566. memset(&cmd, 0, sizeof(cmd));
  567. cmd.opcode = aclk_command;
  568. cmd.param1 = param;
  569. rrd_rdlock();
  570. RRDHOST *host = find_host_by_node_id(node_id);
  571. if (likely(host))
  572. wc = (struct aclk_database_worker_config *)host->dbsync_worker;
  573. rrd_unlock();
  574. if (wc)
  575. aclk_database_enq_cmd(wc, &cmd);
  576. else {
  577. if (aclk_worker_enq_cmd(node_id, &cmd))
  578. log_access("ACLK STA [%s (N/A)]: ACLK synchronization thread is not active.", node_id);
  579. }
  580. return;
  581. }
  582. void aclk_ack_chart_sequence_id(char *node_id, uint64_t last_sequence_id)
  583. {
  584. if (unlikely(!node_id))
  585. return;
  586. char *hostname = get_hostname_by_node_id(node_id);
  587. log_access("ACLK REQ [%s (%s)]: CHARTS ACKNOWLEDGED upto %" PRIu64, node_id, hostname ? hostname : "N/A",
  588. last_sequence_id);
  589. freez(hostname);
  590. aclk_submit_param_command(node_id, ACLK_DATABASE_CHART_ACK, last_sequence_id);
  591. return;
  592. }
  593. // Start streaming charts / dimensions for node_id
  594. void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at, uint64_t batch_id)
  595. {
  596. UNUSED(created_at);
  597. if (unlikely(!node_id))
  598. return;
  599. // log_access("ACLK REQ [%s (N/A)]: CHARTS STREAM from %"PRIu64" t=%ld batch=%"PRIu64, node_id,
  600. // sequence_id, created_at, batch_id);
  601. uuid_t node_uuid;
  602. if (uuid_parse(node_id, node_uuid)) {
  603. log_access("ACLK REQ [%s (N/A)]: CHARTS STREAM ignored, invalid node id", node_id);
  604. return;
  605. }
  606. struct aclk_database_worker_config *wc = NULL;
  607. rrd_rdlock();
  608. RRDHOST *host = localhost;
  609. while(host) {
  610. if (host->node_id && !(uuid_compare(*host->node_id, node_uuid))) {
  611. rrd_unlock();
  612. wc = (struct aclk_database_worker_config *)host->dbsync_worker ?
  613. (struct aclk_database_worker_config *)host->dbsync_worker :
  614. (struct aclk_database_worker_config *)find_inactive_wc_by_node_id(node_id);
  615. if (likely(wc)) {
  616. wc->chart_reset_count++;
  617. __sync_synchronize();
  618. wc->chart_updates = 0;
  619. wc->batch_id = batch_id;
  620. __sync_synchronize();
  621. wc->batch_created = now_realtime_sec();
  622. log_access(
  623. "ACLK REQ [%s (%s)]: CHARTS STREAM from %" PRIu64 " t=%ld resets=%d",
  624. wc->node_id,
  625. wc->host ? wc->host->hostname : "N/A",
  626. wc->chart_sequence_id,
  627. wc->chart_timestamp,
  628. wc->chart_reset_count);
  629. if (sequence_id > wc->chart_sequence_id || wc->chart_reset_count > 10) {
  630. log_access(
  631. "ACLK RES [%s (%s)]: CHARTS FULL RESYNC REQUEST "
  632. "remote_seq=%" PRIu64 " local_seq=%" PRIu64 " resets=%d ",
  633. wc->node_id,
  634. wc->host ? wc->host->hostname : "N/A",
  635. sequence_id,
  636. wc->chart_sequence_id,
  637. wc->chart_reset_count);
  638. chart_reset_t chart_reset;
  639. chart_reset.claim_id = is_agent_claimed();
  640. if (chart_reset.claim_id) {
  641. chart_reset.node_id = node_id;
  642. chart_reset.reason = SEQ_ID_NOT_EXISTS;
  643. aclk_chart_reset(chart_reset);
  644. freez(chart_reset.claim_id);
  645. wc->chart_reset_count = -1;
  646. }
  647. return;
  648. } else {
  649. struct aclk_database_cmd cmd;
  650. memset(&cmd, 0, sizeof(cmd));
  651. // TODO: handle timestamp
  652. if (sequence_id < wc->chart_sequence_id ||
  653. !sequence_id) { // || created_at != wc->chart_timestamp) {
  654. log_access(
  655. "ACLK REQ [%s (%s)]: CHART RESET from %" PRIu64 " t=%ld batch=%" PRIu64,
  656. wc->node_id,
  657. wc->host ? wc->host->hostname : "N/A",
  658. wc->chart_sequence_id,
  659. wc->chart_timestamp,
  660. wc->batch_id);
  661. cmd.opcode = ACLK_DATABASE_RESET_CHART;
  662. cmd.param1 = sequence_id + 1;
  663. cmd.completion = NULL;
  664. aclk_database_enq_cmd(wc, &cmd);
  665. } else {
  666. // log_access(
  667. // "ACLK RES [%s (%s)]: CHARTS STREAM from %" PRIu64
  668. // " t=%ld resets=%d",
  669. // wc->node_id,
  670. // wc->host ? wc->host->hostname : "N/A",
  671. // wc->chart_sequence_id,
  672. // wc->chart_timestamp,
  673. // wc->chart_reset_count);
  674. wc->chart_reset_count = 0;
  675. wc->chart_updates = 1;
  676. }
  677. }
  678. } else
  679. log_access("ACLK STA [%s (N/A)]: ACLK synchronization thread is not active.", node_id);
  680. return;
  681. }
  682. host = host->next;
  683. }
  684. rrd_unlock();
  685. return;
  686. }
  687. #define SQL_SELECT_HOST_MEMORY_MODE "SELECT memory_mode FROM chart WHERE host_id = @host_id LIMIT 1;"
  688. static RRD_MEMORY_MODE sql_get_host_memory_mode(uuid_t *host_id)
  689. {
  690. int rc;
  691. RRD_MEMORY_MODE memory_mode = RRD_MEMORY_MODE_RAM;
  692. sqlite3_stmt *res = NULL;
  693. rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_MEMORY_MODE, -1, &res, 0);
  694. if (unlikely(rc != SQLITE_OK)) {
  695. error_report("Failed to prepare statement to read host memory mode");
  696. return memory_mode;
  697. }
  698. rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
  699. if (unlikely(rc != SQLITE_OK)) {
  700. error_report("Failed to bind host parameter to fetch host memory mode");
  701. goto failed;
  702. }
  703. while (sqlite3_step(res) == SQLITE_ROW) {
  704. memory_mode = (RRD_MEMORY_MODE)sqlite3_column_int(res, 0);
  705. }
  706. failed:
  707. rc = sqlite3_finalize(res);
  708. if (unlikely(rc != SQLITE_OK))
  709. error_report("Failed to finalize the prepared statement when reading host memory mode");
  710. return memory_mode;
  711. }
  712. #define SELECT_HOST_DIMENSION_LIST \
  713. "SELECT d.dim_id, c.update_every, c.type||'.'||c.id, d.id, d.name FROM chart c, dimension d " \
  714. "WHERE d.chart_id = c.chart_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
  715. #define SELECT_HOST_CHART_LIST \
  716. "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \
  717. "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
  718. void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
  719. {
  720. UNUSED(cmd);
  721. int rc;
  722. if (!aclk_use_new_cloud_arch || !aclk_connected)
  723. return;
  724. char *claim_id = is_agent_claimed();
  725. if (unlikely(!claim_id))
  726. return;
  727. sqlite3_stmt *res = NULL;
  728. RRD_MEMORY_MODE memory_mode;
  729. uuid_t host_uuid;
  730. rc = uuid_parse(wc->host_guid, host_uuid);
  731. if (unlikely(rc)) {
  732. freez(claim_id);
  733. return;
  734. }
  735. if (wc->host)
  736. memory_mode = wc->host->rrd_memory_mode;
  737. else
  738. memory_mode = sql_get_host_memory_mode(&host_uuid);
  739. if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
  740. rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0);
  741. else
  742. rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0);
  743. if (unlikely(rc != SQLITE_OK)) {
  744. error_report("Failed to prepare statement to fetch host dimensions");
  745. freez(claim_id);
  746. return;
  747. }
  748. rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC);
  749. if (unlikely(rc != SQLITE_OK)) {
  750. error_report("Failed to bind host parameter to fetch host dimensions");
  751. goto failed;
  752. }
  753. time_t start_time = LONG_MAX;
  754. time_t first_entry_t;
  755. time_t last_entry_t;
  756. uint32_t update_every = 0;
  757. uint32_t dimension_update_count = 0;
  758. int send_status;
  759. struct retention_updated rotate_data;
  760. memset(&rotate_data, 0, sizeof(rotate_data));
  761. int max_intervals = 32;
  762. rotate_data.interval_duration_count = 0;
  763. rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations));
  764. now_realtime_timeval(&rotate_data.rotation_timestamp);
  765. rotate_data.memory_mode = memory_mode;
  766. rotate_data.claim_id = claim_id;
  767. rotate_data.node_id = strdupz(wc->node_id);
  768. time_t now = now_realtime_sec();
  769. while (sqlite3_step(res) == SQLITE_ROW) {
  770. if (!update_every || update_every != (uint32_t)sqlite3_column_int(res, 1)) {
  771. if (update_every) {
  772. debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
  773. if (start_time == LONG_MAX)
  774. rotate_data.interval_durations[rotate_data.interval_duration_count].retention = 0;
  775. else
  776. rotate_data.interval_durations[rotate_data.interval_duration_count].retention =
  777. rotate_data.rotation_timestamp.tv_sec - start_time;
  778. rotate_data.interval_duration_count++;
  779. }
  780. update_every = (uint32_t)sqlite3_column_int(res, 1);
  781. rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every;
  782. start_time = LONG_MAX;
  783. }
  784. #ifdef ENABLE_DBENGINE
  785. if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
  786. rc =
  787. rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t);
  788. else
  789. #endif
  790. {
  791. if (wc->host) {
  792. RRDSET *st = NULL;
  793. rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1;
  794. if (!rc) {
  795. first_entry_t = rrdset_first_entry_t(st);
  796. last_entry_t = rrdset_last_entry_t(st);
  797. }
  798. } else {
  799. rc = 0;
  800. first_entry_t = rotate_data.rotation_timestamp.tv_sec;
  801. }
  802. }
  803. if (likely(!rc && first_entry_t))
  804. start_time = MIN(start_time, first_entry_t);
  805. if (memory_mode == RRD_MEMORY_MODE_DBENGINE && wc->chart_updates) {
  806. int live = ((now - last_entry_t) < (RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * update_every));
  807. if ((!live || !first_entry_t) && (dimension_update_count < ACLK_MAX_DIMENSION_CLEANUP)) {
  808. (void)aclk_upd_dimension_event(
  809. wc,
  810. claim_id,
  811. (uuid_t *)sqlite3_column_blob(res, 0),
  812. (const char *)(const char *)sqlite3_column_text(res, 3),
  813. (const char *)(const char *)sqlite3_column_text(res, 4),
  814. (const char *)(const char *)sqlite3_column_text(res, 2),
  815. first_entry_t,
  816. live ? 0 : last_entry_t,
  817. &send_status);
  818. if (!send_status)
  819. dimension_update_count++;
  820. }
  821. }
  822. }
  823. if (update_every) {
  824. debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
  825. if (start_time == LONG_MAX)
  826. rotate_data.interval_durations[rotate_data.interval_duration_count].retention = 0;
  827. else
  828. rotate_data.interval_durations[rotate_data.interval_duration_count].retention =
  829. rotate_data.rotation_timestamp.tv_sec - start_time;
  830. rotate_data.interval_duration_count++;
  831. }
  832. #ifdef NETDATA_INTERNAL_CHECKS
  833. for (int i = 0; i < rotate_data.interval_duration_count; ++i)
  834. info(
  835. "Update for host %s (node %s) for %u Retention = %u",
  836. wc->host_guid,
  837. wc->node_id,
  838. rotate_data.interval_durations[i].update_every,
  839. rotate_data.interval_durations[i].retention);
  840. #endif
  841. aclk_retention_updated(&rotate_data);
  842. freez(rotate_data.node_id);
  843. freez(rotate_data.interval_durations);
  844. failed:
  845. freez(claim_id);
  846. rc = sqlite3_finalize(res);
  847. if (unlikely(rc != SQLITE_OK))
  848. error_report("Failed to finalize the prepared statement when reading host dimensions");
  849. return;
  850. }
  851. uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc)
  852. {
  853. char sql[ACLK_SYNC_QUERY_SIZE];
  854. static __thread sqlite3_stmt *res = NULL;
  855. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT count(1) FROM aclk_chart_%s ac WHERE ac.date_submitted IS NULL;", wc->uuid_str);
  856. int rc;
  857. uint32_t chart_payload_count = 0;
  858. if (unlikely(!res)) {
  859. rc = prepare_statement(db_meta, sql, &res);
  860. if (rc != SQLITE_OK) {
  861. error_report("Failed to prepare statement to count pending messages");
  862. return 0;
  863. }
  864. }
  865. while (sqlite3_step(res) == SQLITE_ROW)
  866. chart_payload_count = (uint32_t) sqlite3_column_int(res, 0);
  867. rc = sqlite3_reset(res);
  868. if (unlikely(rc != SQLITE_OK))
  869. error_report("Failed to reset statement when fetching pending messages, rc = %d", rc);
  870. return chart_payload_count;
  871. }
  872. void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)
  873. {
  874. char sql[ACLK_SYNC_QUERY_SIZE];
  875. snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \
  876. "WHERE ac.date_submitted IS NOT NULL ORDER BY ac.sequence_id DESC LIMIT 1;", wc->uuid_str);
  877. int rc;
  878. sqlite3_stmt *res = NULL;
  879. rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
  880. if (rc != SQLITE_OK) {
  881. error_report("Failed to prepare statement to find last chart sequence id");
  882. return;
  883. }
  884. wc->chart_sequence_id = 0;
  885. wc->chart_timestamp = 0;
  886. while (sqlite3_step(res) == SQLITE_ROW) {
  887. wc->chart_sequence_id = (uint64_t)sqlite3_column_int64(res, 0);
  888. wc->chart_timestamp = (time_t)sqlite3_column_int64(res, 1);
  889. }
  890. debug(D_ACLK_SYNC, "Node %s reports last sequence_id=%" PRIu64, wc->node_id, wc->chart_sequence_id);
  891. rc = sqlite3_finalize(res);
  892. if (unlikely(rc != SQLITE_OK))
  893. error_report("Failed to reset statement when fetching chart sequence info, rc = %d", rc);
  894. return;
  895. }
  896. int queue_dimension_to_aclk(RRDDIM *rd)
  897. {
  898. int rc = sql_queue_chart_payload((struct aclk_database_worker_config *) rd->rrdset->rrdhost->dbsync_worker,
  899. rd, ACLK_DATABASE_ADD_DIMENSION);
  900. return rc;
  901. }
  902. void aclk_send_dimension_update(RRDDIM *rd)
  903. {
  904. if (!aclk_use_new_cloud_arch)
  905. return;
  906. char *claim_id = is_agent_claimed();
  907. if (unlikely(!claim_id))
  908. return;
  909. time_t first_entry_t = rrddim_first_entry_t(rd);
  910. time_t last_entry_t = rrddim_last_entry_t(rd);
  911. time_t now = now_realtime_sec();
  912. int live = ((now - rd->last_collected_time.tv_sec) < (RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every));
  913. if (!live || rd->state->aclk_live_status != live || !first_entry_t) {
  914. (void)aclk_upd_dimension_event(
  915. rd->rrdset->rrdhost->dbsync_worker,
  916. claim_id,
  917. &rd->state->metric_uuid,
  918. rd->id,
  919. rd->name,
  920. rd->rrdset->id,
  921. first_entry_t,
  922. live ? 0 : last_entry_t,
  923. NULL);
  924. if (!first_entry_t)
  925. debug(
  926. D_ACLK_SYNC,
  927. "%s: Update dimension chart=%s dim=%s live=%d (%ld, %ld)",
  928. rd->rrdset->rrdhost->hostname,
  929. rd->rrdset->name,
  930. rd->name,
  931. live,
  932. first_entry_t,
  933. last_entry_t);
  934. else
  935. debug(
  936. D_ACLK_SYNC,
  937. "%s: Update dimension chart=%s dim=%s live=%d (%ld, %ld) collected %ld seconds ago",
  938. rd->rrdset->rrdhost->hostname,
  939. rd->rrdset->name,
  940. rd->name,
  941. live,
  942. first_entry_t,
  943. last_entry_t,
  944. now - last_entry_t);
  945. rd->state->aclk_live_status = live;
  946. }
  947. freez(claim_id);
  948. return;
  949. }
  950. #define SQL_SEQ_NULL(result, n) sqlite3_column_type(result, n) == SQLITE_NULL ? 0 : sqlite3_column_int64(result, n)
  951. struct aclk_chart_sync_stats *aclk_get_chart_sync_stats(RRDHOST *host)
  952. {
  953. struct aclk_chart_sync_stats *aclk_statistics = NULL;
  954. struct aclk_database_worker_config *wc = NULL;
  955. wc = (struct aclk_database_worker_config *)host->dbsync_worker;
  956. if (!wc)
  957. return NULL;
  958. aclk_statistics = callocz(1, sizeof(struct aclk_chart_sync_stats));
  959. aclk_statistics->updates = wc->chart_updates;
  960. aclk_statistics->batch_id = wc->batch_id;
  961. char host_uuid_fixed[GUID_LEN + 1];
  962. strncpy(host_uuid_fixed, host->machine_guid, GUID_LEN);
  963. host_uuid_fixed[GUID_LEN] = 0;
  964. host_uuid_fixed[8] = '_';
  965. host_uuid_fixed[13] = '_';
  966. host_uuid_fixed[18] = '_';
  967. host_uuid_fixed[23] = '_';
  968. sqlite3_stmt *res = NULL;
  969. BUFFER *sql = buffer_create(1024);
  970. buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s;", host_uuid_fixed);
  971. buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_submitted IS NULL;", host_uuid_fixed);
  972. buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_submitted IS NOT NULL;", host_uuid_fixed);
  973. buffer_sprintf(sql, "SELECT min(sequence_id), max(sequence_id), 0 FROM aclk_chart_%s WHERE date_updated IS NOT NULL;", host_uuid_fixed);
  974. buffer_sprintf(sql, "SELECT max(date_created), max(date_submitted), max(date_updated), 0 FROM aclk_chart_%s;", host_uuid_fixed);
  975. int rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
  976. if (rc != SQLITE_OK) {
  977. buffer_free(sql);
  978. freez(aclk_statistics);
  979. return NULL;
  980. }
  981. rc = sqlite3_step(res);
  982. if (rc == SQLITE_ROW) {
  983. aclk_statistics->min_seqid = SQL_SEQ_NULL(res, 0);
  984. aclk_statistics->max_seqid = SQL_SEQ_NULL(res, 1);
  985. }
  986. rc = sqlite3_step(res);
  987. if (rc == SQLITE_ROW) {
  988. aclk_statistics->min_seqid_pend = SQL_SEQ_NULL(res, 0);
  989. aclk_statistics->max_seqid_pend = SQL_SEQ_NULL(res, 1);
  990. }
  991. rc = sqlite3_step(res);
  992. if (rc == SQLITE_ROW) {
  993. aclk_statistics->min_seqid_sent = SQL_SEQ_NULL(res, 0);
  994. aclk_statistics->max_seqid_sent = SQL_SEQ_NULL(res, 1);
  995. }
  996. rc = sqlite3_step(res);
  997. if (rc == SQLITE_ROW) {
  998. aclk_statistics->min_seqid_ack = SQL_SEQ_NULL(res, 0);
  999. aclk_statistics->max_seqid_ack = SQL_SEQ_NULL(res, 1);
  1000. }
  1001. rc = sqlite3_step(res);
  1002. if (rc == SQLITE_ROW) {
  1003. aclk_statistics->min_seqid_ack = SQL_SEQ_NULL(res, 0);
  1004. aclk_statistics->max_seqid_ack = SQL_SEQ_NULL(res, 1);
  1005. }
  1006. rc = sqlite3_step(res);
  1007. if (rc == SQLITE_ROW) {
  1008. aclk_statistics->max_date_created = (time_t) SQL_SEQ_NULL(res, 0);
  1009. aclk_statistics->max_date_submitted = (time_t) SQL_SEQ_NULL(res, 1);
  1010. aclk_statistics->max_date_ack = (time_t) SQL_SEQ_NULL(res, 2);
  1011. }
  1012. rc = sqlite3_finalize(res);
  1013. if (unlikely(rc != SQLITE_OK))
  1014. error_report("Failed to finalize statement when fetching aclk sync statistics, rc = %d", rc);
  1015. buffer_free(sql);
  1016. return aclk_statistics;
  1017. }
  1018. void sql_check_chart_liveness(RRDSET *st) {
  1019. RRDDIM *rd;
  1020. if (unlikely(st->state->is_ar_chart))
  1021. return;
  1022. rrdset_rdlock(st);
  1023. if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
  1024. if (likely(st->dimensions && st->counter_done && !queue_chart_to_aclk(st))) {
  1025. debug(D_ACLK_SYNC,"Check chart liveness [%s] submit chart definition", st->name);
  1026. rrdset_flag_set(st, RRDSET_FLAG_ACLK);
  1027. }
  1028. }
  1029. else
  1030. debug(D_ACLK_SYNC,"Check chart liveness [%s] chart definition already submitted", st->name);
  1031. time_t mark = now_realtime_sec();
  1032. debug(D_ACLK_SYNC,"Check chart liveness [%s] scanning dimensions", st->name);
  1033. rrddim_foreach_read(rd, st) {
  1034. if (!rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)) {
  1035. int live = (mark - rd->last_collected_time.tv_sec) < RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every;
  1036. if (unlikely(live != rd->state->aclk_live_status)) {
  1037. if (likely(rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
  1038. if (likely(!queue_dimension_to_aclk(rd))) {
  1039. debug(D_ACLK_SYNC,"Dimension change [%s] on [%s] from live %d --> %d", rd->id, rd->rrdset->name, rd->state->aclk_live_status, live);
  1040. rd->state->aclk_live_status = live;
  1041. rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
  1042. }
  1043. }
  1044. }
  1045. else
  1046. debug(D_ACLK_SYNC,"Dimension check [%s] on [%s] liveness matches", rd->id, st->name);
  1047. }
  1048. }
  1049. rrdset_unlock(st);
  1050. }
  1051. #endif //ENABLE_NEW_CLOUD_PROTOCOL
  1052. // ST is read locked
  1053. int queue_chart_to_aclk(RRDSET *st)
  1054. {
  1055. #ifndef ENABLE_NEW_CLOUD_PROTOCOL
  1056. #ifdef ENABLE_ACLK
  1057. aclk_update_chart(st->rrdhost, st->id, 1);
  1058. #else
  1059. UNUSED(st);
  1060. #endif
  1061. return 0;
  1062. #else
  1063. if (!aclk_use_new_cloud_arch && aclk_connected) {
  1064. aclk_update_chart(st->rrdhost, st->id, 1);
  1065. return 0;
  1066. }
  1067. return sql_queue_chart_payload((struct aclk_database_worker_config *) st->rrdhost->dbsync_worker,
  1068. st, ACLK_DATABASE_ADD_CHART);
  1069. #endif
  1070. }