added comments and information at backends.c
Costa Tsaousis (ktsaou) committed
Feb 26, 2017 at 21:49 UTC
cab47afab39dd529fd06040111d314c2100e5183
1 file changed
+223
-72
src/backends.c
+223
-72
@@ -1,12 +1,48 @@
1
#include "common.h"
2
3
+// ----------------------------------------------------------------------------
4
+// How backends work in netdata:
5
+//
6
+// 1. There is an independent thread that runs at the required interval
7
+// (for example, once every 10 seconds)
8
+//
9
+// 2. Every time it wakes, it calls the backend formatting functions to build
10
+// a buffer of data. This is a very fast, memory only operation.
11
+//
12
+// 3. If the buffer already includes data, the new data are appended.
13
+// If the buffer becomes too big, because the data cannot be sent, a
14
+// log is written and the buffer is discarded.
15
+//
16
+// 4. Then it tries to send all the data. It blocks until all the data are sent
17
+// or the socket returns an error.
18
+// If the time required for this is above the interval, it starts skipping
19
+// intervals, but the calculated values include the entire database, without
20
+// gaps (it remembers the timestamps and continues from where it stopped).
21
+//
22
+// 5. repeats the above forever.
23
+//
24
+
25
#define BACKEND_SOURCE_DATA_AS_COLLECTED 0x00000001
26
#define BACKEND_SOURCE_DATA_AVERAGE 0x00000002
27
#define BACKEND_SOURCE_DATA_SUM 0x00000004
28
7
-static inline calculated_number backend_calculate_value_from_stored_data(RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
29
+
30
+// ----------------------------------------------------------------------------
31
+// helper functions for backends
32
+
33
+// calculate the SUM or AVERAGE of a dimension, for any timeframe
34
+// may return NAN if the database does not have any value in the give timeframe
35
+
36
+static inline calculated_number backend_calculate_value_from_stored_data(
37
+ RRDSET *st // the chart
38
+ , RRDDIM *rd // the dimension
39
+ , time_t after // the start timestamp
40
+ , time_t before // the end timestamp
41
+ , uint32_t options // BACKEND_SOURCE_* bitmap
42
+) {
43
+ // find the edges of the rrd database for this chart
44
time_t first_t = rrdset_first_entry_t(st);
9
- time_t last_t = rrdset_last_entry_t(st);
45
+ time_t last_t = rrdset_last_entry_t(st);
46
47
if(unlikely(before < first_t || after > last_t))
48
// the chart has not been updated in the wanted timeframe
@@ -21,7 +57,7 @@ static inline calculated_number backend_calculate_value_from_stored_data(RRDSET
57
after = first_t;
58
59
if(unlikely(after > before))
24
- // this can happen when the st->update_every > before - after
60
+ // this can happen when st->update_every > before - after
61
before = after;
62
63
if(unlikely(before > last_t))
@@ -35,14 +71,20 @@ static inline calculated_number backend_calculate_value_from_stored_data(RRDSET
71
slot, stop_now = 0;
72
73
for(slot = start_at_slot; !stop_now ; slot--) {
74
+
75
if(unlikely(slot < 0)) slot = st->entries - 1;
76
if(unlikely(slot == stop_at_slot)) stop_now = 1;
77
78
storage_number n = rd->values[slot];
42
- if(unlikely(!does_storage_number_exist(n))) continue;
79
+
80
+ if(unlikely(!does_storage_number_exist(n))) {
81
+ // not collected
82
+ continue;
83
+ }
84
85
calculated_number value = unpack_storage_number(n);
86
sum += value;
87
+
88
counter++;
89
}
90
@@ -55,49 +97,189 @@ static inline calculated_number backend_calculate_value_from_stored_data(RRDSET
97
return sum / (calculated_number)counter;
98
}
99
58
-static inline int format_dimension_collected_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
100
+
101
+// discard a response received by a backend
102
+// after logging a simple of it to error.log
103
+
104
+static inline int discard_response(BUFFER *b, const char *backend) {
105
+ char sample[1024];
106
+ const char *s = buffer_tostring(b);
107
+ char *d = sample, *e = &sample[sizeof(sample) - 1];
108
+
109
+ for(; *s && d < e ;s++) {
110
+ char c = *s;
111
+ if(unlikely(!isprint(c))) c = ' ';
112
+ *d++ = c;
113
+ }
114
+ *d = '\0';
115
+
116
+ info("Received %zu bytes from %s backend. Ignoring them. Sample: '%s'", buffer_strlen(b), backend, sample);
117
+ buffer_flush(b);
118
+ return 0;
119
+}
120
+
121
+
122
+// ----------------------------------------------------------------------------
123
+// graphite backend
124
+
125
+static inline int format_dimension_collected_graphite_plaintext(
126
+ BUFFER *b // the buffer to write data to
127
+ , const char *prefix // the prefix to use
128
+ , RRDHOST *host // the host this chart comes from
129
+ , const char *hostname // the hostname (to override host->hostname)
130
+ , RRDSET *st // the chart
131
+ , RRDDIM *rd // the dimension
132
+ , time_t after // the start timestamp
133
+ , time_t before // the end timestamp
134
+ , uint32_t options // BACKEND_SOURCE_* bitmap
135
+) {
136
(void)host;
137
(void)after;
138
(void)before;
139
(void)options;
63
- buffer_sprintf(b, "%s.%s.%s.%s " COLLECTED_NUMBER_FORMAT " %u\n", prefix, hostname, st->id, rd->id, rd->last_collected_value, (uint32_t)rd->last_collected_time.tv_sec);
140
+
141
+ buffer_sprintf(
142
+ b
143
+ , "%s.%s.%s.%s " COLLECTED_NUMBER_FORMAT " %u\n"
144
+ , prefix
145
+ , hostname
146
+ , st->id
147
+ , rd->id
148
+ , rd->last_collected_value
149
+ , (uint32_t)rd->last_collected_time.tv_sec
150
+ );
151
+
152
return 1;
153
}
154
67
-static inline int format_dimension_stored_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
155
+static inline int format_dimension_stored_graphite_plaintext(
156
+ BUFFER *b // the buffer to write data to
157
+ , const char *prefix // the prefix to use
158
+ , RRDHOST *host // the host this chart comes from
159
+ , const char *hostname // the hostname (to override host->hostname)
160
+ , RRDSET *st // the chart
161
+ , RRDDIM *rd // the dimension
162
+ , time_t after // the start timestamp
163
+ , time_t before // the end timestamp
164
+ , uint32_t options // BACKEND_SOURCE_* bitmap
165
+) {
166
(void)host;
167
+
168
calculated_number value = backend_calculate_value_from_stored_data(st, rd, after, before, options);
169
+
170
if(!isnan(value)) {
71
- buffer_sprintf(b, "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n", prefix, hostname, st->id, rd->id, value, (uint32_t) before);
171
+
172
+ buffer_sprintf(
173
+ b
174
+ , "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n"
175
+ , prefix
176
+ , hostname
177
+ , st->id
178
+ , rd->id
179
+ , value
180
+ , (uint32_t) before
181
+ );
182
+
183
return 1;
184
}
185
return 0;
186
}
187
77
-static inline int format_dimension_collected_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
188
+static inline int process_graphite_response(BUFFER *b) {
189
+ return discard_response(b, "graphite");
190
+}
191
+
192
+
193
+// ----------------------------------------------------------------------------
194
+// opentsdb backend
195
+
196
+static inline int format_dimension_collected_opentsdb_telnet(
197
+ BUFFER *b // the buffer to write data to
198
+ , const char *prefix // the prefix to use
199
+ , RRDHOST *host // the host this chart comes from
200
+ , const char *hostname // the hostname (to override host->hostname)
201
+ , RRDSET *st // the chart
202
+ , RRDDIM *rd // the dimension
203
+ , time_t after // the start timestamp
204
+ , time_t before // the end timestamp
205
+ , uint32_t options // BACKEND_SOURCE_* bitmap
206
+) {
207
(void)host;
208
(void)after;
209
(void)before;
210
(void)options;
82
- 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, hostname);
211
+
212
+ buffer_sprintf(
213
+ b
214
+ , "put %s.%s.%s %u " COLLECTED_NUMBER_FORMAT " host=%s\n"
215
+ , prefix
216
+ , st->id
217
+ , rd->id
218
+ , (uint32_t)rd->last_collected_time.tv_sec
219
+ , rd->last_collected_value
220
+ , hostname
221
+ );
222
+
223
return 1;
224
}
225
86
-static inline int format_dimension_stored_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
226
+static inline int format_dimension_stored_opentsdb_telnet(
227
+ BUFFER *b // the buffer to write data to
228
+ , const char *prefix // the prefix to use
229
+ , RRDHOST *host // the host this chart comes from
230
+ , const char *hostname // the hostname (to override host->hostname)
231
+ , RRDSET *st // the chart
232
+ , RRDDIM *rd // the dimension
233
+ , time_t after // the start timestamp
234
+ , time_t before // the end timestamp
235
+ , uint32_t options // BACKEND_SOURCE_* bitmap
236
+) {
237
(void)host;
238
+
239
calculated_number value = backend_calculate_value_from_stored_data(st, rd, after, before, options);
240
+
241
if(!isnan(value)) {
90
- buffer_sprintf(b, "put %s.%s.%s %u " CALCULATED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t) before, value, hostname);
242
+
243
+ buffer_sprintf(
244
+ b
245
+ , "put %s.%s.%s %u " CALCULATED_NUMBER_FORMAT " host=%s\n"
246
+ , prefix
247
+ , st->id
248
+ , rd->id
249
+ , (uint32_t) before
250
+ , value
251
+ , hostname
252
+ );
253
+
254
return 1;
255
}
256
return 0;
257
}
258
96
-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) {
259
+static inline int process_opentsdb_response(BUFFER *b) {
260
+ return discard_response(b, "opentsdb");
261
+}
262
+
263
+
264
+// ----------------------------------------------------------------------------
265
+// json backend
266
+
267
+static inline int format_dimension_collected_json_plaintext(
268
+ BUFFER *b // the buffer to write data to
269
+ , const char *prefix // the prefix to use
270
+ , RRDHOST *host // the host this chart comes from
271
+ , const char *hostname // the hostname (to override host->hostname)
272
+ , RRDSET *st // the chart
273
+ , RRDDIM *rd // the dimension
274
+ , time_t after // the start timestamp
275
+ , time_t before // the end timestamp
276
+ , uint32_t options // BACKEND_SOURCE_* bitmap
277
+) {
278
(void)host;
279
(void)after;
280
(void)before;
281
(void)options;
282
+
283
buffer_sprintf(b, "{"
284
"\"prefix\":\"%s\","
285
"\"hostname\":\"%s\","
@@ -116,7 +298,7 @@ static inline int format_dimension_collected_json_plaintext(BUFFER *b, const cha
298
"\"timestamp\": %u}\n",
299
prefix,
300
hostname,
119
-
301
+
302
st->id,
303
st->name,
304
st->family,
@@ -126,17 +308,29 @@ static inline int format_dimension_collected_json_plaintext(BUFFER *b, const cha
308
309
rd->id,
310
rd->name,
129
- rd->last_collected_value,
130
-
311
+ rd->last_collected_value,
312
+
313
(uint32_t)rd->last_collected_time.tv_sec
314
);
315
316
return 1;
317
}
318
137
-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) {
319
+static inline int format_dimension_stored_json_plaintext(
320
+ BUFFER *b // the buffer to write data to
321
+ , const char *prefix // the prefix to use
322
+ , RRDHOST *host // the host this chart comes from
323
+ , const char *hostname // the hostname (to override host->hostname)
324
+ , RRDSET *st // the chart
325
+ , RRDDIM *rd // the dimension
326
+ , time_t after // the start timestamp
327
+ , time_t before // the end timestamp
328
+ , uint32_t options // BACKEND_SOURCE_* bitmap
329
+) {
330
(void)host;
331
+
332
calculated_number value = backend_calculate_value_from_stored_data(st, rd, after, before, options);
333
+
334
if(!isnan(value)) {
335
buffer_sprintf(b, "{"
336
"\"prefix\":\"%s\","
@@ -176,56 +370,13 @@ static inline int format_dimension_stored_json_plaintext(BUFFER *b, const char *
370
return 0;
371
}
372
179
-static inline int process_graphite_response(BUFFER *b) {
180
- char sample[1024];
181
- const char *s = buffer_tostring(b);
182
- char *d = sample, *e = &sample[sizeof(sample) - 1];
183
-
184
- for(; *s && d < e ;s++) {
185
- char c = *s;
186
- if(unlikely(!isprint(c))) c = ' ';
187
- *d++ = c;
188
- }
189
- *d = '\0';
190
-
191
- info("Received %zu bytes from graphite backend. Ignoring them. Sample: '%s'", buffer_strlen(b), sample);
192
- buffer_flush(b);
193
- return 0;
194
-}
195
-
373
static inline int process_json_response(BUFFER *b) {
197
- char sample[1024];
198
- const char *s = buffer_tostring(b);
199
- char *d = sample, *e = &sample[sizeof(sample) - 1];
200
-
201
- for(; *s && d < e ;s++) {
202
- char c = *s;
203
- if(unlikely(!isprint(c))) c = ' ';
204
- *d++ = c;
205
- }
206
- *d = '\0';
207
-
208
- info("Received %zu bytes from json backend. Ignoring them. Sample: '%s'", buffer_strlen(b), sample);
209
- buffer_flush(b);
210
- return 0;
374
+ return discard_response(b, "json");
375
}
376
213
-static inline int process_opentsdb_response(BUFFER *b) {
214
- char sample[1024];
215
- const char *s = buffer_tostring(b);
216
- char *d = sample, *e = &sample[sizeof(sample) - 1];
217
-
218
- for(; *s && d < e ;s++) {
219
- char c = *s;
220
- if(unlikely(!isprint(c))) c = ' ';
221
- *d++ = c;
222
- }
223
- *d = '\0';
377
225
- info("Received %zu bytes from opentsdb backend. Ignoring them. Sample: '%s'", buffer_strlen(b), sample);
226
- buffer_flush(b);
227
- return 0;
228
-}
378
+// ----------------------------------------------------------------------------
379
+// the backend thread
380
381
void *backends_main(void *ptr) {
382
int default_port = 0;
@@ -295,37 +446,37 @@ void *backends_main(void *ptr) {
446
// select the backend type
447
448
if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
449
+
450
default_port = 2003;
451
+ backend_response_checker = process_graphite_response;
452
+
453
if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
454
backend_request_formatter = format_dimension_collected_graphite_plaintext;
455
else
456
backend_request_formatter = format_dimension_stored_graphite_plaintext;
457
304
- backend_response_checker = process_graphite_response;
458
}
459
else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
460
+
461
default_port = 4242;
462
+ backend_response_checker = process_opentsdb_response;
463
+
464
if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
465
backend_request_formatter = format_dimension_collected_opentsdb_telnet;
466
else
467
backend_request_formatter = format_dimension_stored_opentsdb_telnet;
468
313
- backend_response_checker = process_opentsdb_response;
469
}
315
- else if (!strcmp(type, "json") || !strcmp(type, "json:plaintext"))
316
- {
470
+ else if (!strcmp(type, "json") || !strcmp(type, "json:plaintext")) {
471
+
472
default_port = 5448;
473
+ backend_response_checker = process_json_response;
474
475
if (options == BACKEND_SOURCE_DATA_AS_COLLECTED)
320
- {
476
backend_request_formatter = format_dimension_collected_json_plaintext;
322
- }
477
else
324
- {
478
backend_request_formatter = format_dimension_stored_json_plaintext;
326
- }
479
328
- backend_response_checker = process_json_response;
480
}
481
else {
482
error("Unknown backend type '%s'", type);