time_t first_t = rrdset_first_entry_t(st);
time_t last_t = rrdset_last_entry_t(st);
- if(unlikely(before - after < st->update_every && after != after - after % st->update_every))
- // when st->update_every is bigger than the frequency we send data to backend
- // skip the iterations that are not aligned to the database
+ if(unlikely(before < first_t || after > last_t))
+ // the chart has not been updated in the wanted timeframe
return NAN;
// align the time-frame
return 0;
}
+static inline int format_dimension_collected_json_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
+ (void)host;
+ (void)after;
+ (void)before;
+ (void)options;
+ buffer_sprintf(b, "{"
+ "\"prefix\":\"%s\","
+ "\"hostname\":\"%s\","
+ "\"id\":\"%s\","
+ "\"subid\":\"%s\","
+ "\"value\":" COLLECTED_NUMBER_FORMAT ","
+ "\"timestamp\": %u}\n", prefix, hostname, st->id, rd->id, rd->last_collected_value, (uint32_t)rd->last_collected_time.tv_sec);
+ return 1;
+}
+
+static inline int format_dimension_stored_json_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
+ (void)host;
+ calculated_number value = backend_calculate_value_from_stored_data(st, rd, after, before, options);
+ if(!isnan(value)) {
+ buffer_sprintf(b, "{"
+ "\"prefix\":\"%s\","
+ "\"hostname\":\"%s\","
+ "\"id\":\"%s\","
+ "\"subid\":\"%s\","
+ "\"value\":" CALCULATED_NUMBER_FORMAT ","
+ "\"timestamp\": %u}\n", prefix, hostname, st->id, rd->id, value, (uint32_t) before);
+ return 1;
+ }
+ return 0;
+}
+
static inline int process_graphite_response(BUFFER *b) {
char sample[1024];
const char *s = buffer_tostring(b);
return 0;
}
+static inline int process_json_response(BUFFER *b) {
+ char sample[1024];
+ const char *s = buffer_tostring(b);
+ char *d = sample, *e = &sample[sizeof(sample) - 1];
+
+ for(; *s && d < e ;s++) {
+ char c = *s;
+ if(unlikely(!isprint(c))) c = ' ';
+ *d++ = c;
+ }
+ *d = '\0';
+
+ info("Received %zu bytes from json backend. Ignoring them. Sample: '%s'", buffer_strlen(b), sample);
+ buffer_flush(b);
+ return 0;
+}
+
static inline int process_opentsdb_response(BUFFER *b) {
char sample[1024];
const char *s = buffer_tostring(b);
if(!enabled || frequency < 1)
goto cleanup;
- if(!strsame(source, "as collected")) {
+ if(!strcmp(source, "as collected")) {
options = BACKEND_SOURCE_DATA_AS_COLLECTED;
}
- else if(!strsame(source, "average")) {
+ else if(!strcmp(source, "average")) {
options = BACKEND_SOURCE_DATA_AVERAGE;
}
- else if(!strsame(source, "sum") || !strsame(source, "volume")) {
+ else if(!strcmp(source, "sum") || !strcmp(source, "volume")) {
options = BACKEND_SOURCE_DATA_SUM;
}
else {
goto cleanup;
}
- if(!strsame(type, "graphite") || !strsame(type, "graphite:plaintext")) {
+ if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
default_port = 2003;
if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
backend_request_formatter = format_dimension_collected_graphite_plaintext;
backend_response_checker = process_graphite_response;
}
- else if(!strsame(type, "opentsdb") || !strsame(type, "opentsdb:telnet")) {
+ else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
default_port = 4242;
if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
backend_request_formatter = format_dimension_collected_opentsdb_telnet;
backend_response_checker = process_opentsdb_response;
}
+ else if (!strcmp(type, "json") || !strcmp(type, "json:plaintext"))
+ {
+ default_port = 5448;
+
+ if (options == BACKEND_SOURCE_DATA_AS_COLLECTED)
+ {
+ backend_request_formatter = format_dimension_collected_json_plaintext;
+ }
+ else
+ {
+ backend_request_formatter = format_dimension_stored_json_plaintext;
+ }
+
+ backend_response_checker = process_json_response;
+ }
else {
error("Unknown backend type '%s'", type);
goto cleanup;
info("BACKEND configured ('%s' on '%s' sending '%s' data, every %d seconds, as host '%s', with prefix '%s')", type, destination, source, frequency, hostname, prefix);
usec_t step_ut = frequency * USEC_PER_SEC;
- usec_t random_ut = now_realtime_usec() % (step_ut / 2);
- time_t before = (time_t)((now_realtime_usec() - step_ut) / USEC_PER_SEC);
- time_t after = before;
+ time_t after = now_realtime_sec();
int failures = 0;
+ heartbeat_t hb;
+ heartbeat_init(&hb);
for(;;) {
// ------------------------------------------------------------------------
- // wait for the next iteration point
-
- usec_t now_ut = now_realtime_usec();
- usec_t next_ut = now_ut - (now_ut % step_ut) + step_ut;
- before = (time_t)(next_ut / USEC_PER_SEC);
-
- // add a little delay (1/4 of the step) plus some randomness
- next_ut += (step_ut / 4) + random_ut;
-
- while(now_ut < next_ut) {
- sleep_usec(next_ut - now_ut);
- now_ut = now_realtime_usec();
- }
+ // Wait for the next iteration point.
+ heartbeat_next(&hb, step_ut);
+ time_t before = now_realtime_sec();
// ------------------------------------------------------------------------
// add to the buffer the data we need to send to the backend
-
RRDSET *st;
int pthreadoldcancelstate;
//fprintf(stderr, "\nBACKEND BEGIN:\n%s\nBACKEND END\n", buffer_tostring(b)); // FIXME
//fprintf(stderr, "after = %lu, before = %lu\n", after, before);
+ // prepare for the next iteration
+ // to add incrementally data to buffer
+ after = before;
+
// ------------------------------------------------------------------------
// if we are connected, receive a response, without blocking
// if we are not connected, connect to a backend server
if(unlikely(sock == -1)) {
- usec_t start_ut = now_realtime_usec();
+ usec_t start_ut = now_monotonic_usec();
const char *s = destination;
while(*s) {
const char *e = s;
if(sock != -1) break;
s = e;
}
- chart_backend_latency += now_realtime_usec() - start_ut;
+ chart_backend_latency += now_monotonic_usec() - start_ut;
}
if(unlikely(netdata_exit)) break;
if(likely(sock != -1)) {
size_t len = buffer_strlen(b);
- usec_t start_ut = now_realtime_usec();
+ usec_t start_ut = now_monotonic_usec();
int flags = 0;
#ifdef MSG_NOSIGNAL
flags += MSG_NOSIGNAL;
#endif
ssize_t written = send(sock, buffer_tostring(b), len, flags);
- chart_backend_latency += now_realtime_usec() - start_ut;
+ chart_backend_latency += now_monotonic_usec() - start_ut;
if(written != -1 && (size_t)written == len) {
// we sent the data successfully
chart_transmission_successes++;
close(sock);
sock = -1;
}
-
- // either the buffer is empty
- // or is holding the data we couldn't send
- // so, make sure the next iteration will continue
- // from where we are now
- after = before;
}
else {
error("Failed to update database backend '%s'", destination);
projects / netdata.git / blobdiff