@cryptotaxi247 / netdata-1 / commits / 84ddf0d48

Contexts Loading (#19336)

* do not load instances and dimensions for unknown contexts - instead of creating contexts and adding instances and dimensions * node updates to NC * unify all liveness calls * prevent wanted cache size from getting negative

Costa Tsaousis committed Jan 7, 2025 at 21:24 UTC 84ddf0d48c0c42a07c104e9e1a6caabc2241d2f2

14 files changed +92 -67

src/aclk/aclk.c

+29 -6

@@ -833,6 +833,29 @@ exit:

833 return NULL;

834 }

835

836 +bool aclk_host_state_update_auto(RRDHOST *host) {

837 + int live;

838 + switch(rrdhost_ingestion_status(host)) {

839 + case RRDHOST_INGEST_STATUS_ARCHIVED:

840 + case RRDHOST_INGEST_STATUS_INITIALIZING:

841 + case RRDHOST_INGEST_STATUS_OFFLINE:

842 + live = 0;

843 + break;

844 +

845 + case RRDHOST_INGEST_STATUS_REPLICATING:

846 + // receiving replication

847 + // no need to send this to NC

848 + return false;

849 +

850 + case RRDHOST_INGEST_STATUS_ONLINE:

851 + // currently collecting data

852 + live = 1;

853 + break;

854 + }

855 + aclk_host_state_update(host, live, 1);

856 + return true;

857 +}

858 +

859 void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

860 {

861 ND_UUID node_id;

@@ -858,7 +881,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

881

882 node_instance_creation_t node_instance_creation = {

883 .claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL,

861 - .hops = rrdhost_system_info_hops(host->system_info),

884 + .hops = rrdhost_ingestion_hops(host),

885 .hostname = rrdhost_hostname(host),

886 .machine_guid = host->machine_guid};

887

@@ -869,7 +892,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

892 create_query->data.bin_payload.msg_name = "CreateNodeInstance";

893 nd_log(NDLS_DAEMON, NDLP_DEBUG,

894 "Registering host=%s, hops=%d", host->machine_guid,

872 - rrdhost_system_info_hops(host->system_info));

895 + rrdhost_ingestion_hops(host));

896

897 aclk_execute_query(create_query);

898 return;

@@ -878,7 +901,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

901

902 aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);

903 node_instance_connection_t node_state_update = {

881 - .hops = rrdhost_system_info_hops(host->system_info),

904 + .hops = rrdhost_ingestion_hops(host),

905 .live = cmd,

906 .queryable = queryable,

907 .session_id = aclk_session_newarch

@@ -895,7 +918,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)

918 nd_log(NDLS_DAEMON, NDLP_DEBUG,

919 "Queuing status update for node=%s, live=%d, hops=%d, queryable=%d",

920 (char*)node_state_update.node_id, cmd,

898 - rrdhost_system_info_hops(host->system_info), queryable);

921 + rrdhost_ingestion_hops(host), queryable);

922

923 freez((void*)node_state_update.node_id);

924 query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";

@@ -1067,7 +1090,7 @@ char *aclk_state(void)

1090 }

1091

1092 buffer_sprintf(wb, "\tStreaming Hops: %d\n\tRelationship: %s",

1070 - rrdhost_system_info_hops(host->system_info),

1093 + rrdhost_ingestion_hops(host),

1094 host == localhost ? "self" : "child");

1095

1096 if (host != localhost)

@@ -1202,7 +1225,7 @@ char *aclk_state_json(void)

1225 json_object_object_add(nodeinstance, "node-id", tmp);

1226 }

1227

1205 - tmp = json_object_new_int(rrdhost_system_info_hops(host->system_info));

1228 + tmp = json_object_new_int(rrdhost_ingestion_hops(host));

1229 json_object_object_add(nodeinstance, "streaming-hops", tmp);

1230

1231 tmp = json_object_new_string(host == localhost ? "self" : "child");

src/aclk/aclk.h

@@ -83,6 +83,8 @@ extern struct aclk_shared_state {

83 } aclk_shared_state;

85 void aclk_host_state_update(RRDHOST *host, int cmd, int queryable);

86 +bool aclk_host_state_update_auto(RRDHOST *host);

87 +

88 void aclk_send_node_instances(void);

90 void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname);

