allow streaming to work on FreeBSD; #2683
Costa Tsaousis (ktsaou) committed
Sep 7, 2017 at 02:12 UTC
3b3421c7da2012fa1df48ef6005cb98f1120f218
1 file changed
+90
-59
src/rrdpush.c
+90
-59
@@ -113,12 +113,12 @@ static inline void send_chart_definition(RRDSET *st) {
113
114
// sends the current chart dimensions
115
static inline void send_chart_metrics(RRDSET *st) {
116
- buffer_sprintf(st->rrdhost->rrdpush_buffer, "BEGIN %s %llu\n", st->id, (st->upstream_resync_time > st->last_collected_time.tv_sec)?st->usec_since_last_update:0);
116
+ buffer_sprintf(st->rrdhost->rrdpush_buffer, "BEGIN \"%s\" %llu\n", st->id, (st->upstream_resync_time > st->last_collected_time.tv_sec)?st->usec_since_last_update:0);
117
118
RRDDIM *rd;
119
rrddim_foreach_read(rd, st) {
120
if(rd->updated && rd->exposed)
121
- buffer_sprintf(st->rrdhost->rrdpush_buffer, "SET %s = " COLLECTED_NUMBER_FORMAT "\n"
121
+ buffer_sprintf(st->rrdhost->rrdpush_buffer, "SET \"%s\" = " COLLECTED_NUMBER_FORMAT "\n"
122
, rd->id
123
, rd->collected_value
124
);
@@ -380,7 +380,7 @@ void *rrdpush_sender_thread(void *ptr) {
380
// allow appending data into rrdpush_buffer
381
host->rrdpush_connected = 1;
382
383
- debug(D_STREAM, "Connected...");
383
+ debug(D_STREAM, "STREAM: Connected on fd %d...", host->rrdpush_socket);
384
}
385
386
ifd->fd = host->rrdpush_pipe[PIPE_READ];
@@ -389,13 +389,13 @@ void *rrdpush_sender_thread(void *ptr) {
389
390
ofd->fd = host->rrdpush_socket;
391
ofd->revents = 0;
392
- if(begin < buffer_strlen(host->rrdpush_buffer)) {
393
- debug(D_STREAM, "STREAM: Requesting data output on streaming socket...");
392
+ if(ofd->fd != -1 && begin < buffer_strlen(host->rrdpush_buffer)) {
393
+ debug(D_STREAM, "STREAM: Requesting data output on streaming socket %d...", ofd->fd);
394
ofd->events = POLLOUT;
395
fdmax = 2;
396
}
397
else {
398
- debug(D_STREAM, "STREAM: Not requesting data output on streaming socket (nothing to send now)...");
398
+ debug(D_STREAM, "STREAM: Not requesting data output on streaming socket %d (nothing to send now)...", ofd->fd);
399
ofd->events = 0;
400
fdmax = 1;
401
}
@@ -419,7 +419,7 @@ void *rrdpush_sender_thread(void *ptr) {
419
break;
420
}
421
else if(likely(retval)) {
422
- if (ifd->revents & POLLIN) {
422
+ if (ifd->revents & POLLIN || ifd->revents & POLLPRI) {
423
debug(D_STREAM, "STREAM: Data added to send buffer (current buffer length %zu bytes)...", buffer_strlen(host->rrdpush_buffer));
424
425
char buffer[1000 + 1];
@@ -427,67 +427,98 @@ void *rrdpush_sender_thread(void *ptr) {
427
error("STREAM %s [send to %s]: cannot read from internal pipe.", host->hostname, connected_to);
428
}
429
430
- if (ofd->revents & POLLOUT && begin < buffer_strlen(host->rrdpush_buffer)) {
431
- debug(D_STREAM, "STREAM: Sending data (current buffer length %zu bytes)...", buffer_strlen(host->rrdpush_buffer));
432
-
433
- // BEGIN RRDPUSH LOCKED SESSION
434
-
435
- // during this session, data collectors
436
- // will not be able to append data to our buffer
437
- // but the socket is in non-blocking mode
438
- // so, we will not block at send()
439
-
440
- if (pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL) != 0)
441
- error("STREAM %s [send]: cannot set pthread cancel state to DISABLE.", host->hostname);
442
-
443
- debug(D_STREAM, "STREAM: Getting exclusive lock on host...");
444
- rrdpush_lock(host);
445
-
446
- debug(D_STREAM, "STREAM: Sending data, starting from %zu, size %zu...", begin, buffer_strlen(host->rrdpush_buffer));
447
- ssize_t ret = send(host->rrdpush_socket, &host->rrdpush_buffer->buffer[begin], buffer_strlen(host->rrdpush_buffer) - begin, MSG_DONTWAIT);
448
- if (unlikely(ret == -1)) {
449
- if (errno != EAGAIN && errno != EINTR && errno != EWOULDBLOCK) {
450
- debug(D_STREAM, "STREAM: Send failed - closing socket...");
451
- 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);
452
- close(host->rrdpush_socket);
453
- host->rrdpush_socket = -1;
430
+ if (ofd->revents & POLLOUT) {
431
+ if (begin < buffer_strlen(host->rrdpush_buffer)) {
432
+ debug(D_STREAM, "STREAM: Sending data (current buffer length %zu bytes, begin = %zu)...", buffer_strlen(host->rrdpush_buffer), begin);
433
+
434
+ // BEGIN RRDPUSH LOCKED SESSION
435
+
436
+ // during this session, data collectors
437
+ // will not be able to append data to our buffer
438
+ // but the socket is in non-blocking mode
439
+ // so, we will not block at send()
440
+
441
+ if (pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL) != 0)
442
+ error("STREAM %s [send]: cannot set pthread cancel state to DISABLE.", host->hostname);
443
+
444
+ debug(D_STREAM, "STREAM: Getting exclusive lock on host...");
445
+ rrdpush_lock(host);
446
+
447
+ debug(D_STREAM, "STREAM: Sending data, starting from %zu, size %zu...", begin, buffer_strlen(host->rrdpush_buffer));
448
+ ssize_t ret = send(host->rrdpush_socket, &host->rrdpush_buffer->buffer[begin], buffer_strlen(host->rrdpush_buffer) - begin, MSG_DONTWAIT);
449
+ if (unlikely(ret == -1)) {
450
+ if (errno != EAGAIN && errno != EINTR && errno != EWOULDBLOCK) {
451
+ debug(D_STREAM, "STREAM: Send failed - closing socket...");
452
+ 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);
453
+ close(host->rrdpush_socket);
454
+ host->rrdpush_socket = -1;
455
+ }
456
+ else {
457
+ debug(D_STREAM, "STREAM: Send failed - will retry...");
458
+ }
459
+ }
460
+ else if (likely(ret > 0)) {
461
+ // DEBUG - dump the scring to see it
462
+ //char c = host->rrdpush_buffer->buffer[begin + ret];
463
+ //host->rrdpush_buffer->buffer[begin + ret] = '\0';
464
+ //debug(D_STREAM, "STREAM: sent from %zu to %zd:\n%s\n", begin, ret, &host->rrdpush_buffer->buffer[begin]);
465
+ //host->rrdpush_buffer->buffer[begin + ret] = c;
466
+
467
+ sent_connection += ret;
468
+ sent_bytes += ret;
469
+ begin += ret;
470
+
471
+ if (begin == buffer_strlen(host->rrdpush_buffer)) {
472
+ // we send it all
473
+
474
+ debug(D_STREAM, "STREAM: Sent %zd bytes (the whole buffer)...", ret);
475
+ buffer_flush(host->rrdpush_buffer);
476
+ begin = 0;
477
+ }
478
+ else {
479
+ debug(D_STREAM, "STREAM: Sent %zd bytes (part of the data buffer)...", ret);
480
+ }
481
+
482
+ last_sent_t = now_monotonic_sec();
483
}
484
else {
456
- debug(D_STREAM, "STREAM: Send failed - will retry...");
485
+ debug(D_STREAM, "STREAM: send() returned %zd - closing the socket...", ret);
486
+ error("STREAM %s [send to %s]: failed to send metrics (send() returned %zd) - closing connection - we have sent %zu bytes on this connection.",
487
+ host->hostname, connected_to, ret, sent_connection);
488
+ close(host->rrdpush_socket);
489
+ host->rrdpush_socket = -1;
490
}
458
- }
459
- else if(likely(ret > 0)) {
460
- sent_connection += ret;
461
- sent_bytes += ret;
462
- begin += ret;
491
464
- if (begin == buffer_strlen(host->rrdpush_buffer)) {
465
- // we send it all
492
+ debug(D_STREAM, "STREAM: Releasing exclusive lock on host...");
493
+ rrdpush_unlock(host);
494
467
- debug(D_STREAM, "STREAM: Sent %zd bytes (the whole buffer)...", ret);
468
- buffer_flush(host->rrdpush_buffer);
469
- begin = 0;
470
- }
471
- else {
472
- debug(D_STREAM, "STREAM: Sent %zd bytes (part of the data buffer)...", ret);
473
- }
495
+ if (pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
496
+ error("STREAM %s [send]: cannot set pthread cancel state to ENABLE.", host->hostname);
497
475
- last_sent_t = now_monotonic_sec();
498
+ // END RRDPUSH LOCKED SESSION
499
}
500
else {
478
- debug(D_STREAM, "STREAM: send() returned %zd - closing the socket...", ret);
479
- error("STREAM %s [send to %s]: failed to send metrics (send() returned %zd) - closing connection - we have sent %zu bytes on this connection.", host->hostname, connected_to, ret, sent_connection);
480
- close(host->rrdpush_socket);
481
- host->rrdpush_socket = -1;
501
+ debug(D_STREAM, "STREAM: we have sent the entire buffer, but we received POLLOUT...");
502
}
503
+ }
504
484
- debug(D_STREAM, "STREAM: Releasing exclusive lock on host...");
485
- rrdpush_unlock(host);
486
-
487
- if (pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
488
- error("STREAM %s [send]: cannot set pthread cancel state to ENABLE.", host->hostname);
489
-
490
- // END RRDPUSH LOCKED SESSION
505
+ if(unlikely(ofd->revents & POLLERR)) {
506
+ debug(D_STREAM, "STREAM: Send failed (POLLERR) - closing socket...");
507
+ error("STREAM %s [send to %s]: connection reports errors (POLLERR) - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
508
+ close(host->rrdpush_socket);
509
+ host->rrdpush_socket = -1;
510
+ }
511
+ else if(unlikely(ofd->revents & POLLHUP)) {
512
+ debug(D_STREAM, "STREAM: Send failed (POLLHUP) - closing socket...");
513
+ error("STREAM %s [send to %s]: connection closed (POLLHUP) - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
514
+ close(host->rrdpush_socket);
515
+ host->rrdpush_socket = -1;
516
+ }
517
+ else if(unlikely(ofd->revents & POLLNVAL)) {
518
+ debug(D_STREAM, "STREAM: Send failed (POLLNVAL) - closing socket...");
519
+ error("STREAM %s [send to %s]: connection is invalid (POLLNVAL) - we have sent %zu bytes on this connection.", host->hostname, connected_to, sent_connection);
520
+ close(host->rrdpush_socket);
521
+ host->rrdpush_socket = -1;
522
}
523
}
524
else {
@@ -602,7 +633,7 @@ static int rrdpush_receive(int fd, const char *key, const char *hostname, const
633
, host->rrd_history_entries
634
, rrd_memory_mode_name(host->rrd_memory_mode)
635
, (health_enabled == CONFIG_BOOLEAN_NO)?"disabled":((health_enabled == CONFIG_BOOLEAN_YES)?"enabled":"auto")
605
- , host->tags
636
+ , host->tags?host->tags:""
637
);
638
#endif // NETDATA_INTERNAL_CHECKS
639