@cryptotaxi247 / netdata-1 / commits / 05ef02a81

Apply some logic to possible streaming destinations (#12866)

* replace connect_to_one_of with connect_to_one_of_destinations * move functions from socket.c * use sizeof * move current destination pointer to host * formatting * use snprintfz * get entries in same order * handle single destination as before (or when it is the last of the list), instead of skiping it every other loop * try other destinations on ssl problem

Emmanuel Vasilakis committed May 20, 2022 at 09:02 UTC 05ef02a817c24ff4c5eae2d813387301b093e220

7 files changed +167 -4

database/rrd.h

@@ -766,6 +766,8 @@ struct rrdhost {

766 unsigned int rrdpush_send_enabled; // 1 when this host sends metrics to another netdata

767 char *rrdpush_send_destination; // where to send metrics to

768 char *rrdpush_send_api_key; // the api key at the receiving netdata

769 + struct rrdpush_destinations *destinations; // a linked list of possible destinations

770 + struct rrdpush_destinations *destination; // the current destination from the above list

771

772 // the following are state information for the threading

773 // streaming metrics from this netdata to an upstream netdata

database/rrdhost.c

@@ -181,6 +181,8 @@ RRDHOST *rrdhost_create(const char *hostname,

181

182 host->rrdpush_send_enabled = (rrdpush_enabled && rrdpush_destination && *rrdpush_destination && rrdpush_api_key && *rrdpush_api_key) ? 1 : 0;

183 host->rrdpush_send_destination = (host->rrdpush_send_enabled)?strdupz(rrdpush_destination):NULL;

184 + if (host->rrdpush_send_destination)

185 + host->destinations = destinations_init(host->rrdpush_send_destination);

186 host->rrdpush_send_api_key = (host->rrdpush_send_enabled)?strdupz(rrdpush_api_key):NULL;

187 host->rrdpush_send_charts_matching = simple_pattern_create(rrdpush_send_charts_matching, NULL, SIMPLE_PATTERN_EXACT);

188

streaming/receiver.c

+16 -2

@@ -466,9 +466,23 @@ static int rrdpush_receive(struct receiver_state *rpt)

466

467 if (strcmp(rpt->machine_guid, localhost->machine_guid) == 0) {

468 log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->machine_guid, rpt->hostname, "DENIED - ATTEMPT TO RECEIVE METRICS FROM MACHINE_GUID IDENTICAL TO PARENT");

469 - error("STREAM %s [receive from %s:%s]: denied to receive metrics, machine GUID [%s] is my own. Did you copy the parent/proxy machine GUID to a child?", rpt->hostname, rpt->client_ip, rpt->client_port, rpt->machine_guid);

469 + error("STREAM %s [receive from %s:%s]: denied to receive metrics, machine GUID [%s] is my own. Did you copy the parent/proxy machine GUID to a child, or is this an inter-agent loop?", rpt->hostname, rpt->client_ip, rpt->client_port, rpt->machine_guid);

470 + char initial_response[HTTP_HEADER_SIZE + 1];

471 + snprintfz(initial_response, HTTP_HEADER_SIZE, "%s", START_STREAMING_ERROR_SAME_LOCALHOST);

472 +#ifdef ENABLE_HTTPS

473 + rpt->host->stream_ssl.conn = rpt->ssl.conn;

474 + rpt->host->stream_ssl.flags = rpt->ssl.flags;

475 + if(send_timeout(&rpt->ssl, rpt->fd, initial_response, strlen(initial_response), 0, 60) != (ssize_t)strlen(initial_response)) {

476 +#else

477 + if(send_timeout(rpt->fd, initial_response, strlen(initial_response), 0, 60) != strlen(initial_response)) {

478 +#endif

479 + log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rpt->host->hostname, "FAILED - CANNOT REPLY");

480 + error("STREAM %s [receive from [%s]:%s]: cannot send command.", rpt->host->hostname, rpt->client_ip, rpt->client_port);

481 + close(rpt->fd);

482 + return 0;

483 + }

484 close(rpt->fd);

471 - return 1;

485 + return 0;

486 }

487

488 if (rpt->host==NULL) {

streaming/rrdpush.c

+79

@@ -417,6 +417,85 @@ void rrdpush_claimed_id(RRDHOST *host)

417 error("STREAM %s [send]: cannot write to internal pipe", host->hostname);

418 }

419

420 +int connect_to_one_of_destinations(

421 + struct rrdpush_destinations *destinations,

422 + int default_port,

423 + struct timeval *timeout,

424 + size_t *reconnects_counter,

425 + char *connected_to,

426 + size_t connected_to_size,

427 + struct rrdpush_destinations **destination)

428 +{

429 + int sock = -1;

430 +

431 + for (struct rrdpush_destinations *d = destinations; d; d = d->next) {

432 + if (d->disabled_no_proper_reply) {

433 + d->disabled_no_proper_reply = 0;

434 + continue;

435 + } else if (d->disabled_because_of_localhost) {

436 + continue;

437 + } else if (d->disabled_already_streaming && (d->disabled_already_streaming + 30 > now_realtime_sec())) {

438 + continue;

439 + } else if (d->disabled_because_of_denied_access) {

440 + continue;

441 + }

442 +

443 + if (reconnects_counter)

444 + *reconnects_counter += 1;

445 + sock = connect_to_this(d->destination, default_port, timeout);

446 + if (sock != -1) {

447 + if (connected_to && connected_to_size) {

448 + strncpy(connected_to, d->destination, connected_to_size);

449 + connected_to[connected_to_size - 1] = '\0';

450 + }

451 + *destination = d;

452 + break;

453 + }

454 + }

455 +

456 + return sock;

457 +}

458 +

459 +struct rrdpush_destinations *destinations_init(const char *dests) {

460 + const char *s = dests;

461 + struct rrdpush_destinations *destinations = NULL, *prev = NULL;

462 + while(*s) {

463 + const char *e = s;

464 +

465 + // skip path, moving both s(tart) and e(nd)

466 + if(*e == '/')

467 + while(!isspace(*e) && *e != ',') s = ++e;

468 +

469 + // skip separators, moving both s(tart) and e(nd)

470 + while(isspace(*e) || *e == ',') s = ++e;

471 +

472 + // move e(nd) to the first separator

473 + while(*e && !isspace(*e) && *e != ',' && *e != '/') e++;

474 +

475 + // is there anything?

476 + if(!*s || s == e) break;

477 +

478 + char buf[e - s + 1];

479 + strncpyz(buf, s, e - s);

480 + struct rrdpush_destinations *d = callocz(1, sizeof(struct rrdpush_destinations));

481 + strncpyz(d->destination, buf, sizeof(d->destination)-1);

482 + d->disabled_no_proper_reply = 0;

483 + d->disabled_because_of_localhost = 0;

484 + d->disabled_already_streaming = 0;

485 + d->disabled_because_of_denied_access = 0;

486 + d->next = NULL;

487 + if (!destinations) {

488 + destinations = d;

489 + } else {

490 + prev->next = d;

491 + }

492 + prev = d;

493 +

494 + s = e;

495 + }

496 + return destinations;

497 +}

498 +

499 // ----------------------------------------------------------------------------

500 // rrdpush sender thread

501

streaming/rrdpush.h

+21

@@ -26,6 +26,10 @@

26 #define START_STREAMING_PROMPT_V2 "Hit me baby, push them over and bring the host labels..."

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

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

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

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

32 +

33 #define HTTP_HEADER_SIZE 8192

35 typedef enum {

@@ -138,6 +142,14 @@ struct receiver_state {

142 #endif

143 };

144

145 +struct rrdpush_destinations {

146 + char destination[CONNECTED_TO_SIZE + 1];

147 + int disabled_no_proper_reply;

148 + int disabled_because_of_localhost;

149 + time_t disabled_already_streaming;

150 + int disabled_because_of_denied_access;

151 + struct rrdpush_destinations *next;

152 +};

153

154 extern unsigned int default_rrdpush_enabled;

155 #ifdef ENABLE_COMPRESSION

@@ -149,6 +161,7 @@ extern char *default_rrdpush_send_charts_matching;

161 extern unsigned int remote_clock_resync_iterations;

162

163 extern void sender_init(struct sender_state *s, RRDHOST *parent);

164 +extern struct rrdpush_destinations *destinations_init(const char *destinations);

165 void sender_start(struct sender_state *s);

166 void sender_commit(struct sender_state *s);

167 extern int rrdpush_init();

@@ -164,6 +177,14 @@ extern void rrdpush_sender_thread_stop(RRDHOST *host);

177

178 extern void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, RRDVAR *rv);

179 extern void log_stream_connection(const char *client_ip, const char *client_port, const char *api_key, const char *machine_guid, const char *host, const char *msg);

180 +extern int connect_to_one_of_destinations(

181 + struct rrdpush_destinations *destinations,

182 + int default_port,

183 + struct timeval *timeout,

184 + size_t *reconnects_counter,

185 + char *connected_to,

186 + size_t connected_to_size,

187 + struct rrdpush_destinations **destination);

188

189 #ifdef ENABLE_COMPRESSION

190 struct compressor_state *create_compressor();

streaming/sender.c

+43 -2

@@ -203,6 +203,18 @@ void rrdpush_clean_encoded(stream_encoded_t *se)

203 freez(se->kernel_version);

204 }

205

206 +static inline long int parse_stream_version_for_errors(char *http)

207 +{

208 + if (!memcmp(http, START_STREAMING_ERROR_SAME_LOCALHOST, sizeof(START_STREAMING_ERROR_SAME_LOCALHOST)))

209 + return -2;

210 + else if (!memcmp(http, START_STREAMING_ERROR_ALREADY_STREAMING, sizeof(START_STREAMING_ERROR_ALREADY_STREAMING)))

211 + return -3;

212 + else if (!memcmp(http, START_STREAMING_ERROR_NOT_PERMITTED, sizeof(START_STREAMING_ERROR_NOT_PERMITTED)))

213 + return -4;

214 + else

215 + return -1;

216 +}

217 +

218 static inline long int parse_stream_version(RRDHOST *host, char *http)

219 {

220 long int stream_version = -1;

@@ -227,6 +239,9 @@ static inline long int parse_stream_version(RRDHOST *host, char *http)

239 host->labels.labels_flag |= LABEL_FLAG_STOP_STREAM;

240 host->labels.labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;

241 }

242 + else {

243 + stream_version = parse_stream_version_for_errors(http);

244 + }

245 }