src/aclk/aclk_rx_msgs.c

+3 -7

@@ -281,14 +281,10 @@ int create_node_instance_result(const char *msg, size_t msg_len)

281

282 RRDHOST *host = rrdhost_find_by_guid(res.machine_guid);

283 if (likely(host)) {

284 - if (host == localhost) {

285 - node_state_update.live = 1;

286 - node_state_update.hops = 0;

287 - } else {

288 - node_state_update.live = (!rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN));

289 - node_state_update.hops = rrdhost_system_info_hops(host->system_info);

290 - }

284 + node_state_update.live = rrdhost_is_local(host) ? 1 : 0;

285 + node_state_update.hops = rrdhost_ingestion_hops(host);

286 node_state_update.capabilities = aclk_get_node_instance_capas(host);

287 + schedule_node_state_update(host, 5000);

288 }

289

290 CLAIM_ID claim_id = claim_id_get();

src/database/contexts/worker.c

+9 -14

@@ -33,7 +33,7 @@ static void rrdinstance_load_dimension_callback(SQL_DIMENSION_DATA *sd, void *da

33 RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item(host->rrdctx.contexts, sd->context);

34 if(!rca) {

35 nd_log(NDLS_DAEMON, NDLP_ERR,

36 - "RRDCONTEXT: context '%s' is not found in host '%s'",

36 + "RRDCONTEXT: context '%s' is not found in host '%s' - not loading dimensions",

37 sd->context, rrdhost_hostname(host));

38 return;

39 }

@@ -43,7 +43,7 @@ static void rrdinstance_load_dimension_callback(SQL_DIMENSION_DATA *sd, void *da

43 if(!ria) {

44 rrdcontext_release(rca);

45 nd_log(NDLS_DAEMON, NDLP_ERR,

46 - "RRDCONTEXT: instance '%s' of context '%s' is not found in host '%s'",

46 + "RRDCONTEXT: instance '%s' of context '%s' is not found in host '%s' - not loading dimensions",

47 sd->chart_id, sd->context, rrdhost_hostname(host));

48 return;

49 }

@@ -67,18 +67,13 @@ static void rrdinstance_load_dimension_callback(SQL_DIMENSION_DATA *sd, void *da

67 static void rrdinstance_load_instance_callback(SQL_CHART_DATA *sc, void *data) {

68 RRDHOST *host = data;

70 - RRDCONTEXT tc = {

71 - .id = string_strdupz(sc->context),

72 - .title = string_strdupz(sc->title),

73 - .units = string_strdupz(sc->units),

74 - .family = string_strdupz(sc->family),

75 - .priority = sc->priority,

76 - .chart_type = sc->chart_type,

77 - .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics

78 - .rrdhost = host,

79 - };

80 -

81 - RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_set_and_acquire_item(host->rrdctx.contexts, string2str(tc.id), &tc, sizeof(tc));

70 + RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_get_and_acquire_item(host->rrdctx.contexts, sc->context);

71 + if(!rca) {

72 + nd_log(NDLS_DAEMON, NDLP_ERR,

73 + "RRDCONTEXT: context '%s' is not found in host '%s' - not loadings instances",

74 + sc->context, rrdhost_hostname(host));

75 + return;

76 + }

77 RRDCONTEXT *rc = rrdcontext_acquired_value(rca);

79 RRDINSTANCE tri = {

src/database/engine/cache.c

+5 -2

@@ -403,7 +403,7 @@ static inline size_t cache_usage_per1000(PGC *cache, size_t *size_to_evict) {

403 wanted_cache_size = referenced_size + dirty;

404

405 // if we don't have enough clean pages, there is no reason to be aggressive or critical

406 - if(wanted_cache_size < current_cache_size - clean)

406 + if(wanted_cache_size < (current_cache_size - clean) && current_cache_size > clean)

407 wanted_cache_size = current_cache_size - clean;

408

409 if(cache->config.out_of_memory_protection_bytes) {

@@ -415,7 +415,10 @@ static inline size_t cache_usage_per1000(PGC *cache, size_t *size_to_evict) {

415 const uint64_t min_available = cache->config.out_of_memory_protection_bytes;

416 if (sm.ram_available_bytes < min_available) {

417 // we must shrink

418 - wanted_cache_size = current_cache_size - (min_available - sm.ram_available_bytes);

418 + if(current_cache_size > (min_available - sm.ram_available_bytes))

419 + wanted_cache_size = current_cache_size - (min_available - sm.ram_available_bytes);

420 + else

421 + wanted_cache_size = hot + dirty;

422 }

423 else if(cache->config.use_all_ram) {

424 // we can grow

src/database/rrd.h

+6 -2

@@ -1263,9 +1263,13 @@ extern RRDHOST *localhost;

1263 #define rrdhost_sender_replicating_charts_minus_one(host) (__atomic_sub_fetch(&((host)->stream.snd.status.replication.charts), 1, __ATOMIC_RELAXED))

1264 #define rrdhost_sender_replicating_charts_zero(host) (__atomic_store_n(&((host)->stream.snd.status.replication.charts), 0, __ATOMIC_RELAXED))

1265

1266 -#define rrdhost_is_online(host) ( \

1266 +#define rrdhost_is_local(host) ( \

1267 (host) == localhost || \

1268 - rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST) || \

1268 + rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST) \

1269 + )

1270 +

1271 +#define rrdhost_is_online(host) ( \

1272 + rrdhost_is_local(host) || \

1273 (rrdhost_flag_check(host, RRDHOST_FLAG_COLLECTOR_ONLINE) && !rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN)) \

1274 )

1275

src/database/rrdfunctions.c

+1 -1

@@ -296,7 +296,7 @@ int rrd_functions_find_by_name(RRDHOST *host, BUFFER *wb, const char *name, size

296 rrd_collector_running(rdcf->collector) ? "yes" : "no",

297 host->stream.rcv.status.tid, host->stream.snd.status.tid,

298 state_id, rdcf->rrdhost_state_id,

299 - rrdhost_system_info_hops(host->system_info)

299 + rrdhost_ingestion_hops(host)

300 );

301

302 dictionary_acquired_item_release(host->functions, *item);

src/database/sqlite/sqlite_aclk.c

+5 -11

@@ -333,11 +333,8 @@ static void node_update_timer_cb(uv_timer_t *handle)

333 struct aclk_sync_cfg_t *ahc = handle->data;

334 RRDHOST *host = ahc->host;

335

336 - rrdhost_receiver_lock(host);

337 - int live = (host == localhost || host->receiver || !(rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN))) ? 1 : 0;

338 - rrdhost_receiver_unlock(host);

339 - nd_log(NDLS_ACLK, NDLP_DEBUG,"Timer: Sending node update info for %s, LIVE = %d", rrdhost_hostname(host), live);

340 - aclk_host_state_update(host, live, 1);

336 + if(aclk_host_state_update_auto(host))

337 + uv_timer_stop(&ahc->timer);

338 }

339

340 static void close_callback(uv_handle_t *handle, void *data __maybe_unused)

@@ -426,19 +423,16 @@ static void aclk_synchronization(void *arg)

423 uv_timer_stop(&ahc->timer);

424

425 ahc->timer.data = ahc;

429 - int rc = uv_timer_start(&ahc->timer, node_update_timer_cb, schedule_time, 0);

426 + int rc = uv_timer_start(&ahc->timer, node_update_timer_cb, schedule_time, 5000);

427 if (!rc)

428 break; // Timer started, exit

429 }

430 }

