dequeue backends responses from the socket; #1432
Costa Tsaousis (ktsaou) committed
Dec 24, 2016 at 20:04 UTC
d6d37b5b2f80bb0026a88d98b2500e13df9f171e
2 files changed
+89
-10
src/backends.c
+88
-9
@@ -94,11 +94,24 @@ static inline int format_dimension_stored_opentsdb_telnet(BUFFER *b, const char
94
return 0;
95
}
96
97
+static inline int process_graphite_response(BUFFER *b) {
98
+ info("Received %zu bytes from graphite backend. Ignoring them.", buffer_strlen(b));
99
+ buffer_flush(b);
100
+ return 0;
101
+}
102
+
103
+static inline int process_opentsdb_response(BUFFER *b) {
104
+ info("Received %zu bytes from opentsdb backend. Ignoring them.", buffer_strlen(b));
105
+ buffer_flush(b);
106
+ return 0;
107
+}
108
+
109
void *backends_main(void *ptr) {
110
(void)ptr;
111
100
- BUFFER *b = buffer_create(1);
101
- int (*formatter)(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options);
112
+ BUFFER *b = buffer_create(1), *response = buffer_create(1);
113
+ int (*backend_request_formatter)(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) = NULL;
114
+ int (*backend_response_checker)(BUFFER *b) = NULL;
115
116
info("BACKEND thread created with task id %d", gettid());
117
@@ -151,22 +164,31 @@ void *backends_main(void *ptr) {
164
if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
165
default_port = 2003;
166
if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
154
- formatter = format_dimension_collected_graphite_plaintext;
167
+ backend_request_formatter = format_dimension_collected_graphite_plaintext;
168
else
156
- formatter = format_dimension_stored_graphite_plaintext;
169
+ backend_request_formatter = format_dimension_stored_graphite_plaintext;
170
+
171
+ backend_response_checker = process_graphite_response;
172
}
173
else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
174
default_port = 4242;
175
if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
161
- formatter = format_dimension_collected_opentsdb_telnet;
176
+ backend_request_formatter = format_dimension_collected_opentsdb_telnet;
177
else
163
- formatter = format_dimension_stored_opentsdb_telnet;
178
+ backend_request_formatter = format_dimension_stored_opentsdb_telnet;
179
+
180
+ backend_response_checker = process_opentsdb_response;
181
}
182
else {
183
error("Unknown backend type '%s'", type);
184
goto cleanup;
185
}
186
187
+ if(backend_request_formatter == NULL || backend_response_checker == NULL) {
188
+ error("backend is misconfigured - disabling it.");
189
+ goto cleanup;
190
+ }
191
+
192
if(timeoutms < 1) {
193
error("BACKED invalid timeout %ld ms given. Assuming %d ms.", timeoutms, frequency * 2 * 1000);
194
timeoutms = frequency * 2 * 1000;
@@ -273,7 +295,7 @@ void *backends_main(void *ptr) {
295
RRDDIM *rd;
296
for(rd = st->dimensions; rd ;rd = rd->next) {
297
if(rd->last_collected_time.tv_sec >= after)
276
- chart_buffered_metrics += formatter(b, prefix, &localhost, hostname, st, rd, after, before, options);
298
+ chart_buffered_metrics += backend_request_formatter(b, prefix, &localhost, hostname, st, rd, after, before, options);
299
}
300
301
pthread_rwlock_unlock(&st->rwlock);
@@ -283,6 +305,8 @@ void *backends_main(void *ptr) {
305
if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
306
error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
307
308
+ // ------------------------------------------------------------------------
309
+
310
chart_buffered_bytes = (collected_number)buffer_strlen(b);
311
312
// reset the monitoring chart counters
@@ -302,7 +326,58 @@ void *backends_main(void *ptr) {
326
//fprintf(stderr, "after = %lu, before = %lu\n", after, before);
327
328
// ------------------------------------------------------------------------
305
- // connect to a backend server
329
+ // if we are connected, receive a response, without blocking
330
+
331
+ if(likely(sock != -1)) {
332
+ if(likely(fcntl(sock, F_SETFL, O_NONBLOCK) >= 0)) {
333
+ errno = 0;
334
+
335
+ // loop through to collect all data
336
+ while(sock != -1 && errno != EWOULDBLOCK) {
337
+ buffer_need_bytes(response, 4096);
338
+
339
+ ssize_t r = recv(sock, &response->buffer[response->len], response->size - response->len
340
+ , O_NONBLOCK);
341
+ if(likely(r > 0)) {
342
+ // we received some data
343
+ response->len += r;
344
+ }
345
+ else if(r == 0) {
346
+ error("Backend '%s' closed the socket", destination);
347
+ close(sock);
348
+ sock = -1;
349
+ }
350
+ else {
351
+ // failed to receive data
352
+ if(errno != EAGAIN && errno != EWOULDBLOCK)
353
+ error("Cannot receive data from backend '%s'.", destination);
354
+ }
355
+ }
356
+
357
+ if(sock != -1) {
358
+ // unset O_NONBLOCK
359
+ int val = fcntl(sock, F_GETFL, 0);
360
+ if(likely(val >= 0)) {
361
+ int flags = O_NONBLOCK;
362
+ val &= ~flags;
363
+ if(unlikely(fcntl(sock, F_SETFL, val) < 0))
364
+ error("Cannot unset O_NONBLOCK from backend's '%s' socket.", destination);
365
+ }
366
+ else
367
+ error("Cannot get active flags from backend's '%s' socket.", destination);
368
+ }
369
+
370
+ // if we received data, process them
371
+ if(buffer_strlen(response))
372
+ backend_response_checker(response);
373
+ }
374
+ else {
375
+ error("Cannot set O_NONBLOCK from backend's '%s' socket. Receiving data from backend skipped.", destination);
376
+ }
377
+ }
378
+
379
+ // ------------------------------------------------------------------------
380
+ // if we are not connected, connect to a backend server
381
382
if(unlikely(sock == -1)) {
383
usec_t start_ut = now_realtime_usec();
@@ -332,7 +407,7 @@ void *backends_main(void *ptr) {
407
if(unlikely(netdata_exit)) break;
408
409
// ------------------------------------------------------------------------
335
- // send our buffer to the backend server
410
+ // if we are connected, send our buffer to the backend server
411
412
if(likely(sock != -1)) {
413
size_t len = buffer_strlen(b);
@@ -341,6 +416,7 @@ void *backends_main(void *ptr) {
416
#ifdef MSG_NOSIGNAL
417
flags += MSG_NOSIGNAL;
418
#endif
419
+
420
ssize_t written = send(sock, buffer_tostring(b), len, flags);
421
chart_backend_latency += now_realtime_usec() - start_ut;
422
if(written != -1 && (size_t)written == len) {
@@ -439,6 +515,9 @@ cleanup:
515
if(sock != -1)
516
close(sock);
517
518
+ buffer_free(b);
519
+ buffer_free(response);
520
+
521
info("BACKEND thread exiting");
522
523
pthread_exit(NULL);
src/socket.c
+1
-1
@@ -85,7 +85,7 @@ int connect_to(const char *definition, int default_port, struct timeval *timeout
85
86
int ai_err = getaddrinfo(host, service, &hints, &ai_head);
87
if (ai_err != 0) {
88
- error("Cannot resolve host '%s', port '%s': %s\n", host, service, gai_strerror(ai_err));
88
+ error("Cannot resolve host '%s', port '%s': %s", host, service, gai_strerror(ai_err));
89
return -1;
90
}
91