Fix race condition with orphan hosts (#9862)
* Fix race condition between orphan host cleanup and new streaming connections. * Remove health enabling from log replay, it will be handled at streaming connection time.
Markos Fountoulakis committed
Sep 4, 2020 at 13:03 UTC
adf253f710f62106193bbcf5f46543ba6fd93d2a
4 files changed
+22
-9
database/engine/metadata_log/metalogpluginsd.c
+1
-5
@@ -12,7 +12,6 @@ PARSER_RC metalog_pluginsd_host_action(
12
{
13
int history = 5;
14
RRD_MEMORY_MODE mode = RRD_MEMORY_MODE_DBENGINE;
15
- int health_enabled = default_health_enabled;
15
int rrdpush_enabled = default_rrdpush_enabled;
16
char *rrdpush_destination = default_rrdpush_destination;
17
char *rrdpush_api_key = default_rrdpush_api_key;
@@ -49,9 +48,6 @@ PARSER_RC metalog_pluginsd_host_action(
48
update_every = (int)appconfig_get_number(&stream_config, machine_guid, "update every", update_every);
49
if(update_every < 0) update_every = 1;
50
52
- //health_enabled = appconfig_get_boolean_ondemand(&stream_config, rpt->key, "health enabled by default", health_enabled);
53
- health_enabled = appconfig_get_boolean_ondemand(&stream_config, machine_guid, "health enabled", health_enabled);
54
-
51
//rrdpush_enabled = appconfig_get_boolean(&stream_config, rpt->key, "default proxy enabled", rrdpush_enabled);
52
rrdpush_enabled = appconfig_get_boolean(&stream_config, machine_guid, "proxy enabled", rrdpush_enabled);
53
@@ -77,7 +73,7 @@ PARSER_RC metalog_pluginsd_host_action(
73
, update_every
74
, history // entries
75
, mode
80
- , health_enabled // health enabled
76
+ , 0 // health enabled
77
, rrdpush_enabled // Push enabled
78
, rrdpush_destination //destination
79
, rrdpush_api_key // api key
database/rrdhost.c
+6
@@ -573,6 +573,12 @@ RRDHOST *rrdhost_find_or_create(
573
, rrdpush_send_charts_matching
574
, system_info);
575
}
576
+ if (host) {
577
+ rrdhost_wrlock(host);
578
+ rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
579
+ host->senders_disconnected_time = 0;
580
+ rrdhost_unlock(host);
581
+ }
582
583
rrdhost_cleanup_orphan_hosts_nolock(host);
584
streaming/receiver.c
+6
-4
@@ -421,9 +421,7 @@ static int rrdpush_receive(struct receiver_state *rpt)
421
}
422
*/
423
424
- rrdhost_flag_clear(rpt->host, RRDHOST_FLAG_ORPHAN);
424
// rpt->host->connected_senders++;
426
- rpt->host->senders_disconnected_time = 0;
425
rpt->host->labels_flag = (rpt->stream_version > 0)?LABEL_FLAG_UPDATE_STREAM:LABEL_FLAG_STOP_STREAM;
426
427
if(health_enabled != CONFIG_BOOLEAN_NO) {
@@ -453,17 +451,21 @@ static int rrdpush_receive(struct receiver_state *rpt)
451
452
// During a shutdown there is cleanup code in rrdhost that will cancel the sender thread
453
if (!netdata_exit && rpt->host) {
454
+ rrd_rdlock();
455
+ rrdhost_wrlock(rpt->host);
456
netdata_mutex_lock(&rpt->host->receiver_lock);
457
if (rpt->host->receiver == rpt) {
458
- rrdhost_wrlock(rpt->host);
458
rpt->host->senders_disconnected_time = now_realtime_sec();
459
rrdhost_flag_set(rpt->host, RRDHOST_FLAG_ORPHAN);
460
if(health_enabled == CONFIG_BOOLEAN_AUTO)
461
rpt->host->health_enabled = 0;
463
- rrdhost_unlock(rpt->host);
462
+ }
463
+ if (rpt->host->receiver == rpt) {
464
rrdpush_sender_thread_stop(rpt->host);
465
}
466
netdata_mutex_unlock(&rpt->host->receiver_lock);
467
+ rrdhost_unlock(rpt->host);
468
+ rrd_unlock();
469
}
470
471
// cleanup
streaming/rrdpush.c
+9
@@ -632,11 +632,16 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
632
* lookup to the now-attached structure).
633
*/
634
struct receiver_state *rpt = callocz(1, sizeof(*rpt));
635
+
636
+ rrd_rdlock();
637
RRDHOST *host = rrdhost_find_by_guid(machine_guid, 0);
638
if (unlikely(host && rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))) /* Ignore archived hosts. */
639
host = NULL;
640
if (host) {
641
+ rrdhost_wrlock(host);
642
netdata_mutex_lock(&host->receiver_lock);
643
+ rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
644
+ host->senders_disconnected_time = 0;
645
if (host->receiver != NULL) {
646
time_t age = now_realtime_sec() - host->receiver->last_msg_t;
647
if (age > 30) {
@@ -647,6 +652,8 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
652
}
653
else {
654
netdata_mutex_unlock(&host->receiver_lock);
655
+ rrdhost_unlock(host);
656
+ rrd_unlock();
657
log_stream_connection(w->client_ip, w->client_port, key, host->machine_guid, host->hostname,
658
"REJECTED - ALREADY CONNECTED");
659
info("STREAM %s [receive from [%s]:%s]: multiple connections for same host detected - existing connection is active (within last %ld sec), rejecting new connection.", host->hostname, w->client_ip, w->client_port, age);
@@ -659,7 +666,9 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
666
}
667
host->receiver = rpt;
668
netdata_mutex_unlock(&host->receiver_lock);
669
+ rrdhost_unlock(host);
670
}
671
+ rrd_unlock();
672
673
rpt->last_msg_t = now_realtime_sec();
674