diff options
Diffstat (limited to 'streaming/rrdpush.c')
| -rw-r--r-- | streaming/rrdpush.c | 619 |
1 files changed, 352 insertions, 267 deletions
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c index a57f1b080..256fa8282 100644 --- a/streaming/rrdpush.c +++ b/streaming/rrdpush.c @@ -108,7 +108,7 @@ int rrdpush_init() { default_rrdpush_seconds_to_replicate = config_get_number(CONFIG_SECTION_DB, "seconds to replicate", default_rrdpush_seconds_to_replicate); default_rrdpush_replication_step = config_get_number(CONFIG_SECTION_DB, "seconds per replication step", default_rrdpush_replication_step); - rrdhost_free_orphan_time = config_get_number(CONFIG_SECTION_DB, "cleanup orphan hosts after secs", rrdhost_free_orphan_time); + rrdhost_free_orphan_time_s = config_get_number(CONFIG_SECTION_DB, "cleanup orphan hosts after secs", rrdhost_free_orphan_time_s); #ifdef ENABLE_COMPRESSION default_compression_enabled = (unsigned int)appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM, @@ -295,40 +295,14 @@ static inline bool rrdpush_send_chart_definition(BUFFER *wb, RRDSET *st) { rrdsetvar_print_to_streaming_custom_chart_variables(st, wb); if (stream_has_capability(host->sender, STREAM_CAP_REPLICATION)) { - time_t first_entry_local = rrdset_first_entry_t_of_tier(st, 0); - time_t last_entry_local = st->last_updated.tv_sec; - - if(unlikely(!last_entry_local)) - last_entry_local = rrdset_last_entry_t(st); + time_t db_first_time_t, db_last_time_t; time_t now = now_realtime_sec(); - if(unlikely(last_entry_local > now)) { - internal_error(true, - "RRDSET REPLAY ERROR: 'host:%s/chart:%s' last updated time %ld is in the future, adjusting it to now %ld", - rrdhost_hostname(st->rrdhost), rrdset_id(st), - last_entry_local, now); - last_entry_local = now; - } - - if(unlikely(first_entry_local && last_entry_local && first_entry_local >= last_entry_local)) { - internal_error(true, - "RRDSET REPLAY ERROR: 'host:%s/chart:%s' first updated time %ld is equal or bigger than last updated time %ld, adjusting it last updated time - update every", - rrdhost_hostname(st->rrdhost), rrdset_id(st), - first_entry_local, last_entry_local); - first_entry_local = last_entry_local - st->update_every; - } - - if(unlikely(!first_entry_local && last_entry_local)) { - internal_error(true, - "RRDSET REPLAY ERROR: 'host:%s/chart:%s' first time %ld, last time %ld, setting both to last time", - rrdhost_hostname(st->rrdhost), rrdset_id(st), - first_entry_local, last_entry_local); - first_entry_local = last_entry_local; - } + rrdset_get_retention_of_tier_for_collected_chart(st, &db_first_time_t, &db_last_time_t, now, 0); buffer_sprintf(wb, PLUGINSD_KEYWORD_CHART_DEFINITION_END " %llu %llu %llu\n", - (unsigned long long)first_entry_local, - (unsigned long long)last_entry_local, + (unsigned long long)db_first_time_t, + (unsigned long long)db_last_time_t, (unsigned long long)now); rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS); @@ -342,17 +316,17 @@ static inline bool rrdpush_send_chart_definition(BUFFER *wb, RRDSET *st) { #endif } - st->upstream_resync_time = st->last_collected_time.tv_sec + (remote_clock_resync_iterations * st->update_every); + st->upstream_resync_time_s = st->last_collected_time.tv_sec + (remote_clock_resync_iterations * st->update_every); return replication_progress; } // sends the current chart dimensions -static void rrdpush_send_chart_metrics(BUFFER *wb, RRDSET *st, struct sender_state *s, RRDSET_FLAGS flags) { +static void rrdpush_send_chart_metrics(BUFFER *wb, RRDSET *st, struct sender_state *s __maybe_unused, RRDSET_FLAGS flags) { buffer_fast_strcat(wb, "BEGIN \"", 7); buffer_fast_strcat(wb, rrdset_id(st), string_strlen(st->id)); buffer_fast_strcat(wb, "\" ", 2); - if(stream_has_capability(s, STREAM_CAP_REPLICATION) || st->last_collected_time.tv_sec > st->upstream_resync_time) + if(st->last_collected_time.tv_sec > st->upstream_resync_time_s) buffer_print_llu(wb, st->usec_since_last_update); else buffer_fast_strcat(wb, "0", 1); @@ -399,6 +373,7 @@ bool rrdset_push_chart_definition_now(RRDSET *st) { BUFFER *wb = sender_start(host->sender); rrdpush_send_chart_definition(wb, st); sender_commit(host->sender, wb); + sender_thread_buffer_free(); return true; } @@ -463,6 +438,8 @@ void rrdpush_send_host_labels(RRDHOST *host) { buffer_sprintf(wb, "OVERWRITE %s\n", "labels"); sender_commit(host->sender, wb); + + sender_thread_buffer_free(); } void rrdpush_claimed_id(RRDHOST *host) @@ -480,6 +457,8 @@ void rrdpush_claimed_id(RRDHOST *host) rrdhost_aclk_state_unlock(host); sender_commit(host->sender, wb); + + sender_thread_buffer_free(); } int connect_to_one_of_destinations( @@ -496,20 +475,11 @@ int connect_to_one_of_destinations( for (struct rrdpush_destinations *d = host->destinations; d; d = d->next) { time_t now = now_realtime_sec(); - if(d->postpone_reconnection_until > now) { - info( - "STREAM %s: skipping destination '%s' (default port: %d) due to last error (code: %d, %s), will retry it in %d seconds", - rrdhost_hostname(host), - string2str(d->destination), - default_port, - d->last_handshake, d->last_error?d->last_error:"unset reason description", - (int)(d->postpone_reconnection_until - now)); - + if(d->postpone_reconnection_until > now) continue; - } info( - "STREAM %s: attempting to connect to '%s' (default port: %d)...", + "STREAM %s: connecting to '%s' (default port: %d)...", rrdhost_hostname(host), string2str(d->destination), default_port); @@ -528,8 +498,8 @@ int connect_to_one_of_destinations( // move the current item to the end of the list // without this, this destination will break the loop again and again // not advancing the destinations to find one that may work - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(host->destinations, d, prev, next); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(host->destinations, d, prev, next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(host->destinations, d, prev, next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(host->destinations, d, prev, next); break; } @@ -550,7 +520,9 @@ bool destinations_init_add_one(char *entry, void *data) { struct rrdpush_destinations *d = callocz(1, sizeof(struct rrdpush_destinations)); d->destination = string_strdupz(entry); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(t->list, d, prev, next); + __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_senders, sizeof(struct rrdpush_destinations), __ATOMIC_RELAXED); + + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(t->list, d, prev, next); t->count++; info("STREAM: added streaming destination No %d: '%s' to host '%s'", t->count, string2str(d->destination), rrdhost_hostname(t->host)); @@ -577,9 +549,10 @@ void rrdpush_destinations_init(RRDHOST *host) { void rrdpush_destinations_free(RRDHOST *host) { while (host->destinations) { struct rrdpush_destinations *tmp = host->destinations; - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(host->destinations, tmp, prev, next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(host->destinations, tmp, prev, next); string_freez(tmp->destination); freez(tmp); + __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_senders, sizeof(struct rrdpush_destinations), __ATOMIC_RELAXED); } host->destinations = NULL; @@ -590,25 +563,16 @@ void rrdpush_destinations_free(RRDHOST *host) { // Either the receiver lost the connection or the host is being destroyed. // The sender mutex guards thread creation, any spurious data is wiped on reconnection. -void rrdpush_sender_thread_stop(RRDHOST *host) { - +void rrdpush_sender_thread_stop(RRDHOST *host, const char *reason, bool wait) { if (!host->sender) return; netdata_mutex_lock(&host->sender->mutex); - netdata_thread_t thr = 0; if(rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN)) { - rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN); - - info("STREAM %s [send]: signaling sending thread to stop...", rrdhost_hostname(host)); - - // signal the thread that we want to join it - rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_SENDER_JOIN); - // copy the thread id, so that we will be waiting for the right one - // even if a new one has been spawn - thr = host->rrdpush_sender_thread; + host->sender->exit.shutdown = true; + host->sender->exit.reason = reason; // signal it to cancel netdata_thread_cancel(host->rrdpush_sender_thread); @@ -616,11 +580,14 @@ void rrdpush_sender_thread_stop(RRDHOST *host) { netdata_mutex_unlock(&host->sender->mutex); - if(thr != 0) { - info("STREAM %s [send]: waiting for the sending thread to stop...", rrdhost_hostname(host)); - void *result; - netdata_thread_join(thr, &result); - info("STREAM %s [send]: sending thread has exited.", rrdhost_hostname(host)); + if(wait) { + netdata_mutex_lock(&host->sender->mutex); + while(host->sender->tid) { + netdata_mutex_unlock(&host->sender->mutex); + sleep_usec(10 * USEC_PER_MS); + netdata_mutex_lock(&host->sender->mutex); + } + netdata_mutex_unlock(&host->sender->mutex); } } @@ -638,9 +605,9 @@ static void rrdpush_sender_thread_spawn(RRDHOST *host) { if(!rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN)) { char tag[NETDATA_THREAD_TAG_MAX + 1]; - snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STREAM_SENDER[%s]", rrdhost_hostname(host)); + snprintfz(tag, NETDATA_THREAD_TAG_MAX, THREAD_TAG_STREAM_SENDER "[%s]", rrdhost_hostname(host)); - if(netdata_thread_create(&host->rrdpush_sender_thread, tag, NETDATA_THREAD_OPTION_JOINABLE, rrdpush_sender_thread, (void *) host->sender)) + if(netdata_thread_create(&host->rrdpush_sender_thread, tag, NETDATA_THREAD_OPTION_DEFAULT, rrdpush_sender_thread, (void *) host->sender)) error("STREAM %s [send]: failed to create new thread for client.", rrdhost_hostname(host)); else rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN); @@ -654,7 +621,7 @@ int rrdpush_receiver_permission_denied(struct web_client *w) { // to prevent an attacker from gaining info about the error buffer_flush(w->response.data); buffer_sprintf(w->response.data, "You are not permitted to access this. Check the logs for more info."); - return 401; + return HTTP_RESP_UNAUTHORIZED; } int rrdpush_receiver_too_busy_now(struct web_client *w) { @@ -662,21 +629,42 @@ int rrdpush_receiver_too_busy_now(struct web_client *w) { // to prevent an attacker from gaining info about the error buffer_flush(w->response.data); buffer_sprintf(w->response.data, "The server is too busy now to accept this request. Try later."); - return 503; + return HTTP_RESP_SERVICE_UNAVAILABLE; } void *rrdpush_receiver_thread(void *ptr); int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) { - info("clients wants to STREAM metrics."); - char *key = NULL, *hostname = NULL, *registry_hostname = NULL, *machine_guid = NULL, *os = "unknown", *timezone = "unknown", *abbrev_timezone = "UTC", *tags = NULL; - int32_t utc_offset = 0; - int update_every = default_rrd_update_every; - uint32_t stream_version = UINT_MAX; - char buf[GUID_LEN + 1]; + if(!service_running(ABILITY_STREAMING_CONNECTIONS)) + return rrdpush_receiver_too_busy_now(w); + + struct receiver_state *rpt = callocz(1, sizeof(*rpt)); + rpt->last_msg_t = now_realtime_sec(); + rpt->capabilities = STREAM_CAP_INVALID; + rpt->hops = 1; + + __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_receivers, sizeof(*rpt), __ATOMIC_RELAXED); + __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(struct rrdhost_system_info), __ATOMIC_RELAXED); + + rpt->system_info = callocz(1, sizeof(struct rrdhost_system_info)); + rpt->system_info->hops = rpt->hops; + + rpt->fd = w->ifd; + rpt->client_ip = strdupz(w->client_ip); + rpt->client_port = strdupz(w->client_port); + + rpt->config.update_every = default_rrd_update_every; + +#ifdef ENABLE_HTTPS + rpt->ssl.conn = w->ssl.conn; + rpt->ssl.flags = w->ssl.flags; + + w->ssl.conn = NULL; + w->ssl.flags = NETDATA_SSL_START; +#endif + + // parse the parameters and fill rpt and rpt->system_info - struct rrdhost_system_info *system_info = callocz(1, sizeof(struct rrdhost_system_info)); - system_info->hops = 1; while(url) { char *value = mystrsep(&url, "&"); if(!value || !*value) continue; @@ -685,178 +673,307 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) { if(!name || !*name) continue; if(!value || !*value) continue; - if(!strcmp(name, "key")) - key = value; - else if(!strcmp(name, "hostname")) - hostname = value; - else if(!strcmp(name, "registry_hostname")) - registry_hostname = value; - else if(!strcmp(name, "machine_guid")) - machine_guid = value; + if(!strcmp(name, "key") && !rpt->key) + rpt->key = strdupz(value); + + else if(!strcmp(name, "hostname") && !rpt->hostname) + rpt->hostname = strdupz(value); + + else if(!strcmp(name, "registry_hostname") && !rpt->registry_hostname) + rpt->registry_hostname = strdupz(value); + + else if(!strcmp(name, "machine_guid") && !rpt->machine_guid) + rpt->machine_guid = strdupz(value); + else if(!strcmp(name, "update_every")) - update_every = (int)strtoul(value, NULL, 0); - else if(!strcmp(name, "os")) - os = value; - else if(!strcmp(name, "timezone")) - timezone = value; - else if(!strcmp(name, "abbrev_timezone")) - abbrev_timezone = value; + rpt->config.update_every = (int)strtoul(value, NULL, 0); + + else if(!strcmp(name, "os") && !rpt->os) + rpt->os = strdupz(value); + + else if(!strcmp(name, "timezone") && !rpt->timezone) + rpt->timezone = strdupz(value); + + else if(!strcmp(name, "abbrev_timezone") && !rpt->abbrev_timezone) + rpt->abbrev_timezone = strdupz(value); + else if(!strcmp(name, "utc_offset")) - utc_offset = (int32_t)strtol(value, NULL, 0); + rpt->utc_offset = (int32_t)strtol(value, NULL, 0); + else if(!strcmp(name, "hops")) - system_info->hops = (uint16_t) strtoul(value, NULL, 0); + rpt->hops = rpt->system_info->hops = (uint16_t) strtoul(value, NULL, 0); + else if(!strcmp(name, "ml_capable")) - system_info->ml_capable = strtoul(value, NULL, 0); + rpt->system_info->ml_capable = strtoul(value, NULL, 0); + else if(!strcmp(name, "ml_enabled")) - system_info->ml_enabled = strtoul(value, NULL, 0); + rpt->system_info->ml_enabled = strtoul(value, NULL, 0); + else if(!strcmp(name, "mc_version")) - system_info->mc_version = strtoul(value, NULL, 0); - else if(!strcmp(name, "tags")) - tags = value; - else if(!strcmp(name, "ver")) - stream_version = convert_stream_version_to_capabilities(strtoul(value, NULL, 0)); + rpt->system_info->mc_version = strtoul(value, NULL, 0); + + else if(!strcmp(name, "tags") && !rpt->tags) + rpt->tags = strdupz(value); + + else if(!strcmp(name, "ver") && (rpt->capabilities & STREAM_CAP_INVALID)) + rpt->capabilities = convert_stream_version_to_capabilities(strtoul(value, NULL, 0)); + else { // An old Netdata child does not have a compatible streaming protocol, map to something sane. if (!strcmp(name, "NETDATA_SYSTEM_OS_NAME")) name = "NETDATA_HOST_OS_NAME"; + else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID")) name = "NETDATA_HOST_OS_ID"; + else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID_LIKE")) name = "NETDATA_HOST_OS_ID_LIKE"; + else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION")) name = "NETDATA_HOST_OS_VERSION"; + else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION_ID")) name = "NETDATA_HOST_OS_VERSION_ID"; + else if (!strcmp(name, "NETDATA_SYSTEM_OS_DETECTION")) name = "NETDATA_HOST_OS_DETECTION"; - else if(!strcmp(name, "NETDATA_PROTOCOL_VERSION") && stream_version == UINT_MAX) { - stream_version = convert_stream_version_to_capabilities(1); - } - if (unlikely(rrdhost_set_system_info_variable(system_info, name, value))) { - info("STREAM [receive from [%s]:%s]: request has parameter '%s' = '%s', which is not used.", - w->client_ip, w->client_port, name, value); + else if(!strcmp(name, "NETDATA_PROTOCOL_VERSION") && (rpt->capabilities & STREAM_CAP_INVALID)) + rpt->capabilities = convert_stream_version_to_capabilities(1); + + if (unlikely(rrdhost_set_system_info_variable(rpt->system_info, name, value))) { + info("STREAM '%s' [receive from [%s]:%s]: " + "request has parameter '%s' = '%s', which is not used." + , (rpt->hostname && *rpt->hostname) ? rpt->hostname : "-" + , rpt->client_ip, rpt->client_port + , name, value); } } } - if (stream_version == UINT_MAX) - stream_version = convert_stream_version_to_capabilities(0); + if (rpt->capabilities & STREAM_CAP_INVALID) + // no version is supplied, assume version 0; + rpt->capabilities = convert_stream_version_to_capabilities(0); - if(!key || !*key) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, (key && *key)?key:"-", (machine_guid && *machine_guid)?machine_guid:"-", (hostname && *hostname)?hostname:"-", "ACCESS DENIED - NO KEY"); - error("STREAM [receive from [%s]:%s]: request without an API key. Forbidding access.", w->client_ip, w->client_port); - return rrdpush_receiver_permission_denied(w); + // find the program name and version + if(w->user_agent && w->user_agent[0]) { + char *t = strchr(w->user_agent, '/'); + if(t && *t) { + *t = '\0'; + t++; + } + + rpt->program_name = strdupz(w->user_agent); + if(t && *t) rpt->program_version = strdupz(t); } - if(!hostname || !*hostname) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, (key && *key)?key:"-", (machine_guid && *machine_guid)?machine_guid:"-", (hostname && *hostname)?hostname:"-", "ACCESS DENIED - NO HOSTNAME"); - error("STREAM [receive from [%s]:%s]: request without a hostname. Forbidding access.", w->client_ip, w->client_port); + // check if we should accept this connection + + if(!rpt->key || !*rpt->key) { + rrdpush_receive_log_status( + rpt, + "request without an API key", + "NO API KEY PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } - if(!machine_guid || !*machine_guid) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, (key && *key)?key:"-", (machine_guid && *machine_guid)?machine_guid:"-", (hostname && *hostname)?hostname:"-", "ACCESS DENIED - NO MACHINE GUID"); - error("STREAM [receive from [%s]:%s]: request without a machine GUID. Forbidding access.", w->client_ip, w->client_port); + if(!rpt->hostname || !*rpt->hostname) { + rrdpush_receive_log_status( + rpt, + "request without a hostname", + "NO HOSTNAME PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } - if(regenerate_guid(key, buf) == -1) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - INVALID KEY"); - error("STREAM [receive from [%s]:%s]: API key '%s' is not valid GUID (use the command uuidgen to generate one). Forbidding access.", w->client_ip, w->client_port, key); + if(!rpt->registry_hostname) + rpt->registry_hostname = strdupz(rpt->hostname); + + if(!rpt->machine_guid || !*rpt->machine_guid) { + rrdpush_receive_log_status( + rpt, + "request without a machine GUID", + "NO MACHINE GUID PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } - if(regenerate_guid(machine_guid, buf) == -1) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - INVALID MACHINE GUID"); - error("STREAM [receive from [%s]:%s]: machine GUID '%s' is not GUID. Forbidding access.", w->client_ip, w->client_port, machine_guid); - return rrdpush_receiver_permission_denied(w); + { + char buf[GUID_LEN + 1]; + + if (regenerate_guid(rpt->key, buf) == -1) { + rrdpush_receive_log_status( + rpt, + "API key is not a valid UUID (use the command uuidgen to generate one)", + "INVALID API KEY PERMISSION DENIED"); + + receiver_state_free(rpt); + return rrdpush_receiver_permission_denied(w); + } + + if (regenerate_guid(rpt->machine_guid, buf) == -1) { + rrdpush_receive_log_status( + rpt, + "machine GUID is not a valid UUID", + "INVALID MACHINE GUID PERMISSION DENIED"); + + receiver_state_free(rpt); + return rrdpush_receiver_permission_denied(w); + } } - const char *api_key_type = appconfig_get(&stream_config, key, "type", "api"); + const char *api_key_type = appconfig_get(&stream_config, rpt->key, "type", "api"); if(!api_key_type || !*api_key_type) api_key_type = "unknown"; if(strcmp(api_key_type, "api") != 0) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - API KEY GIVEN IS NOT API KEY"); - error("STREAM [receive from [%s]:%s]: API key '%s' is a %s GUID. Forbidding access.", w->client_ip, w->client_port, key, api_key_type); + rrdpush_receive_log_status( + rpt, + "API key is a machine GUID", + "INVALID API KEY PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } - if(!appconfig_get_boolean(&stream_config, key, "enabled", 0)) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - KEY NOT ENABLED"); - error("STREAM [receive from [%s]:%s]: API key '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, key); + if(!appconfig_get_boolean(&stream_config, rpt->key, "enabled", 0)) { + rrdpush_receive_log_status( + rpt, + "API key is not enabled", + "API KEY DISABLED PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } { - SIMPLE_PATTERN *key_allow_from = simple_pattern_create(appconfig_get(&stream_config, key, "allow from", "*"), NULL, SIMPLE_PATTERN_EXACT); + SIMPLE_PATTERN *key_allow_from = simple_pattern_create( + appconfig_get(&stream_config, rpt->key, "allow from", "*"), + NULL, SIMPLE_PATTERN_EXACT); + if(key_allow_from) { if(!simple_pattern_matches(key_allow_from, w->client_ip)) { simple_pattern_free(key_allow_from); - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - KEY NOT ALLOWED FROM THIS IP"); - error("STREAM [receive from [%s]:%s]: API key '%s' is not permitted from this IP. Forbidding access.", w->client_ip, w->client_port, key); + + rrdpush_receive_log_status( + rpt, + "API key is not allowed from this IP", + "NOT ALLOWED IP PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } + simple_pattern_free(key_allow_from); } } - const char *machine_guid_type = appconfig_get(&stream_config, machine_guid, "type", "machine"); - if(!machine_guid_type || !*machine_guid_type) machine_guid_type = "unknown"; - if(strcmp(machine_guid_type, "machine") != 0) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - MACHINE GUID GIVEN IS NOT A MACHINE GUID"); - error("STREAM [receive from [%s]:%s]: machine GUID '%s' is a %s GUID. Forbidding access.", w->client_ip, w->client_port, machine_guid, machine_guid_type); - return rrdpush_receiver_permission_denied(w); + { + const char *machine_guid_type = appconfig_get(&stream_config, rpt->machine_guid, "type", "machine"); + if (!machine_guid_type || !*machine_guid_type) machine_guid_type = "unknown"; + + if (strcmp(machine_guid_type, "machine") != 0) { + rrdpush_receive_log_status( + rpt, + "machine GUID is an API key", + "INVALID MACHINE GUID PERMISSION DENIED"); + + receiver_state_free(rpt); + return rrdpush_receiver_permission_denied(w); + } } - if(!appconfig_get_boolean(&stream_config, machine_guid, "enabled", 1)) { - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - MACHINE GUID NOT ENABLED"); - error("STREAM [receive from [%s]:%s]: machine GUID '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, machine_guid); + if(!appconfig_get_boolean(&stream_config, rpt->machine_guid, "enabled", 1)) { + rrdpush_receive_log_status( + rpt, + "machine GUID is not enabled", + "MACHINE GUID DISABLED PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } { - SIMPLE_PATTERN *machine_allow_from = simple_pattern_create(appconfig_get(&stream_config, machine_guid, "allow from", "*"), NULL, SIMPLE_PATTERN_EXACT); + SIMPLE_PATTERN *machine_allow_from = simple_pattern_create( + appconfig_get(&stream_config, rpt->machine_guid, "allow from", "*"), + NULL, SIMPLE_PATTERN_EXACT); + if(machine_allow_from) { if(!simple_pattern_matches(machine_allow_from, w->client_ip)) { simple_pattern_free(machine_allow_from); - rrdhost_system_info_free(system_info); - log_stream_connection(w->client_ip, w->client_port, key, machine_guid, hostname, "ACCESS DENIED - MACHINE GUID NOT ALLOWED FROM THIS IP"); - error("STREAM [receive from [%s]:%s]: Machine GUID '%s' is not permitted from this IP. Forbidding access.", w->client_ip, w->client_port, machine_guid); + + rrdpush_receive_log_status( + rpt, + "machine GUID is not allowed from this IP", + "NOT ALLOWED IP PERMISSION DENIED"); + + receiver_state_free(rpt); return rrdpush_receiver_permission_denied(w); } + simple_pattern_free(machine_allow_from); } } + if (strcmp(rpt->machine_guid, localhost->machine_guid) == 0) { + + rrdpush_receive_log_status( + rpt, + "machine GUID is my own", + "LOCALHOST PERMISSION DENIED"); + + char initial_response[HTTP_HEADER_SIZE + 1]; + snprintfz(initial_response, HTTP_HEADER_SIZE, "%s", START_STREAMING_ERROR_SAME_LOCALHOST); + + if(send_timeout( +#ifdef ENABLE_HTTPS + &rpt->ssl, +#endif + rpt->fd, initial_response, strlen(initial_response), 0, 60) != (ssize_t)strlen(initial_response)) { + + error("STREAM '%s' [receive from [%s]:%s]: " + "failed to reply." + , rpt->hostname + , rpt->client_ip, rpt->client_port + ); + } + + close(rpt->fd); + receiver_state_free(rpt); + return web_client_socket_is_now_used_for_streaming(w); + } + if(unlikely(web_client_streaming_rate_t > 0)) { - static netdata_mutex_t stream_rate_mutex = NETDATA_MUTEX_INITIALIZER; - static volatile time_t last_stream_accepted_t = 0; + static SPINLOCK spinlock = NETDATA_SPINLOCK_INITIALIZER; + static time_t last_stream_accepted_t = 0; - netdata_mutex_lock(&stream_rate_mutex); time_t now = now_realtime_sec(); + netdata_spinlock_lock(&spinlock); if(unlikely(last_stream_accepted_t == 0)) last_stream_accepted_t = now; if(now - last_stream_accepted_t < web_client_streaming_rate_t) { - netdata_mutex_unlock(&stream_rate_mutex); - rrdhost_system_info_free(system_info); - error("STREAM [receive from [%s]:%s]: too busy to accept new streaming request. Will be allowed in %ld secs.", w->client_ip, w->client_port, (long)(web_client_streaming_rate_t - (now - last_stream_accepted_t))); + netdata_spinlock_unlock(&spinlock); + + char msg[100 + 1]; + snprintfz(msg, 100, + "rate limit, will accept new connection in %ld secs", + (long)(web_client_streaming_rate_t - (now - last_stream_accepted_t))); + + rrdpush_receive_log_status( + rpt, + msg, + "RATE LIMIT TRY LATER"); + + receiver_state_free(rpt); return rrdpush_receiver_too_busy_now(w); } last_stream_accepted_t = now; - netdata_mutex_unlock(&stream_rate_mutex); + netdata_spinlock_unlock(&spinlock); } /* @@ -867,117 +984,85 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) { * lock to prevent race-hazard (two threads try to create the host concurrently, one wins and the other does a * lookup to the now-attached structure). */ - struct receiver_state *rpt = callocz(1, sizeof(*rpt)); - rrd_rdlock(); - RRDHOST *host = rrdhost_find_by_guid(machine_guid); - if (unlikely(host && rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))) /* Ignore archived hosts. */ - host = NULL; - if (host) { - rrdhost_wrlock(host); - netdata_mutex_lock(&host->receiver_lock); - rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN); - host->senders_disconnected_time = 0; - if (host->receiver != NULL) { - time_t age = now_realtime_sec() - host->receiver->last_msg_t; - if (age > 30) { - host->receiver->shutdown = 1; - shutdown(host->receiver->fd, SHUT_RDWR); - host->receiver = NULL; // Thread holds reference to structure - info( - "STREAM %s [receive from [%s]:%s]: multiple connections for same host detected - " - "existing connection is dead (%"PRId64" sec), accepting new connection.", - rrdhost_hostname(host), - w->client_ip, - w->client_port, - (int64_t)age); - } - else { - netdata_mutex_unlock(&host->receiver_lock); - rrdhost_unlock(host); - rrd_unlock(); - log_stream_connection(w->client_ip, w->client_port, key, host->machine_guid, rrdhost_hostname(host), - "REJECTED - ALREADY CONNECTED"); - info( - "STREAM %s [receive from [%s]:%s]: multiple connections for same host detected - " - "existing connection is active (within last %"PRId64" sec), rejecting new connection.", - rrdhost_hostname(host), - w->client_ip, - w->client_port, - (int64_t)age); - // Have not set WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET - caller should clean up - buffer_flush(w->response.data); - buffer_strcat(w->response.data, "This GUID is already streaming to this server"); - freez(rpt); - return 409; + { + time_t age = 0; + bool receiver_stale = false; + bool receiver_working = false; + + rrd_rdlock(); + RRDHOST *host = rrdhost_find_by_guid(rpt->machine_guid); + if (unlikely(host && rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))) /* Ignore archived hosts. */ + host = NULL; + + if (host) { + netdata_mutex_lock(&host->receiver_lock); + if (host->receiver) { + age = now_realtime_sec() - host->receiver->last_msg_t; + + if (age < 30) + receiver_working = true; + else + receiver_stale = true; } + netdata_mutex_unlock(&host->receiver_lock); } - host->receiver = rpt; - netdata_mutex_unlock(&host->receiver_lock); - rrdhost_unlock(host); - } - rrd_unlock(); - - rpt->last_msg_t = now_realtime_sec(); - - rpt->host = host; - rpt->fd = w->ifd; - rpt->key = strdupz(key); - rpt->hostname = strdupz(hostname); - rpt->registry_hostname = strdupz((registry_hostname && *registry_hostname)?registry_hostname:hostname); - rpt->machine_guid = strdupz(machine_guid); - rpt->os = strdupz(os); - rpt->timezone = strdupz(timezone); - rpt->abbrev_timezone = strdupz(abbrev_timezone); - rpt->utc_offset = utc_offset; - rpt->tags = (tags)?strdupz(tags):NULL; - rpt->client_ip = strdupz(w->client_ip); - rpt->client_port = strdupz(w->client_port); - rpt->update_every = update_every; - rpt->system_info = system_info; - rpt->capabilities = stream_version; -#ifdef ENABLE_HTTPS - rpt->ssl.conn = w->ssl.conn; - rpt->ssl.flags = w->ssl.flags; - - w->ssl.conn = NULL; - w->ssl.flags = NETDATA_SSL_START; -#endif - - if(w->user_agent && w->user_agent[0]) { - char *t = strchr(w->user_agent, '/'); - if(t && *t) { - *t = '\0'; - t++; + rrd_unlock(); + + if (receiver_stale && stop_streaming_receiver(host, "STALE RECEIVER")) { + // we stopped the receiver + // we can proceed with this connection + receiver_stale = false; + + info("STREAM '%s' [receive from [%s]:%s]: " + "stopped previous stale receiver to accept this one." + , rpt->hostname + , rpt->client_ip, rpt->client_port + ); } - rpt->program_name = strdupz(w->user_agent); - if(t && *t) rpt->program_version = strdupz(t); + if (receiver_working || receiver_stale) { + // another receiver is already connected + // try again later + + char msg[200 + 1]; + snprintfz(msg, 200, + "multiple connections for same host, " + "old connection was used %ld secs ago%s", + age, receiver_stale ? " (signaled old receiver to stop)" : " (new connection not accepted)"); + + rrdpush_receive_log_status( + rpt, + msg, + "ALREADY CONNECTED"); + + // Have not set WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET - caller should clean up + buffer_flush(w->response.data); + buffer_strcat(w->response.data, "This GUID is already streaming to this server"); + receiver_state_free(rpt); + return HTTP_RESP_CONFLICT; + } } - - debug(D_SYSTEM, "starting STREAM receive thread."); char tag[FILENAME_MAX + 1]; - snprintfz(tag, FILENAME_MAX, "STREAM_RECEIVER[%s,[%s]:%s]", rpt->hostname, w->client_ip, w->client_port); - - if(netdata_thread_create(&rpt->thread, tag, NETDATA_THREAD_OPTION_DEFAULT, rrdpush_receiver_thread, (void *)rpt)) - error("Failed to create new STREAM receive thread for client."); - - // prevent the caller from closing the streaming socket - if(web_server_mode == WEB_SERVER_MODE_STATIC_THREADED) { - web_client_flag_set(w, WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET); - } - else { - if(w->ifd == w->ofd) - w->ifd = w->ofd = -1; - else - w->ifd = -1; + snprintfz(tag, FILENAME_MAX, THREAD_TAG_STREAM_RECEIVER "[%s,[%s]:%s]", rpt->hostname, w->client_ip, w->client_port); + + if(netdata_thread_create(&rpt->thread, tag, NETDATA_THREAD_OPTION_DEFAULT, rrdpush_receiver_thread, (void *)rpt)) { + rrdpush_receive_log_status( + rpt, + "can't create receiver thread", + "INTERNAL SERVER ERROR"); + + buffer_flush(w->response.data); + buffer_strcat(w->response.data, "Can't handle this request"); + receiver_state_free(rpt); + return HTTP_RESP_INTERNAL_SERVER_ERROR; } - buffer_flush(w->response.data); - return 200; + // prevent the caller from closing the streaming socket + return web_client_socket_is_now_used_for_streaming(w); } static void stream_capabilities_to_string(BUFFER *wb, STREAM_CAPABILITIES caps) { @@ -995,7 +1080,7 @@ static void stream_capabilities_to_string(BUFFER *wb, STREAM_CAPABILITIES caps) } void log_receiver_capabilities(struct receiver_state *rpt) { - BUFFER *wb = buffer_create(100); + BUFFER *wb = buffer_create(100, NULL); stream_capabilities_to_string(wb, rpt->capabilities); info("STREAM %s [receive from [%s]:%s]: established link with negotiated capabilities: %s", @@ -1005,7 +1090,7 @@ void log_receiver_capabilities(struct receiver_state *rpt) { } void log_sender_capabilities(struct sender_state *s) { - BUFFER *wb = buffer_create(100); + BUFFER *wb = buffer_create(100, NULL); stream_capabilities_to_string(wb, s->capabilities); info("STREAM %s [send to %s]: established link with negotiated capabilities: %s", |