ec95f306f · cryptotaxi247/netdata-1

@@ -155,6 +155,7 @@ rrd_flag_add_remove_atomic(RRD_FLAGS *flags, RRD_FLAGS check, RRD_FLAGS conditio

155 | RRD_FLAG_DELETED \

156 | RRD_FLAG_UPDATE_REASON_STOPPED_BEING_COLLECTED \

157 | RRD_FLAG_UPDATE_REASON_ZERO_RETENTION \

158 + | RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD \

159 )

160

161 #define rrd_flag_set_archived(obj) \

@@ -272,9 +273,9 @@ typedef struct rrdinstance {

273 DICTIONARY *rrdmetrics;

274

275 struct {

275 - uint32_t collected_metrics; // a temporary variable to detect BEGIN/END without SET

276 - // don't use it for other purposes

277 - // it goes up and then resets to zero, on every iteration

276 + uint32_t collected_metrics_count; // a temporary variable to detect BEGIN/END without SET

277 + // don't use it for other purposes

278 + // it goes up and then resets to zero, on every iteration

279 } internal;

280 } RRDINSTANCE;

281

@@ -297,12 +298,20 @@ typedef struct rrdcontext {

298 DICTIONARY *rrdinstances;

299 RRDHOST *rrdhost;

300

301 + struct {

302 + RRD_FLAGS queued_flags; // the last flags that triggered the post-processing

303 + usec_t queued_ut; // the last time this was queued

304 + usec_t dequeued_ut; // the last time we sent (or deduplicated) this context

305 + size_t executions; // how many times this context has been processed

306 + } pp;

307 +

308 struct {

309 RRD_FLAGS queued_flags; // the last flags that triggered the queueing

310 usec_t queued_ut; // the last time this was queued

311 usec_t delay_calc_ut; // the last time we calculated the scheduled_dispatched_ut

312 usec_t scheduled_dispatch_ut; // the time it was/is scheduled to be sent

305 - usec_t dequeued_ut; // the last time we sent (or deduped) this context

313 + usec_t dequeued_ut; // the last time we sent (or deduplicated) this context

314 + size_t dispatches; // the number of times this has been dispatched to hub

315 } queue;

316

317 netdata_mutex_t mutex;

@@ -639,7 +648,7 @@ static inline void rrdmetric_collected_rrddim(RRDDIM *rd) {

648 rrd_flag_set_collected(rm);

649

650 // we use this variable to detect BEGIN/END without SET

642 - rm->ri->internal.collected_metrics++;

651 + rm->ri->internal.collected_metrics_count++;

652

653 rrdmetric_trigger_updates(rm, __FUNCTION__ );

654 }

@@ -1079,11 +1088,11 @@ static inline void rrdinstance_collected_rrdset(RRDSET *st) {

1088

1089 rrdinstance_updated_rrdset_flags_no_action(ri, st);

1090

1082 - if(unlikely(ri->internal.collected_metrics && !rrd_flag_is_collected(ri)))

1091 + if(unlikely(ri->internal.collected_metrics_count && !rrd_flag_is_collected(ri)))

1092 rrd_flag_set_collected(ri);

1093

1094 // we use this variable to detect BEGIN/END without SET

1086 - ri->internal.collected_metrics = 0;

1095 + ri->internal.collected_metrics_count = 0;

1096

1097 rrdinstance_trigger_updates(ri, __FUNCTION__ );

1098 }

@@ -1273,11 +1282,31 @@ static bool rrdcontext_hub_queue_conflict_callback(const DICTIONARY_ITEM *item _

1282 static void rrdcontext_post_processing_queue_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {

1283 RRDCONTEXT *rc = context;

1284 rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING);

1285 + rc->pp.queued_flags = rc->flags;

1286 + rc->pp.queued_ut = now_realtime_usec();

1287 }

1288

1289 static void rrdcontext_post_processing_queue_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {

1290 RRDCONTEXT *rc = context;

1291 rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING);

1292 + rc->pp.dequeued_ut = now_realtime_usec();

1293 +}

1294 +

1295 +static bool rrdcontext_post_processing_queue_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *new_context __maybe_unused, void *nothing __maybe_unused) {

1296 + RRDCONTEXT *rc = context;

1297 + bool changed = false;

1298 +

1299 + if(!(rc->flags & RRD_FLAG_QUEUED_FOR_POST_PROCESSING)) {

1300 + rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_POST_PROCESSING);

1301 + changed = true;

1302 + }

1303 +

