| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "stream.h" |
| 4 | #include "stream-thread.h" |
| 5 | #include "stream-receiver-internals.h" |
| 6 | #include "stream-replication-sender.h" |
| 7 | |
| 8 | #if defined(__APPLE__) && !defined(TCP_KEEPIDLE) |
| 9 | #define TCP_KEEPIDLE TCP_KEEPALIVE |
| 10 | #endif |
| 11 | |
| 12 | #define CONNECTION_PROBE_AFTER_SECONDS (30) |
| 13 | #define CONNECTION_PROBE_INTERVAL_SECONDS (10) |
| 14 | #define CONNECTION_PROBE_COUNT (3) |
| 15 | |
| 16 | void svc_rrdhost_obsolete_all_charts(RRDHOST *host); |
| 17 | |
| 18 | // -------------------------------------------------------------------------------------------------------------------- |
| 19 | |
| 20 | static void stream_receiver_connected_msg(RRDHOST *host, char *dst, size_t len) { |
| 21 | time_t now = now_realtime_sec(); |
| 22 | time_t last_db_entry = 0; |
| 23 | rrdhost_retention(host, now, false, NULL, &last_db_entry); |
| 24 | |
| 25 | if(now < last_db_entry) |
| 26 | last_db_entry = now; |
| 27 | |
| 28 | if(!last_db_entry) |
| 29 | strncpyz(dst, "connected and ready to receive data, new node", len - 1); |
| 30 | else if(last_db_entry == now) |
| 31 | strncpyz(dst, "connected and ready to receive data, last sample in the db just now", len - 1); |
| 32 | else { |
| 33 | char buf[128]; |
| 34 | duration_snprintf(buf, sizeof(buf), now - last_db_entry, "s", true); |
| 35 | snprintfz(dst, len, "connected and ready to receive data, last sample in the db %s ago", buf); |
| 36 | } |
| 37 | } |
| 38 | |
| 39 | void stream_receiver_log_status(struct receiver_state *rpt, const char *msg, STREAM_HANDSHAKE reason, ND_LOG_FIELD_PRIORITY priority) { |
| 40 | // this function may be called BEFORE we spawn the receiver thread |
| 41 | // so, we need to add the fields again (it does not harm) |
| 42 | ND_LOG_STACK lgs[] = { |
| 43 | ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip), |
| 44 | ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port), |
| 45 | ND_LOG_FIELD_TXT(NDF_NIDL_NODE, (rpt->hostname && *rpt->hostname) ? rpt->hostname : ""), |
| 46 | ND_LOG_FIELD_I64(NDF_RESPONSE_CODE, stream_handshake_error_to_response_code(reason)), |
| 47 | ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_from_child_msgid), |
| 48 | ND_LOG_FIELD_END(), |
| 49 | }; |
| 50 | ND_LOG_STACK_PUSH(lgs); |
| 51 | |
| 52 | nd_log(NDLS_ACCESS, priority, "api_key:'%s' machine_guid:'%s' node:'%s' msg:'%s' reason:'%s'" |
| 53 | , (rpt->key && *rpt->key)? rpt->key : "" |
| 54 | , (rpt->machine_guid && *rpt->machine_guid) ? rpt->machine_guid : "" |
| 55 | , (rpt->hostname && *rpt->hostname) ? rpt->hostname : "" |
| 56 | , msg |
| 57 | , stream_handshake_error_to_string(reason)); |
| 58 | |
| 59 | nd_log(NDLS_DAEMON, priority, "STREAM RCV '%s' [from [%s]:%s]: %s %s%s%s" |
| 60 | , (rpt->hostname && *rpt->hostname) ? rpt->hostname : "" |
| 61 | , rpt->remote_ip, rpt->remote_port |
| 62 | , msg |
| 63 | , reason != STREAM_HANDSHAKE_NEVER?" (":"" |
| 64 | , stream_handshake_error_to_string(reason) |
| 65 | , reason != STREAM_HANDSHAKE_NEVER?")":"" |
| 66 | ); |
| 67 | |
| 68 | if(reason < 0) |
| 69 | pulse_parent_receiver_rejected(reason); |
| 70 | } |
| 71 | |
| 72 | // -------------------------------------------------------------------------------------------------------------------- |
| 73 | |
| 74 | void stream_receiver_free(struct receiver_state *rpt) { |
| 75 | nd_sock_close(&rpt->sock); |
| 76 | stream_decompressor_destroy(&rpt->thread.compressed.decompressor); |
| 77 | |
| 78 | if(rpt->system_info) |
| 79 | rrdhost_system_info_free(rpt->system_info); |
| 80 | |
| 81 | __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_receivers, sizeof(*rpt), __ATOMIC_RELAXED); |
| 82 | |
| 83 | freez(rpt->key); |
| 84 | freez(rpt->hostname); |
| 85 | freez(rpt->registry_hostname); |
| 86 | freez(rpt->machine_guid); |
| 87 | freez(rpt->os); |
| 88 | freez(rpt->timezone); |
| 89 | freez(rpt->abbrev_timezone); |
| 90 | freez(rpt->remote_ip); |
| 91 | freez(rpt->remote_port); |
| 92 | freez(rpt->program_name); |
| 93 | freez(rpt->program_version); |
| 94 | |
| 95 | string_freez(rpt->config.send.api_key); |
| 96 | string_freez(rpt->config.send.parents); |
| 97 | string_freez(rpt->config.send.charts_matching); |
| 98 | |
| 99 | buffer_free(rpt->thread.line_buffer); |
| 100 | rpt->thread.line_buffer = NULL; |
| 101 | |
| 102 | freez(rpt->thread.compressed.buf); |
| 103 | rpt->thread.compressed.buf = NULL; |
| 104 | rpt->thread.compressed.size = 0; |
| 105 | |
| 106 | rpt->thread.send_to_child.msg.session = 0; |
| 107 | rpt->thread.send_to_child.msg.meta = NULL; |
| 108 | stream_circular_buffer_destroy(rpt->thread.send_to_child.scb); |
| 109 | rpt->thread.send_to_child.scb = NULL; |
| 110 | |
| 111 | string_freez(rpt->thread.cd.id); |
| 112 | string_freez(rpt->thread.cd.filename); |
| 113 | string_freez(rpt->thread.cd.fullfilename); |
| 114 | string_freez(rpt->thread.cd.cmd); |
| 115 | rpt->thread.cd.id = NULL; |
| 116 | rpt->thread.cd.filename = NULL; |
| 117 | rpt->thread.cd.fullfilename = NULL; |
| 118 | rpt->thread.cd.cmd = NULL; |
| 119 | |
| 120 | #ifdef NETDATA_LOG_STREAM_RECEIVER |
| 121 | if(rpt->log.fp) |
| 122 | fclose(rpt->log.fp); |
| 123 | #endif |
| 124 | |
| 125 | freez(rpt); |
| 126 | } |
| 127 | |
| 128 | // -------------------------------------------------------------------------------------------------------------------- |
| 129 | |
| 130 | static int stream_receiver_response_permission_denied(struct web_client *w) { |
| 131 | // we always respond with the same message and error code |
| 132 | // to prevent an attacker from gaining info about the error |
| 133 | buffer_flush(w->response.data); |
| 134 | buffer_strcat(w->response.data, START_STREAMING_ERROR_NOT_PERMITTED); |
| 135 | return HTTP_RESP_UNAUTHORIZED; |
| 136 | } |
| 137 | |
| 138 | static int stream_receiver_response_too_busy_now(struct web_client *w) { |
| 139 | // we always respond with the same message and error code |
| 140 | // to prevent an attacker from gaining info about the error |
| 141 | buffer_flush(w->response.data); |
| 142 | buffer_strcat(w->response.data, START_STREAMING_ERROR_BUSY_TRY_LATER); |
| 143 | return HTTP_RESP_SERVICE_UNAVAILABLE; |
| 144 | } |
| 145 | |
| 146 | static void stream_receiver_takeover_web_connection(struct web_client *w, struct receiver_state *rpt) { |
| 147 | // Set the file descriptor and ssl from the web client |
| 148 | rpt->sock.fd = w->fd; |
| 149 | rpt->sock.ssl = w->ssl; |
| 150 | |
| 151 | w->ssl = NETDATA_SSL_UNSET_CONNECTION; |
| 152 | |
| 153 | WEB_CLIENT_IS_DEAD(w); |
| 154 | |
| 155 | if(web_server_mode == WEB_SERVER_MODE_STATIC_THREADED) { |
| 156 | web_client_flag_set(w, WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET); |
| 157 | } |
| 158 | else |
| 159 | w->fd = -1; |
| 160 | |
| 161 | buffer_flush(w->response.data); |
| 162 | |
| 163 | web_server_remove_current_socket_from_poll(); |
| 164 | } |
| 165 | |
| 166 | static void stream_send_error_on_taken_over_connection(struct receiver_state *rpt, const char *msg) { |
| 167 | nd_sock_send_timeout(&rpt->sock, (char *)msg, strlen(msg), 0, 5); |
| 168 | } |
| 169 | |
| 170 | static bool stream_receiver_send_first_response(struct receiver_state *rpt) { |
| 171 | // find the host for this receiver |
| 172 | { |
| 173 | // this will also update the host with our system_info |
| 174 | RRDHOST *host = rrdhost_find_or_create( |
| 175 | rpt->hostname, |
| 176 | rpt->registry_hostname, |
| 177 | rpt->machine_guid, |
| 178 | rpt->os, |
| 179 | rpt->timezone, |
| 180 | rpt->abbrev_timezone, |
| 181 | rpt->utc_offset, |
| 182 | rpt->program_name, |
| 183 | rpt->program_version, |
| 184 | rpt->config.update_every, |
| 185 | rpt->config.history, |
| 186 | rpt->config.mode, |
| 187 | rpt->config.health.enabled != CONFIG_BOOLEAN_NO, |
| 188 | rpt->config.send.enabled && rpt->config.send.parents && rpt->config.send.api_key, |
| 189 | rpt->config.send.parents, |
| 190 | rpt->config.send.api_key, |
| 191 | rpt->config.send.charts_matching, |
| 192 | rpt->config.replication.enabled, |
| 193 | rpt->config.replication.period, |
| 194 | rpt->config.replication.step, |
| 195 | rpt->system_info, |
| 196 | 0); |
| 197 | |
| 198 | rrdhost_system_info_free(rpt->system_info); |
| 199 | rpt->system_info = NULL; |
| 200 | |
| 201 | if(!host) { |
| 202 | stream_receiver_log_status( |
| 203 | rpt, |
| 204 | "rejecting streaming connection; failed to find or create the required host structure", |
| 205 | STREAM_HANDSHAKE_PARENT_INTERNAL_ERROR, NDLP_ERR); |
| 206 | |
| 207 | stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INTERNAL_ERROR); |
| 208 | return false; |
| 209 | } |
| 210 | |
| 211 | if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) { |
| 212 | stream_receiver_log_status( |
| 213 | rpt, |
| 214 | "rejecting streaming connection; host is initializing, retry later", |
| 215 | STREAM_HANDSHAKE_PARENT_IS_INITIALIZING, NDLP_NOTICE); |
| 216 | |
| 217 | stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INITIALIZATION); |
| 218 | return false; |
| 219 | } |
| 220 | |
| 221 | // this is not needed since we now have a waiting list for nodes |
| 222 | // if (unlikely(!stream_control_children_should_be_accepted())) { |
| 223 | // stream_receiver_log_status( |
| 224 | // rpt, |
| 225 | // "rejecting streaming connection; the system is backfilling higher tiers with high-resolution data, retry later", |
| 226 | // STREAM_HANDSHAKE_PARENT_IS_INITIALIZING, NDLP_NOTICE); |
| 227 | // |
| 228 | // stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INITIALIZATION); |
| 229 | // return false; |
| 230 | // } |
| 231 | |
| 232 | RRDHOST_SET_RECEIVER_RESULT result = rrdhost_set_receiver(host, rpt); |
| 233 | if (result == RRDHOST_SET_RECEIVER_CLEANUP_BUSY) { |
| 234 | stream_receiver_log_status( |
| 235 | rpt, |
| 236 | "rejecting streaming connection; internal cleanup is in progress for this node, please retry shortly", |
| 237 | STREAM_HANDSHAKE_PARENT_BUSY_TRY_LATER, NDLP_INFO); |
| 238 | |
| 239 | stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_BUSY_TRY_LATER); |
| 240 | return false; |
| 241 | } |
| 242 | if (result == RRDHOST_SET_RECEIVER_ALREADY_ATTACHED) { |
| 243 | stream_receiver_log_status( |
| 244 | rpt, |
| 245 | "rejecting streaming connection; host is already served by another receiver", |
| 246 | STREAM_HANDSHAKE_PARENT_NODE_ALREADY_CONNECTED, NDLP_INFO); |
| 247 | |
| 248 | stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_ALREADY_STREAMING); |
| 249 | return false; |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | #ifdef NETDATA_INTERNAL_CHECKS |
| 254 | netdata_log_info("STREAM RCV '%s' [from [%s]:%s]: " |
| 255 | "client willing to stream metrics for host '%s' with machine_guid '%s': " |
| 256 | "update every = %d, history = %d, memory mode = %s, health %s,%s" |
| 257 | , rpt->hostname |
| 258 | , rpt->remote_ip, rpt->remote_port, rrdhost_hostname(rpt->host) |
| 259 | , rpt->host->machine_guid |
| 260 | , rpt->host->rrd_update_every |
| 261 | , rpt->host->rrd_history_entries |
| 262 | , rrd_memory_mode_name(rpt->host->rrd_memory_mode) |
| 263 | , (rpt->config.health.enabled == CONFIG_BOOLEAN_NO)?"disabled":((rpt->config.health.enabled == CONFIG_BOOLEAN_YES)?"enabled":"auto") |
| 264 | , (rpt->sock.ssl.conn != NULL) ? " SSL," : "" |
| 265 | ); |
| 266 | #endif // NETDATA_INTERNAL_CHECKS |
| 267 | |
| 268 | stream_select_receiver_compression_algorithm(rpt); |
| 269 | |
| 270 | { |
| 271 | // netdata_log_info("STREAM RCV %s [from [%s]:%s]: initializing communication...", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port); |
| 272 | char initial_response[HTTP_HEADER_SIZE]; |
| 273 | if (stream_has_capability(rpt, STREAM_CAP_VCAPS)) { |
| 274 | log_receiver_capabilities(rpt); |
| 275 | sprintf(initial_response, "%s%u", START_STREAMING_PROMPT_VN, rpt->capabilities); |
| 276 | } |
| 277 | else if (stream_has_capability(rpt, STREAM_CAP_VN)) { |
| 278 | log_receiver_capabilities(rpt); |
| 279 | sprintf(initial_response, "%s%d", START_STREAMING_PROMPT_VN, stream_capabilities_to_vn(rpt->capabilities)); |
| 280 | } |
| 281 | else if (stream_has_capability(rpt, STREAM_CAP_V2)) { |
| 282 | log_receiver_capabilities(rpt); |
| 283 | sprintf(initial_response, "%s", START_STREAMING_PROMPT_V2); |
| 284 | } |
| 285 | else { // stream_has_capability(rpt, STREAM_CAP_V1) |
| 286 | log_receiver_capabilities(rpt); |
| 287 | sprintf(initial_response, "%s", START_STREAMING_PROMPT_V1); |
| 288 | } |
| 289 | |
| 290 | // OUR FIRST RESPONSE IS READY! |
| 291 | |
| 292 | // web server sockets are non-blocking - set them to blocking mode |
| 293 | { |
| 294 | // remove the non-blocking flag from the socket |
| 295 | if(sock_setnonblock(rpt->sock.fd, false) != 0) |
| 296 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 297 | "STREAM RCV '%s' [from [%s]:%s]: cannot remove the non-blocking flag from socket %d", |
| 298 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd); |
| 299 | |
| 300 | struct timeval timeout; |
| 301 | timeout.tv_sec = 600; |
| 302 | timeout.tv_usec = 0; |
| 303 | if (unlikely(setsockopt(rpt->sock.fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout) != 0)) |
| 304 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 305 | "STREAM RCV '%s' [from [%s]:%s]: cannot set timeout for socket %d", |
| 306 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd); |
| 307 | |
| 308 | // Enable TCP keepalive to detect dead connections faster |
| 309 | // When a child vanishes (e.g., VM powered off), the socket won't close normally. |
| 310 | // TCP keepalive will probe the connection and detect it's dead. |
| 311 | int enable = 1; |
| 312 | int idle = CONNECTION_PROBE_AFTER_SECONDS; |
| 313 | int interval = CONNECTION_PROBE_INTERVAL_SECONDS; |
| 314 | int count = CONNECTION_PROBE_COUNT; |
| 315 | |
| 316 | if (setsockopt(rpt->sock.fd, SOL_SOCKET, SO_KEEPALIVE, &enable, sizeof(enable)) != 0) |
| 317 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 318 | "STREAM RCV '%s' [from [%s]:%s]: cannot enable SO_KEEPALIVE on socket %d", |
| 319 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd); |
| 320 | #ifdef TCP_KEEPIDLE |
| 321 | if (setsockopt(rpt->sock.fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle)) != 0) |
| 322 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 323 | "STREAM RCV '%s' [from [%s]:%s]: cannot set TCP_KEEPIDLE on socket %d", |
| 324 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd); |
| 325 | if (setsockopt(rpt->sock.fd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(interval)) != 0) |
| 326 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 327 | "STREAM RCV '%s' [from [%s]:%s]: cannot set TCP_KEEPINTVL on socket %d", |
| 328 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd); |
| 329 | if (setsockopt(rpt->sock.fd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(count)) != 0) |
| 330 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 331 | "STREAM RCV '%s' [from [%s]:%s]: cannot set TCP_KEEPCNT on socket %d", |
| 332 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd); |
| 333 | #endif |
| 334 | } |
| 335 | |
| 336 | netdata_log_debug(D_STREAM, "Initial response to %s: %s", rpt->remote_ip, initial_response); |
| 337 | ssize_t bytes_sent = nd_sock_send_timeout(&rpt->sock, initial_response, strlen(initial_response), 0, 60); |
| 338 | |
| 339 | if(bytes_sent != (ssize_t)strlen(initial_response)) { |
| 340 | internal_error(true, "Cannot send response, got %zd bytes, expecting %zu bytes", bytes_sent, strlen(initial_response)); |
| 341 | stream_receiver_log_status( |
| 342 | rpt, |
| 343 | "cannot reply back, dropping connection", |
| 344 | STREAM_HANDSHAKE_CONNECT_SEND_TIMEOUT, NDLP_ERR); |
| 345 | rrdhost_clear_receiver(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_WRITE_FAILED); |
| 346 | return false; |
| 347 | } |
| 348 | } |
| 349 | |
| 350 | return true; |
| 351 | } |
| 352 | |
| 353 | int stream_receiver_accept_connection(struct web_client *w, char *decoded_query_string) { |
| 354 | pulse_parent_receiver_request(); |
| 355 | |
| 356 | if(!service_running(ABILITY_STREAMING_CONNECTIONS)) |
| 357 | return stream_receiver_response_too_busy_now(w); |
| 358 | |
| 359 | struct receiver_state *rpt = callocz(1, sizeof(*rpt)); |
| 360 | rpt->thread.compressed.size = COMPRESSION_MAX_CHUNK; |
| 361 | rpt->thread.compressed.buf = mallocz(rpt->thread.compressed.size); |
| 362 | rpt->connected_since_s = now_realtime_sec(); |
| 363 | rpt->thread.last_traffic_ut = now_monotonic_usec(); |
| 364 | rpt->hops = 1; |
| 365 | |
| 366 | rpt->capabilities = STREAM_CAP_INVALID; |
| 367 | |
| 368 | __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_receivers, sizeof(*rpt), __ATOMIC_RELAXED); |
| 369 | |
| 370 | rpt->system_info = rrdhost_system_info_create(); |
| 371 | rrdhost_system_info_hops_set(rpt->system_info, rpt->hops); |
| 372 | |
| 373 | nd_sock_init(&rpt->sock, netdata_ssl_web_server_ctx, false); |
| 374 | rpt->remote_ip = strdupz(w->user_auth.client_ip); |
| 375 | rpt->remote_port = strdupz(w->client_port); |
| 376 | |
| 377 | rpt->config.update_every = nd_profile.update_every; |
| 378 | |
| 379 | // parse the parameters and fill rpt and rpt->system_info |
| 380 | |
| 381 | while(decoded_query_string) { |
| 382 | char *value = strsep_skip_consecutive_separators(&decoded_query_string, "&"); |
| 383 | if(!value || !*value) continue; |
| 384 | |
| 385 | char *name = strsep_skip_consecutive_separators(&value, "="); |
| 386 | if(!name || !*name) continue; |
| 387 | if(!value || !*value) continue; |
| 388 | |
| 389 | if(!strcmp(name, "key") && !rpt->key) |
| 390 | rpt->key = strdupz(value); |
| 391 | |
| 392 | else if(!strcmp(name, "hostname") && !rpt->hostname) |
| 393 | rpt->hostname = strdupz(value); |
| 394 | |
| 395 | else if(!strcmp(name, "registry_hostname") && !rpt->registry_hostname) |
| 396 | rpt->registry_hostname = strdupz(value); |
| 397 | |
| 398 | else if(!strcmp(name, "machine_guid") && !rpt->machine_guid) |
| 399 | rpt->machine_guid = strdupz(value); |
| 400 | |
| 401 | else if(!strcmp(name, "update_every")) |
| 402 | rpt->config.update_every = (int)strtoul(value, NULL, 0); |
| 403 | |
| 404 | else if(!strcmp(name, "os") && !rpt->os) |
| 405 | rpt->os = strdupz(value); |
| 406 | |
| 407 | else if(!strcmp(name, "timezone") && !rpt->timezone) |
| 408 | rpt->timezone = strdupz(value); |
| 409 | |
| 410 | else if(!strcmp(name, "abbrev_timezone") && !rpt->abbrev_timezone) |
| 411 | rpt->abbrev_timezone = strdupz(value); |
| 412 | |
| 413 | else if(!strcmp(name, "utc_offset")) |
| 414 | rpt->utc_offset = (int32_t)strtol(value, NULL, 0); |
| 415 | |
| 416 | else if(!strcmp(name, "hops")) { |
| 417 | rpt->hops = (int16_t)strtol(value, NULL, 0); |
| 418 | rrdhost_system_info_hops_set(rpt->system_info, rpt->hops); |
| 419 | } |
| 420 | |
| 421 | else if(!strcmp(name, "ml_capable")) |
| 422 | rrdhost_system_info_ml_capable_set(rpt->system_info, str2i(value)); |
| 423 | |
| 424 | else if(!strcmp(name, "ml_enabled")) |
| 425 | rrdhost_system_info_ml_enabled_set(rpt->system_info, str2i(value)); |
| 426 | |
| 427 | else if(!strcmp(name, "mc_version")) |
| 428 | rrdhost_system_info_mc_version_set(rpt->system_info, str2i(value)); |
| 429 | |
| 430 | else if(!strcmp(name, "ver") && (rpt->capabilities & STREAM_CAP_INVALID)) |
| 431 | rpt->capabilities = convert_stream_version_to_capabilities(strtoul(value, NULL, 0), NULL, false); |
| 432 | |
| 433 | else { |
| 434 | // An old Netdata child does not have a compatible streaming protocol, map to something sane. |
| 435 | if (!strcmp(name, "NETDATA_SYSTEM_OS_NAME")) |
| 436 | name = "NETDATA_HOST_OS_NAME"; |
| 437 | |
| 438 | else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID")) |
| 439 | name = "NETDATA_HOST_OS_ID"; |
| 440 | |
| 441 | else if (!strcmp(name, "NETDATA_SYSTEM_OS_ID_LIKE")) |
| 442 | name = "NETDATA_HOST_OS_ID_LIKE"; |
| 443 | |
| 444 | else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION")) |
| 445 | name = "NETDATA_HOST_OS_VERSION"; |
| 446 | |
| 447 | else if (!strcmp(name, "NETDATA_SYSTEM_OS_VERSION_ID")) |
| 448 | name = "NETDATA_HOST_OS_VERSION_ID"; |
| 449 | |
| 450 | else if (!strcmp(name, "NETDATA_SYSTEM_OS_DETECTION")) |
| 451 | name = "NETDATA_HOST_OS_DETECTION"; |
| 452 | |
| 453 | else if(!strcmp(name, "NETDATA_PROTOCOL_VERSION") && (rpt->capabilities & STREAM_CAP_INVALID)) |
| 454 | rpt->capabilities = convert_stream_version_to_capabilities(1, NULL, false); |
| 455 | |
| 456 | if (unlikely(rrdhost_system_info_set_by_name(rpt->system_info, name, value))) { |
| 457 | nd_log_daemon(NDLP_NOTICE, "STREAM RCV '%s' [from [%s]:%s]: " |
| 458 | "request has parameter '%s' = '%s', which is not used." |
| 459 | , (rpt->hostname && *rpt->hostname) ? rpt->hostname : "-" |
| 460 | , rpt->remote_ip, rpt->remote_port, name, value); |
| 461 | } |
| 462 | } |
| 463 | } |
| 464 | |
| 465 | if (rpt->capabilities & STREAM_CAP_INVALID) |
| 466 | // no version is supplied, assume version 0; |
| 467 | rpt->capabilities = convert_stream_version_to_capabilities(0, NULL, false); |
| 468 | |
| 469 | // find the program name and version |
| 470 | if(w->user_agent && w->user_agent[0]) { |
| 471 | char *t = strchr(w->user_agent, '/'); |
| 472 | if(t && *t) { |
| 473 | *t = '\0'; |
| 474 | t++; |
| 475 | } |
| 476 | |
| 477 | rpt->program_name = strdupz(w->user_agent); |
| 478 | if(t && *t) rpt->program_version = strdupz(t); |
| 479 | } |
| 480 | |
| 481 | // check if we should accept this connection |
| 482 | |
| 483 | if(!rpt->key || !*rpt->key) { |
| 484 | stream_receiver_log_status( |
| 485 | rpt, |
| 486 | "rejecting streaming connection; request without an API key", |
| 487 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 488 | |
| 489 | stream_receiver_free(rpt); |
| 490 | return stream_receiver_response_permission_denied(w); |
| 491 | } |
| 492 | |
| 493 | if(!rpt->hostname || !*rpt->hostname) { |
| 494 | stream_receiver_log_status( |
| 495 | rpt, |
| 496 | "rejecting streaming connection; request without a hostname", |
| 497 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 498 | |
| 499 | stream_receiver_free(rpt); |
| 500 | return stream_receiver_response_permission_denied(w); |
| 501 | } |
| 502 | |
| 503 | if(!rpt->registry_hostname) |
| 504 | rpt->registry_hostname = strdupz(rpt->hostname); |
| 505 | |
| 506 | if(!rpt->machine_guid || !*rpt->machine_guid) { |
| 507 | stream_receiver_log_status( |
| 508 | rpt, |
| 509 | "rejecting streaming connection; request without a machine UUID", |
| 510 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 511 | |
| 512 | stream_receiver_free(rpt); |
| 513 | return stream_receiver_response_permission_denied(w); |
| 514 | } |
| 515 | |
| 516 | { |
| 517 | char buf[GUID_LEN + 1]; |
| 518 | |
| 519 | if (regenerate_guid(rpt->key, buf) == -1) { |
| 520 | stream_receiver_log_status( |
| 521 | rpt, |
| 522 | "rejecting streaming connection; API key is not a valid UUID (use the command uuidgen to generate one)", |
| 523 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 524 | |
| 525 | stream_receiver_free(rpt); |
| 526 | return stream_receiver_response_permission_denied(w); |
| 527 | } |
| 528 | |
| 529 | if (regenerate_guid(rpt->machine_guid, buf) == -1) { |
| 530 | stream_receiver_log_status( |
| 531 | rpt, |
| 532 | "rejecting streaming connection; machine UUID is not a valid UUID", |
| 533 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 534 | |
| 535 | stream_receiver_free(rpt); |
| 536 | return stream_receiver_response_permission_denied(w); |
| 537 | } |
| 538 | } |
| 539 | |
| 540 | if(!stream_conf_is_key_type(rpt->key, "api")) { |
| 541 | stream_receiver_log_status( |
| 542 | rpt, |
| 543 | "rejecting streaming connection; API key provided is a machine UUID (did you mix them up?)", |
| 544 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 545 | |
| 546 | stream_receiver_free(rpt); |
| 547 | return stream_receiver_response_permission_denied(w); |
| 548 | } |
| 549 | |
| 550 | // the default for api keys is false, so that users |
| 551 | // have to enable them manually |
| 552 | if(!stream_conf_api_key_is_enabled(rpt->key, false)) { |
| 553 | stream_receiver_log_status( |
| 554 | rpt, |
| 555 | "rejecting streaming connection; API key is not enabled in stream.conf", |
| 556 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 557 | |
| 558 | stream_receiver_free(rpt); |
| 559 | return stream_receiver_response_permission_denied(w); |
| 560 | } |
| 561 | |
| 562 | if(!stream_conf_api_key_allows_client(rpt->key, w->user_auth.client_ip)) { |
| 563 | stream_receiver_log_status( |
| 564 | rpt, |
| 565 | "rejecting streaming connection; API key is not allowed from this IP", |
| 566 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 567 | |
| 568 | stream_receiver_free(rpt); |
| 569 | return stream_receiver_response_permission_denied(w); |
| 570 | } |
| 571 | |
| 572 | if (!stream_conf_is_key_type(rpt->machine_guid, "machine")) { |
| 573 | stream_receiver_log_status( |
| 574 | rpt, |
| 575 | "rejecting streaming connection; machine UUID is an API key (did you mix them up?)", |
| 576 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 577 | |
| 578 | stream_receiver_free(rpt); |
| 579 | return stream_receiver_response_permission_denied(w); |
| 580 | } |
| 581 | |
| 582 | // the default for machine guids is true, so that users do not |
| 583 | // have to enable them manually |
| 584 | if(!stream_conf_api_key_is_enabled(rpt->machine_guid, true)) { |
| 585 | stream_receiver_log_status( |
| 586 | rpt, |
| 587 | "rejecting streaming connection; machine UUID is not enabled in stream.conf", |
| 588 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 589 | |
| 590 | stream_receiver_free(rpt); |
| 591 | return stream_receiver_response_permission_denied(w); |
| 592 | } |
| 593 | |
| 594 | if(!stream_conf_api_key_allows_client(rpt->machine_guid, w->user_auth.client_ip)) { |
| 595 | stream_receiver_log_status( |
| 596 | rpt, |
| 597 | "rejecting streaming connection; machine UUID is not allowed from this IP", |
| 598 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 599 | |
| 600 | stream_receiver_free(rpt); |
| 601 | return stream_receiver_response_permission_denied(w); |
| 602 | } |
| 603 | |
| 604 | if (strcmp(rpt->machine_guid, localhost->machine_guid) == 0) { |
| 605 | stream_receiver_takeover_web_connection(w, rpt); |
| 606 | |
| 607 | stream_receiver_log_status( |
| 608 | rpt, |
| 609 | "rejecting streaming connection; machine UUID is my own", |
| 610 | STREAM_HANDSHAKE_PARENT_IS_LOCALHOST, NDLP_DEBUG); |
| 611 | |
| 612 | char initial_response[HTTP_HEADER_SIZE + 1]; |
| 613 | snprintfz(initial_response, HTTP_HEADER_SIZE, "%s", START_STREAMING_ERROR_SAME_LOCALHOST); |
| 614 | |
| 615 | if(nd_sock_send_timeout(&rpt->sock, initial_response, strlen(initial_response), 0, 60) != |
| 616 | (ssize_t)strlen(initial_response)) { |
| 617 | |
| 618 | nd_log_daemon(NDLP_ERR, "STREAM RCV '%s' [from [%s]:%s]: failed to reply.", |
| 619 | rpt->hostname, rpt->remote_ip, rpt->remote_port); |
| 620 | } |
| 621 | |
| 622 | stream_receiver_free(rpt); |
| 623 | return HTTP_RESP_OK; |
| 624 | } |
| 625 | |
| 626 | { |
| 627 | RRDHOST *existing = rrdhost_find_by_guid(rpt->machine_guid); |
| 628 | // RRDHOST_OPTION_VIRTUAL_HOST is only set by local collectors (pluginsd_host_define_end), |
| 629 | // never by streaming. The stale detection in pluginsd_host() clears it when a vnode stops |
| 630 | // being collected, so archived/orphaned vnodes will not have this flag set. |
| 631 | // No additional checks for RRDHOST_FLAG_COLLECTOR_ONLINE or RRDHOST_FLAG_ARCHIVED are needed. |
| 632 | if(existing && rrdhost_is_virtual(existing)) { |
| 633 | stream_receiver_takeover_web_connection(w, rpt); |
| 634 | |
| 635 | stream_receiver_log_status( |
| 636 | rpt, |
| 637 | "rejecting streaming connection; this is a locally collected vnode", |
| 638 | STREAM_HANDSHAKE_PARENT_VNODE_IS_LOCAL, NDLP_DEBUG); |
| 639 | |
| 640 | char initial_response[HTTP_HEADER_SIZE + 1]; |
| 641 | snprintfz(initial_response, HTTP_HEADER_SIZE, "%s", START_STREAMING_ERROR_LOCAL_VNODE); |
| 642 | |
| 643 | if(nd_sock_send_timeout(&rpt->sock, initial_response, strlen(initial_response), 0, 60) != |
| 644 | (ssize_t)strlen(initial_response)) { |
| 645 | nd_log_daemon(NDLP_ERR, "STREAM RCV '%s' [from [%s]:%s]: failed to reply.", |
| 646 | rpt->hostname, rpt->remote_ip, rpt->remote_port); |
| 647 | } |
| 648 | |
| 649 | stream_receiver_free(rpt); |
| 650 | return HTTP_RESP_OK; |
| 651 | } |
| 652 | } |
| 653 | |
| 654 | if(unlikely(web_client_streaming_rate_t > 0)) { |
| 655 | static SPINLOCK spinlock = SPINLOCK_INITIALIZER; |
| 656 | static time_t last_stream_accepted_t = 0; |
| 657 | |
| 658 | time_t now = now_realtime_sec(); |
| 659 | spinlock_lock(&spinlock); |
| 660 | |
| 661 | if(unlikely(last_stream_accepted_t == 0)) |
| 662 | last_stream_accepted_t = now; |
| 663 | |
| 664 | if(now - last_stream_accepted_t < web_client_streaming_rate_t) { |
| 665 | spinlock_unlock(&spinlock); |
| 666 | |
| 667 | char msg[100 + 1]; |
| 668 | snprintfz(msg, sizeof(msg) - 1, |
| 669 | "rejecting streaming connection; rate limit, will accept new connection in %ld secs", |
| 670 | (long)(web_client_streaming_rate_t - (now - last_stream_accepted_t))); |
| 671 | |
| 672 | stream_receiver_log_status(rpt, msg, STREAM_HANDSHAKE_PARENT_BUSY_TRY_LATER, NDLP_NOTICE); |
| 673 | |
| 674 | stream_receiver_free(rpt); |
| 675 | return stream_receiver_response_too_busy_now(w); |
| 676 | } |
| 677 | |
| 678 | last_stream_accepted_t = now; |
| 679 | spinlock_unlock(&spinlock); |
| 680 | } |
| 681 | |
| 682 | /* |
| 683 | * Quick path for rejecting multiple connections. The lock taken is fine-grained - it only protects the receiver |
| 684 | * pointer within the host (if a host exists). This protects against multiple concurrent web requests hitting |
| 685 | * separate threads within the web-server and landing here. The lock guards the thread-shutdown sequence that |
| 686 | * detaches the receiver from the host. If the host is being created (first time-access) then we also use the |
| 687 | * lock to prevent race-hazard (two threads try to create the host concurrently, one wins and the other does a |
| 688 | * lookup to the now-attached structure). |
| 689 | */ |
| 690 | |
| 691 | { |
| 692 | time_t age = 0; |
| 693 | bool receiver_stale = false; |
| 694 | bool receiver_working = false; |
| 695 | |
| 696 | rrd_rdlock(); |
| 697 | RRDHOST *host = rrdhost_find_by_guid(rpt->machine_guid); |
| 698 | if (unlikely(host && rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))) /* Ignore archived hosts. */ |
| 699 | host = NULL; |
| 700 | |
| 701 | if (host) { |
| 702 | rrdhost_receiver_lock(host); |
| 703 | if (host->receiver) { |
| 704 | age =(time_t)((now_monotonic_usec() - host->receiver->thread.last_traffic_ut) / USEC_PER_SEC); |
| 705 | |
| 706 | if (age < 30) |
| 707 | receiver_working = true; |
| 708 | else |
| 709 | receiver_stale = true; |
| 710 | } |
| 711 | rrdhost_receiver_unlock(host); |
| 712 | } |
| 713 | rrd_rdunlock(); |
| 714 | |
| 715 | if (receiver_stale && string_strcmp(host->hostname, rpt->hostname) != 0) { |
| 716 | stream_receiver_log_status( |
| 717 | rpt, |
| 718 | "rejecting streaming connection; machine GUID is connected with a different hostname", |
| 719 | STREAM_HANDSHAKE_PARENT_DENIED_ACCESS, NDLP_WARNING); |
| 720 | |
| 721 | stream_receiver_free(rpt); |
| 722 | return stream_receiver_response_permission_denied(w); |
| 723 | } |
| 724 | |
| 725 | if (receiver_stale && |
| 726 | stream_receiver_signal_to_stop_and_wait(host, STREAM_HANDSHAKE_RCV_DISCONNECT_STALE_RECEIVER)) { |
| 727 | // we stopped the receiver |
| 728 | // we can proceed with this connection |
| 729 | receiver_stale = false; |
| 730 | |
| 731 | nd_log_daemon(NDLP_NOTICE, "STREAM RCV '%s' [from [%s]:%s]: " |
| 732 | "stopped previous stale receiver to accept this one." |
| 733 | , rpt->hostname |
| 734 | , rpt->remote_ip, rpt->remote_port); |
| 735 | } |
| 736 | |
| 737 | if (receiver_working || receiver_stale) { |
| 738 | // another receiver is already connected |
| 739 | // try again later |
| 740 | |
| 741 | char msg[200 + 1]; |
| 742 | snprintfz(msg, sizeof(msg) - 1, |
| 743 | "rejecting streaming connection; multiple connections for the same host, " |
| 744 | "old connection was last used %ld secs ago%s", |
| 745 | age, receiver_stale ? " (signaled old receiver to stop)" : " (new connection not accepted)"); |
| 746 | |
| 747 | stream_receiver_log_status(rpt, msg, STREAM_HANDSHAKE_PARENT_NODE_ALREADY_CONNECTED, NDLP_WARNING); |
| 748 | |
| 749 | // Have not set WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET - caller should clean up |
| 750 | buffer_flush(w->response.data); |
| 751 | buffer_strcat(w->response.data, START_STREAMING_ERROR_ALREADY_STREAMING); |
| 752 | stream_receiver_free(rpt); |
| 753 | return HTTP_RESP_CONFLICT; |
| 754 | } |
| 755 | } |
| 756 | |
| 757 | stream_receiver_takeover_web_connection(w, rpt); |
| 758 | |
| 759 | // after this point, our response code is irrelevant |
| 760 | // the socket is now ours... |
| 761 | |
| 762 | // read the configuration for this receiver |
| 763 | stream_conf_receiver_config(rpt, &rpt->config, rpt->key, rpt->machine_guid); |
| 764 | |
| 765 | if(stream_receiver_send_first_response(rpt)) { |
| 766 | // we are the receiver of the node |
| 767 | |
| 768 | // mark all charts as obsolete |
| 769 | svc_rrdhost_obsolete_all_charts(rpt->host); |
| 770 | |
| 771 | char msg[256]; |
| 772 | stream_receiver_connected_msg(rpt->host, msg, sizeof(msg)); |
| 773 | stream_receiver_log_status(rpt, msg, 0, NDLP_INFO); |
| 774 | |
| 775 | // in case we have cloud connection we inform cloud a new child connected |
| 776 | schedule_node_state_update(rpt->host, 300); |
| 777 | rrdhost_set_is_parent_label(); |
| 778 | |
| 779 | // let it reconnect to parents asap |
| 780 | stream_parents_host_reset(rpt->host, STREAM_HANDSHAKE_SP_PREPARING); |
| 781 | |
| 782 | // add it to a stream thread queue |
| 783 | stream_receiver_add_to_queue(rpt); |
| 784 | } |
| 785 | else { |
| 786 | // we are not the receiver of the node |
| 787 | // the child has been notified (or we couldn't send a message to it) |
| 788 | stream_receiver_free(rpt); |
| 789 | } |
| 790 | |
| 791 | return HTTP_RESP_OK; |
| 792 | } |