| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "stream-thread.h" |
| 4 | #include "stream-sender-internals.h" |
| 5 | #include "stream-replication-sender.h" |
| 6 | |
| 7 | // help the IDE detect use after free |
| 8 | #define stream_sender_move_running_to_connector_or_remove(sth, s, reason, receiver_reason, reconnect) do { \ |
| 9 | stream_sender_move_running_to_connector_or_remove_internal(sth, s, reason, receiver_reason, reconnect); \ |
| 10 | (s) = NULL; \ |
| 11 | } while(0) |
| 12 | |
| 13 | static void stream_sender_move_running_to_connector_or_remove_internal(struct stream_thread *sth, struct sender_state *s, STREAM_HANDSHAKE reason, STREAM_HANDSHAKE receiver_reason, bool reconnect); |
| 14 | |
| 15 | // -------------------------------------------------------------------------------------------------------------------- |
| 16 | |
| 17 | #ifdef NETDATA_LOG_STREAM_SENDER |
| 18 | void stream_sender_log_payload(struct sender_state *s, BUFFER *payload, STREAM_TRAFFIC_TYPE type __maybe_unused, bool inbound) { |
| 19 | spinlock_lock(&s->log.spinlock); |
| 20 | |
| 21 | if (!s->log.fp) { |
| 22 | char filename[FILENAME_MAX + 1]; |
| 23 | snprintfz( |
| 24 | filename, FILENAME_MAX, "/tmp/stream-sender-%s.txt", s->host ? rrdhost_hostname(s->host) : "unknown"); |
| 25 | |
| 26 | s->log.fp = fopen(filename, "w"); |
| 27 | |
| 28 | // Align first_call to wall clock time |
| 29 | clock_gettime(CLOCK_REALTIME, &s->log.first_call); |
| 30 | s->log.first_call.tv_nsec = 0; // Align to the start of the second |
| 31 | } |
| 32 | |
| 33 | if (s->log.fp) { |
| 34 | struct timespec now; |
| 35 | clock_gettime(CLOCK_REALTIME, &now); |
| 36 | |
| 37 | time_t elapsed_sec = now.tv_sec - s->log.first_call.tv_sec; |
| 38 | long elapsed_nsec = now.tv_nsec - s->log.first_call.tv_nsec; |
| 39 | |
| 40 | if (elapsed_nsec < 0) { |
| 41 | elapsed_sec--; |
| 42 | elapsed_nsec += 1000000000; |
| 43 | } |
| 44 | |
| 45 | uint16_t days = elapsed_sec / 86400; |
| 46 | uint8_t hours = (elapsed_sec % 86400) / 3600; |
| 47 | uint8_t minutes = (elapsed_sec % 3600) / 60; |
| 48 | uint8_t seconds = elapsed_sec % 60; |
| 49 | uint16_t milliseconds = elapsed_nsec / 1000000; |
| 50 | |
| 51 | char prefix[30]; |
| 52 | snprintf(prefix, sizeof(prefix), "%03ud.%02u:%02u:%02u.%03u ", |
| 53 | days, hours, minutes, seconds, milliseconds); |
| 54 | |
| 55 | const char *line_start = buffer_tostring(payload); |
| 56 | const char *line_end; |
| 57 | |
| 58 | while (line_start && *line_start) { |
| 59 | line_end = strchr(line_start, '\n'); |
| 60 | if (line_end) { |
| 61 | fprintf(s->log.fp, "%s%s%.*s\n", prefix, inbound ? "> " : "< ", (int)(line_end - line_start), line_start); |
| 62 | line_start = line_end + 1; |
| 63 | } else { |
| 64 | fprintf(s->log.fp, "%s%s%s\n", prefix, inbound ? "> " : "< ", line_start); |
| 65 | break; |
| 66 | } |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | // fflush(s->log.fp); |
| 71 | spinlock_unlock(&s->log.spinlock); |
| 72 | } |
| 73 | #endif |
| 74 | |
| 75 | // -------------------------------------------------------------------------------------------------------------------- |
| 76 | |
| 77 | void stream_sender_charts_and_replication_reset(struct sender_state *s) { |
| 78 | // stop all replication commands inflight |
| 79 | replication_sender_delete_pending_requests(s); |
| 80 | |
| 81 | // reset the state of all charts |
| 82 | RRDSET *st; |
| 83 | rrdset_foreach_read(st, s->host) { |
| 84 | // Decrement only when this chart actually contributed +1 to the host |
| 85 | // counter, i.e. when IN_PROGRESS was set. The previous condition |
| 86 | // (!FINISHED) over-decremented initial-state charts (no flags set, no |
| 87 | // prior +1) and relied on a force-zero safety net below to compensate. |
| 88 | // Force-zero is unsafe against concurrent claim-before-publish in |
| 89 | // stream_sender_send_rrdset_definition: a sender that has just |
| 90 | // incremented but not yet published IN_PROGRESS would be desynced from |
| 91 | // the counter we forcibly cleared. Use the precise condition instead. |
| 92 | // Pulse status is intentionally not flipped here -- the surrounding |
| 93 | // sender connect/disconnect lifecycle drives it (e.g. SND_DISCONNECTED). |
| 94 | RRDSET_FLAGS old = rrdset_flag_set_and_clear(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS); |
| 95 | if(old & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS) |
| 96 | rrdhost_sender_replicating_charts_minus_one(st->rrdhost); |
| 97 | |
| 98 | #ifdef REPLICATION_TRACKING |
| 99 | st->stream.snd.who = REPLAY_WHO_UNKNOWN; |
| 100 | #endif |
| 101 | |
| 102 | st->stream.snd.resync_time_s = 0; |
| 103 | |
| 104 | RRDDIM *rd; |
| 105 | rrddim_foreach_read(rd, st) |
| 106 | rrddim_metadata_exposed_upstream_clear(rd); |
| 107 | rrddim_foreach_done(rd); |
| 108 | |
| 109 | rrdset_metadata_updated(st); |
| 110 | } |
| 111 | rrdset_foreach_done(st); |
| 112 | |
| 113 | // Observability only. The per-chart loop now precisely balances |
| 114 | // contributions; a non-zero residual either reflects a concurrent |
| 115 | // claim-before-publish in flight (will resolve) or a real accounting bug |
| 116 | // worth investigating. Do NOT force-zero: that would desynchronize the |
| 117 | // counter from any in-flight sender's not-yet-published IN_PROGRESS flag. |
| 118 | size_t residual = rrdhost_sender_replicating_charts(s->host); |
| 119 | if(residual != 0) { |
| 120 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 121 | "STREAM REPLAY: sender replicating-charts counter is %zu after reset " |
| 122 | "(expected 0); leaving it untouched to preserve any concurrent " |
| 123 | "claim-before-publish in flight", |
| 124 | residual); |
| 125 | } |
| 126 | |
| 127 | stream_sender_replicating_charts_zero(s); |
| 128 | |
| 129 | __atomic_store_n(&s->host->stream.snd.status.replication.counter_in, 0, __ATOMIC_RELAXED); |
| 130 | __atomic_store_n(&s->host->stream.snd.status.replication.counter_out, 0, __ATOMIC_RELAXED); |
| 131 | } |
| 132 | |
| 133 | // -------------------------------------------------------------------------------------------------------------------- |
| 134 | |
| 135 | static void stream_sender_on_connect_and_disconnect(struct sender_state *s) { |
| 136 | stream_sender_execute_commands_cleanup(s); |
| 137 | stream_sender_charts_and_replication_reset(s); |
| 138 | |
| 139 | stream_sender_lock(s); |
| 140 | stream_circular_buffer_flush_unsafe(s->scb, stream_send.buffer_max_size); |
| 141 | stream_sender_unlock(s); |
| 142 | } |
| 143 | |
| 144 | void stream_sender_on_connect(struct sender_state *s) { |
| 145 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 146 | "STREAM SND [%s]: running on-connect hooks...", |
| 147 | rrdhost_hostname(s->host)); |
| 148 | |
| 149 | rrdhost_flag_set(s->host, RRDHOST_FLAG_STREAM_SENDER_CONNECTED); |
| 150 | |
| 151 | stream_sender_on_connect_and_disconnect(s); |
| 152 | |
| 153 | s->thread.last_traffic_ut = now_monotonic_usec(); |
| 154 | |
| 155 | freez(s->thread.rbuf.b); |
| 156 | s->thread.rbuf.size = PLUGINSD_LINE_MAX + 1; |
| 157 | s->thread.rbuf.b = mallocz(s->thread.rbuf.size); |
| 158 | s->thread.rbuf.b[0] = '\0'; |
| 159 | s->thread.rbuf.read_len = 0; |
| 160 | } |
| 161 | |
| 162 | static void stream_sender_on_ready_to_dispatch(struct sender_state *s) { |
| 163 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 164 | "STREAM SND '%s': running ready-to-dispatch hooks...", |
| 165 | rrdhost_hostname(s->host)); |
| 166 | |
| 167 | // set this flag before sending any data, or the data will not be sent |
| 168 | rrdhost_flag_set(s->host, RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS); |
| 169 | |
| 170 | // send our global metadata to the parent |
| 171 | stream_sender_send_custom_host_variables(s->host); |
| 172 | stream_path_send_to_parent(s->host); |
| 173 | stream_sender_send_claimed_id(s->host); |
| 174 | stream_send_host_labels(s->host); |
| 175 | stream_send_global_functions(s->host); |
| 176 | } |
| 177 | |
| 178 | void stream_sender_on_disconnect(struct sender_state *s) { |
| 179 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 180 | "STREAM SND '%s': running on-disconnect hooks...", |
| 181 | rrdhost_hostname(s->host)); |
| 182 | |
| 183 | // Stop new metadata pushes BEFORE the reset. New collectors that haven't |
| 184 | // yet entered stream_sender_send_rrdset_definition will fail the |
| 185 | // rrdhost_can_stream_metadata_to_parent() predicate and skip the |
| 186 | // bookkeeping entirely; in-flight collectors that already passed the |
| 187 | // predicate are caught by the post-CAS recheck in |
| 188 | // stream_sender_send_rrdset_definition (which then rolls back via atomic |
| 189 | // CAS, so the reset's per-chart accounting and the rollback do not |
| 190 | // double-decrement). The duplicate clear later in |
| 191 | // stream_sender_move_running_to_connector_or_remove_internal / |
| 192 | // stream_sender_remove is idempotent for atomic flag ops. |
| 193 | rrdhost_flag_clear(s->host, RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS); |
| 194 | |
| 195 | stream_sender_on_connect_and_disconnect(s); |
| 196 | |
| 197 | // update the child (the receiver side) for this parent |
| 198 | stream_path_parent_disconnected(s->host); |
| 199 | stream_receiver_send_node_and_claim_id_to_child(s->host); |
| 200 | |
| 201 | freez(s->thread.rbuf.b); |
| 202 | s->thread.rbuf.size = 0; |
| 203 | s->thread.rbuf.b = NULL; |
| 204 | s->thread.rbuf.read_len = 0; |
| 205 | } |
| 206 | |
| 207 | // -------------------------------------------------------------------------------------------------------------------- |
| 208 | |
| 209 | static bool stream_sender_log_capabilities(BUFFER *wb, void *ptr) { |
| 210 | struct sender_state *state = ptr; |
| 211 | if(!state) |
| 212 | return false; |
| 213 | |
| 214 | stream_capabilities_to_string(wb, state->capabilities); |
| 215 | return true; |
| 216 | } |
| 217 | |
| 218 | static bool stream_sender_log_transport(BUFFER *wb, void *ptr) { |
| 219 | struct sender_state *state = ptr; |
| 220 | if(!state) |
| 221 | return false; |
| 222 | |
| 223 | buffer_strcat(wb, nd_sock_is_ssl(&state->sock) ? "https" : "http"); |
| 224 | return true; |
| 225 | } |
| 226 | |
| 227 | static bool stream_sender_log_dst_ip(BUFFER *wb, void *ptr) { |
| 228 | struct sender_state *state = ptr; |
| 229 | if(!state || state->sock.fd == -1) |
| 230 | return false; |
| 231 | |
| 232 | SOCKET_PEERS peers = nd_sock_socket_peers(&state->sock); |
| 233 | buffer_strcat(wb, peers.peer.ip); |
| 234 | return true; |
| 235 | } |
| 236 | |
| 237 | static bool stream_sender_log_dst_port(BUFFER *wb, void *ptr) { |
| 238 | struct sender_state *state = ptr; |
| 239 | if(!state || state->sock.fd == -1) |
| 240 | return false; |
| 241 | |
| 242 | SOCKET_PEERS peers = nd_sock_socket_peers(&state->sock); |
| 243 | buffer_print_uint64(wb, peers.peer.port); |
| 244 | return true; |
| 245 | } |
| 246 | |
| 247 | // -------------------------------------------------------------------------------------------------------------------- |
| 248 | // opcodes |
| 249 | |
| 250 | ALWAYS_INLINE |
| 251 | void stream_sender_handle_op(struct stream_thread *sth, struct sender_state *s, struct stream_opcode *msg) { |
| 252 | ND_LOG_STACK lgs[] = { |
| 253 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname), |
| 254 | ND_LOG_FIELD_CB(NDF_DST_IP, stream_sender_log_dst_ip, s), |
| 255 | ND_LOG_FIELD_CB(NDF_DST_PORT, stream_sender_log_dst_port, s), |
| 256 | ND_LOG_FIELD_CB(NDF_DST_TRANSPORT, stream_sender_log_transport, s), |
| 257 | ND_LOG_FIELD_CB(NDF_DST_CAPABILITIES, stream_sender_log_capabilities, s), |
| 258 | ND_LOG_FIELD_END(), |
| 259 | }; |
| 260 | ND_LOG_STACK_PUSH(lgs); |
| 261 | |
| 262 | if(msg->opcode & STREAM_OPCODE_SENDER_BUFFER_OVERFLOW) { |
| 263 | worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_OVERFLOW); |
| 264 | errno_clear(); |
| 265 | stream_sender_lock(s); |
| 266 | // copy the statistics |
| 267 | STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(s->scb); |
| 268 | stream_sender_unlock(s); |
| 269 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 270 | "STREAM SND[%zu] '%s' [to %s]: send buffer is full (buffer size %u, max %u, used %u, available %u). " |
| 271 | "Restarting connection.", |
| 272 | sth->id, rrdhost_hostname(s->host), s->remote_ip, |
| 273 | stats.bytes_size, stats.bytes_max_size, stats.bytes_outstanding, stats.bytes_available); |
| 274 | |
| 275 | stream_sender_move_running_to_connector_or_remove( |
| 276 | sth, s, STREAM_HANDSHAKE_DISCONNECT_BUFFER_OVERFLOW, 0, true); |
| 277 | return; |
| 278 | } |
| 279 | |
| 280 | if(msg->opcode & STREAM_OPCODE_SENDER_STOP_RECEIVER_LEFT) { |
| 281 | worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_RECEIVER_LEFT); |
| 282 | stream_sender_move_running_to_connector_or_remove( |
| 283 | sth, s, STREAM_HANDSHAKE_SND_DISCONNECT_RECEIVER_LEFT, msg->reason, false); |
| 284 | |
| 285 | // at this point we also have access to the receiver exit reason as msg->reason |
| 286 | |
| 287 | return; |
| 288 | } |
| 289 | |
| 290 | if(msg->opcode & STREAM_OPCODE_SENDER_RECONNECT_WITHOUT_COMPRESSION) { |
| 291 | worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_COMPRESSION_ERROR); |
| 292 | errno_clear(); |
| 293 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 294 | "STREAM SND[%zu] '%s' [to %s]: restarting connection without compression.", |
| 295 | sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 296 | |
| 297 | stream_sender_move_running_to_connector_or_remove( |
| 298 | sth, s, STREAM_HANDSHAKE_SND_DISCONNECT_COMPRESSION_FAILED, 0, true); |
| 299 | return; |
| 300 | } |
| 301 | |
| 302 | if(msg->opcode & STREAM_OPCODE_SENDER_STOP_HOST_CLEANUP) { |
| 303 | worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_HOST_CLEANUP); |
| 304 | stream_sender_move_running_to_connector_or_remove( |
| 305 | sth, s, STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP, 0, false); |
| 306 | return; |
| 307 | } |
| 308 | |
| 309 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 310 | "STREAM SND[%zu]: invalid msg id %u", sth->id, (unsigned)msg->opcode); |
| 311 | } |
| 312 | |
| 313 | |
| 314 | // -------------------------------------------------------------------------------------------------------------------- |
| 315 | |
| 316 | void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth) { |
| 317 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 318 | |
| 319 | // process the queue |
| 320 | Word_t idx = 0; |
| 321 | for(struct sender_state *s = SENDERS_FIRST(&sth->queue.senders, &idx); |
| 322 | s; |
| 323 | s = SENDERS_NEXT(&sth->queue.senders, &idx)) { |
| 324 | worker_is_busy(WORKER_STREAM_JOB_DEQUEUE); |
| 325 | |
| 326 | SENDERS_DEL(&sth->queue.senders, idx); |
| 327 | |
| 328 | ND_LOG_STACK lgs[] = { |
| 329 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname), |
| 330 | ND_LOG_FIELD_CB(NDF_DST_IP, stream_sender_log_dst_ip, s), |
| 331 | ND_LOG_FIELD_CB(NDF_DST_PORT, stream_sender_log_dst_port, s), |
| 332 | ND_LOG_FIELD_CB(NDF_DST_TRANSPORT, stream_sender_log_transport, s), |
| 333 | ND_LOG_FIELD_CB(NDF_DST_CAPABILITIES, stream_sender_log_capabilities, s), |
| 334 | ND_LOG_FIELD_END(), |
| 335 | }; |
| 336 | ND_LOG_STACK_PUSH(lgs); |
| 337 | |
| 338 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 339 | "STREAM SND[%zu] '%s' [to %s]: moving host from dispatcher queue to dispatcher running...", |
| 340 | sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 341 | |
| 342 | if(sock_setnonblock(s->sock.fd, true) != 1) |
| 343 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 344 | "STREAM SND[%zu] '%s' [to %s]: failed to set non-blocking mode on socket %d", |
| 345 | sth->id, rrdhost_hostname(s->host), s->remote_ip, s->sock.fd); |
| 346 | |
| 347 | sock_setcloexec(s->sock.fd, true); |
| 348 | sock_enlarge_rcv_buf(s->sock.fd); |
| 349 | sock_enlarge_snd_buf(s->sock.fd); |
| 350 | sock_setcork(s->sock.fd, false); |
| 351 | |
| 352 | stream_sender_lock(s); |
| 353 | s->thread.meta.type = POLLFD_TYPE_SENDER; |
| 354 | s->thread.meta.s = s; |
| 355 | |
| 356 | s->thread.msg.thread_slot = (int32_t)sth->id; |
| 357 | s->thread.msg.session = os_random32(); |
| 358 | s->thread.msg.meta = &s->thread.meta; |
| 359 | |
| 360 | __atomic_store_n(&s->host->stream.snd.status.tid, gettid_cached(), __ATOMIC_RELAXED); |
| 361 | s->host->stream.snd.status.connections++; |
| 362 | s->last_state_since_t = now_realtime_sec(); |
| 363 | |
| 364 | s->replication.last_progress_ut = now_monotonic_usec(); |
| 365 | |
| 366 | stream_circular_buffer_flush_unsafe(s->scb, stream_send.buffer_max_size); |
| 367 | replication_sender_recalculate_buffer_used_ratio_unsafe(s); |
| 368 | stream_sender_unlock(s); |
| 369 | |
| 370 | internal_fatal(META_GET(&sth->run.meta, (Word_t)&s->thread.meta) != NULL, "Sender already exists in meta list"); |
| 371 | META_SET(&sth->run.meta, (Word_t)&s->thread.meta, &s->thread.meta); |
| 372 | |
| 373 | s->thread.wanted = ND_POLL_READ; |
| 374 | if(!nd_poll_add(sth->run.ndpl, s->sock.fd, s->thread.wanted, &s->thread.meta)) |
| 375 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 376 | "STREAM SND[%zu] '%s' [to %s]: failed to add sender socket to nd_poll()", |
| 377 | sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 378 | |
| 379 | stream_sender_on_ready_to_dispatch(s); |
| 380 | |
| 381 | pulse_host_status(s->host, PULSE_HOST_STATUS_SND_RUNNING, 0); |
| 382 | } |
| 383 | } |
| 384 | |
| 385 | void stream_sender_remove(struct sender_state *s, STREAM_HANDSHAKE reason) { |
| 386 | // THIS FUNCTION IS USED BY THE CONNECTOR TOO |
| 387 | // when it gives up on a certain node |
| 388 | |
| 389 | stream_sender_lock(s); |
| 390 | |
| 391 | if(reason == STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP && s->exit.reason) |
| 392 | reason = s->exit.reason; |
| 393 | |
| 394 | s->exit.reason = 0; |
| 395 | |
| 396 | __atomic_store_n(&s->exit.shutdown, false, __ATOMIC_RELAXED); |
| 397 | rrdhost_flag_clear(s->host, RRDHOST_FLAG_STREAM_SENDER_ADDED | RRDHOST_FLAG_STREAM_SENDER_CONNECTED | RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS); |
| 398 | |
| 399 | s->last_state_since_t = now_realtime_sec(); |
| 400 | stream_parent_set_host_disconnect_reason(s->host, reason, s->last_state_since_t); |
| 401 | s->connector.id = -1; |
| 402 | |
| 403 | stream_sender_unlock(s); |
| 404 | |
| 405 | stream_parents_host_reset(s->host, reason); |
| 406 | |
| 407 | #ifdef NETDATA_LOG_STREAM_SENDER |
| 408 | spinlock_lock(&s->log.spinlock); |
| 409 | if (s->log.fp) { |
| 410 | fclose(s->log.fp); |
| 411 | s->log.fp = NULL; |
| 412 | } |
| 413 | buffer_free(s->log.received); |
| 414 | s->log.received = NULL; |
| 415 | spinlock_unlock(&s->log.spinlock); |
| 416 | #endif |
| 417 | } |
| 418 | |
| 419 | static void stream_sender_log_disconnection(struct stream_thread *sth, struct sender_state *s, STREAM_HANDSHAKE reason, STREAM_HANDSHAKE receiver_reason) { |
| 420 | ND_LOG_STACK lgs[] = { |
| 421 | ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid), |
| 422 | ND_LOG_FIELD_END(), |
| 423 | }; |
| 424 | ND_LOG_STACK_PUSH(lgs); |
| 425 | |
| 426 | if(reason == STREAM_HANDSHAKE_SND_DISCONNECT_RECEIVER_LEFT && receiver_reason) |
| 427 | nd_log(NDLS_DAEMON, NDLP_NOTICE, |
| 428 | "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s (receiver left due to: %s)", |
| 429 | sth->id, rrdhost_hostname(s->host), s->remote_ip, |
| 430 | stream_handshake_error_to_string(reason), |
| 431 | stream_handshake_error_to_string(receiver_reason)); |
| 432 | else |
| 433 | nd_log(NDLS_DAEMON, NDLP_NOTICE, |
| 434 | "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s", |
| 435 | sth->id, rrdhost_hostname(s->host), s->remote_ip, stream_handshake_error_to_string(reason)); |
| 436 | } |
| 437 | |
| 438 | static void stream_sender_move_running_to_connector_or_remove_internal(struct stream_thread *sth, struct sender_state *s, STREAM_HANDSHAKE reason, STREAM_HANDSHAKE receiver_reason, bool reconnect) { |
| 439 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 440 | |
| 441 | ND_LOG_STACK lgs[] = { |
| 442 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname), |
| 443 | ND_LOG_FIELD_CB(NDF_DST_IP, stream_sender_log_dst_ip, s), |
| 444 | ND_LOG_FIELD_CB(NDF_DST_PORT, stream_sender_log_dst_port, s), |
| 445 | ND_LOG_FIELD_CB(NDF_DST_TRANSPORT, stream_sender_log_transport, s), |
| 446 | ND_LOG_FIELD_CB(NDF_DST_CAPABILITIES, stream_sender_log_capabilities, s), |
| 447 | ND_LOG_FIELD_END(), |
| 448 | }; |
| 449 | ND_LOG_STACK_PUSH(lgs); |
| 450 | |
| 451 | internal_fatal(META_GET(&sth->run.meta, (Word_t)&s->thread.meta) == NULL, "Sender to be removed is not in the list of senders"); |
| 452 | META_DEL(&sth->run.meta, (Word_t)&s->thread.meta); |
| 453 | |
| 454 | s->thread.wanted = 0; |
| 455 | if(!nd_poll_del(sth->run.ndpl, s->sock.fd)) |
| 456 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 457 | "STREAM SND[%zu] '%s' [to %s]: failed to delete sender socket from nd_poll()", |
| 458 | sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 459 | |
| 460 | // clear this flag asap, to stop other threads from pushing metrics for this node |
| 461 | rrdhost_flag_clear(s->host, RRDHOST_FLAG_STREAM_SENDER_CONNECTED | RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS); |
| 462 | |
| 463 | // clear these asap, to make sender_commit() stop processing data for this host |
| 464 | stream_sender_lock(s); |
| 465 | |
| 466 | if(reason == STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP && s->exit.reason) |
| 467 | reason = s->exit.reason; |
| 468 | |
| 469 | s->exit.reason = reason; |
| 470 | s->thread.msg.session = 0; |
| 471 | s->thread.msg.meta = NULL; |
| 472 | |
| 473 | __atomic_store_n(&s->host->stream.snd.status.tid, 0, __ATOMIC_RELAXED); |
| 474 | stream_sender_unlock(s); |
| 475 | |
| 476 | stream_sender_log_disconnection(sth, s, reason, receiver_reason); |
| 477 | |
| 478 | // IMPORTANT: make sure it REMOVED from nd_poll() before closing the socket |
| 479 | // otherwise, undefined things will happen due to socket reuse and epoll() |
| 480 | nd_sock_close(&s->sock); |
| 481 | |
| 482 | stream_parent_set_host_disconnect_reason(s->host, reason, now_realtime_sec()); |
| 483 | stream_sender_clear_parent_claim_id(s->host); |
| 484 | sender_host_buffer_free(s->host); |
| 485 | |
| 486 | pulse_host_status(s->host, PULSE_HOST_STATUS_SND_OFFLINE, reason); |
| 487 | |
| 488 | stream_thread_node_removed(s->host); |
| 489 | |
| 490 | stream_connector_requeue( |
| 491 | s, reconnect && !stream_connector_is_signaled_to_stop(s) ? STRCNT_CMD_CONNECT : STRCNT_CMD_REMOVE); |
| 492 | } |
| 493 | |
| 494 | void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut) { |
| 495 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 496 | |
| 497 | size_t bytes_uncompressed = 0; |
| 498 | size_t bytes_compressed = 0; |
| 499 | NETDATA_DOUBLE overall_buffer_ratio = 0.0; |
| 500 | |
| 501 | Word_t idx = 0; |
| 502 | for(struct pollfd_meta *m = META_FIRST(&sth->run.meta, &idx); |
| 503 | m; |
| 504 | m = META_NEXT(&sth->run.meta, &idx)) { |
| 505 | if(m->type != POLLFD_TYPE_SENDER) continue; |
| 506 | struct sender_state *s = m->s; |
| 507 | |
| 508 | stream_sender_lock(s); |
| 509 | // copy the statistics |
| 510 | STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(s->scb); |
| 511 | stream_sender_unlock(s); |
| 512 | |
| 513 | if (stats.buffer_ratio > overall_buffer_ratio) |
| 514 | overall_buffer_ratio = stats.buffer_ratio; |
| 515 | |
| 516 | if(unlikely(stats.bytes_outstanding && |
| 517 | s->thread.last_traffic_ut + stream_send.parents.timeout_s * USEC_PER_SEC < now_ut && |
| 518 | !stream_sender_pending_replication_requests(s) && |
| 519 | !stream_sender_replicating_charts(s))) { |
| 520 | |
| 521 | ND_LOG_STACK lgs[] = { |
| 522 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname), |
| 523 | ND_LOG_FIELD_CB(NDF_DST_IP, stream_sender_log_dst_ip, s), |
| 524 | ND_LOG_FIELD_CB(NDF_DST_PORT, stream_sender_log_dst_port, s), |
| 525 | ND_LOG_FIELD_CB(NDF_DST_TRANSPORT, stream_sender_log_transport, s), |
| 526 | ND_LOG_FIELD_CB(NDF_DST_CAPABILITIES, stream_sender_log_capabilities, s), |
| 527 | ND_LOG_FIELD_END(), |
| 528 | }; |
| 529 | ND_LOG_STACK_PUSH(lgs); |
| 530 | |
| 531 | worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_TIMEOUT); |
| 532 | |
| 533 | char duration[RFC3339_MAX_LENGTH]; |
| 534 | duration_snprintf(duration, sizeof(duration), (int64_t)(now_monotonic_usec() - s->thread.last_traffic_ut), "us", true); |
| 535 | |
| 536 | char pending[64] = "0"; |
| 537 | if(stats.bytes_outstanding) |
| 538 | size_snprintf(pending, sizeof(pending), stats.bytes_outstanding, "B", false); |
| 539 | |
| 540 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 541 | "STREAM SND[%zu] '%s' [to %s]: there was not traffic for %ld seconds - closing connection - " |
| 542 | "we have sent %zu bytes in %zu operations, it is idle for %s, and we have %s pending to send " |
| 543 | "(buffer is used %.2f%%).", |
| 544 | sth->id, rrdhost_hostname(s->host), s->remote_ip, stream_send.parents.timeout_s, |
| 545 | stats.bytes_sent, stats.sends, |
| 546 | duration, pending, stats.buffer_ratio); |
| 547 | |
| 548 | stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_TIMEOUT, 0, true); |
| 549 | continue; |
| 550 | } |
| 551 | |
| 552 | bytes_compressed += stats.bytes_added; |
| 553 | bytes_uncompressed += stats.bytes_uncompressed; |
| 554 | |
| 555 | nd_poll_event_t wanted = ND_POLL_READ | (stats.bytes_outstanding ? ND_POLL_WRITE : 0); |
| 556 | if(unlikely(s->thread.wanted != wanted)) { |
| 557 | // nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 558 | // "STREAM SND[%zu] '%s' [to %s]: nd_poll() wanted events mismatch.", |
| 559 | // sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 560 | |
| 561 | s->thread.wanted = wanted; |
| 562 | if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, s->thread.wanted)) |
| 563 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 564 | "STREAM SND[%zu] '%s' [to %s]: failed to update nd_poll().", |
| 565 | sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 566 | } |
| 567 | } |
| 568 | |
| 569 | if (bytes_compressed && bytes_uncompressed) { |
| 570 | NETDATA_DOUBLE compression_ratio = 100.0 - ((NETDATA_DOUBLE)bytes_compressed * 100.0 / (NETDATA_DOUBLE)bytes_uncompressed); |
| 571 | worker_set_metric(WORKER_SENDER_JOB_BYTES_COMPRESSION_RATIO, compression_ratio); |
| 572 | } |
| 573 | |
| 574 | worker_set_metric(WORKER_SENDER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_uncompressed); |
| 575 | worker_set_metric(WORKER_SENDER_JOB_BYTES_COMPRESSED, (NETDATA_DOUBLE)bytes_compressed); |
| 576 | worker_set_metric(WORKER_SENDER_JOB_BUFFER_RATIO, overall_buffer_ratio); |
| 577 | } |
| 578 | |
| 579 | static bool stream_sender_did_replication_progress(struct sender_state *s) { |
| 580 | RRDHOST *host = s->host; |
| 581 | |
| 582 | size_t host_counter_sum = |
| 583 | __atomic_load_n(&host->stream.snd.status.replication.counter_in, __ATOMIC_RELAXED) + |
| 584 | __atomic_load_n(&host->stream.snd.status.replication.counter_out, __ATOMIC_RELAXED); |
| 585 | |
| 586 | if(s->replication.last_counter_sum != host_counter_sum) { |
| 587 | // there has been some progress |
| 588 | s->replication.last_counter_sum = host_counter_sum; |
| 589 | s->replication.last_progress_ut = now_monotonic_usec(); |
| 590 | return true; |
| 591 | } |
| 592 | |
| 593 | if(!host_counter_sum) |
| 594 | // we have not started yet |
| 595 | return true; |
| 596 | |
| 597 | if(dictionary_entries(s->replication.requests)) |
| 598 | // we still have requests to execute |
| 599 | return true; |
| 600 | |
| 601 | return (now_monotonic_usec() - s->replication.last_progress_ut < 10ULL * 60 * USEC_PER_SEC); |
| 602 | } |
| 603 | |
| 604 | void stream_sender_replication_check_from_poll(struct stream_thread *sth, usec_t now_ut __maybe_unused) { |
| 605 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__); |
| 606 | |
| 607 | Word_t idx = 0; |
| 608 | for(struct pollfd_meta *m = META_FIRST(&sth->run.meta, &idx); |
| 609 | m; |
| 610 | m = META_NEXT(&sth->run.meta, &idx)) { |
| 611 | if (m->type != POLLFD_TYPE_SENDER) continue; |
| 612 | struct sender_state *s = m->s; |
| 613 | RRDHOST *host = s->host; |
| 614 | |
| 615 | if(stream_sender_did_replication_progress(s)) { |
| 616 | s->replication.last_checked_ut = 0; |
| 617 | continue; |
| 618 | } |
| 619 | |
| 620 | if(s->replication.last_checked_ut == s->replication.last_progress_ut) |
| 621 | continue; |
| 622 | |
| 623 | ND_LOG_STACK lgs[] = { |
| 624 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, host->hostname), |
| 625 | ND_LOG_FIELD_CB(NDF_DST_IP, stream_sender_log_dst_ip, s), |
| 626 | ND_LOG_FIELD_CB(NDF_DST_PORT, stream_sender_log_dst_port, s), |
| 627 | ND_LOG_FIELD_CB(NDF_DST_TRANSPORT, stream_sender_log_transport, s), |
| 628 | ND_LOG_FIELD_CB(NDF_DST_CAPABILITIES, stream_sender_log_capabilities, s), |
| 629 | ND_LOG_FIELD_END(), |
| 630 | }; |
| 631 | ND_LOG_STACK_PUSH(lgs); |
| 632 | |
| 633 | size_t stalled = 0, finished = 0; |
| 634 | RRDSET *st; |
| 635 | rrdset_foreach_read(st, host) { |
| 636 | RRDSET_FLAGS st_flags = rrdset_flag_get(st); |
| 637 | if(st_flags & (RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_UPSTREAM_IGNORE)) |
| 638 | continue; |
| 639 | |
| 640 | if(st_flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED) { |
| 641 | finished++; |
| 642 | continue; |
| 643 | } |
| 644 | |
| 645 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 646 | "STREAM SND[%zu] '%s' [to %s]: REPLICATION STALLED: instance '%s' %s replication yet.", |
| 647 | sth->id, rrdhost_hostname(host), s->remote_ip, |
| 648 | rrdset_id(st), |
| 649 | (st_flags & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS) ? "has not finished" : "has not started"); |
| 650 | |
| 651 | stalled++; |
| 652 | } |
| 653 | rrdset_foreach_done(st); |
| 654 | |
| 655 | if(stalled && !stream_sender_did_replication_progress(s)) { |
| 656 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 657 | "STREAM SND[%zu] '%s' [to %s]: REPLICATION EXCEPTIONS SUMMARY: node has %zu stalled replication requests (%zu completed)." |
| 658 | "We have received %u and sent %u replication commands. " |
| 659 | "Disconnecting node to restore streaming.", |
| 660 | sth->id, rrdhost_hostname(s->host), s->remote_ip, |
| 661 | stalled, finished, |
| 662 | __atomic_load_n(&host->stream.snd.status.replication.counter_in, __ATOMIC_RELAXED), |
| 663 | __atomic_load_n(&host->stream.snd.status.replication.counter_out, __ATOMIC_RELAXED)); |
| 664 | |
| 665 | stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_REPLICATION_STALLED, 0, true); |
| 666 | continue; |
| 667 | } |
| 668 | |
| 669 | s->replication.last_checked_ut = s->replication.last_progress_ut; |
| 670 | } |
| 671 | } |
| 672 | |
| 673 | bool stream_sender_send_data(struct stream_thread *sth, struct sender_state *s, usec_t now_ut, bool process_opcodes_and_enable_removal) { |
| 674 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 675 | |
| 676 | EVLOOP_STATUS status = EVLOOP_STATUS_CONTINUE; |
| 677 | while(status == EVLOOP_STATUS_CONTINUE) { |
| 678 | waitq_acquire(&s->waitq, WAITQ_PRIO_URGENT); |
| 679 | stream_sender_lock(s); |
| 680 | |
| 681 | STREAM_CIRCULAR_BUFFER_STATS *stats = stream_circular_buffer_stats_unsafe(s->scb); |
| 682 | char *chunk; |
| 683 | size_t outstanding = stream_circular_buffer_get_unsafe(s->scb, &chunk); |
| 684 | |
| 685 | if(!outstanding) { |
| 686 | status = EVLOOP_STATUS_NO_MORE_DATA; |
| 687 | stream_sender_unlock(s); |
| 688 | waitq_release(&s->waitq); |
| 689 | continue; |
| 690 | } |
| 691 | |
| 692 | ssize_t rc = nd_sock_send_nowait(&s->sock, chunk, outstanding); |
| 693 | if (likely(rc > 0)) { |
| 694 | pulse_stream_sent_bytes(rc); |
| 695 | stream_circular_buffer_del_unsafe(s->scb, rc, now_ut); |
| 696 | replication_sender_recalculate_buffer_used_ratio_unsafe(s); |
| 697 | s->thread.last_traffic_ut = now_ut; |
| 698 | sth->snd.bytes_sent += rc; |
| 699 | |
| 700 | if (!stats->bytes_outstanding) { |
| 701 | // we sent them all - remove ND_POLL_WRITE |
| 702 | s->thread.wanted = ND_POLL_READ; |
| 703 | if (!nd_poll_upd(sth->run.ndpl, s->sock.fd, s->thread.wanted)) |
| 704 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 705 | "STREAM SND[%zu] '%s' [to %s]: failed to update nd_poll().", |
| 706 | sth->id, rrdhost_hostname(s->host), s->remote_ip); |
| 707 | |
| 708 | // recreate the circular buffer if we have to |
| 709 | stream_circular_buffer_recreate_timed_unsafe(s->scb, now_ut, false); |
| 710 | status = EVLOOP_STATUS_NO_MORE_DATA; |
| 711 | } |
| 712 | } |
| 713 | else if (rc == 0 || errno == ECONNRESET) |
| 714 | status = EVLOOP_STATUS_SOCKET_CLOSED; |
| 715 | |
| 716 | else if (rc < 0) { |
| 717 | if(errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR) |
| 718 | status = EVLOOP_STATUS_SOCKET_FULL; |
| 719 | else |
| 720 | status = EVLOOP_STATUS_SOCKET_ERROR; |
| 721 | } |
| 722 | stream_sender_unlock(s); |
| 723 | waitq_release(&s->waitq); |
| 724 | |
| 725 | if (status == EVLOOP_STATUS_SOCKET_ERROR || status == EVLOOP_STATUS_SOCKET_CLOSED) { |
| 726 | const char *disconnect_reason = NULL; |
| 727 | STREAM_HANDSHAKE reason; |
| 728 | |
| 729 | if(status == EVLOOP_STATUS_SOCKET_ERROR) { |
| 730 | worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_SEND_ERROR); |
| 731 | disconnect_reason = "socket reports error while writing"; |
| 732 | reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_WRITE_FAILED; |
| 733 | } |
| 734 | else /* if(status == EVLOOP_STATUS_SOCKET_CLOSED) */ { |
| 735 | worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED); |
| 736 | disconnect_reason = "socket reports EOF (closed by parent)"; |
| 737 | reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE; |
| 738 | } |
| 739 | |
| 740 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 741 | "STREAM SND[%zu] '%s' [to %s]: %s (%zd, on fd %d) - restarting connection - " |
| 742 | "we have sent %zu bytes in %zu operations.", |
| 743 | sth->id, rrdhost_hostname(s->host), s->remote_ip, disconnect_reason, rc, s->sock.fd, |
| 744 | stats->bytes_sent, stats->sends); |
| 745 | |
| 746 | if(process_opcodes_and_enable_removal) { |
| 747 | // this is not executed from the opcode handling mechanism |
| 748 | // so we can safely remove the sender |
| 749 | stream_sender_move_running_to_connector_or_remove(sth, s, reason, 0, true); |
| 750 | break; |
| 751 | } |
| 752 | else { |
| 753 | // protection against this case: |
| 754 | // |
| 755 | // 1. sender gets a function request |
| 756 | // 2. sender executes the request |
| 757 | // 3. response is immediately available |
| 758 | // 4. sender_commit() appends the data to the sending circular buffer |
| 759 | // 5. sender_commit() sends opcode to enable sending |
| 760 | // 6. opcode bypasses the signal and runs this function inline to dispatch immediately |
| 761 | // 7. sending fails (remote disconnected) |
| 762 | // 8. sender is removed |
| 763 | // |
| 764 | // Point 2 above crashes. The sender is no longer there (freed at point 8) |
| 765 | // and there is no way for point 2 to know... |
| 766 | } |
| 767 | } |
| 768 | else if(process_opcodes_and_enable_removal && |
| 769 | status == EVLOOP_STATUS_CONTINUE && |
| 770 | stream_thread_process_opcodes(sth, &s->thread.meta)) |
| 771 | status = EVLOOP_STATUS_OPCODE_ON_ME; |
| 772 | } |
| 773 | |
| 774 | return EVLOOP_STATUS_STILL_ALIVE(status); |
| 775 | } |
| 776 | |
| 777 | bool stream_sender_receive_data(struct stream_thread *sth, struct sender_state *s, usec_t now_ut, bool process_opcodes) { |
| 778 | EVLOOP_STATUS status = EVLOOP_STATUS_CONTINUE; |
| 779 | while(status == EVLOOP_STATUS_CONTINUE) { |
| 780 | ssize_t rc = nd_sock_revc_nowait(&s->sock, s->thread.rbuf.b + s->thread.rbuf.read_len, s->thread.rbuf.size - s->thread.rbuf.read_len - 1); |
| 781 | if (likely(rc > 0)) { |
| 782 | s->thread.rbuf.read_len += rc; |
| 783 | |
| 784 | s->thread.last_traffic_ut = now_ut; |
| 785 | sth->snd.bytes_received += rc; |
| 786 | pulse_stream_received_bytes(rc); |
| 787 | |
| 788 | worker_is_busy(WORKER_SENDER_JOB_EXECUTE); |
| 789 | stream_sender_execute_commands(s); |
| 790 | } |
| 791 | else if (rc == 0 || errno == ECONNRESET) |
| 792 | status = EVLOOP_STATUS_SOCKET_CLOSED; |
| 793 | |
| 794 | else if (rc < 0) { |
| 795 | if(errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR) |
| 796 | status = EVLOOP_STATUS_SOCKET_FULL; |
| 797 | else |
| 798 | status = EVLOOP_STATUS_SOCKET_ERROR; |
| 799 | } |
| 800 | |
| 801 | if(status == EVLOOP_STATUS_SOCKET_ERROR || status == EVLOOP_STATUS_SOCKET_CLOSED) { |
| 802 | const char *disconnect_reason; |
| 803 | STREAM_HANDSHAKE reason; |
| 804 | |
| 805 | if(status == EVLOOP_STATUS_SOCKET_ERROR) { |
| 806 | worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_RECEIVE_ERROR); |
| 807 | reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED; |
| 808 | disconnect_reason = "error during receive"; |
| 809 | } |
| 810 | else /* if(status == EVLOOP_STATUS_SOCKET_CLOSED) */ { |
| 811 | worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED); |
| 812 | reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE; |
| 813 | disconnect_reason = "socket reports EOF (closed by parent)"; |
| 814 | } |
| 815 | |
| 816 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 817 | "STREAM SND[%zu] '%s' [to %s]: %s (fd %d) - restarting sender connection.", |
| 818 | sth->id, rrdhost_hostname(s->host), s->remote_ip, disconnect_reason, s->sock.fd); |
| 819 | |
| 820 | stream_sender_move_running_to_connector_or_remove(sth, s, reason, 0, true); |
| 821 | break; |
| 822 | } |
| 823 | else if(status == EVLOOP_STATUS_CONTINUE && process_opcodes && stream_thread_process_opcodes(sth, &s->thread.meta)) |
| 824 | status = EVLOOP_STATUS_OPCODE_ON_ME; |
| 825 | } |
| 826 | |
| 827 | return EVLOOP_STATUS_STILL_ALIVE(status); |
| 828 | } |
| 829 | |
| 830 | // process poll() events for streaming senders |
| 831 | // returns true when the sender is still there, false if it removed it |
| 832 | bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_state *s, nd_poll_event_t events, usec_t now_ut) { |
| 833 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 834 | |
| 835 | ND_LOG_STACK lgs[] = { |
| 836 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname), |
| 837 | ND_LOG_FIELD_CB(NDF_DST_IP, stream_sender_log_dst_ip, s), |
| 838 | ND_LOG_FIELD_CB(NDF_DST_PORT, stream_sender_log_dst_port, s), |
| 839 | ND_LOG_FIELD_CB(NDF_DST_TRANSPORT, stream_sender_log_transport, s), |
| 840 | ND_LOG_FIELD_CB(NDF_DST_CAPABILITIES, stream_sender_log_capabilities, s), |
| 841 | ND_LOG_FIELD_END(), |
| 842 | }; |
| 843 | ND_LOG_STACK_PUSH(lgs); |
| 844 | |
| 845 | if(unlikely(events & (ND_POLL_ERROR|ND_POLL_HUP|ND_POLL_INVALID))) { |
| 846 | // we have errors on this socket |
| 847 | |
| 848 | char *error = "unknown error"; |
| 849 | |
| 850 | if (events & ND_POLL_ERROR) |
| 851 | error = "socket reports errors"; |
| 852 | else if (events & ND_POLL_HUP) |
| 853 | error = "connection closed by remote end (HUP)"; |
| 854 | else if (events & ND_POLL_INVALID) |
| 855 | error = "connection is invalid"; |
| 856 | |
| 857 | worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_SOCKET_ERROR); |
| 858 | |
| 859 | stream_sender_lock(s); |
| 860 | // copy the statistics |
| 861 | STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(s->scb); |
| 862 | stream_sender_unlock(s); |
| 863 | |
| 864 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 865 | "STREAM SND[%zu] '%s' [to %s]: %s restarting connection - %zu bytes transmitted in %zu operations.", |
| 866 | sth->id, rrdhost_hostname(s->host), s->remote_ip, error, stats.bytes_sent, stats.sends); |
| 867 | |
| 868 | stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR, 0, true); |
| 869 | return false; |
| 870 | } |
| 871 | |
| 872 | if(events & ND_POLL_READ) { |
| 873 | worker_is_busy(WORKER_STREAM_JOB_SOCKET_RECEIVE); |
| 874 | if(!stream_sender_receive_data(sth, s, now_ut, true)) |
| 875 | return false; |
| 876 | } |
| 877 | |
| 878 | if(events & ND_POLL_WRITE) { |
| 879 | worker_is_busy(WORKER_STREAM_JOB_SOCKET_SEND); |
| 880 | if(!stream_sender_send_data(sth, s, now_ut, true)) |
| 881 | return false; |
| 882 | } |
| 883 | |
| 884 | return true; |
| 885 | } |
| 886 | |
| 887 | void stream_sender_cleanup(struct stream_thread *sth) { |
| 888 | // stop all hosts |
| 889 | Word_t idx = 0; |
| 890 | for(struct pollfd_meta *m = META_FIRST(&sth->run.meta, &idx); |
| 891 | m; |
| 892 | m = META_NEXT(&sth->run.meta, &idx)) { |
| 893 | if(m->type != POLLFD_TYPE_SENDER) continue; |
| 894 | struct sender_state *s = m->s; |
| 895 | |
| 896 | s->exit.reason = STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN; |
| 897 | s->exit.shutdown = true; |
| 898 | stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN, 0, false); |
| 899 | } |
| 900 | } |