eliminate any possibility of 100% cpu at netdata slaves and proxies, while sending metrics
Costa Tsaousis (ktsaou) committed
Oct 1, 2017 at 22:13 UTC
4151275df7013d3dc53f630d8e08f0972c4afa9d
1 file changed
+25
-34
src/rrdpush.c
+25
-34
@@ -254,6 +254,15 @@ void rrdpush_sender_thread_stop(RRDHOST *host) {
254
}
255
}
256
257
+static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
258
+ host->rrdpush_sender_connected = 0;
259
+
260
+ if(host->rrdpush_sender_socket != -1) {
261
+ close(host->rrdpush_sender_socket);
262
+ host->rrdpush_sender_socket = -1;
263
+ }
264
+}
265
+
266
static void rrdpush_sender_thread_cleanup_callback(void *ptr) {
267
RRDHOST *host = (RRDHOST *)ptr;
268
@@ -261,12 +270,8 @@ static void rrdpush_sender_thread_cleanup_callback(void *ptr) {
270
rrdhost_wrlock(host);
271
272
info("STREAM %s [send]: sending thread cleans up...", host->hostname);
264
- host->rrdpush_sender_connected = 0;
273
266
- if(host->rrdpush_sender_socket != -1) {
267
- close(host->rrdpush_sender_socket);
268
- host->rrdpush_sender_socket = -1;
269
- }
274
+ rrdpush_sender_thread_close_socket(host);
275
276
// close the pipe
277
if(host->rrdpush_sender_pipe[PIPE_READ] != -1) {
@@ -349,11 +354,13 @@ void *rrdpush_sender_thread(void *ptr) {
354
// check for outstanding cancellation requests
355
pthread_testcancel();
356
357
+ if(host->rrdpush_sender_socket == -1)
358
+ sleep(reconnect_delay);
359
+
360
debug(D_STREAM, "STREAM: Checking if we need to timeout the connection...");
361
if(host->rrdpush_sender_socket != -1 && now_monotonic_sec() - last_sent_t > timeout) {
362
error("STREAM %s [send to %s]: could not send metrics for %d seconds - closing connection - we have sent %zu bytes on this connection.", host->hostname, connected_to, timeout, sent_connection);
355
- close(host->rrdpush_sender_socket);
356
- host->rrdpush_sender_socket = -1;
363
+ rrdpush_sender_thread_close_socket(host);
364
}
365
366
if(unlikely(host->rrdpush_sender_socket == -1)) {
@@ -368,7 +375,6 @@ void *rrdpush_sender_thread(void *ptr) {
375
376
if(unlikely(host->rrdpush_sender_socket == -1)) {
377
error("STREAM %s [send to %s]: failed to connect", host->hostname, host->rrdpush_send_destination);
371
- sleep(reconnect_delay);
378
continue;
379
}
380
@@ -391,28 +397,22 @@ void *rrdpush_sender_thread(void *ptr) {
397
);
398
399
if(send_timeout(host->rrdpush_sender_socket, http, strlen(http), 0, timeout) == -1) {
394
- close(host->rrdpush_sender_socket);
395
- host->rrdpush_sender_socket = -1;
400
error("STREAM %s [send to %s]: failed to send http header to netdata", host->hostname, connected_to);
397
- sleep(reconnect_delay);
401
+ rrdpush_sender_thread_close_socket(host);
402
continue;
403
}
404
405
info("STREAM %s [send to %s]: waiting response from remote netdata...", host->hostname, connected_to);
406
407
if(recv_timeout(host->rrdpush_sender_socket, http, HTTP_HEADER_SIZE, 0, timeout) == -1) {
404
- close(host->rrdpush_sender_socket);
405
- host->rrdpush_sender_socket = -1;
408
error("STREAM %s [send to %s]: failed to initialize communication", host->hostname, connected_to);
407
- sleep(reconnect_delay);
409
+ rrdpush_sender_thread_close_socket(host);
410
continue;
411
}
412
413
if(strncmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT))) {
412
- close(host->rrdpush_sender_socket);
413
- host->rrdpush_sender_socket = -1;
414
error("STREAM %s [send to %s]: server is not replying properly.", host->hostname, connected_to);
415
- sleep(reconnect_delay);
415
+ rrdpush_sender_thread_close_socket(host);
416
continue;
417
}
418
@@ -465,8 +465,7 @@ void *rrdpush_sender_thread(void *ptr) {
465
}
466
467
error("STREAM %s [send to %s]: failed to poll().", host->hostname, connected_to);
468
- close(host->rrdpush_sender_socket);
469
- host->rrdpush_sender_socket = -1;
468
+ rrdpush_sender_thread_close_socket(host);
469
break;
470
}
471
else if(likely(retval)) {
@@ -501,15 +500,14 @@ void *rrdpush_sender_thread(void *ptr) {
500
if (errno != EAGAIN && errno != EINTR && errno != EWOULDBLOCK) {
501
debug(D_STREAM, "STREAM: Send failed - closing socket...");
502
error("STREAM %s [send to %s]: failed to send metrics - closing connection - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
504
- close(host->rrdpush_sender_socket);
505
- host->rrdpush_sender_socket = -1;
503
+ rrdpush_sender_thread_close_socket(host);
504
}
505
else {
506
debug(D_STREAM, "STREAM: Send failed - will retry...");
507
}
508
}
509
else if (likely(ret > 0)) {
512
- // DEBUG - dump the scring to see it
510
+ // DEBUG - dump the string to see it
511
//char c = host->rrdpush_sender_buffer->buffer[begin + ret];
512
//host->rrdpush_sender_buffer->buffer[begin + ret] = '\0';
513
//debug(D_STREAM, "STREAM: sent from %zu to %zd:\n%s\n", begin, ret, &host->rrdpush_sender_buffer->buffer[begin]);
@@ -536,8 +534,7 @@ void *rrdpush_sender_thread(void *ptr) {
534
debug(D_STREAM, "STREAM: send() returned %zd - closing the socket...", ret);
535
error("STREAM %s [send to %s]: failed to send metrics (send() returned %zd) - closing connection - we have sent %zu bytes on this connection.",
536
host->hostname, connected_to, ret, sent_connection);
539
- close(host->rrdpush_sender_socket);
540
- host->rrdpush_sender_socket = -1;
537
+ rrdpush_sender_thread_close_socket(host);
538
}
539
540
debug(D_STREAM, "STREAM: Releasing exclusive lock on host...");
@@ -556,20 +553,17 @@ void *rrdpush_sender_thread(void *ptr) {
553
if(unlikely(ofd->revents & POLLERR)) {
554
debug(D_STREAM, "STREAM: Send failed (POLLERR) - closing socket...");
555
error("STREAM %s [send to %s]: connection reports errors (POLLERR), closing it - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
559
- close(host->rrdpush_sender_socket);
560
- host->rrdpush_sender_socket = -1;
556
+ rrdpush_sender_thread_close_socket(host);
557
}
558
else if(unlikely(ofd->revents & POLLHUP)) {
559
debug(D_STREAM, "STREAM: Send failed (POLLHUP) - closing socket...");
560
error("STREAM %s [send to %s]: connection closed by remote end (POLLHUP) - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
565
- close(host->rrdpush_sender_socket);
566
- host->rrdpush_sender_socket = -1;
561
+ rrdpush_sender_thread_close_socket(host);
562
}
563
else if(unlikely(ofd->revents & POLLNVAL)) {
564
debug(D_STREAM, "STREAM: Send failed (POLLNVAL) - closing socket...");
565
error("STREAM %s [send to %s]: connection is invalid (POLLNVAL), closing it - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
571
- close(host->rrdpush_sender_socket);
572
- host->rrdpush_sender_socket = -1;
566
+ rrdpush_sender_thread_close_socket(host);
567
}
568
}
569
else {
@@ -581,10 +575,7 @@ void *rrdpush_sender_thread(void *ptr) {
575
debug(D_STREAM, "STREAM: Buffer is too big (%zu bytes), bigger than the max (%zu) - flushing it...", buffer_strlen(host->rrdpush_sender_buffer), max_size);
576
errno = 0;
577
error("STREAM %s [send to %s]: too many data pending - buffer is %zu bytes long, %zu unsent - we have sent %zu bytes in total, %zu on this connection. Closing connection to flush the data.", host->hostname, connected_to, host->rrdpush_sender_buffer->len, host->rrdpush_sender_buffer->len - begin, sent_bytes, sent_connection);
584
- if(host->rrdpush_sender_socket != -1) {
585
- close(host->rrdpush_sender_socket);
586
- host->rrdpush_sender_socket = -1;
587
- }
578
+ rrdpush_sender_thread_close_socket(host);
579
}
580
}
581