@cryptotaxi247 / netdata-1 / commits / 61b5afbb9

Reject incoming streaming connection for own vnode (#21889)

* Reject locally collected vnode streaming attempts - Add logic to detect and block streaming attempts for vnodes already collected locally. - Introduce new error code and response for such cases during the streaming handshake process. - Ensure proper handling and logging of rejected connections. * Add error handling for locally collected vnode streaming rejections - Introduce response definition for `START_STREAMING_ERROR_LOCAL_VNODE` to address cases where the vnode is already collected locally. - Set appropriate error messages, reconnection delays, and logging mechanisms. * Add clarification comment for virtual host handling in streaming connections - Document `RRDHOST_OPTION_VIRTUAL_HOST` behavior to highlight its use by local collectors and clarify its relationship with stale detection for archived/orphaned vnodes.

Stelios Fragkakis committed Mar 6, 2026 at 09:40 UTC 61b5afbb934bbee7bc09a7231db6dd939fcd03f4

5 files changed +47 -1

src/plugins.d/pluginsd_parser.c

@@ -245,6 +245,11 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si

245 parser->user.host_define.rrdlabels = NULL;

246 }

247

248 + if(SERVING_PLUGINSD(parser)) {

249 + rrdlabels_add(host->rrdlabels, "_collector_machine_guid",

250 + localhost->machine_guid, RRDLABEL_SRC_AUTO);

251 + }

252 +

253 pluginsd_update_host_ephemerality(host);

254 pluginsd_host_define_cleanup(parser);

255

src/streaming/stream-connector.c

+10

@@ -52,6 +52,16 @@ static struct {

52 .postpone_reconnect_seconds = 60 * 60, // the IP may change, try it every hour

53 .priority = NDLP_DEBUG,

54 },

55 + {

56 + .response = START_STREAMING_ERROR_LOCAL_VNODE,

57 + .length = sizeof(START_STREAMING_ERROR_LOCAL_VNODE) - 1,

58 + .version = STREAM_HANDSHAKE_PARENT_VNODE_IS_LOCAL,

59 + .dynamic = false,

60 + .error = "remote server rejected this stream, the vnode is collected locally on that server",

61 + .worker_job_id = WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_BAD_HANDSHAKE,

62 + .postpone_reconnect_seconds = 60 * 60, // the vnode may stop being collected, try every hour

63 + .priority = NDLP_DEBUG,

64 + },

65 {

66 .response = START_STREAMING_ERROR_ALREADY_STREAMING,

67 .length = sizeof(START_STREAMING_ERROR_ALREADY_STREAMING) - 1,

src/streaming/stream-handshake.c

@@ -57,6 +57,7 @@ static struct {

57 {STREAM_HANDSHAKE_SP_CONNECTED, "SOCKET CONNECTED", 200}, // OK

58 {STREAM_HANDSHAKE_SP_NO_STREAM_INFO, "NO STREAM INFO", 404}, // Not Found

59 {STREAM_HANDSHAKE_SP_NO_DESTINATION, "NO PARENT TO SEND TO", 502}, // Bad Gateway

60 + {STREAM_HANDSHAKE_PARENT_VNODE_IS_LOCAL, "LOCAL VNODE", 101},

62 { 0, NULL, 0 },

63 };

src/streaming/stream-handshake.h

+3 -1

@@ -11,6 +11,7 @@

11 #define START_STREAMING_PROMPT_VN "Hit me baby, push them over with the version="

13 #define START_STREAMING_ERROR_SAME_LOCALHOST "Don't hit me baby, you are trying to stream my localhost back"

14 +#define START_STREAMING_ERROR_LOCAL_VNODE "Don't hit me baby, you are trying to stream my vnode back"

15 #define START_STREAMING_ERROR_ALREADY_STREAMING "This GUID is already streaming to this server"

16 #define START_STREAMING_ERROR_NOT_PERMITTED "You are not permitted to access this. Check the logs for more info."

17 #define START_STREAMING_ERROR_BUSY_TRY_LATER "The server is too busy now to accept this request. Try later."

@@ -72,9 +73,10 @@ typedef enum {

73 STREAM_HANDSHAKE_SP_CONNECTED = -36,

74 STREAM_HANDSHAKE_SP_NO_STREAM_INFO = -37,

75 STREAM_HANDSHAKE_SP_NO_DESTINATION = -38,

76 + STREAM_HANDSHAKE_PARENT_VNODE_IS_LOCAL = -39, // sent by parent - DO NOT CHANGE

78 // terminator - keep this positive, bigger than all negative values

77 - STREAM_HANDSHAKE_NEGATIVE_MAX = 39,

79 + STREAM_HANDSHAKE_NEGATIVE_MAX = 40,

80 } STREAM_HANDSHAKE;

82 const char *stream_handshake_error_to_string(STREAM_HANDSHAKE reason);

src/streaming/stream-receiver-connection.c

+28

@@ -613,6 +613,34 @@ int stream_receiver_accept_connection(struct web_client *w, char *decoded_query_

613 return HTTP_RESP_OK;

614 }

615

616 + {

617 + RRDHOST *existing = rrdhost_find_by_guid(rpt->machine_guid);

618 + // RRDHOST_OPTION_VIRTUAL_HOST is only set by local collectors (pluginsd_host_define_end),

619 + // never by streaming. The stale detection in pluginsd_host() clears it when a vnode stops

620 + // being collected, so archived/orphaned vnodes will not have this flag set.

621 + // No additional checks for RRDHOST_FLAG_COLLECTOR_ONLINE or RRDHOST_FLAG_ARCHIVED are needed.

622 + if(existing && rrdhost_is_virtual(existing)) {

623 + stream_receiver_takeover_web_connection(w, rpt);

624 +

625 + stream_receiver_log_status(

626 + rpt,

627 + "rejecting streaming connection; this is a locally collected vnode",

628 + STREAM_HANDSHAKE_PARENT_VNODE_IS_LOCAL, NDLP_DEBUG);

629 +

630 + char initial_response[HTTP_HEADER_SIZE + 1];

631 + snprintfz(initial_response, HTTP_HEADER_SIZE, "%s", START_STREAMING_ERROR_LOCAL_VNODE);

632 +

633 + if(nd_sock_send_timeout(&rpt->sock, initial_response, strlen(initial_response), 0, 60) !=

634 + (ssize_t)strlen(initial_response)) {

635 + nd_log_daemon(NDLP_ERR, "STREAM RCV '%s' [from [%s]:%s]: failed to reply.",

636 + rpt->hostname, rpt->remote_ip, rpt->remote_port);

637 + }

638 +

639 + stream_receiver_free(rpt);

640 + return HTTP_RESP_OK;

641 + }

642 + }

643 +

644 if(unlikely(web_client_streaming_rate_t > 0)) {

645 static SPINLOCK spinlock = SPINLOCK_INITIALIZER;

646 static time_t last_stream_accepted_t = 0;