@cryptotaxi247 / netdata-1 / commits / b91307e29

Improve streaming connection loss detection (#21430)

* Enable TCP keepalive and implement dead connection detection in stream receiver. * Add netdata_ssl_peek() and nd_sock_peek_nowait() to support non-blocking socket peeking for SSL and plain TCP * Refine TCP keepalive configuration and improve socket error handling in stream receiver. * Handle ECONNRESET in socket peek and refine error checks in stream receiver. * Refine SSL error handling and set `errno` for incomplete handshakes.

Stelios Fragkakis committed Dec 12, 2025 at 22:18 UTC b91307e29245672f91d899f4311fbfc88755c59e

5 files changed +166 -4

src/libnetdata/socket/nd-sock.h

@@ -124,6 +124,14 @@ static ssize_t nd_sock_revc_nowait(ND_SOCK *s, void *buf, size_t num) {

124 return recv(s->fd, buf, num, MSG_DONTWAIT);

125 }

126

127 +ALWAYS_INLINE

128 +static ssize_t nd_sock_peek_nowait(ND_SOCK *s, void *buf, size_t num) {

129 + if (nd_sock_is_ssl(s))

130 + return netdata_ssl_peek(&s->ssl, buf, num);

131 + else

132 + return recv(s->fd, buf, num, MSG_PEEK | MSG_DONTWAIT);

133 +}

134 +

135 ALWAYS_INLINE

136 static ssize_t nd_sock_send_nowait(ND_SOCK *s, void *buf, size_t num) {

137 if (nd_sock_is_ssl(s))

src/libnetdata/socket/security.c

+70 -4

@@ -286,8 +286,10 @@ ssize_t netdata_ssl_read(NETDATA_SSL *ssl, void *buf, size_t num) {

286 errno = 0;

287 ssl->ssl_errno = 0;

288

289 - if(unlikely(!is_handshake_complete(ssl, "read")))

289 + if(unlikely(!is_handshake_complete(ssl, "read"))) {

290 + errno = ENOTCONN;

291 return -1;

292 + }

293

294 int bytes = SSL_read(ssl->conn, buf, (int)num);

295

@@ -302,8 +304,14 @@ ssize_t netdata_ssl_read(NETDATA_SSL *ssl, void *buf, size_t num) {

304 ssl->ssl_errno = err;

305 errno = EWOULDBLOCK;

306 }

305 - else

307 + else {

308 + // For SSL_ERROR_SYSCALL, errno contains the underlying socket error

309 + // (e.g., ECONNRESET). Save it before calling netdata_ssl_log_error_queue()

310 + // which may corrupt errno through subsequent function calls.

311 + int saved_errno = errno;

312 netdata_ssl_log_error_queue("SSL_read", ssl, err);

313 + errno = saved_errno;

314 + }

315

316 bytes = -1; // according to read() or recv()

317 }

@@ -311,6 +319,56 @@ ssize_t netdata_ssl_read(NETDATA_SSL *ssl, void *buf, size_t num) {

319 return bytes;

320 }

321

322 +/*

323 + * netdata_ssl_peek() - peek at incoming SSL data without consuming it

324 + *

325 + * This function is identical to netdata_ssl_read() but uses SSL_peek()

326 + * instead of SSL_read(), leaving the data in the SSL buffer for a

327 + * subsequent read operation. Useful for probing connection status.

328 + *

329 + * Returns:

330 + * > 0: Number of bytes available to peek

331 + * 0: Connection closed (SSL_ERROR_ZERO_RETURN)

332 + * -1: Error (check errno: EWOULDBLOCK means no data available)

333 + */

334 +ALWAYS_INLINE

335 +ssize_t netdata_ssl_peek(NETDATA_SSL *ssl, void *buf, size_t num) {

336 + errno = 0;

337 + ssl->ssl_errno = 0;

338 +

339 + if(unlikely(!is_handshake_complete(ssl, "peek"))) {

340 + errno = ENOTCONN;

341 + return -1;

342 + }

343 +

344 + int bytes = SSL_peek(ssl->conn, buf, (int)num);

345 +

346 + if(unlikely(bytes <= 0)) {

347 + int err = SSL_get_error(ssl->conn, bytes);

348 + if (err == SSL_ERROR_ZERO_RETURN) {

349 + ssl->ssl_errno = err;

350 + return 0; // Connection closed

351 + }

352 +

353 + if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {

354 + ssl->ssl_errno = err;

355 + errno = EWOULDBLOCK;

356 + }

357 + else {

358 + // For SSL_ERROR_SYSCALL, errno contains the underlying socket error

359 + // (e.g., ECONNRESET). Save it before calling netdata_ssl_log_error_queue()

360 + // which may corrupt errno through subsequent function calls.

361 + int saved_errno = errno;

362 + netdata_ssl_log_error_queue("SSL_peek", ssl, err);

363 + errno = saved_errno;

364 + }

365 +

366 + bytes = -1;

367 + }

368 +

369 + return bytes;

370 +}

371 +

372 /*

373 * netdata_ssl_write() should return the same as write():

374 *

@@ -329,8 +387,10 @@ ssize_t netdata_ssl_write(NETDATA_SSL *ssl, const void *buf, size_t num) {

387 errno = 0;

388 ssl->ssl_errno = 0;

389

332 - if(unlikely(!is_handshake_complete(ssl, "write")))

390 + if(unlikely(!is_handshake_complete(ssl, "write"))) {

391 + errno = ENOTCONN;

392 return -1;

393 + }

394

395 int bytes = SSL_write(ssl->conn, (uint8_t *)buf, (int)num);

396

@@ -340,8 +400,14 @@ ssize_t netdata_ssl_write(NETDATA_SSL *ssl, const void *buf, size_t num) {

400 ssl->ssl_errno = err;

401 errno = EWOULDBLOCK;

402 }

343 - else

403 + else {

404 + // For SSL_ERROR_SYSCALL, errno contains the underlying socket error

405 + // (e.g., ECONNRESET). Save it before calling netdata_ssl_log_error_queue()

406 + // which may corrupt errno through subsequent function calls.

407 + int saved_errno = errno;

408 netdata_ssl_log_error_queue("SSL_write", ssl, err);

409 + errno = saved_errno;

410 + }

411

412 bytes = -1; // according to write() or send()

413 }

src/libnetdata/socket/security.h

@@ -48,6 +48,7 @@ void netdata_ssl_close(NETDATA_SSL *ssl);

49 ssize_t netdata_ssl_read(NETDATA_SSL *ssl, void *buf, size_t num);

50 ssize_t netdata_ssl_write(NETDATA_SSL *ssl, const void *buf, size_t num);

51 +ssize_t netdata_ssl_peek(NETDATA_SSL *ssl, void *buf, size_t num);

53 ssize_t netdata_ssl_pending(NETDATA_SSL *ssl);

54 bool netdata_ssl_has_pending(NETDATA_SSL *ssl);

src/streaming/stream-receiver-connection.c

+35

@@ -5,6 +5,14 @@

5 #include "stream-receiver-internals.h"

6 #include "stream-replication-sender.h"

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);

18 // --------------------------------------------------------------------------------------------------------------------

@@ -286,6 +294,33 @@ static bool stream_receiver_send_first_response(struct receiver_state *rpt) {

294 nd_log(NDLS_DAEMON, NDLP_ERR,

295 "STREAM RCV '%s' [from [%s]:%s]: cannot set timeout for socket %d",

296 rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);

297 +

298 + // Enable TCP keepalive to detect dead connections faster

299 + // When a child vanishes (e.g., VM powered off), the socket won't close normally.

300 + // TCP keepalive will probe the connection and detect it's dead.

301 + int enable = 1;

302 + int idle = CONNECTION_PROBE_AFTER_SECONDS;

303 + int interval = CONNECTION_PROBE_INTERVAL_SECONDS;

304 + int count = CONNECTION_PROBE_COUNT;

305 +

306 + if (setsockopt(rpt->sock.fd, SOL_SOCKET, SO_KEEPALIVE, &enable, sizeof(enable)) != 0)

307 + nd_log(NDLS_DAEMON, NDLP_WARNING,

308 + "STREAM RCV '%s' [from [%s]:%s]: cannot enable SO_KEEPALIVE on socket %d",

309 + rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);

310 +#ifdef TCP_KEEPIDLE

311 + if (setsockopt(rpt->sock.fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle)) != 0)

312 + nd_log(NDLS_DAEMON, NDLP_WARNING,

313 + "STREAM RCV '%s' [from [%s]:%s]: cannot set TCP_KEEPIDLE on socket %d",

314 + rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);

315 + if (setsockopt(rpt->sock.fd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(interval)) != 0)

316 + nd_log(NDLS_DAEMON, NDLP_WARNING,

317 + "STREAM RCV '%s' [from [%s]:%s]: cannot set TCP_KEEPINTVL on socket %d",

318 + rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);

319 + if (setsockopt(rpt->sock.fd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(count)) != 0)

320 + nd_log(NDLS_DAEMON, NDLP_WARNING,

321 + "STREAM RCV '%s' [from [%s]:%s]: cannot set TCP_KEEPCNT on socket %d",

322 + rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);

323 +#endif

324 }

325

326 netdata_log_debug(D_STREAM, "Initial response to %s: %s", rpt->remote_ip, initial_response);

src/streaming/stream-receiver.c

+52

@@ -896,6 +896,58 @@ void stream_receiver_check_all_nodes_from_poll(struct stream_thread *sth, usec_t

896 if (m->type != POLLFD_TYPE_RECEIVER) continue;

897 struct receiver_state *rpt = m->rpt;

898

899 + // Probe socket to detect dead connections (e.g., from TCP keepalive)

900 + // Uses nd_sock_peek_nowait() which handles both SSL and plain TCP:

901 + // - For SSL: uses SSL_peek() to avoid corrupting SSL state

902 + // - For plain TCP: uses recv(MSG_PEEK | MSG_DONTWAIT)

903 + char probe_byte;

904 + ssize_t probe_rc = nd_sock_peek_nowait(&rpt->sock, &probe_byte, 1);

905 + if (probe_rc == 0 || (probe_rc < 0 && errno == ECONNRESET)) {

906 + // Connection closed by remote (gracefully or via reset)

907 + ND_LOG_STACK lgs[] = {

908 + ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),

909 + ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port),

910 + ND_LOG_FIELD_TXT(NDF_NIDL_NODE, rpt->hostname),

911 + ND_LOG_FIELD_CB(NDF_SRC_TRANSPORT, stream_receiver_log_transport, rpt),

912 + ND_LOG_FIELD_CB(NDF_SRC_CAPABILITIES, stream_receiver_log_capabilities, rpt),

913 + ND_LOG_FIELD_END(),

914 + };

915 + ND_LOG_STACK_PUSH(lgs);

916 +

917 + worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED);

918 + nd_log(NDLS_DAEMON, NDLP_ERR,

919 + "STREAM RCV[%zu] '%s' [from %s]: socket closed by remote - closing connection",

920 + sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip);

921 +

922 + stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE);

923 + continue;

924 + }

925 + if (probe_rc < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {

926 + // Socket error detected (keepalive timeout, etc.)

927 + // Save errno immediately as subsequent calls may modify it

928 + int saved_errno = errno;

929 +

930 + ND_LOG_STACK lgs[] = {

931 + ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),

932 + ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port),

933 + ND_LOG_FIELD_TXT(NDF_NIDL_NODE, rpt->hostname),

934 + ND_LOG_FIELD_CB(NDF_SRC_TRANSPORT, stream_receiver_log_transport, rpt),

935 + ND_LOG_FIELD_CB(NDF_SRC_CAPABILITIES, stream_receiver_log_capabilities, rpt),

936 + ND_LOG_FIELD_END(),

937 + };

938 + ND_LOG_STACK_PUSH(lgs);

939 +

940 + worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_SOCKET_ERROR);

941 + nd_log(NDLS_DAEMON, NDLP_ERR,

942 + "STREAM RCV[%zu] '%s' [from %s]: socket error detected: %s - closing connection",

943 + sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, strerror(saved_errno));

944 +

945 + stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR);

946 + continue;

947 + }

948 + // probe_rc > 0: data available (normal)

949 + // probe_rc < 0 with EAGAIN/EWOULDBLOCK: no data but connection alive

950 +

951 spinlock_lock(&rpt->thread.send_to_child.spinlock);

952 STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(rpt->thread.send_to_child.scb);

953 spinlock_unlock(&rpt->thread.send_to_child.spinlock);