@cryptotaxi247 / netdata-1 / commits / b092845a2

new mechanism for two stage processing of web output data - still in progress

Costa Tsaousis (ktsaou) committed Oct 25, 2015 at 01:46 UTC b092845a22b4469e08456717ca5e042884801b4e
4 files changed +395 -36
src/Makefile
+1 -1
@@ -29,7 +29,7 @@ endif
29 COMMON_FLAGS = BIN_DIR='$(BIN_DIR)' CONFIG_DIR='$(CONFIG_DIR)' LOG_DIR='$(LOG_DIR)' PLUGINS_DIR='$(PLUGINS_DIR)'
30
31 proc_sources = proc_net_dev.c proc_net_ip_vs_stats.c proc_diskstats.c proc_meminfo.c proc_net_netstat.c proc_net_snmp.c proc_net_stat_conntrack.c proc_stat.c proc_vmstat.c proc_net_rpc_nfsd.c proc_sys_kernel_random_entropy_avail.c proc_interrupts.c
32 -sources = avl.c dictionary.c procfile.c common.c log.c popen.c url.c config.c web_buffer.c storage_number.c web_client.c global_statistics.c rrd.c rrd2json.c web_server.c plugins_d.c daemon.c plugin_tc.c plugin_checks.c plugin_idlejitter.c plugin_proc.c unit_test.c main.c
32 +sources = rrd.c rrd2json.c avl.c dictionary.c procfile.c common.c log.c popen.c url.c config.c web_buffer.c storage_number.c web_client.c global_statistics.c web_server.c plugins_d.c daemon.c plugin_tc.c plugin_checks.c plugin_idlejitter.c plugin_proc.c unit_test.c main.c
33 libs = -pthread
34
35 ifdef debug
src/rrd.c
-16
@@ -1141,15 +1141,8 @@ unsigned long long rrdset_done(RRDSET *st)
1141 // reset the storage flags for the next point, if any;
1142 storage_flags = SN_EXISTS;
1143
1144 - if(unlikely(st->first_entry_t && st->counter >= (unsigned long long)st->entries)) {
1145 - // the db is overwriting values
1146 - // add the value we will overwrite
1147 - st->first_entry_t += st->update_every * 1000000ULL;
1148 - }
1149 -
1144 st->counter++;
1145 st->current_entry = ((st->current_entry + 1) >= st->entries) ? 0 : st->current_entry + 1;
1152 - if(unlikely(!st->first_entry_t)) st->first_entry_t = next_ut;
1146 last_ut = next_ut;
1147 }
1148
@@ -1229,12 +1222,3 @@ unsigned long long rrdset_done(RRDSET *st)
1222
1223 return(st->usec_since_last_update);
1224 }
1232 -
1233 -
1234 -// find the oldest entry in the data, skipping all empty slots
1235 -time_t rrdset_first_entry_t(RRDSET *st)
1236 -{
1237 - if(unlikely(!st->first_entry_t)) return st->last_updated.tv_sec;
1238 -
1239 - return st->first_entry_t / 1000000;
1240 -}
src/rrd.h
+31 -5
@@ -21,8 +21,8 @@ extern int rrd_delete_unupdated_dimensions;
21
22 #define RRD_ID_LENGTH_MAX 1024
23
24 -#define RRDSET_MAGIC "NETDATA RRD SET FILE V012"
25 -#define RRDDIMENSION_MAGIC "NETDATA RRD DIMENSION FILE V012"
24 +#define RRDSET_MAGIC "NETDATA RRD SET FILE V013"
25 +#define RRDDIMENSION_MAGIC "NETDATA RRD DIMENSION FILE V013"
26
27 typedef long long total_number;
28 #define TOTAL_NUMBER_FORMAT "%lld"
@@ -211,8 +211,6 @@ struct rrdset {
211 unsigned long counter; // the number of times we added values to this rrd
212 unsigned long counter_done; // the number of times we added values to this rrd
213
214 - unsigned long long first_entry_t; // the timestamp (in microseconds) of the oldest entry in the db
215 -
214 uint32_t hash; // a simple hash on the id, to speed up searching
215 // we first compare hashes, and only if the hashes are equal we do string comparisons
216
@@ -271,8 +269,33 @@ extern void rrdset_next_plugins(RRDSET *st);
269
270 extern unsigned long long rrdset_done(RRDSET *st);
271
274 -extern time_t rrdset_first_entry_t(RRDSET *st);
272 +// get the total duration in seconds of the round robin database
273 +#define rrdset_duration(st) ((time_t)( (((st)->counter >= ((unsigned long)(st)->entries))?(unsigned long)(st)->entries:(st)->counter) * (st)->update_every ))
274 +
275 +// get the timestamp of the last entry in the round robin database
276 +#define rrdset_last_entry_t(st) ((time_t)(((st)->last_updated.tv_sec)))
277 +
278 +// get the timestamp of first entry in the round robin database
279 +#define rrdset_first_entry_t(st) ((time_t)(rrdset_last_entry_t(st) - rrdset_duration(st)))
280 +
281 +// get the last slot updated in the round robin database
282 +#define rrdset_last_slot(st) (((st)->current_entry == 0) ? (st)->entries - 1 : (st)->current_entry - 1)
283 +
284 +// get the first / oldest slot updated in the round robin database
285 +#define rrdset_first_slot(st) ((st)->current_entry)
286
287 +// get the slot of the round robin database, for the given timestamp (t)
288 +// it always returns a valid slot, although may not be for the time requested if the time is outside the round robin database
289 +#define rrdset_time2slot(st, t) ( \
290 + ( (t) >= rrdset_last_entry_t(st)) ? ( rrdset_last_slot(st) ) : \
291 + ( ((t) <= rrdset_first_entry_t(st)) ? rrdset_first_slot(st) : \
292 + ( (rrdset_last_slot(st) >= ((long)(rrdset_last_entry_t(st) - (t)) / (long)((st)->update_every)) ) ? \
293 + (rrdset_last_slot(st) - ((long)(rrdset_last_entry_t(st) - (t)) / (long)((st)->update_every)) ) : \
294 + (rrdset_last_slot(st) - ((long)(rrdset_last_entry_t(st) - (t)) / (long)((st)->update_every)) + (st)->entries ) \
295 + )))
296 +
297 +// get the timestamp of a specific slot in the round robin database
298 +#define rrdset_slot2time(st, slot) ()
299
300 // ----------------------------------------------------------------------------
301 // RRD DIMENSION functions
@@ -290,4 +313,7 @@ extern int rrddim_unhide(RRDSET *st, const char *id);
313 extern void rrddim_set_by_pointer(RRDSET *st, RRDDIM *rd, collected_number value);
314 extern int rrddim_set(RRDSET *st, const char *id, collected_number value);
315
316 +
317 +
318 +
319 #endif /* NETDATA_RRD_H */
src/rrd2json.c
+363 -14
@@ -53,9 +53,9 @@ unsigned long rrd_stats_one_json(RRDSET *st, char *options, struct web_buffer *w
53 , st->counter
54 , st->entries
55 , rrdset_first_entry_t(st)
56 - , st->current_entry
57 - , st->last_updated.tv_sec
58 - , now - (st->last_updated.tv_sec > now) ? now : st->last_updated.tv_sec
56 + , rrdset_last_slot(st)
57 + , rrdset_last_entry_t(st)
58 + , (now < rrdset_last_entry_t(st)) ? (time_t)0 : now - rrdset_last_entry_t(st)
59 , st->update_every
60 , st->isdetail
61 , st->usec_since_last_update
@@ -190,15 +190,10 @@ unsigned long rrd_stats_json(int type, RRDSET *st, struct web_buffer *wb, int en
190 if(entries_to_show < 1) entries_to_show = 1;
191 if(group < 1) group = 1;
192
193 - // make sure current_entry is within limits
194 - long current_entry = (long)st->current_entry - (long)1;
195 - if(current_entry < 0) current_entry = 0;
196 - else if(current_entry >= st->entries) current_entry = st->entries - 1;
197 -
193 // find the oldest entry of the round-robin
194 long max_entries_init = (st->counter < (unsigned long)st->entries) ? st->counter : (unsigned long)st->entries;
195
201 - time_t time_init = st->last_updated.tv_sec;
196 + time_t time_init = rrdset_last_entry_t(st);
197
198 if(before == 0 || before > time_init) before = time_init;
199 if(after == 0) after = rrdset_first_entry_t(st);
@@ -242,8 +237,8 @@ unsigned long rrd_stats_json(int type, RRDSET *st, struct web_buffer *wb, int en
237 debug(D_RRD_STATS, "%s first_entry_t = %lu, last_entry_t = %lu, duration = %lu, after = %lu, before = %lu, duration = %lu, entries_to_show = %lu, group = %lu, max_entries = %ld"
238 , st->id
239 , rrdset_first_entry_t(st)
245 - , st->last_updated.tv_sec
246 - , st->last_updated.tv_sec - rrdset_first_entry_t(st)
240 + , rrdset_last_entry_t(st)
241 + , rrdset_last_entry_t(st) - rrdset_first_entry_t(st)
242 , after
243 , before
244 , before - after
@@ -264,19 +259,22 @@ unsigned long rrd_stats_json(int type, RRDSET *st, struct web_buffer *wb, int en
259 // temp arrays for keeping values per dimension
260
261 calculated_number group_values[dimensions]; // keep sums when grouping
267 - calculated_number print_values[dimensions]; // keep the final value to be printed
262 int print_hidden[dimensions]; // keep hidden flags
263 int found_non_zero[dimensions];
264 int found_non_existing[dimensions];
265
266 // initialize them
267 for( rd = st->dimensions, c = 0 ; rd && c < dimensions ; rd = rd->next, c++) {
274 - group_values[c] = print_values[c] = 0;
268 + group_values[c] = 0;
269 print_hidden[c] = rd->hidden;
270 found_non_zero[c] = 0;
271 found_non_existing[c] = 0;
272 }
273
274 +
275 + // error("OLD: points=%d after=%d before=%d group=%d, duration=%d", entries_to_show, before - (st->update_every * group * entries_to_show), before, group, before - after + 1);
276 + // rrd2array(st, entries_to_show, before - (st->update_every * group * entries_to_show), before, group_method, only_non_zero);
277 +
278 // -------------------------------------------------------------------------
279 // remove dimensions that contain only zeros
280
@@ -329,7 +327,10 @@ unsigned long rrd_stats_json(int type, RRDSET *st, struct web_buffer *wb, int en
327
328 long count = 0, printed = 0, group_count = 0;
329 last_timestamp = 0;
332 - for(t = current_entry; max_entries ; now -= st->update_every, t--, max_entries--) {
330 +
331 + long expected_to_start_at_slot = rrdset_time2slot(st, before);
332 +
333 + for(t = rrdset_last_slot(st); max_entries ; now -= st->update_every, t--, max_entries--) {
334 if(t < 0) t = st->entries - 1;
335
336 int print_this = 0;
@@ -350,6 +351,11 @@ unsigned long rrd_stats_json(int type, RRDSET *st, struct web_buffer *wb, int en
351 // make sure we return data in the proper time range
352 if(now < after || now > before) continue;
353
354 + if(expected_to_start_at_slot != -999999) {
355 + error("%s: Expected to start on slot %ld, started on %ld, %s", st->id, expected_to_start_at_slot, t, (t == expected_to_start_at_slot)?"OK":"ERROR");
356 + expected_to_start_at_slot = -999999;
357 + }
358 +
359 count++;
360 group_count++;
361
@@ -488,6 +494,349 @@ unsigned long rrd_stats_json(int type, RRDSET *st, struct web_buffer *wb, int en
494 }
495
496
497 +// ----------------------------------------------------------------------------
498 +
499 +// RRDR options
500 +#define RRDR_EMPTY 0x01
501 +#define RRDR_RESET 0x02
502 +#define RRDR_HIDDEN 0x04
503 +#define RRDR_NONZERO 0x08
504 +
505 +
506 +typedef struct rrdresult {
507 + RRDSET *st; // the chart this result refers to
508 +
509 + int d; // the number of dimensions
510 + int n; // the number of values in the arrays
511 +
512 + time_t *t; // array of timestamps
513 + calculated_number *v; // array n x d values
514 + uint8_t *o; // array n x d options
515 +
516 + int c; // current line (n)
517 +
518 + int has_st_lock; // if st is read locked by us
519 +} RRDR;
520 +
521 +inline static calculated_number *rrdr_line_values(RRDR *r)
522 +{
523 + return &r->v[ r->c * r->d ];
524 +}
525 +
526 +inline static uint8_t *rrdr_line_options(RRDR *r)
527 +{
528 + return &r->o[ r->c * r->d ];
529 +}
530 +
531 +inline static int rrdr_line_next(RRDR *r, time_t t)
532 +{
533 + // save the time
534 + r->t[r->c] = t;
535 +
536 + r->c++;
537 + if(unlikely(r->c >= r->n)) {
538 + r->c--;
539 + return 0;
540 + }
541 +
542 + return 1;
543 +}
544 +
545 +inline static void rrdr_lock_rrdset(RRDR *r) {
546 + if(unlikely(!r)) {
547 + error("NULL value given!");
548 + return;
549 + }
550 +
551 + pthread_rwlock_rdlock(&r->st->rwlock);
552 + r->has_st_lock = 1;
553 +}
554 +
555 +inline static void rrdr_unlock_rrdset(RRDR *r) {
556 + if(unlikely(!r)) {
557 + error("NULL value given!");
558 + return;
559 + }
560 +
561 + if(likely(r->has_st_lock)) {
562 + pthread_rwlock_unlock(&r->st->rwlock);
563 + r->has_st_lock = 0;
564 + }
565 +}
566 +
567 +inline static void rrdr_free(RRDR *r)
568 +{
569 + if(unlikely(!r)) {
570 + error("NULL value given!");
571 + return;
572 + }
573 +
574 + rrdr_unlock_rrdset(r);
575 + if(likely(r->t)) free(r->t);
576 + if(likely(r->v)) free(r->v);
577 + if(likely(r->o)) free(r->o);
578 + free(r);
579 +}
580 +
581 +static RRDR *rrdr_create(RRDSET *st, int n)
582 +{
583 + if(unlikely(!st)) {
584 + error("NULL value given!");
585 + return NULL;
586 + }
587 +
588 + RRDR *r = calloc(1, sizeof(RRDR));
589 + if(unlikely(!r)) goto cleanup;
590 +
591 + r->st = st;
592 +
593 + rrdr_lock_rrdset(r);
594 +
595 + RRDDIM *rd;
596 + for(rd = st->dimensions ; rd ; rd = rd->next) r->d++;
597 +
598 + r->n = n;
599 + r->t = malloc(sizeof(time_t) * n);
600 + if(unlikely(!r->t)) goto cleanup;
601 +
602 + r->t = malloc(sizeof(time_t) * n);
603 + if(unlikely(!r->t)) goto cleanup;
604 +
605 + r->v = malloc(sizeof(calculated_number) * n * r->d);
606 + if(unlikely(!r->v)) goto cleanup;
607 +
608 + r->o = malloc(sizeof(calculated_number) * n * r->d);
609 + if(unlikely(!r->o)) goto cleanup;
610 +
611 + return r;
612 +
613 +cleanup:
614 + error("Cannot allocate memory");
615 + if(likely(r)) rrdr_free(r);
616 + return NULL;
617 +}
618 +
619 +RRDR *rrd2rrdr(RRDSET *st, long points, time_t after, time_t before, int group_method)
620 +{
621 + time_t first_entry_t = rrdset_first_entry_t(st);
622 + time_t last_entry_t = rrdset_last_entry_t(st);
623 +
624 + // allow relative for before and after
625 + if(before <= st->update_every * st->entries) before = last_entry_t + before;
626 + if(after <= st->update_every * st->entries) after = last_entry_t + after;
627 +
628 + // make sure they are within our timeframe
629 + if(before > last_entry_t) before = last_entry_t;
630 + if(before < first_entry_t) before = first_entry_t;
631 +
632 + if(after > last_entry_t) after = last_entry_t;
633 + if(after < first_entry_t) after = first_entry_t;
634 +
635 + // check if they are upside down
636 + if(after > before) {
637 + time_t t = before;
638 + before = after;
639 + after = t;
640 + }
641 +
642 + // the duration of the chart
643 + time_t duration = before - after;
644 + if(duration <= 0) return NULL;
645 +
646 + // check the required points
647 + if(points <= 0) points = duration;
648 +
649 + // calculate proper grouping of source data
650 + int group = duration / points;
651 + if(group <= 0) group = 1;
652 + if(duration / group > points) group++;
653 +
654 + // align timestamps to group
655 + before -= before % group;
656 + after -= after % group;
657 + duration = before - after;
658 +
659 + // error("NEW: points=%d after=%d before=%d group=%d, duration=%d", points, after, before, group, duration);
660 +
661 + // Now we have:
662 + // before = the end time of the calculation
663 + // after = the start time of the calculation
664 + // duration = the duration of the calculation
665 + // group = the number of source points to aggregate / group together
666 + // method = the method of grouping source points
667 + // points = the number of points to generate
668 +
669 +
670 + // -------------------------------------------------------------------------
671 + // initialize our result set
672 +
673 + RRDR *r = rrdr_create(st, points);
674 + if(!r) return NULL;
675 + if(!r->d) {
676 + rrdr_free(r);
677 + return NULL;
678 + }
679 +
680 + // find how many dimensions we have
681 + long dimensions = r->d;
682 +
683 +
684 + // -------------------------------------------------------------------------
685 +
686 + // how many entries can we use from the source data?
687 + long max_entries = (st->counter < (unsigned long)st->entries) ? st->counter : (unsigned long)st->entries;
688 +
689 +
690 + // -------------------------------------------------------------------------
691 + // checks for debugging
692 +
693 + if(st->debug) {
694 + debug(D_RRD_STATS, "%s first_entry_t = %lu, last_entry_t = %lu, duration = %lu, after = %lu, before = %lu, duration = %lu, entries_to_show = %lu, group = %lu, max_entries = %ld"
695 + , st->id
696 + , first_entry_t
697 + , last_entry_t
698 + , last_entry_t - first_entry_t
699 + , after
700 + , before
701 + , duration
702 + , points
703 + , group
704 + , max_entries
705 + );
706 + }
707 +
708 +
709 + // -------------------------------------------------------------------------
710 + // temp arrays for keeping values per dimension
711 +
712 + calculated_number group_values[dimensions]; // keep sums when grouping
713 + long group_counts[dimensions]; // keep the number of values added to group_values
714 + uint8_t group_options[dimensions];
715 + uint8_t found_non_zero[dimensions];
716 +
717 +
718 + // initialize them
719 + RRDDIM *rd;
720 + long c;
721 + for( rd = st->dimensions, c = 0 ; rd && c < dimensions ; rd = rd->next, c++) {
722 + group_values[c] = 0;
723 + group_counts[c] = 0;
724 + group_options[c] = 0;
725 + found_non_zero[c] = 0;
726 + }
727 +
728 +
729 + // -------------------------------------------------------------------------
730 + // the main loop
731 +
732 + int debug = st->debug;
733 +
734 + time_t now = last_entry_t,
735 + dt = st->update_every;
736 +
737 + long t = rrdset_time2slot(st, before), // rrdset_last_slot(st),
738 + count = 0,
739 + added = 0,
740 + group_count = group,
741 + add_this = 0;
742 +
743 + for( ; max_entries ; now -= dt, t--, max_entries--) {
744 + if(unlikely(t < 0)) c = st->entries - 1;
745 +
746 + if(unlikely(debug)) debug(D_RRD_STATS, "%s c = %ld, count = %ld, group_count = %ld, added = %ld, now = %lu, %s %s"
747 + , st->id
748 + , t
749 + , count + 1
750 + , group_count + 1
751 + , added
752 + , now
753 + , (group_count == 0)?"PRINT":" - "
754 + , (now >= after && now <= before)?"RANGE":" - "
755 + );
756 +
757 + // make sure we return data in the proper time range
758 + if(unlikely(now < after || now > before)) continue;
759 +
760 + count++;
761 + group_count++;
762 +
763 + if(unlikely(group_count == group)) {
764 + if(unlikely(added >= points)) break;
765 + add_this = 1;
766 + }
767 +
768 + // do the calculations
769 + for(rd = st->dimensions, c = 0 ; likely(rd && c < dimensions) ; rd = rd->next, c++) {
770 + storage_number n = rd->values[t];
771 + if(unlikely(!does_storage_number_exist(n))) continue;
772 +
773 + group_counts[c]++;
774 +
775 + calculated_number value = unpack_storage_number(n);
776 + if(value != 0.0) {
777 + group_options[c] |= RRDR_NONZERO;
778 + found_non_zero[c] = 1;
779 + }
780 +
781 + if(unlikely(did_storage_number_reset(n)))
782 + group_options[c] |= RRDR_RESET;
783 +
784 + switch(group_method) {
785 + case GROUP_MAX:
786 + if(unlikely(abs(value) > abs(group_values[c])))
787 + group_values[c] = value;
788 + break;
789 +
790 + default:
791 + case GROUP_SUM:
792 + case GROUP_AVERAGE:
793 + group_values[c] += value;
794 + break;
795 +
796 + group_values[c] += value;
797 + break;
798 + }
799 + }
800 +
801 + // added it
802 + if(unlikely(add_this)) {
803 + calculated_number *cn = rrdr_line_values(r);
804 + uint8_t *co = rrdr_line_options(r);
805 +
806 + for(rd = st->dimensions, c = 0 ; likely(rd && c < dimensions) ; rd = rd->next, c++) {
807 + if(rd->hidden) group_options[c] |= RRDR_HIDDEN;
808 +
809 + co[c] = group_options[c];
810 +
811 + if(group_counts[c] == 0) {
812 + cn[c] = 0.0;
813 + co[c] |= RRDR_EMPTY;
814 + }
815 + else if(unlikely(group_method == GROUP_AVERAGE)) {
816 + cn[c] = group_values[c] / group_counts[c];
817 + }
818 + else {
819 + cn[c] = group_values[c];
820 + }
821 +
822 + // reset them for the next loop
823 + group_values[c] = 0;
824 + group_counts[c] = 0;
825 + group_options[c] = 0;
826 + }
827 +
828 + added++;
829 + group_count = 0;
830 + add_this = 0;
831 + }
832 +
833 + rrdr_line_next(r, now);
834 + }
835 +
836 + return r;
837 +}
838 +
839 +
840 /*
841 unsigned long rrdset2json(int type, RRDSET *st, struct web_buffer *wb, int entries_to_show, int group, int group_method, time_t after, time_t before, int only_non_zero) {
842