246 }

247 return stream_version;

@@ -246,13 +261,14 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po

261 debug(D_STREAM, "STREAM: Attempting to connect...");

262 info("STREAM %s [send to %s]: connecting...", host->hostname, host->rrdpush_send_destination);

263

249 - host->rrdpush_sender_socket = connect_to_one_of(

250 - host->rrdpush_send_destination

264 + host->rrdpush_sender_socket = connect_to_one_of_destinations(

265 + host->destinations

266 , default_port

267 , &tv

268 , &s->reconnects_counter

269 , s->connected_to

270 , sizeof(s->connected_to)-1

271 + , &host->destination

272 );

273

274 if(unlikely(host->rrdpush_sender_socket == -1)) {

@@ -411,6 +427,8 @@ if(!s->rrdpush_compression)

427 if (netdata_use_ssl_on_stream == NETDATA_SSL_FORCE) {

428 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);

429 rrdpush_sender_thread_close_socket(host);

430 + if (host->destination->next)

431 + host->destination->disabled_no_proper_reply = 1;

432 return 0;

433 }else {

434 host->ssl.flags = NETDATA_SSL_NO_HANDSHAKE;

@@ -423,6 +441,8 @@ if(!s->rrdpush_compression)

441 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SSL_ERROR);

442 error("Closing the stream connection, because the server SSL certificate is not valid.");

443 rrdpush_sender_thread_close_socket(host);

444 + if (host->destination->next)

445 + host->destination->disabled_no_proper_reply = 1;

446 return 0;

447 }

