| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "sqlite_functions.h" |
| 4 | #include "sqlite_aclk_node.h" |
| 5 | |
| 6 | #include "../../aclk/aclk_contexts_api.h" |
| 7 | #include "../../aclk/aclk_capas.h" |
| 8 | #include "../../aclk/aclk_query_queue.h" |
| 9 | |
| 10 | DICTIONARY *collectors_from_charts(RRDHOST *host, DICTIONARY *dict) { |
| 11 | RRDSET *st; |
| 12 | char name[500]; |
| 13 | |
| 14 | rrdset_foreach_read(st, host) |
| 15 | { |
| 16 | if (rrdset_is_available_for_viewers(st)) { |
| 17 | struct collector_info col = {.plugin = rrdset_plugin_name(st), .module = rrdset_module_name(st)}; |
| 18 | snprintfz(name, sizeof(name) - 1, "%s:%s", col.plugin, col.module); |
| 19 | dictionary_set(dict, name, &col, sizeof(struct collector_info)); |
| 20 | } |
| 21 | } |
| 22 | rrdset_foreach_done(st); |
| 23 | |
| 24 | return dict; |
| 25 | } |
| 26 | |
| 27 | static void build_node_collectors(RRDHOST *host) |
| 28 | { |
| 29 | struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE); |
| 30 | |
| 31 | struct update_node_collectors upd_node_collectors; |
| 32 | DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED); |
| 33 | |
| 34 | CLAIM_ID claim_id = claim_id_get(); |
| 35 | upd_node_collectors.node_id = aclk_host_config->node_id; |
| 36 | upd_node_collectors.claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL; |
| 37 | |
| 38 | upd_node_collectors.node_collectors = collectors_from_charts(host, dict); |
| 39 | aclk_update_node_collectors(&upd_node_collectors); |
| 40 | |
| 41 | dictionary_destroy(dict); |
| 42 | |
| 43 | nd_log(NDLS_ACCESS, NDLP_DEBUG, |
| 44 | "ACLK RES [%s (%s)]: NODE COLLECTORS SENT", |
| 45 | aclk_host_config->node_id, rrdhost_hostname(host)); |
| 46 | } |
| 47 | |
| 48 | static void build_node_info(RRDHOST *host, struct aclk_sync_completion *sync_completion) |
| 49 | { |
| 50 | struct update_node_info node_info; |
| 51 | |
| 52 | struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE); |
| 53 | |
| 54 | CLAIM_ID claim_id = claim_id_get(); |
| 55 | |
| 56 | rrd_rdlock(); |
| 57 | node_info.node_id = aclk_host_config->node_id; |
| 58 | node_info.claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL; |
| 59 | node_info.machine_guid = host->machine_guid; |
| 60 | node_info.child = (host != localhost); |
| 61 | node_info.ml_info.ml_capable = ml_capable(); |
| 62 | node_info.ml_info.ml_enabled = ml_enabled(host); |
| 63 | |
| 64 | node_info.node_instance_capabilities = aclk_get_node_instance_capas(host); |
| 65 | |
| 66 | now_realtime_timeval(&node_info.updated_at); |
| 67 | |
| 68 | char *host_version = NULL; |
| 69 | bool is_virtual_host = (rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST) || IS_VIRTUAL_HOST_OS(host)); |
| 70 | |
| 71 | if (host != localhost && !is_virtual_host) |
| 72 | host_version = stream_receiver_program_version_strdupz(host); |
| 73 | |
| 74 | RRDHOST_TZ host_tz = rrdhost_tz_get(host); |
| 75 | |
| 76 | node_info.data.name = rrdhost_hostname(host); |
| 77 | node_info.data.os = rrdhost_os(host); |
| 78 | node_info.data.version = host_version ? host_version : NETDATA_VERSION; |
| 79 | node_info.data.release_channel = get_release_channel(); |
| 80 | node_info.data.timezone = host_tz.abbrev_timezone; |
| 81 | node_info.data.custom_info = inicfg_get(&netdata_config, CONFIG_SECTION_WEB, "custom dashboard_info.js", ""); |
| 82 | node_info.data.machine_guid = host->machine_guid; |
| 83 | node_info.node_capabilities = (struct capability *)aclk_get_agent_capas(); |
| 84 | node_info.data.host_labels_ptr = host->rrdlabels; |
| 85 | |
| 86 | rrdhost_system_info_to_node_info(host->system_info, &node_info); |
| 87 | |
| 88 | aclk_update_node_info(&node_info, sync_completion); |
| 89 | nd_log( |
| 90 | NDLS_ACCESS, |
| 91 | NDLP_DEBUG, |
| 92 | "ACLK RES [%s (%s)]: NODE INFO SENT for guid [%s] (%s)", |
| 93 | aclk_host_config->node_id, |
| 94 | rrdhost_hostname(host), |
| 95 | host->machine_guid, |
| 96 | host == localhost ? "parent" : "child"); |
| 97 | |
| 98 | rrd_rdunlock(); |
| 99 | rrdhost_tz_free(&host_tz); |
| 100 | freez(node_info.node_instance_capabilities); |
| 101 | freez(host_version); |
| 102 | |
| 103 | aclk_host_config->node_collectors_send = now_realtime_sec(); |
| 104 | } |
| 105 | |
| 106 | void send_node_info_with_wait(RRDHOST *host) |
| 107 | { |
| 108 | if (unlikely(!host || !__atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE))) |
| 109 | return; |
| 110 | |
| 111 | // No node_id means cloud doesn't know about this node - nothing to update |
| 112 | if (uuid_is_null(host->node_id.uuid)) |
| 113 | return; |
| 114 | |
| 115 | if (!aclk_online()) |
| 116 | return; |
| 117 | |
| 118 | struct aclk_sync_completion *sc = aclk_sync_completion_create(); |
| 119 | |
| 120 | build_node_info(host, sc); |
| 121 | |
| 122 | bool success = aclk_sync_completion_timedwait(sc, 30); |
| 123 | if (!success) { |
| 124 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 125 | "Timed out waiting for node info update for host '%s'", |
| 126 | rrdhost_hostname(host)); |
| 127 | } |
| 128 | // sc is automatically freed when both waiter and query release their references |
| 129 | } |
| 130 | |
| 131 | void send_node_update_with_wait(RRDHOST *host, int live, int queryable) |
| 132 | { |
| 133 | if (unlikely(!host || !__atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE))) |
| 134 | return; |
| 135 | |
| 136 | // No node_id means cloud doesn't know about this node - nothing to update |
| 137 | if (uuid_is_null(host->node_id.uuid)) |
| 138 | return; |
| 139 | |
| 140 | if (!aclk_online()) |
| 141 | return; |
| 142 | |
| 143 | struct aclk_sync_completion *sc = aclk_sync_completion_create(); |
| 144 | |
| 145 | aclk_host_state_update(host, live, queryable, sc); |
| 146 | |
| 147 | bool success = aclk_sync_completion_timedwait(sc, 30); |
| 148 | if (!success) { |
| 149 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 150 | "Timed out waiting for node state update for host '%s'", |
| 151 | rrdhost_hostname(host)); |
| 152 | } |
| 153 | // sc is automatically freed when both waiter and query release their references |
| 154 | } |
| 155 | |
| 156 | void aclk_check_node_info_and_collectors(void) |
| 157 | { |
| 158 | RRDHOST *host; |
| 159 | |
| 160 | if (unlikely(!aclk_online_for_nodes())) |
| 161 | return; |
| 162 | |
| 163 | size_t context_loading = 0; |
| 164 | size_t replicating_rcv = 0; |
| 165 | size_t replicating_snd = 0; |
| 166 | size_t context_pp = 0; |
| 167 | |
| 168 | STRING *context_loading_host = NULL; |
| 169 | STRING *replicating_rcv_host = NULL; |
| 170 | STRING *replicating_snd_host = NULL; |
| 171 | STRING *context_pp_host = NULL; |
| 172 | |
| 173 | #ifdef REPLICATION_TRACKING |
| 174 | struct replay_who_counters replay_counters = { 0 }; |
| 175 | #endif |
| 176 | |
| 177 | time_t now = now_realtime_sec(); |
| 178 | dfe_start_reentrant(rrdhost_root_index, host) |
| 179 | { |
| 180 | struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE); |
| 181 | if (unlikely(!aclk_host_config)) |
| 182 | continue; |
| 183 | |
| 184 | if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) { |
| 185 | internal_error(true, "ACLK SYNC: Context still pending for %s", rrdhost_hostname(host)); |
| 186 | context_loading++; |
| 187 | context_loading_host = host->hostname; |
| 188 | continue; |
| 189 | } |
| 190 | |
| 191 | if (!aclk_host_config->node_info_send_time && !aclk_host_config->node_collectors_send) |
| 192 | continue; |
| 193 | |
| 194 | if (unlikely(rrdhost_receiver_replicating_charts(host))) { |
| 195 | internal_error(true, "ACLK SYNC: Host %s is still replicating in", rrdhost_hostname(host)); |
| 196 | replicating_rcv++; |
| 197 | replicating_rcv_host = host->hostname; |
| 198 | } |
| 199 | |
| 200 | if (unlikely(rrdhost_sender_replicating_charts(host))) { |
| 201 | internal_error(true, "ACLK SYNC: Host %s is still replicating out", rrdhost_hostname(host)); |
| 202 | replicating_snd++; |
| 203 | replicating_snd_host = host->hostname; |
| 204 | } |
| 205 | |
| 206 | #ifdef REPLICATION_TRACKING |
| 207 | replication_tracking_counters(host, &replay_counters); |
| 208 | #endif |
| 209 | |
| 210 | if(replicating_rcv) |
| 211 | continue; |
| 212 | |
| 213 | bool pp_queue_empty = !rrdcontext_queue_entries(&host->rrdctx.pp_queue); |
| 214 | |
| 215 | if (!pp_queue_empty && (aclk_host_config->node_info_send_time || aclk_host_config->node_collectors_send)) { |
| 216 | context_pp++; |
| 217 | context_pp_host = host->hostname; |
| 218 | } |
| 219 | |
| 220 | if (pp_queue_empty && aclk_host_config->node_info_send_time && |
| 221 | aclk_host_config->node_info_send_time + 30 < now) { |
| 222 | aclk_host_config->node_info_send_time = 0; |
| 223 | build_node_info(host, NULL); |
| 224 | schedule_node_state_update(host, 10000); |
| 225 | internal_error(true, "ACLK SYNC: Sending node info for %s", rrdhost_hostname(host)); |
| 226 | } |
| 227 | |
| 228 | if (pp_queue_empty && aclk_host_config->node_collectors_send && |
| 229 | aclk_host_config->node_collectors_send + 30 < now) { |
| 230 | build_node_collectors(host); |
| 231 | internal_error(true, "ACLK SYNC: Sending collectors for %s", rrdhost_hostname(host)); |
| 232 | aclk_host_config->node_collectors_send = 0; |
| 233 | } |
| 234 | } |
| 235 | dfe_done(host); |
| 236 | |
| 237 | if (context_loading || replicating_rcv || replicating_snd || context_pp) { |
| 238 | #ifdef REPLICATION_TRACKING |
| 239 | char replay_counters_txt[1024]; |
| 240 | snprintfz(replay_counters_txt, sizeof(replay_counters_txt), |
| 241 | " - REPLAY WHO RCV { %zu unknown, %zu me, %zu them, %zu finished } - " |
| 242 | "REPLAY WHO SND { %zu unknown, %zu me, %zu them, %zu finished }", |
| 243 | replay_counters.rcv[REPLAY_WHO_UNKNOWN], replay_counters.rcv[REPLAY_WHO_ME], replay_counters.rcv[REPLAY_WHO_THEM], replay_counters.rcv[REPLAY_WHO_FINISHED], |
| 244 | replay_counters.snd[REPLAY_WHO_UNKNOWN], replay_counters.snd[REPLAY_WHO_ME], replay_counters.snd[REPLAY_WHO_THEM], replay_counters.snd[REPLAY_WHO_FINISHED] |
| 245 | ); |
| 246 | #else |
| 247 | char *replay_counters_txt = ""; |
| 248 | #endif |
| 249 | |
| 250 | const char *context_loading_pre = "", *context_loading_body = "", *context_loading_post = ""; |
| 251 | if(context_loading == 1) { |
| 252 | context_loading_pre = " (host '"; |
| 253 | context_loading_body = string2str(context_loading_host); |
| 254 | context_loading_post = "')"; |
| 255 | } |
| 256 | |
| 257 | const char *replicating_rcv_pre = "", *replicating_rcv_body = "", *replicating_rcv_post = ""; |
| 258 | if(replicating_rcv == 1) { |
| 259 | replicating_rcv_pre = " (host '"; |
| 260 | replicating_rcv_body = string2str(replicating_rcv_host); |
| 261 | replicating_rcv_post = "')"; |
| 262 | } |
| 263 | |
| 264 | const char *replicating_snd_pre = "", *replicating_snd_body = "", *replicating_snd_post = ""; |
| 265 | if(replicating_snd == 1) { |
| 266 | replicating_snd_pre = " (host '"; |
| 267 | replicating_snd_body = string2str(replicating_snd_host); |
| 268 | replicating_snd_post = "')"; |
| 269 | } |
| 270 | |
| 271 | const char *context_pp_pre = "", *context_pp_body = "", *context_pp_post = ""; |
| 272 | if(context_pp == 1) { |
| 273 | context_pp_pre = " (host '"; |
| 274 | context_pp_body = string2str(context_pp_host); |
| 275 | context_pp_post = "')"; |
| 276 | } |
| 277 | |
| 278 | nd_log_limit_static_global_var(erl, 10, 100 * USEC_PER_MS); |
| 279 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO, |
| 280 | "NODES INFO: %zu nodes loading contexts%s%s%s, %zu receiving replication%s%s%s, %zu sending replication%s%s%s, %zu pending context post processing%s%s%s%s", |
| 281 | context_loading, context_loading_pre, context_loading_body, context_loading_post, |
| 282 | replicating_rcv, replicating_rcv_pre, replicating_rcv_body, replicating_rcv_post, |
| 283 | replicating_snd, replicating_snd_pre, replicating_snd_body, replicating_snd_post, |
| 284 | context_pp, context_pp_pre, context_pp_body, context_pp_post, |
| 285 | replay_counters_txt |
| 286 | ); |
| 287 | } |
| 288 | } |