unified streaming logging
Costa Tsaousis (ktsaou) committed
Feb 23, 2017 at 20:35 UTC
4c0f57dbe7699faeb52b4eed04f7604db846f459
1 file changed
+52
-56
src/rrdpush.c
+52
-56
@@ -3,7 +3,7 @@
3
int rrdpush_enabled = 0;
4
int rrdpush_exclusive = 1;
5
6
-static char *central_netdata = NULL;
6
+static char *remote_netdata_config = NULL;
7
static char *api_key = NULL;
8
9
#define CONNECTED_TO_SIZE 100
@@ -25,7 +25,7 @@ static BUFFER *rrdpush_buffer = NULL;
25
// (rrdpush_pipe[PIPE_WRITE], rrdpush_buffer
26
static pthread_mutex_t rrdpush_mutex = PTHREAD_MUTEX_INITIALIZER;
27
28
-// if the streaming thread is connected to a central netdata
28
+// if the streaming thread is connected to a remote netdata
29
// this is set to 1, otherwise 0.
30
static volatile int rrdpush_connected = 0;
31
@@ -131,15 +131,18 @@ void rrdset_done_push(RRDSET *st) {
131
rrdpush_lock();
132
133
if(unlikely(!rrdpush_buffer || !rrdpush_connected)) {
134
- if(!error_shown)
135
- error("STREAM: not ready - discarding collected metrics.");
134
+ if(unlikely(!error_shown))
135
+ error("STREAM [send]: not ready - discarding collected metrics.");
136
137
error_shown = 1;
138
139
rrdpush_unlock();
140
return;
141
}
142
- error_shown = 0;
142
+ else if(unlikely(error_shown)) {
143
+ error("STREAM [send]: ready - sending metrics...");
144
+ error_shown = 0;
145
+ }
146
147
rrdset_rdlock(st);
148
if(need_to_send_chart_definition(st))
@@ -150,7 +153,7 @@ void rrdset_done_push(RRDSET *st) {
153
154
// signal the sender there are more data
155
if(write(rrdpush_pipe[PIPE_WRITE], " ", 1) == -1)
153
- error("STREAM: cannot write to internal pipe");
156
+ error("STREAM [send]: cannot write to internal pipe");
157
158
rrdpush_unlock();
159
}
@@ -158,7 +161,7 @@ void rrdset_done_push(RRDSET *st) {
161
static inline void rrdpush_flush(void) {
162
rrdpush_lock();
163
if(buffer_strlen(rrdpush_buffer))
161
- error("STREAM: discarding %zu bytes of metrics data already in the buffer.", buffer_strlen(rrdpush_buffer));
164
+ error("STREAM [send]: discarding %zu bytes of metrics already in the buffer.", buffer_strlen(rrdpush_buffer));
165
166
buffer_flush(rrdpush_buffer);
167
reset_all_charts();
@@ -168,10 +171,10 @@ static inline void rrdpush_flush(void) {
171
int rrdpush_init() {
172
rrdpush_enabled = config_get_boolean("stream", "enabled", rrdpush_enabled);
173
rrdpush_exclusive = config_get_boolean("stream", "exclusive", rrdpush_exclusive);
171
- central_netdata = config_get("stream", "stream metrics to", "");
174
+ remote_netdata_config = config_get("stream", "stream metrics to", "");
175
api_key = config_get("stream", "api key", "");
176
174
- if(!rrdpush_enabled || !central_netdata || !*central_netdata || !api_key || !*api_key) {
177
+ if(!rrdpush_enabled || !remote_netdata_config || !*remote_netdata_config || !api_key || !*api_key) {
178
rrdpush_enabled = 0;
179
rrdpush_exclusive = 0;
180
}
@@ -182,13 +185,13 @@ int rrdpush_init() {
185
void *rrdpush_sender_thread(void *ptr) {
186
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
187
185
- info("STREAM: central netdata push thread created with task id %d", gettid());
188
+ info("STREAM [send]: thread created (task id %d)", gettid());
189
190
if(pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
188
- error("STREAM: cannot set pthread cancel type to DEFERRED.");
191
+ error("STREAM [send]: cannot set pthread cancel type to DEFERRED.");
192
193
if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
191
- error("STREAM: cannot set pthread cancel state to ENABLE.");
194
+ error("STREAM [send]: cannot set pthread cancel state to ENABLE.");
195
196
int timeout = (int)config_get_number("stream", "timeout seconds", 60);
197
int default_port = (int)config_get_number("stream", "default port", 19999);
@@ -196,14 +199,15 @@ void *rrdpush_sender_thread(void *ptr) {
199
unsigned int reconnect_delay = (unsigned int)config_get_number("stream", "reconnect delay seconds", 5);
200
remote_clock_resync_iterations = (unsigned int)config_get_number("stream", "initial clock resync iterations", remote_clock_resync_iterations);
201
int sock = -1;
202
+ char connected_to[CONNECTED_TO_SIZE + 1] = "";
203
200
- if(!rrdpush_enabled || !central_netdata || !*central_netdata || !api_key || !*api_key)
204
+ if(!rrdpush_enabled || !remote_netdata_config || !*remote_netdata_config || !api_key || !*api_key)
205
goto cleanup;
206
207
// initialize rrdpush globals
208
rrdpush_buffer = buffer_create(1);
209
rrdpush_connected = 0;
206
- if(pipe(rrdpush_pipe) == -1) fatal("STREAM: cannot create required pipe.");
210
+ if(pipe(rrdpush_pipe) == -1) fatal("STREAM [send]: cannot create required pipe.");
211
212
// initialize local variables
213
size_t begin = 0;
@@ -222,8 +226,6 @@ void *rrdpush_sender_thread(void *ptr) {
226
ifd = &fds[0];
227
ofd = &fds[1];
228
225
- char connected_to[CONNECTED_TO_SIZE + 1];
226
-
229
for(;;) {
230
if(netdata_exit) break;
231
@@ -232,16 +234,16 @@ void *rrdpush_sender_thread(void *ptr) {
234
// they will be lost, so there is no point to do it
235
rrdpush_connected = 0;
236
235
- info("STREAM: connecting to central netdata at: %s", central_netdata);
236
- sock = connect_to_one_of(central_netdata, default_port, &tv, &reconnects_counter, connected_to, CONNECTED_TO_SIZE);
237
+ info("STREAM [send to %s]: connecting...", remote_netdata_config);
238
+ sock = connect_to_one_of(remote_netdata_config, default_port, &tv, &reconnects_counter, connected_to, CONNECTED_TO_SIZE);
239
240
if(unlikely(sock == -1)) {
239
- error("STREAM: failed to connect to central netdata at: %s", central_netdata);
241
+ error("STREAM [send to %s]: failed to connect", remote_netdata_config);
242
sleep(reconnect_delay);
243
continue;
244
}
245
244
- info("STREAM: initializing communication to central netdata at: %s", connected_to);
246
+ info("STREAM [send to %s]: initializing communication...", connected_to);
247
248
char http[1000 + 1];
249
snprintfz(http, 1000,
@@ -259,17 +261,17 @@ void *rrdpush_sender_thread(void *ptr) {
261
if(send_timeout(sock, http, strlen(http), 0, timeout) == -1) {
262
close(sock);
263
sock = -1;
262
- error("STREAM: failed to send http header to netdata at: %s", connected_to);
264
+ error("STREAM [send to %s]: failed to send http header to netdata", connected_to);
265
sleep(reconnect_delay);
266
continue;
267
}
268
267
- info("STREAM: Waiting for STREAM from central netdata at: %s", connected_to);
269
+ info("STREAM [send to %s]: waiting response from remote netdata...", connected_to);
270
271
if(recv_timeout(sock, http, 1000, 0, timeout) == -1) {
272
close(sock);
273
sock = -1;
272
- error("STREAM: failed to receive STREAM from netdata at: %s", connected_to);
274
+ error("STREAM [send to %s]: failed to initialize communication", connected_to);
275
sleep(reconnect_delay);
276
continue;
277
}
@@ -277,15 +279,15 @@ void *rrdpush_sender_thread(void *ptr) {
279
if(strncmp(http, "STREAM", 6)) {
280
close(sock);
281
sock = -1;
280
- error("STREAM: server at %s, did not send STREAM", connected_to);
282
+ error("STREAM [send to %s]: server is not replying properly.", connected_to);
283
sleep(reconnect_delay);
284
continue;
285
}
286
285
- info("STREAM: Established communication with central netdata at: %s - sending metrics...", connected_to);
287
+ info("STREAM [send to %s]: established communication - sending metrics...", connected_to);
288
289
if(fcntl(sock, F_SETFL, O_NONBLOCK) < 0)
288
- error("STREAM: cannot set non-blocking mode for socket.");
290
+ error("STREAM [send to %s]: cannot set non-blocking mode for socket.", connected_to);
291
292
rrdpush_flush();
293
sent_connection = 0;
@@ -317,7 +319,7 @@ void *rrdpush_sender_thread(void *ptr) {
319
if(errno == EAGAIN || errno == EINTR)
320
continue;
321
320
- error("STREAM: Failed to poll().");
322
+ error("STREAM [send to %s]: failed to poll().", connected_to);
323
close(sock);
324
sock = -1;
325
break;
@@ -330,21 +332,15 @@ void *rrdpush_sender_thread(void *ptr) {
332
if(ifd->revents & POLLIN) {
333
char buffer[1000 + 1];
334
if(read(rrdpush_pipe[PIPE_READ], buffer, 1000) == -1)
333
- error("STREAM: Cannot read from internal pipe.");
335
+ error("STREAM [send to %s]: cannot read from internal pipe.", connected_to);
336
}
337
338
if(ofd->revents & POLLOUT && begin < buffer_strlen(rrdpush_buffer)) {
337
- // info("STREAM: send buffer is ready, sending %zu bytes starting at %zu", buffer_strlen(rrdpush_buffer) - begin, begin);
338
-
339
- // fprintf(stderr, "PUSH BEGIN\n");
340
- // fwrite(&rrdpush_buffer->buffer[begin], 1, buffer_strlen(rrdpush_buffer) - begin, stderr);
341
- // fprintf(stderr, "\nPUSH END\n");
342
-
339
rrdpush_lock();
340
ssize_t ret = send(sock, &rrdpush_buffer->buffer[begin], buffer_strlen(rrdpush_buffer) - begin, MSG_DONTWAIT);
341
if(ret == -1) {
342
if(errno != EAGAIN && errno != EINTR) {
347
- error("STREAM: failed to send metrics to central netdata at %s. We have sent %zu bytes on this connection.", connected_to, sent_connection);
343
+ error("STREAM [send to %s]: failed to send metrics - closing connection - we have sent %zu bytes on this connection.", connected_to, sent_connection);
344
close(sock);
345
sock = -1;
346
}
@@ -364,7 +360,7 @@ void *rrdpush_sender_thread(void *ptr) {
360
// protection from overflow
361
if(rrdpush_buffer->len > max_size) {
362
errno = 0;
367
- error("STREAM: 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.", rrdpush_buffer->len, rrdpush_buffer->len - begin, sent_bytes, sent_connection);
363
+ error("STREAM [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.", connected_to, rrdpush_buffer->len, rrdpush_buffer->len - begin, sent_bytes, sent_connection);
364
if(sock != -1) {
365
close(sock);
366
sock = -1;
@@ -373,7 +369,7 @@ void *rrdpush_sender_thread(void *ptr) {
369
}
370
371
cleanup:
376
- debug(D_WEB_CLIENT, "STREAM: central netdata push thread exits.");
372
+ debug(D_WEB_CLIENT, "STREAM [send]: sending thread exits.");
373
374
// make sure the data collection threads do not write data
375
rrdpush_connected = 0;
@@ -427,7 +423,7 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *m
423
else
424
host = rrdhost_find_or_create(hostname, machine_guid, os, update_every, history, mode, health_enabled?1:0);
425
430
- info("STREAM request from client '%s:%s' for host '%s' with machine_guid '%s': update every = %d, history = %d, memory mode = %s, health %s",
426
+ info("STREAM [receive from [%s]:%s]: metrics for host '%s' with machine_guid '%s': update every = %d, history = %d, memory mode = %s, health %s",
427
client_ip, client_port,
428
hostname, machine_guid,
429
update_every,
@@ -453,20 +449,20 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *m
449
snprintfz(cd.fullfilename, FILENAME_MAX, "%s:%s", client_ip, client_port);
450
snprintfz(cd.cmd, PLUGINSD_CMD_MAX, "%s:%s", client_ip, client_port);
451
456
- info("STREAM [%s]:%s: sending STREAM to initiate streaming...", client_ip, client_port);
452
+ info("STREAM [receive from [%s]:%s]: initializing communication...", client_ip, client_port);
453
if(send_timeout(fd, "STREAM", 6, 0, 60) != 6) {
458
- error("STREAM [%s]:%s: cannot send STREAM.", client_ip, client_port);
454
+ error("STREAM [receive from [%s]:%s]: cannot send STREAM command.", client_ip, client_port);
455
return 0;
456
}
457
458
// remove the non-blocking flag from the socket
459
if(fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) & ~O_NONBLOCK) == -1)
464
- error("STREAM [%s]:%s: cannot remove the non-blocking flag from socket %d", client_ip, client_port, fd);
460
+ error("STREAM [receive from [%s]:%s]: cannot remove the non-blocking flag from socket %d", client_ip, client_port, fd);
461
462
// convert the socket to a FILE *
463
FILE *fp = fdopen(fd, "r");
464
if(!fp) {
469
- error("STREAM [%s]:%s: failed to get a FILE for FD %d.", client_ip, client_port, fd);
465
+ error("STREAM [receive from [%s]:%s]: failed to get a FILE for FD %d.", client_ip, client_port, fd);
466
return 0;
467
}
468
@@ -477,9 +473,9 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *m
473
rrdhost_unlock(host);
474
475
// call the plugins.d processor to receive the metrics
480
- info("STREAM [%s]:%s: connecting client to plugins.d (host '%s', machine GUID '%s').", client_ip, client_port, host->hostname, host->machine_guid);
476
+ info("STREAM [receive from [%s]:%s]: connecting client to plugins.d (host '%s', machine GUID '%s').", client_ip, client_port, host->hostname, host->machine_guid);
477
size_t count = pluginsd_process(host, &cd, fp, 1);
482
- error("STREAM [%s]:%s: client disconnected (host '%s', machine GUID '%s', completed updates %zu).", client_ip, client_port, host->hostname, host->machine_guid, count);
478
+ error("STREAM [receive from [%s]:%s]: client disconnected (host '%s', machine GUID '%s', completed updates %zu).", client_ip, client_port, host->hostname, host->machine_guid, count);
479
480
rrdhost_wrlock(host);
481
host->use_counter--;
@@ -508,15 +504,15 @@ void *rrdpush_receiver_thread(void *ptr) {
504
struct rrdpush_thread *rpt = (struct rrdpush_thread *)ptr;
505
506
if (pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
511
- error("STREAM: cannot set pthread cancel type to DEFERRED.");
507
+ error("STREAM [receive]: cannot set pthread cancel type to DEFERRED.");
508
509
if (pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
514
- error("STREAM: cannot set pthread cancel state to ENABLE.");
510
+ error("STREAM [receive]: cannot set pthread cancel state to ENABLE.");
511
512
513
info("STREAM [%s]:%s: receive thread created (task id %d)", rpt->client_ip, rpt->client_port, gettid());
514
rrdpush_receive(rpt->fd, rpt->key, rpt->hostname, rpt->machine_guid, rpt->os, rpt->update_every, rpt->client_ip, rpt->client_port);
519
- info("STREAM [%s]:%s: receive thread ended (task id %d)", rpt->client_ip, rpt->client_port, gettid());
515
+ info("STREAM [receive from [%s]:%s]: receive thread ended (task id %d)", rpt->client_ip, rpt->client_port, gettid());
516
517
close(rpt->fd);
518
freez(rpt->key);
@@ -538,7 +534,7 @@ static inline int rrdpush_receive_validate_api_key(const char *key) {
534
int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url) {
535
(void)host;
536
541
- info("STREAM [%s]:%s: client connection.", w->client_ip, w->client_port);
537
+ info("STREAM [receive from [%s]:%s]: new client connection.", w->client_ip, w->client_port);
538
539
char *key = NULL, *hostname = NULL, *machine_guid = NULL, *os = NULL;
540
int update_every = default_rrd_update_every;
@@ -564,35 +560,35 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
560
}
561
562
if(!key || !*key) {
567
- error("STREAM [%s]:%s: request without an API key. Forbidding access.", w->client_ip, w->client_port);
563
+ error("STREAM [receive from [%s]:%s]: request without an API key. Forbidding access.", w->client_ip, w->client_port);
564
buffer_flush(w->response.data);
565
buffer_sprintf(w->response.data, "You need an API key for this request.");
566
return 401;
567
}
568
569
if(!hostname || !*hostname) {
574
- error("STREAM [%s]:%s: request without a hostname. Forbidding access.", w->client_ip, w->client_port);
570
+ error("STREAM [receive from [%s]:%s]: request without a hostname. Forbidding access.", w->client_ip, w->client_port);
571
buffer_flush(w->response.data);
572
buffer_sprintf(w->response.data, "You need to send a hostname too.");
573
return 400;
574
}
575
576
if(!machine_guid || !*machine_guid) {
581
- error("STREAM [%s]:%s: request without a machine GUID. Forbidding access.", w->client_ip, w->client_port);
577
+ error("STREAM [receive from [%s]:%s]: request without a machine GUID. Forbidding access.", w->client_ip, w->client_port);
578
buffer_flush(w->response.data);
579
buffer_sprintf(w->response.data, "You need to send a machine GUID too.");
580
return 400;
581
}
582
583
if(!rrdpush_receive_validate_api_key(key)) {
588
- error("STREAM [%s]:%s: API key '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, key);
584
+ error("STREAM [receive from [%s]:%s]: API key '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, key);
585
buffer_flush(w->response.data);
586
buffer_sprintf(w->response.data, "Your API key is not permitted access.");
587
return 401;
588
}
589
590
if(!appconfig_get_boolean(&stream_config, machine_guid, "enabled", 1)) {
595
- error("STREAM [%s]:%s: machine GUID '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, machine_guid);
591
+ error("STREAM [receive from [%s]:%s]: machine GUID '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, machine_guid);
592
buffer_flush(w->response.data);
593
buffer_sprintf(w->response.data, "Your machine guide is not permitted access.");
594
return 404;
@@ -609,13 +605,13 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
605
rpt->update_every = update_every;
606
pthread_t thread;
607
612
- debug(D_SYSTEM, "Starting STREAM thread for client [%s]:%s.", w->client_ip, w->client_port);
608
+ debug(D_SYSTEM, "STREAM [receive from [%s]:%s]: starting receiving thread.", w->client_ip, w->client_port);
609
610
if(pthread_create(&thread, NULL, rrdpush_receiver_thread, (void *)rpt))
615
- error("failed to create new STREAM thread for client [%s]:%s.", w->client_ip, w->client_port);
611
+ error("STREAM [receive from [%s]:%s]: failed to create new thread for client.", w->client_ip, w->client_port);
612
613
else if(pthread_detach(thread))
618
- error("Cannot request detach newly created thread for client [%s]:%s.", w->client_ip, w->client_port);
614
+ error("STREAM [receive from [%s]:%s]: cannot request detach newly created thread.", w->client_ip, w->client_port);
615
616
// prevent the caller from closing the streaming socket
617
if(w->ifd == w->ofd)