448 }

@@ -462,6 +482,27 @@ if(!s->rrdpush_compression)

482 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE);

483 error("STREAM %s [send to %s]: server is not replying properly (is it a netdata?).", host->hostname, s->connected_to);

484 rrdpush_sender_thread_close_socket(host);

485 + //catch other reject reasons and force to check other destinations

486 + if (host->destination->next)

487 + host->destination->disabled_no_proper_reply = 1;

488 + return 0;

489 + }

490 + else if(version == -2) {

491 + error("STREAM %s [send to %s]: remote server is the localhost for [%s].", host->hostname, s->connected_to, host->hostname);

492 + rrdpush_sender_thread_close_socket(host);

493 + host->destination->disabled_because_of_localhost = 1;

494 + return 0;

495 + }

496 + else if(version == -3) {

497 + error("STREAM %s [send to %s]: remote server already receives metrics for [%s].", host->hostname, s->connected_to, host->hostname);

498 + rrdpush_sender_thread_close_socket(host);

499 + host->destination->disabled_already_streaming = now_realtime_sec();

500 + return 0;

501 + }

502 + else if(version == -4) {

503 + error("STREAM %s [send to %s]: remote server denied access for [%s].", host->hostname, s->connected_to, host->hostname);

504 + rrdpush_sender_thread_close_socket(host);

505 + host->destination->disabled_because_of_denied_access = 1;

506 return 0;

507 }

508 s->version = version;

web/server/static/static-threaded.c

@@ -303,6 +303,10 @@ static int web_server_rcv_callback(POLLINFO *pi, short int *events) {

303 worker_is_busy(WORKER_JOB_PROCESS);

304 web_client_process_request(w);

305

306 + if (unlikely(w->mode == WEB_CLIENT_MODE_STREAM)) {

307 + web_client_send(w);

308 + }

309 +

310 if(unlikely(w->mode == WEB_CLIENT_MODE_FILECOPY)) {

311 if(w->pollinfo_filecopy_slot == 0) {

312 debug(D_WEB_CLIENT, "%llu: FILECOPY DETECTED ON FD %d", w->id, pi->fd);