431

432 // This is fallback if timer fails

436 - rrdhost_receiver_lock(host);

437 - int live = (host == localhost || host->receiver || !(rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN))) ? 1 : 0;

438 - rrdhost_receiver_unlock(host);

439 - aclk_host_state_update(host, live, 1);

440 - nd_log(NDLS_ACLK, NDLP_DEBUG,"Sending node update info for %s, LIVE = %d", rrdhost_hostname(host), live);

433 + aclk_host_state_update_auto(host);

434 break;

435 +

436 case ACLK_DATABASE_NODE_UNREGISTER:

437 sql_unregister_node(cmd.param[0]);

438 break;

src/database/sqlite/sqlite_aclk_node.c

+1 -5

@@ -135,27 +135,23 @@ void aclk_check_node_info_and_collectors(void)

135 if (!wc->node_info_send_time && !wc->node_collectors_send)

136 continue;

137

138 - bool replicating = false;

139 -

138 if (unlikely(rrdhost_receiver_replicating_charts(host))) {

139 internal_error(true, "ACLK SYNC: Host %s is still replicating in", rrdhost_hostname(host));

140 replicating_rcv++;

141 replicating_rcv_host = host->hostname;

144 - replicating = true;

142 }

143

144 if (unlikely(rrdhost_sender_replicating_charts(host))) {

145 internal_error(true, "ACLK SYNC: Host %s is still replicating out", rrdhost_hostname(host));

146 replicating_snd++;

147 replicating_snd_host = host->hostname;

151 - replicating = true;

148 }