1304 + if(rc->pp.queued_flags != rc->flags) {

1305 + rc->pp.queued_flags |= rc->flags;

1306 + changed = true;

1307 + }

1308 +

1309 + return changed;

1310 }

1311

1312 void rrdhost_create_rrdcontexts(RRDHOST *host) {

@@ -1291,15 +1320,14 @@ void rrdhost_create_rrdcontexts(RRDHOST *host) {

1320 dictionary_register_react_callback((DICTIONARY *)host->rrdctx, rrdcontext_react_callback, host);

1321

1322 host->rrdctx_hub_queue = (RRDCONTEXTS *)dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_VALUE_LINK_DONT_CLONE);

1294 -

1323 dictionary_register_insert_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_hub_queue_insert_callback, NULL);

1324 dictionary_register_delete_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_hub_queue_delete_callback, NULL);

1325 dictionary_register_conflict_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_hub_queue_conflict_callback, NULL);

1326

1327 host->rrdctx_post_processing_queue = (RRDCONTEXTS *)dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_VALUE_LINK_DONT_CLONE);

1300 -

1301 - dictionary_register_insert_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_post_processing_queue_insert_callback, NULL);

1302 - dictionary_register_delete_callback((DICTIONARY *)host->rrdctx_hub_queue, rrdcontext_post_processing_queue_delete_callback, NULL);

1328 + dictionary_register_insert_callback((DICTIONARY *)host->rrdctx_post_processing_queue, rrdcontext_post_processing_queue_insert_callback, NULL);

1329 + dictionary_register_delete_callback((DICTIONARY *)host->rrdctx_post_processing_queue, rrdcontext_post_processing_queue_delete_callback, NULL);

1330 + dictionary_register_conflict_callback((DICTIONARY *)host->rrdctx_post_processing_queue, rrdcontext_post_processing_queue_conflict_callback, NULL);

1331 }

1332

1333 void rrdhost_destroy_rrdcontexts(RRDHOST *host) {

@@ -1396,7 +1424,6 @@ void rrdcontext_host_child_connected(RRDHOST *host) {

1424 }

1425

1426 void rrdcontext_host_child_disconnected(RRDHOST *host) {

1399 -

1427 rrdcontext_recalculate_host_retention(host, RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD, false);

1428 }

1429

@@ -1894,14 +1921,29 @@ static inline int rrdcontext_to_json_callback(const DICTIONARY_ITEM *item, void

1921 ",\n\t\t\t\"last_queued\":%llu"

1922 ",\n\t\t\t\"scheduled_dispatch\":%llu"

1923 ",\n\t\t\t\"last_dequeued\":%llu"

1924 + ",\n\t\t\t\"dispatches\":%zu"

1925 ",\n\t\t\t\"hub_version\":%"PRIu64""

1926 ",\n\t\t\t\"version\":%"PRIu64""

1927 , rc->queue.queued_ut / USEC_PER_SEC

1928 , rc->queue.scheduled_dispatch_ut / USEC_PER_SEC

1929 , rc->queue.dequeued_ut / USEC_PER_SEC

1930 + , rc->queue.dispatches

1931 , rc->hub.version

1932 , rc->version

1933 );

1934 +

1935 + buffer_strcat(wb, ",\n\t\t\t\"pp_reasons\":\"");

1936 + rrd_reasons_to_buffer(rc->pp.queued_flags, wb);

1937 + buffer_strcat(wb, "\"");

1938 +

1939 + buffer_sprintf(wb,

1940 + ",\n\t\t\t\"pp_last_queued\":%llu"

1941 + ",\n\t\t\t\"pp_last_dequeued\":%llu"

1942 + ",\n\t\t\t\"pp_executed\":%zu"

1943 + , rc->pp.queued_ut / USEC_PER_SEC

1944 + , rc->pp.dequeued_ut / USEC_PER_SEC

1945 + , rc->pp.executions

1946 + );

1947 }

1948

1949 rrdcontext_unlock(rc);

@@ -2609,6 +2651,7 @@ static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG

2651 }

2652

2653 rrdcontext_lock(rc);

2654 + rc->pp.executions++;

2655

2656 if(unlikely(!instances_active)) {

2657 // we had some instances, but they are gone now...

@@ -2931,6 +2974,7 @@ static void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now

2974 rrdcontext_message_send_unsafe(rc, false, bundle);

2975 messages_added++;

2976

2977 + rc->queue.dispatches++;

2978 rc->queue.dequeued_ut = now_ut;

2979 }

2980 else

fix post-processing of contexts (#13807)