allow average or sum/volume to be sent to the backend
Costa Tsaousis (ktsaou) committed
Nov 20, 2016 at 12:59 UTC
b6408db2829c34da32b3593e74502de65df2fab4
1 file changed
+45
-23
src/backends.c
+45
-23
@@ -1,5 +1,9 @@
1
#include "common.h"
2
3
+#define BACKEND_SOURCE_DATA_AS_COLLECTED 0x00000001
4
+#define BACKEND_SOURCE_DATA_AVERAGE 0x00000002
5
+#define BACKEND_SOURCE_DATA_SUM 0x00000003
6
+
7
int connect_to_socket4(const char *ip, int port) {
8
int sock;
9
@@ -154,7 +158,7 @@ static inline int connect_to_one(const char *definition, int default_port) {
158
return fd;
159
}
160
157
-static inline calculated_number backend_duration_average(RRDSET *st, RRDDIM *rd, time_t after, time_t before) {
161
+static inline calculated_number backend_duration_average(RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
162
time_t first_t = rrdset_first_entry_t(st);
163
time_t last_t = rrdset_last_entry_t(st);
164
@@ -197,36 +201,37 @@ static inline calculated_number backend_duration_average(RRDSET *st, RRDDIM *rd,
201
counter++;
202
}
203
200
- if(!counter)
204
+ if(unlikely(!counter))
205
return NAN;
206
207
+ if(unlikely(options & BACKEND_SOURCE_DATA_SUM))
208
+ return sum;
209
+
210
return sum / (calculated_number)counter;
211
}
212
206
-static inline void format_dimension_collected_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before) {
213
+static inline void format_dimension_collected_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
214
+ (void)after;
215
(void)before;
208
-
209
- time_t last = rd->last_collected_time.tv_sec;
210
- if(likely(last > after))
211
- buffer_sprintf(b, "%s.%s.%s.%s " COLLECTED_NUMBER_FORMAT " %u\n", prefix, host->hostname, st->id, rd->id, rd->last_collected_value, (uint32_t)last);
216
+ (void)options;
217
+ buffer_sprintf(b, "%s.%s.%s.%s " COLLECTED_NUMBER_FORMAT " %u\n", prefix, host->hostname, st->id, rd->id, rd->last_collected_value, (uint32_t)rd->last_collected_time.tv_sec);
218
}
219
214
-static inline void format_dimension_stored_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before) {
215
- calculated_number value = backend_duration_average(st, rd, after, before);
220
+static inline void format_dimension_stored_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
221
+ calculated_number value = backend_duration_average(st, rd, after, before, options);
222
if(!isnan(value))
223
buffer_sprintf(b, "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n", prefix, host->hostname, st->id, rd->id, value, (uint32_t)before);
224
}
225
220
-static inline void format_dimension_collected_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before) {
226
+static inline void format_dimension_collected_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
227
+ (void)after;
228
(void)before;
222
-
223
- time_t last = rd->last_collected_time.tv_sec;
224
- if(likely(last > after))
225
- buffer_sprintf(b, "put %s.%s.%s %u " COLLECTED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t)last, rd->last_collected_value, host->hostname);
229
+ (void)options;
230
+ buffer_sprintf(b, "put %s.%s.%s %u " COLLECTED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t)rd->last_collected_time.tv_sec, rd->last_collected_value, host->hostname);
231
}
232
228
-static inline void format_dimension_stored_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before) {
229
- calculated_number value = backend_duration_average(st, rd, after, before);
233
+static inline void format_dimension_stored_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
234
+ calculated_number value = backend_duration_average(st, rd, after, before, options);
235
if(!isnan(value))
236
buffer_sprintf(b, "put %s.%s.%s %u " CALCULATED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t)before, value, host->hostname);
237
}
@@ -235,7 +240,7 @@ void *backends_main(void *ptr) {
240
(void)ptr;
241
242
BUFFER *b = buffer_create(1);
238
- void (*formatter)(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before);
243
+ void (*formatter)(BUFFER *b, const char *prefix, RRDHOST *host, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options);
244
245
info("BACKEND thread created with task id %d", gettid());
246
@@ -247,27 +252,42 @@ void *backends_main(void *ptr) {
252
253
int default_port = 0;
254
int sock = -1;
255
+ uint32_t options = BACKEND_SOURCE_DATA_AVERAGE;
256
int enabled = config_get_boolean("backend", "enable", 0);
251
- const char *source = config_get("backend", "data source", "as stored");
257
+ const char *source = config_get("backend", "data source", "average");
258
const char *type = config_get("backend", "type", "graphite");
259
const char *destination = config_get("backend", "destination", "localhost");
260
const char *prefix = config_get("backend", "prefix", "netdata");
261
int frequency = (int)config_get_number("backend", "update every", 30);
262
int buffer_on_failures = (int)config_get_number("backend", "buffer on failures", 10);
263
258
- if(!enabled)
264
+ if(!enabled || frequency < 1)
265
+ goto cleanup;
266
+
267
+ if(!strcmp(source, "as collected")) {
268
+ options = BACKEND_SOURCE_DATA_AS_COLLECTED;
269
+ }
270
+ else if(!strcmp(source, "average")) {
271
+ options = BACKEND_SOURCE_DATA_AVERAGE;
272
+ }
273
+ else if(!strcmp(source, "sum") || !strcmp(source, "volume")) {
274
+ options = BACKEND_SOURCE_DATA_SUM;
275
+ }
276
+ else {
277
+ error("Invalid data source method '%s' for backend given. Disabling backed.", source);
278
goto cleanup;
279
+ }
280
281
if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
282
default_port = 2003;
263
- if(!strcmp(source, "as collected"))
283
+ if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
284
formatter = format_dimension_collected_graphite_plaintext;
285
else
286
formatter = format_dimension_stored_graphite_plaintext;
287
}
288
else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
289
default_port = 4242;
270
- if(!strcmp(source, "as collected"))
290
+ if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
291
formatter = format_dimension_collected_opentsdb_telnet;
292
else
293
formatter = format_dimension_stored_opentsdb_telnet;
@@ -307,8 +327,10 @@ void *backends_main(void *ptr) {
327
pthread_rwlock_rdlock(&st->rwlock);
328
329
RRDDIM *rd;
310
- for(rd = st->dimensions; rd ;rd = rd->next)
311
- formatter(b, prefix, &localhost, st, rd, after, before);
330
+ for(rd = st->dimensions; rd ;rd = rd->next) {
331
+ if(rd->last_collected_time.tv_sec >= after)
332
+ formatter(b, prefix, &localhost, st, rd, after, before, options);
333
+ }
334
335
pthread_rwlock_unlock(&st->rwlock);
336
}