149

150 #ifdef REPLICATION_TRACKING

151 replication_tracking_counters(host, &replay_counters);

152 #endif

153

158 - if(replicating)

154 + if(replicating_rcv)

155 continue;

156

157 bool pp_queue_empty = !(host->rrdctx.pp_queue && dictionary_entries(host->rrdctx.pp_queue));

src/database/sqlite/sqlite_metadata.c

+3 -5

@@ -458,10 +458,8 @@ struct node_instance_list *get_node_list(void)

458

459 uuid_copy(node_list[row].host_id, *host_id);

460 node_list[row].queryable = 1;

461 - node_list[row].live =

462 - (host == localhost || host->receiver || !(rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN))) ? 1 : 0;

463 - node_list[row].hops = host->system_info ? rrdhost_system_info_hops(host->system_info) :

464 - uuid_eq(*host_id, localhost->host_id.uuid) ? 0 : 1;

461 + node_list[row].live = rrdhost_ingestion_status(host) == RRDHOST_INGEST_STATUS_ONLINE ? 1 : 0;

462 + node_list[row].hops = rrdhost_ingestion_hops(host);

463 node_list[row].hostname =

464 sqlite3_column_bytes(res, 2) ? strdupz((char *)sqlite3_column_text(res, 2)) : NULL;

465 }

@@ -938,7 +936,7 @@ static int store_host_metadata(RRDHOST *host)

936 SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_os(host), 1));

937 SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_timezone(host), 1));

938 SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, "", 1));

941 - SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, host->system_info ? rrdhost_system_info_hops(host->system_info) : 0));

939 + SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, rrdhost_ingestion_hops(host)));

940 SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, host->rrd_memory_mode));

941 SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_abbrev_timezone(host), 1));

942 SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, host->utc_offset));

src/streaming/rrdhost-status.c

+19 -7

@@ -96,6 +96,18 @@ static NETDATA_DOUBLE rrdhost_sender_replication_completion_unsafe(RRDHOST *host

96 return completion;

97 }

99 +RRDHOST_INGEST_STATUS rrdhost_ingestion_status(RRDHOST *host) {

100 + RRDHOST_STATUS status;

101 + rrdhost_status(host, now_realtime_sec(), &status);

102 + return status.ingest.status;

103 +}

104 +

105 +int16_t rrdhost_ingestion_hops(RRDHOST *host) {

106 + if(rrdhost_is_local(host)) return 0;

107 + if(!host->system_info) return 1;

108 + return rrdhost_system_info_hops(host->system_info);

109 +}

110 +

