3 int connect_to_socket4(const char *ip, int port) {
6 debug(D_LISTENER, "IPv4 connecting to ip '%s' port %d", ip, port);
8 sock = socket(AF_INET, SOCK_STREAM, 0);
10 error("IPv4 socket() on ip '%s' port %d failed.", ip, port);
14 struct sockaddr_in name;
15 memset(&name, 0, sizeof(struct sockaddr_in));
16 name.sin_family = AF_INET;
17 name.sin_port = htons(port);
19 int ret = inet_pton(AF_INET, ip, (void *)&name.sin_addr.s_addr);
21 error("Failed to convert '%s' to a valid IPv4 address.", ip);
26 if(connect(sock, (struct sockaddr *) &name, sizeof(name)) < 0) {
28 error("IPv4 failed to connect to '%s', port %d", ip, port);
32 debug(D_LISTENER, "Connected to IPv4 ip '%s' port %d", ip, port);
36 int connect_to_socket6(const char *ip, int port) {
40 debug(D_LISTENER, "IPv6 connecting to ip '%s' port %d", ip, port);
42 sock = socket(AF_INET6, SOCK_STREAM, 0);
44 error("IPv6 socket() on ip '%s' port %d failed.", ip, port);
49 if(setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (void*)&ipv6only, sizeof(ipv6only)) != 0)
50 error("Cannot set IPV6_V6ONLY on ip '%s' port's %d.", ip, port);
52 struct sockaddr_in6 name;
53 memset(&name, 0, sizeof(struct sockaddr_in6));
54 name.sin6_family = AF_INET6;
55 name.sin6_port = htons ((uint16_t) port);
57 int ret = inet_pton(AF_INET6, ip, (void *)&name.sin6_addr.s6_addr);
59 error("Failed to convert IP '%s' to a valid IPv6 address.", ip);
64 name.sin6_scope_id = 0;
66 if(connect(sock, (struct sockaddr *)&name, sizeof(name)) < 0) {
68 error("IPv6 failed to connect to '%s', port %d", ip, port);
72 debug(D_LISTENER, "Connected to IPv6 ip '%s' port %d", ip, port);
77 static inline int connect_to_one(const char *definition, int default_port) {
78 struct addrinfo hints;
79 struct addrinfo *result = NULL, *rp = NULL;
81 char buffer[strlen(definition) + 1];
82 strcpy(buffer, definition);
85 snprintfz(buffer2, 10, "%d", default_port);
87 char *ip = buffer, *port = buffer2;
92 while(*e && *e != ']') e++;
99 while(*e && *e != ':') e++;
113 memset(&hints, 0, sizeof(struct addrinfo));
114 hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */
115 hints.ai_socktype = SOCK_DGRAM; /* Datagram socket */
116 hints.ai_flags = AI_PASSIVE; /* For wildcard IP address */
117 hints.ai_protocol = 0; /* Any protocol */
118 hints.ai_canonname = NULL;
119 hints.ai_addr = NULL;
120 hints.ai_next = NULL;
122 int r = getaddrinfo(ip, port, &hints, &result);
124 error("Cannot resolve host '%s', port '%s': %s\n", ip, port, gai_strerror(r));
129 for (rp = result; rp != NULL && fd == -1; rp = rp->ai_next) {
130 char rip[INET_ADDRSTRLEN + INET6_ADDRSTRLEN] = "INVALID";
133 switch (rp->ai_addr->sa_family) {
135 struct sockaddr_in *sin = (struct sockaddr_in *) rp->ai_addr;
136 inet_ntop(AF_INET, &sin->sin_addr, rip, INET_ADDRSTRLEN);
137 rport = ntohs(sin->sin_port);
138 fd = connect_to_socket4(rip, rport);
143 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *) rp->ai_addr;
144 inet_ntop(AF_INET6, &sin6->sin6_addr, rip, INET6_ADDRSTRLEN);
145 rport = ntohs(sin6->sin6_port);
146 fd = connect_to_socket6(rip, rport);
152 freeaddrinfo(result);
157 static inline calculated_number backend_duration_average(RRDSET *st, RRDDIM *rd, time_t after, time_t before) {
158 time_t first_t = rrdset_first_entry_t(st);
159 time_t last_t = rrdset_last_entry_t(st);
161 if(unlikely(before - after < st->update_every && after != after - after % st->update_every))
162 // when st->update_every is bigger than the frequency we send data to backend
163 // skip the iterations that are not aligned to the database
166 // align the time-frame
167 // for 'after' also skip the first value by adding st->update_every
168 after = after - after % st->update_every + st->update_every;
169 before = before - before % st->update_every;
171 if(unlikely(after < first_t))
174 if(unlikely(after > before))
175 // this can happen when the st->update_every > before - after
178 if(unlikely(before > last_t))
182 calculated_number sum = 0;
184 long start_at_slot = rrdset_time2slot(st, before),
185 stop_at_slot = rrdset_time2slot(st, after),
188 for(slot = start_at_slot; !stop_now ; slot--) {
189 if(unlikely(slot < 0)) slot = st->entries - 1;
190 if(unlikely(slot == stop_at_slot)) stop_now = 1;
192 storage_number n = rd->values[slot];
193 if(unlikely(!does_storage_number_exist(n))) continue;
195 calculated_number value = unpack_storage_number(n);
203 return sum / (calculated_number)counter;
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) {
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);
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);
217 buffer_sprintf(b, "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n", prefix, host->hostname, st->id, rd->id, value, (uint32_t)before);
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) {
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);
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);
231 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);
234 void *backends_main(void *ptr) {
237 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);
240 info("BACKENDs thread created with task id %d", gettid());
242 if(pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
243 error("Cannot set pthread cancel type to DEFERRED.");
245 if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
246 error("Cannot set pthread cancel state to ENABLE.");
248 int default_port = 0;
250 int enabled = config_get_boolean("backends", "enable", 0);
251 const char *source = config_get("backends", "data source", "as stored");
252 const char *type = config_get("backends", "type", "graphite");
253 const char *destination = config_get("backends", "destination", "localhost");
254 const char *prefix = config_get("backends", "prefix", "netdata");
255 int frequency = (int)config_get_number("backends", "update every", 30);
256 int buffer_on_failures = (int)config_get_number("backends", "buffer on failures", 10);
261 if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
263 if(!strcmp(source, "as collected"))
264 formatter = format_dimension_collected_graphite_plaintext;
266 formatter = format_dimension_stored_graphite_plaintext;
268 else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
270 if(!strcmp(source, "as collected"))
271 formatter = format_dimension_collected_opentsdb_telnet;
273 formatter = format_dimension_stored_opentsdb_telnet;
276 error("Unknown backend type '%s'", type);
280 unsigned long long step_ut = frequency * 1000000ULL;
281 unsigned long long random_ut = time_usec() % (step_ut / 2);
282 time_t before = (time_t)((time_usec() - step_ut) / 10000000ULL);
283 time_t after = before;
287 unsigned long long now_ut = time_usec();
288 unsigned long long next_ut = now_ut - (now_ut % step_ut) + step_ut;
289 before = (time_t)(next_ut / 1000000ULL);
291 // add a little delay (1/4 of the step) plus some randomness
292 next_ut += (step_ut / 4) + random_ut;
294 while(now_ut < next_ut) {
295 sleep_usec(next_ut - now_ut);
296 now_ut = time_usec();
300 int pthreadoldcancelstate;
302 if(unlikely(pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &pthreadoldcancelstate) != 0))
303 error("Cannot set pthread cancel state to DISABLE.");
305 rrdhost_rdlock(&localhost);
306 for(st = localhost.rrdset_root; st ;st = st->next) {
307 pthread_rwlock_rdlock(&st->rwlock);
310 for(rd = st->dimensions; rd ;rd = rd->next)
311 formatter(b, prefix, &localhost, st, rd, after, before);
313 pthread_rwlock_unlock(&st->rwlock);
315 rrdhost_unlock(&localhost);
317 if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
318 error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
320 if(unlikely(netdata_exit)) break;
322 //fprintf(stderr, "\nBACKEND BEGIN:\n%s\nBACKEND END\n", buffer_tostring(b));
323 //fprintf(stderr, "after = %lu, before = %lu\n", after, before);
325 if(unlikely(sock == -1)) {
326 const char *s = destination;
330 // skip separators, moving both s(tart) and e(nd)
331 while(isspace(*e) || *e == ',') s = ++e;
333 // move e(nd) to the first separator
334 while(*e && !isspace(*e) && *e != ',') e++;
336 // is there anything?
337 if(!*s || s == e) break;
340 strncpyz(buf, s, e - s);
341 sock = connect_to_one(buf, default_port);
342 if(sock != -1) break;
347 if(unlikely(netdata_exit)) break;
349 if(likely(sock != -1)) {
350 size_t len = buffer_strlen(b);
351 ssize_t written = write(sock, buffer_tostring(b), len);
352 if(written != -1 && (size_t)written == len) {
358 error("Failed to write data to database backend '%s'. Willing to write %zu bytes, wrote %zd bytes. Will re-connect.", destination, len, written);
367 error("Failed to update database backend '%s'", destination);
370 if(failures > buffer_on_failures) {
371 error("Reached %d backend failures. Flushing buffers to protect this host - this results in data loss on back-end server '%s'", failures, destination);
376 if(unlikely(netdata_exit)) break;
383 info("BACKENDs thread exiting");