111 void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {

112 memset(s, 0, sizeof(*s));

113

@@ -130,7 +142,7 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {

142 s->ingest.reason = (online) ? STREAM_HANDSHAKE_NEVER : host->stream.rcv.status.exit_reason;

143

144 rrdhost_receiver_lock(host);

133 - s->ingest.hops = (int16_t)(host->system_info ? rrdhost_system_info_hops(host->system_info) : (host == localhost) ? 0 : 1);

145 + s->ingest.hops = rrdhost_ingestion_hops(host);

146 bool has_receiver = false;

147 if (host->receiver && rrdhost_flag_check(host, RRDHOST_FLAG_COLLECTOR_ONLINE)) {

148 has_receiver = true;

@@ -144,16 +156,20 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {

156 }

157 rrdhost_receiver_unlock(host);

158

159 + s->ingest.collected.metrics = __atomic_load_n(&host->collected.metrics_count, __ATOMIC_RELAXED);

160 + s->ingest.collected.instances = __atomic_load_n(&host->collected.instances_count, __ATOMIC_RELAXED);

161 + s->ingest.collected.contexts = __atomic_load_n(&host->collected.contexts_count, __ATOMIC_RELAXED);

162 +

163 if (online) {

164 if(s->db.status == RRDHOST_DB_STATUS_INITIALIZING)

165 s->ingest.status = RRDHOST_INGEST_STATUS_INITIALIZING;

166

151 - else if (host == localhost || rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST)) {

167 + else if (rrdhost_is_local(host)) {

168 s->ingest.status = RRDHOST_INGEST_STATUS_ONLINE;

169 s->ingest.since = netdata_start_time;

170 }

171

156 - else if (s->ingest.replication.in_progress)

172 + else if (s->ingest.replication.in_progress || !s->ingest.collected.metrics)

173 s->ingest.status = RRDHOST_INGEST_STATUS_REPLICATING;

174

175 else

@@ -169,10 +185,6 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {

185 s->ingest.status = RRDHOST_INGEST_STATUS_OFFLINE;

186 }

187

172 - s->ingest.collected.metrics = __atomic_load_n(&host->collected.metrics_count, __ATOMIC_RELAXED);

173 - s->ingest.collected.instances = __atomic_load_n(&host->collected.instances_count, __ATOMIC_RELAXED);

174 - s->ingest.collected.contexts = __atomic_load_n(&host->collected.contexts_count, __ATOMIC_RELAXED);

175 -

188 if(host == localhost)

189 s->ingest.type = RRDHOST_INGEST_TYPE_LOCALHOST;

190 else if(has_receiver)

src/streaming/rrdhost-status.h

+7 -5

@@ -16,11 +16,11 @@ typedef enum __attribute__((packed)) {

16 } RRDHOST_DB_LIVENESS;

18 typedef enum __attribute__((packed)) {

19 - RRDHOST_INGEST_STATUS_ARCHIVED = 0,

20 - RRDHOST_INGEST_STATUS_INITIALIZING,

21 - RRDHOST_INGEST_STATUS_REPLICATING,

22 - RRDHOST_INGEST_STATUS_ONLINE,

23 - RRDHOST_INGEST_STATUS_OFFLINE,

19 + RRDHOST_INGEST_STATUS_ARCHIVED = 0, // an old host in the database (never connected during this session)

20 + RRDHOST_INGEST_STATUS_INITIALIZING, // contexts are still loading

21 + RRDHOST_INGEST_STATUS_REPLICATING, // receiving replication

22 + RRDHOST_INGEST_STATUS_ONLINE, // currently collecting data

23 + RRDHOST_INGEST_STATUS_OFFLINE, // a disconnected node

24 } RRDHOST_INGEST_STATUS;

26 typedef enum __attribute__((packed)) {

@@ -166,5 +166,7 @@ typedef struct rrdhost_status_t {

166 } RRDHOST_STATUS;

167

168 void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s);

169 +RRDHOST_INGEST_STATUS rrdhost_ingestion_status(RRDHOST *host);

170 +int16_t rrdhost_ingestion_hops(RRDHOST *host);

171

172 #endif //NETDATA_RRDHOST_STATUS_H

src/streaming/stream-connector.c

+1 -1

@@ -274,7 +274,7 @@ bool stream_connect(struct sender_state *s, uint16_t default_port, time_t timeou

274 // make sure the socket is closed

275 nd_sock_close(&s->sock);

276

277 - s->hops = (int16_t)(rrdhost_system_info_hops(host->system_info) + 1);

277 + s->hops = (int16_t)(rrdhost_ingestion_hops(s->host) + 1);

278

279 // reset this to make sure we have its current value

280 s->sock.verify_certificate = netdata_ssl_validate_certificate_sender;

src/web/api/v1/api_v1_info.c

+1 -1

@@ -38,7 +38,7 @@ static inline void web_client_api_request_v1_info_mirrored_hosts_status(BUFFER *

38 buffer_json_add_array_item_object(wb);

40 buffer_json_member_add_string(wb, "hostname", rrdhost_hostname(host));

41 - buffer_json_member_add_uint64(wb, "hops", host->system_info ? rrdhost_system_info_hops(host->system_info) : (host == localhost) ? 0 : 1);

41 + buffer_json_member_add_int64(wb, "hops", rrdhost_ingestion_hops(host));

42 buffer_json_member_add_boolean(wb, "reachable", (host == localhost || !rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN)));

44 buffer_json_member_add_string(wb, "guid", host->machine_guid);