1 /*****************************************************************************
2 * RRDtool 1.2.13 Copyright by Tobi Oetiker, 1997-2006
3 *****************************************************************************
4 * rrd_update.c RRD Update Function
5 *****************************************************************************
7 *****************************************************************************/
10 #include <sys/types.h>
16 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
17 #include <sys/locking.h>
23 #include "rrd_rpncalc.h"
25 #include "rrd_is_thread_safe.h"
28 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
30 * WIN32 does not have gettimeofday and struct timeval. This is a quick and dirty
33 #include <sys/timeb.h>
37 time_t tv_sec; /* seconds */
38 long tv_usec; /* microseconds */
43 int tz_minuteswest; /* minutes W of Greenwich */
44 int tz_dsttime; /* type of dst correction */
47 static int gettimeofday(struct timeval *t, struct __timezone *tz) {
49 struct _timeb current_time;
51 _ftime(¤t_time);
53 t->tv_sec = current_time.time;
54 t->tv_usec = current_time.millitm * 1000;
61 * normilize time as returned by gettimeofday. usec part must
64 static void normalize_time(struct timeval *t)
68 t->tv_usec += 1000000L;
72 /* Local prototypes */
73 int LockRRD(FILE *rrd_file);
75 info_t *write_RRA_row (rrd_t *rrd, unsigned long rra_idx,
76 unsigned long *rra_current,
77 unsigned short CDP_scratch_idx,
79 FILE UNUSED(*rrd_file),
83 info_t *pcdp_summary, time_t *rra_time, void *rrd_mmaped_file);
85 info_t *write_RRA_row (rrd_t *rrd, unsigned long rra_idx,
86 unsigned long *rra_current,
87 unsigned short CDP_scratch_idx, FILE *rrd_file,
88 info_t *pcdp_summary, time_t *rra_time);
90 int rrd_update_r(char *filename, char *tmplt, int argc, char **argv);
91 int _rrd_update(char *filename, char *tmplt, int argc, char **argv,
94 #define IFDNAN(X,Y) (isnan(X) ? (Y) : (X));
99 main(int argc, char **argv){
100 rrd_update(argc,argv);
101 if (rrd_test_error()) {
102 printf("RRDtool " PACKAGE_VERSION " Copyright by Tobi Oetiker, 1997-2005\n\n"
103 "Usage: rrdupdate filename\n"
104 "\t\t\t[--template|-t ds-name:ds-name:...]\n"
105 "\t\t\ttime|N:value[:value...]\n\n"
106 "\t\t\tat-time@value[:value...]\n\n"
107 "\t\t\t[ time:value[:value...] ..]\n\n");
109 printf("ERROR: %s\n",rrd_get_error());
117 info_t *rrd_update_v(int argc, char **argv)
120 info_t *result = NULL;
122 optind = 0; opterr = 0; /* initialize getopt */
125 static struct option long_options[] =
127 {"template", required_argument, 0, 't'},
130 int option_index = 0;
132 opt = getopt_long(argc, argv, "t:",
133 long_options, &option_index);
144 rrd_set_error("unknown option '%s'",argv[optind-1]);
150 /* need at least 2 arguments: filename, data. */
151 if (argc-optind < 2) {
152 rrd_set_error("Not enough arguments");
156 result = info_push(NULL,sprintf_alloc("return_value"),RD_I_INT,rc);
157 rc.u_int = _rrd_update(argv[optind], tmplt,
158 argc - optind - 1, argv + optind + 1, result);
159 result->value.u_int = rc.u_int;
165 rrd_update(int argc, char **argv)
169 optind = 0; opterr = 0; /* initialize getopt */
172 static struct option long_options[] =
174 {"template", required_argument, 0, 't'},
177 int option_index = 0;
179 opt = getopt_long(argc, argv, "t:",
180 long_options, &option_index);
191 rrd_set_error("unknown option '%s'",argv[optind-1]);
196 /* need at least 2 arguments: filename, data. */
197 if (argc-optind < 2) {
198 rrd_set_error("Not enough arguments");
203 rc = rrd_update_r(argv[optind], tmplt,
204 argc - optind - 1, argv + optind + 1);
209 rrd_update_r(char *filename, char *tmplt, int argc, char **argv)
211 return _rrd_update(filename, tmplt, argc, argv, NULL);
215 _rrd_update(char *filename, char *tmplt, int argc, char **argv,
216 info_t *pcdp_summary)
221 unsigned long i,ii,iii=1;
223 unsigned long rra_begin; /* byte pointer to the rra
224 * area in the rrd file. this
225 * pointer never changes value */
226 unsigned long rra_start; /* byte pointer to the rra
227 * area in the rrd file. this
228 * pointer changes as each rrd is
230 unsigned long rra_current; /* byte pointer to the current write
231 * spot in the rrd file. */
232 unsigned long rra_pos_tmp; /* temporary byte pointer. */
234 pre_int,post_int; /* interval between this and
236 unsigned long proc_pdp_st; /* which pdp_st was the last
238 unsigned long occu_pdp_st; /* when was the pdp_st
239 * before the last update
241 unsigned long proc_pdp_age; /* how old was the data in
242 * the pdp prep area when it
243 * was last updated */
244 unsigned long occu_pdp_age; /* how long ago was the last
246 rrd_value_t *pdp_new; /* prepare the incoming data
247 * to be added the the
249 rrd_value_t *pdp_temp; /* prepare the pdp values
250 * to be added the the
253 long *tmpl_idx; /* index representing the settings
254 transported by the tmplt index */
255 unsigned long tmpl_cnt = 2; /* time and data */
259 time_t current_time = 0;
260 time_t rra_time = 0; /* time of update for a RRA */
261 unsigned long current_time_usec=0;/* microseconds part of current time */
262 struct timeval tmp_time; /* used for time conversion */
265 int schedule_smooth = 0;
266 rrd_value_t *seasonal_coef = NULL, *last_seasonal_coef = NULL;
267 /* a vector of future Holt-Winters seasonal coefs */
268 unsigned long elapsed_pdp_st;
269 /* number of elapsed PDP steps since last update */
270 unsigned long *rra_step_cnt = NULL;
271 /* number of rows to be updated in an RRA for a data
273 unsigned long start_pdp_offset;
274 /* number of PDP steps since the last update that
275 * are assigned to the first CDP to be generated
276 * since the last update. */
277 unsigned short scratch_idx;
278 /* index into the CDP scratch array */
279 enum cf_en current_cf;
280 /* numeric id of the current consolidation function */
281 rpnstack_t rpnstack; /* used for COMPUTE DS */
282 int version; /* rrd version */
283 char *endptr; /* used in the conversion */
285 void *rrd_mmaped_file;
286 unsigned long rrd_filesize;
289 rpnstack_init(&rpnstack);
291 /* need at least 1 arguments: data. */
293 rrd_set_error("Not enough arguments");
299 if(rrd_open(filename,&rrd_file,&rrd, RRD_READWRITE)==-1){
302 /* initialize time */
303 version = atoi(rrd.stat_head->version);
304 gettimeofday(&tmp_time, 0);
305 normalize_time(&tmp_time);
306 current_time = tmp_time.tv_sec;
308 current_time_usec = tmp_time.tv_usec;
311 current_time_usec = 0;
314 rra_current = rra_start = rra_begin = ftell(rrd_file);
315 /* This is defined in the ANSI C standard, section 7.9.5.3:
317 When a file is opened with udpate mode ('+' as the second
318 or third character in the ... list of mode argument
319 variables), both input and ouptut may be performed on the
320 associated stream. However, ... input may not be directly
321 followed by output without an intervening call to a file
322 positioning function, unless the input oepration encounters
325 fseek(rrd_file, 0, SEEK_END);
326 rrd_filesize = ftell(rrd_file);
327 fseek(rrd_file, rra_current, SEEK_SET);
329 fseek(rrd_file, 0, SEEK_CUR);
333 /* get exclusive lock to whole file.
334 * lock gets removed when we close the file.
336 if (LockRRD(rrd_file) != 0) {
337 rrd_set_error("could not lock RRD");
343 if((updvals = malloc( sizeof(char*) * (rrd.stat_head->ds_cnt+1)))==NULL){
344 rrd_set_error("allocating updvals pointer array");
350 if ((pdp_temp = malloc(sizeof(rrd_value_t)
351 *rrd.stat_head->ds_cnt))==NULL){
352 rrd_set_error("allocating pdp_temp ...");
359 if ((tmpl_idx = malloc(sizeof(unsigned long)
360 *(rrd.stat_head->ds_cnt+1)))==NULL){
361 rrd_set_error("allocating tmpl_idx ...");
368 /* initialize tmplt redirector */
369 /* default config example (assume DS 1 is a CDEF DS)
370 tmpl_idx[0] -> 0; (time)
371 tmpl_idx[1] -> 1; (DS 0)
372 tmpl_idx[2] -> 3; (DS 2)
373 tmpl_idx[3] -> 4; (DS 3) */
374 tmpl_idx[0] = 0; /* time */
375 for (i = 1, ii = 1 ; i <= rrd.stat_head->ds_cnt ; i++)
377 if (dst_conv(rrd.ds_def[i-1].dst) != DST_CDEF)
383 /* we should work on a writeable copy here */
385 unsigned int tmpl_len;
386 tmplt = strdup(tmplt);
388 tmpl_cnt = 1; /* the first entry is the time */
389 tmpl_len = strlen(tmplt);
390 for(i=0;i<=tmpl_len ;i++) {
391 if (tmplt[i] == ':' || tmplt[i] == '\0') {
393 if (tmpl_cnt>rrd.stat_head->ds_cnt){
394 rrd_set_error("tmplt contains more DS definitions than RRD");
395 free(updvals); free(pdp_temp);
396 free(tmpl_idx); rrd_free(&rrd);
397 fclose(rrd_file); return(-1);
399 if ((tmpl_idx[tmpl_cnt++] = ds_match(&rrd,dsname)) == -1){
400 rrd_set_error("unknown DS name '%s'",dsname);
401 free(updvals); free(pdp_temp);
403 free(tmpl_idx); rrd_free(&rrd);
404 fclose(rrd_file); return(-1);
406 /* the first element is always the time */
407 tmpl_idx[tmpl_cnt-1]++;
408 /* go to the next entry on the tmplt */
409 dsname = &tmplt[i+1];
410 /* fix the damage we did before */
420 if ((pdp_new = malloc(sizeof(rrd_value_t)
421 *rrd.stat_head->ds_cnt))==NULL){
422 rrd_set_error("allocating pdp_new ...");
432 rrd_mmaped_file = mmap(0,
434 PROT_READ | PROT_WRITE,
438 if (rrd_mmaped_file == MAP_FAILED) {
439 rrd_set_error("error mmapping file %s", filename);
448 /* loop through the arguments. */
449 for(arg_i=0; arg_i<argc;arg_i++) {
450 char *stepper = malloc((strlen(argv[arg_i])+1)*sizeof(char));
451 char *step_start = stepper;
453 char *parsetime_error = NULL;
454 enum {atstyle, normal} timesyntax;
455 struct rrd_time_value ds_tv;
456 if (stepper == NULL){
457 rrd_set_error("failed duplication argv entry");
463 munmap(rrd_mmaped_file, rrd_filesize);
468 /* initialize all ds input to unknown except the first one
469 which has always got to be set */
470 for(ii=1;ii<=rrd.stat_head->ds_cnt;ii++) updvals[ii] = "U";
471 strcpy(stepper,argv[arg_i]);
473 /* separate all ds elements; first must be examined separately
474 due to alternate time syntax */
475 if ((p=strchr(stepper,'@'))!=NULL) {
476 timesyntax = atstyle;
479 } else if ((p=strchr(stepper,':'))!=NULL) {
484 rrd_set_error("expected timestamp not found in data source from %s:...",
490 updvals[tmpl_idx[ii]] = stepper;
492 if (*stepper == ':') {
496 updvals[tmpl_idx[ii]] = stepper+1;
502 if (ii != tmpl_cnt-1) {
503 rrd_set_error("expected %lu data source readings (got %lu) from %s:...",
504 tmpl_cnt-1, ii, argv[arg_i]);
509 /* get the time from the reading ... handle N */
510 if (timesyntax == atstyle) {
511 if ((parsetime_error = parsetime(updvals[0], &ds_tv))) {
512 rrd_set_error("ds time: %s: %s", updvals[0], parsetime_error );
516 if (ds_tv.type == RELATIVE_TO_END_TIME ||
517 ds_tv.type == RELATIVE_TO_START_TIME) {
518 rrd_set_error("specifying time relative to the 'start' "
519 "or 'end' makes no sense here: %s",
525 current_time = mktime(&ds_tv.tm) + ds_tv.offset;
526 current_time_usec = 0; /* FIXME: how to handle usecs here ? */
528 } else if (strcmp(updvals[0],"N")==0){
529 gettimeofday(&tmp_time, 0);
530 normalize_time(&tmp_time);
531 current_time = tmp_time.tv_sec;
532 current_time_usec = tmp_time.tv_usec;
535 tmp = strtod(updvals[0], 0);
536 current_time = floor(tmp);
537 current_time_usec = (long)((tmp-(double)current_time) * 1000000.0);
539 /* dont do any correction for old version RRDs */
541 current_time_usec = 0;
543 if(current_time < rrd.live_head->last_up ||
544 (current_time == rrd.live_head->last_up &&
545 (long)current_time_usec <= (long)rrd.live_head->last_up_usec)) {
546 rrd_set_error("illegal attempt to update using time %ld when "
547 "last update time is %ld (minimum one second step)",
548 current_time, rrd.live_head->last_up);
554 /* seek to the beginning of the rra's */
555 if (rra_current != rra_begin) {
557 if(fseek(rrd_file, rra_begin, SEEK_SET) != 0) {
558 rrd_set_error("seek error in rrd");
563 rra_current = rra_begin;
565 rra_start = rra_begin;
567 /* when was the current pdp started */
568 proc_pdp_age = rrd.live_head->last_up % rrd.stat_head->pdp_step;
569 proc_pdp_st = rrd.live_head->last_up - proc_pdp_age;
571 /* when did the last pdp_st occur */
572 occu_pdp_age = current_time % rrd.stat_head->pdp_step;
573 occu_pdp_st = current_time - occu_pdp_age;
575 /* interval = current_time - rrd.live_head->last_up; */
576 interval = (double)(current_time - rrd.live_head->last_up)
577 + (double)((long)current_time_usec - (long)rrd.live_head->last_up_usec)/1000000.0;
579 if (occu_pdp_st > proc_pdp_st){
580 /* OK we passed the pdp_st moment*/
581 pre_int = (long)occu_pdp_st - rrd.live_head->last_up; /* how much of the input data
582 * occurred before the latest
584 pre_int -= ((double)rrd.live_head->last_up_usec)/1000000.0; /* adjust usecs */
585 post_int = occu_pdp_age; /* how much after it */
586 post_int += ((double)current_time_usec)/1000000.0; /* adjust usecs */
600 "post_int %lf\n", proc_pdp_age, proc_pdp_st,
601 occu_pdp_age, occu_pdp_st,
602 interval, pre_int, post_int);
605 /* process the data sources and update the pdp_prep
606 * area accordingly */
607 for(i=0;i<rrd.stat_head->ds_cnt;i++){
609 dst_idx= dst_conv(rrd.ds_def[i].dst);
611 /* make sure we do not build diffs with old last_ds values */
612 if(rrd.ds_def[i].par[DS_mrhb_cnt].u_cnt < interval
613 && ( dst_idx == DST_COUNTER || dst_idx == DST_DERIVE)){
614 strncpy(rrd.pdp_prep[i].last_ds,"U",LAST_DS_LEN-1);
617 /* NOTE: DST_CDEF should never enter this if block, because
618 * updvals[i+1][0] is initialized to 'U'; unless the caller
619 * accidently specified a value for the DST_CDEF. To handle
620 * this case, an extra check is required. */
622 if((updvals[i+1][0] != 'U') &&
623 (dst_idx != DST_CDEF) &&
624 rrd.ds_def[i].par[DS_mrhb_cnt].u_cnt >= interval) {
626 /* the data source type defines how to process the data */
627 /* pdp_new contains rate * time ... eg the bytes
628 * transferred during the interval. Doing it this way saves
629 * a lot of math operations */
635 if(rrd.pdp_prep[i].last_ds[0] != 'U'){
636 for(ii=0;updvals[i+1][ii] != '\0';ii++){
637 if(updvals[i+1][ii] < '0' || updvals[i+1][ii] > '9' || (ii==0 && updvals[i+1][ii] == '-')){
638 rrd_set_error("not a simple integer: '%s'",updvals[i+1]);
642 if (rrd_test_error()){
645 pdp_new[i]= rrd_diff(updvals[i+1],rrd.pdp_prep[i].last_ds);
646 if(dst_idx == DST_COUNTER) {
647 /* simple overflow catcher suggested by Andres Kroonmaa */
648 /* this will fail terribly for non 32 or 64 bit counters ... */
649 /* are there any others in SNMP land ? */
650 if (pdp_new[i] < (double)0.0 )
651 pdp_new[i] += (double)4294967296.0 ; /* 2^32 */
652 if (pdp_new[i] < (double)0.0 )
653 pdp_new[i] += (double)18446744069414584320.0; /* 2^64-2^32 */;
655 rate = pdp_new[i] / interval;
663 pdp_new[i] = strtod(updvals[i+1],&endptr);
665 rrd_set_error("converting '%s' to float: %s",updvals[i+1],rrd_strerror(errno));
668 if (endptr[0] != '\0'){
669 rrd_set_error("conversion of '%s' to float not complete: tail '%s'",updvals[i+1],endptr);
672 rate = pdp_new[i] / interval;
676 pdp_new[i] = strtod(updvals[i+1],&endptr) * interval;
678 rrd_set_error("converting '%s' to float: %s",updvals[i+1],rrd_strerror(errno));
681 if (endptr[0] != '\0'){
682 rrd_set_error("conversion of '%s' to float not complete: tail '%s'",updvals[i+1],endptr);
685 rate = pdp_new[i] / interval;
688 rrd_set_error("rrd contains unknown DS type : '%s'",
692 /* break out of this for loop if the error string is set */
693 if (rrd_test_error()){
696 /* make sure pdp_temp is neither too large or too small
697 * if any of these occur it becomes unknown ...
699 if ( ! isnan(rate) &&
700 (( ! isnan(rrd.ds_def[i].par[DS_max_val].u_val) &&
701 rate > rrd.ds_def[i].par[DS_max_val].u_val ) ||
702 ( ! isnan(rrd.ds_def[i].par[DS_min_val].u_val) &&
703 rate < rrd.ds_def[i].par[DS_min_val].u_val ))){
707 /* no news is news all the same */
712 /* make a copy of the command line argument for the next run */
720 rrd.pdp_prep[i].last_ds,
721 updvals[i+1], pdp_new[i]);
723 if(dst_idx == DST_COUNTER || dst_idx == DST_DERIVE){
724 strncpy(rrd.pdp_prep[i].last_ds,
725 updvals[i+1],LAST_DS_LEN-1);
726 rrd.pdp_prep[i].last_ds[LAST_DS_LEN-1]='\0';
729 /* break out of the argument parsing loop if the error_string is set */
730 if (rrd_test_error()){
734 /* has a pdp_st moment occurred since the last run ? */
736 if (proc_pdp_st == occu_pdp_st){
737 /* no we have not passed a pdp_st moment. therefore update is simple */
739 for(i=0;i<rrd.stat_head->ds_cnt;i++){
740 if(isnan(pdp_new[i])) {
741 /* this is not realy accurate if we use subsecond data arival time
742 should have thought of it when going subsecond resolution ...
743 sorry next format change we will have it! */
744 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt += floor(interval);
746 if (isnan( rrd.pdp_prep[i].scratch[PDP_val].u_val )){
747 rrd.pdp_prep[i].scratch[PDP_val].u_val= pdp_new[i];
749 rrd.pdp_prep[i].scratch[PDP_val].u_val+= pdp_new[i];
758 rrd.pdp_prep[i].scratch[PDP_val].u_val,
759 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
763 /* an pdp_st has occurred. */
765 /* in pdp_prep[].scratch[PDP_val].u_val we have collected rate*seconds which
766 * occurred up to the last run.
767 pdp_new[] contains rate*seconds from the latest run.
768 pdp_temp[] will contain the rate for cdp */
770 for(i=0;i<rrd.stat_head->ds_cnt;i++){
771 /* update pdp_prep to the current pdp_st. */
772 double pre_unknown = 0.0;
773 if(isnan(pdp_new[i]))
774 /* a final bit of unkonwn to be added bevore calculation
775 * we use a tempaorary variable for this so that we
776 * don't have to turn integer lines before using the value */
777 pre_unknown = pre_int;
779 if (isnan( rrd.pdp_prep[i].scratch[PDP_val].u_val )){
780 rrd.pdp_prep[i].scratch[PDP_val].u_val= pdp_new[i]/interval*pre_int;
782 rrd.pdp_prep[i].scratch[PDP_val].u_val+= pdp_new[i]/interval*pre_int;
787 /* if too much of the pdp_prep is unknown we dump it */
789 /* removed because this does not agree with the definition
790 a heart beat can be unknown */
791 /* (rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt
792 > rrd.ds_def[i].par[DS_mrhb_cnt].u_cnt) || */
793 /* if the interval is larger thatn mrhb we get NAN */
794 (interval > rrd.ds_def[i].par[DS_mrhb_cnt].u_cnt) ||
795 (occu_pdp_st-proc_pdp_st <=
796 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt)) {
799 pdp_temp[i] = rrd.pdp_prep[i].scratch[PDP_val].u_val
800 / ((double)(occu_pdp_st - proc_pdp_st
801 - rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt)
805 /* process CDEF data sources; remember each CDEF DS can
806 * only reference other DS with a lower index number */
807 if (dst_conv(rrd.ds_def[i].dst) == DST_CDEF) {
809 rpnp = rpn_expand((rpn_cdefds_t *) &(rrd.ds_def[i].par[DS_cdef]));
810 /* substitue data values for OP_VARIABLE nodes */
811 for (ii = 0; rpnp[ii].op != OP_END; ii++)
813 if (rpnp[ii].op == OP_VARIABLE) {
814 rpnp[ii].op = OP_NUMBER;
815 rpnp[ii].val = pdp_temp[rpnp[ii].ptr];
818 /* run the rpn calculator */
819 if (rpn_calc(rpnp,&rpnstack,0,pdp_temp,i) == -1) {
821 break; /* exits the data sources pdp_temp loop */
825 /* make pdp_prep ready for the next run */
826 if(isnan(pdp_new[i])){
827 /* this is not realy accurate if we use subsecond data arival time
828 should have thought of it when going subsecond resolution ...
829 sorry next format change we will have it! */
830 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt = floor(post_int);
831 rrd.pdp_prep[i].scratch[PDP_val].u_val = DNAN;
833 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt = 0;
834 rrd.pdp_prep[i].scratch[PDP_val].u_val =
835 pdp_new[i]/interval*post_int;
843 "new_unkn_sec %5lu\n",
845 rrd.pdp_prep[i].scratch[PDP_val].u_val,
846 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
850 /* if there were errors during the last loop, bail out here */
851 if (rrd_test_error()){
856 /* compute the number of elapsed pdp_st moments */
857 elapsed_pdp_st = (occu_pdp_st - proc_pdp_st) / rrd.stat_head -> pdp_step;
859 fprintf(stderr,"elapsed PDP steps: %lu\n", elapsed_pdp_st);
861 if (rra_step_cnt == NULL)
863 rra_step_cnt = (unsigned long *)
864 malloc((rrd.stat_head->rra_cnt)* sizeof(unsigned long));
867 for(i = 0, rra_start = rra_begin;
868 i < rrd.stat_head->rra_cnt;
869 rra_start += rrd.rra_def[i].row_cnt * rrd.stat_head -> ds_cnt * sizeof(rrd_value_t),
872 current_cf = cf_conv(rrd.rra_def[i].cf_nam);
873 start_pdp_offset = rrd.rra_def[i].pdp_cnt -
874 (proc_pdp_st / rrd.stat_head -> pdp_step) % rrd.rra_def[i].pdp_cnt;
875 if (start_pdp_offset <= elapsed_pdp_st) {
876 rra_step_cnt[i] = (elapsed_pdp_st - start_pdp_offset) /
877 rrd.rra_def[i].pdp_cnt + 1;
882 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL)
884 /* If this is a bulk update, we need to skip ahead in the seasonal
885 * arrays so that they will be correct for the next observed value;
886 * note that for the bulk update itself, no update will occur to
887 * DEVSEASONAL or SEASONAL; futhermore, HWPREDICT and DEVPREDICT will
889 if (rra_step_cnt[i] > 2)
891 /* skip update by resetting rra_step_cnt[i],
892 * note that this is not data source specific; this is due
893 * to the bulk update, not a DNAN value for the specific data
896 lookup_seasonal(&rrd,i,rra_start,rrd_file,elapsed_pdp_st,
897 &last_seasonal_coef);
898 lookup_seasonal(&rrd,i,rra_start,rrd_file,elapsed_pdp_st + 1,
902 /* periodically run a smoother for seasonal effects */
903 /* Need to use first cdp parameter buffer to track
904 * burnin (burnin requires a specific smoothing schedule).
905 * The CDP_init_seasonal parameter is really an RRA level,
906 * not a data source within RRA level parameter, but the rra_def
907 * is read only for rrd_update (not flushed to disk). */
908 iii = i*(rrd.stat_head -> ds_cnt);
909 if (rrd.cdp_prep[iii].scratch[CDP_init_seasonal].u_cnt
912 if (rrd.rra_ptr[i].cur_row + elapsed_pdp_st
913 > rrd.rra_def[i].row_cnt - 1) {
914 /* mark off one of the burnin cycles */
915 ++(rrd.cdp_prep[iii].scratch[CDP_init_seasonal].u_cnt);
919 /* someone has no doubt invented a trick to deal with this
920 * wrap around, but at least this code is clear. */
921 if (rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt >
922 rrd.rra_ptr[i].cur_row)
924 /* here elapsed_pdp_st = rra_step_cnt[i] because of 1-1
925 * mapping between PDP and CDP */
926 if (rrd.rra_ptr[i].cur_row + elapsed_pdp_st
927 >= rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt)
931 "schedule_smooth 1: cur_row %lu, elapsed_pdp_st %lu, smooth idx %lu\n",
932 rrd.rra_ptr[i].cur_row, elapsed_pdp_st,
933 rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt);
938 /* can't rely on negative numbers because we are working with
940 /* Don't need modulus here. If we've wrapped more than once, only
941 * one smooth is executed at the end. */
942 if (rrd.rra_ptr[i].cur_row + elapsed_pdp_st >= rrd.rra_def[i].row_cnt
943 && rrd.rra_ptr[i].cur_row + elapsed_pdp_st - rrd.rra_def[i].row_cnt
944 >= rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt)
948 "schedule_smooth 2: cur_row %lu, elapsed_pdp_st %lu, smooth idx %lu\n",
949 rrd.rra_ptr[i].cur_row, elapsed_pdp_st,
950 rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt);
957 rra_current = ftell(rrd_file);
958 } /* if cf is DEVSEASONAL or SEASONAL */
960 if (rrd_test_error()) break;
962 /* update CDP_PREP areas */
963 /* loop over data soures within each RRA */
965 ii < rrd.stat_head->ds_cnt;
969 /* iii indexes the CDP prep area for this data source within the RRA */
970 iii=i*rrd.stat_head->ds_cnt+ii;
972 if (rrd.rra_def[i].pdp_cnt > 1) {
974 if (rra_step_cnt[i] > 0) {
975 /* If we are in this block, as least 1 CDP value will be written to
976 * disk, this is the CDP_primary_val entry. If more than 1 value needs
977 * to be written, then the "fill in" value is the CDP_secondary_val
979 if (isnan(pdp_temp[ii]))
981 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt += start_pdp_offset;
982 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = DNAN;
984 /* CDP_secondary value is the RRA "fill in" value for intermediary
985 * CDP data entries. No matter the CF, the value is the same because
986 * the average, max, min, and last of a list of identical values is
987 * the same, namely, the value itself. */
988 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = pdp_temp[ii];
991 if (rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt
992 > rrd.rra_def[i].pdp_cnt*
993 rrd.rra_def[i].par[RRA_cdp_xff_val].u_val)
995 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = DNAN;
996 /* initialize carry over */
997 if (current_cf == CF_AVERAGE) {
998 if (isnan(pdp_temp[ii])) {
999 rrd.cdp_prep[iii].scratch[CDP_val].u_val = DNAN;
1001 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii] *
1002 ((elapsed_pdp_st - start_pdp_offset) % rrd.rra_def[i].pdp_cnt);
1005 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1008 rrd_value_t cum_val, cur_val;
1009 switch (current_cf) {
1011 cum_val = IFDNAN(rrd.cdp_prep[iii].scratch[CDP_val].u_val, 0.0);
1012 cur_val = IFDNAN(pdp_temp[ii],0.0);
1013 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val =
1014 (cum_val + cur_val * start_pdp_offset) /
1015 (rrd.rra_def[i].pdp_cnt
1016 -rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt);
1017 /* initialize carry over value */
1018 if (isnan(pdp_temp[ii])) {
1019 rrd.cdp_prep[iii].scratch[CDP_val].u_val = DNAN;
1021 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii] *
1022 ((elapsed_pdp_st - start_pdp_offset) % rrd.rra_def[i].pdp_cnt);
1026 cum_val = IFDNAN(rrd.cdp_prep[iii].scratch[CDP_val].u_val, -DINF);
1027 cur_val = IFDNAN(pdp_temp[ii],-DINF);
1029 if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val) &&
1030 isnan(pdp_temp[ii])) {
1032 "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!",
1037 if (cur_val > cum_val)
1038 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cur_val;
1040 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cum_val;
1041 /* initialize carry over value */
1042 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1045 cum_val = IFDNAN(rrd.cdp_prep[iii].scratch[CDP_val].u_val, DINF);
1046 cur_val = IFDNAN(pdp_temp[ii],DINF);
1048 if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val) &&
1049 isnan(pdp_temp[ii])) {
1051 "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!",
1056 if (cur_val < cum_val)
1057 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cur_val;
1059 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cum_val;
1060 /* initialize carry over value */
1061 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1065 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = pdp_temp[ii];
1066 /* initialize carry over value */
1067 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1070 } /* endif meets xff value requirement for a valid value */
1071 /* initialize carry over CDP_unkn_pdp_cnt, this must after CDP_primary_val
1072 * is set because CDP_unkn_pdp_cnt is required to compute that value. */
1073 if (isnan(pdp_temp[ii]))
1074 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt =
1075 (elapsed_pdp_st - start_pdp_offset) % rrd.rra_def[i].pdp_cnt;
1077 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt = 0;
1078 } else /* rra_step_cnt[i] == 0 */
1081 if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val)) {
1082 fprintf(stderr,"schedule CDP_val update, RRA %lu DS %lu, DNAN\n",
1085 fprintf(stderr,"schedule CDP_val update, RRA %lu DS %lu, %10.2f\n",
1086 i,ii,rrd.cdp_prep[iii].scratch[CDP_val].u_val);
1089 if (isnan(pdp_temp[ii])) {
1090 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt += elapsed_pdp_st;
1091 } else if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val))
1093 if (current_cf == CF_AVERAGE) {
1094 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii] *
1097 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1100 fprintf(stderr,"Initialize CDP_val for RRA %lu DS %lu: %10.2f\n",
1101 i,ii,rrd.cdp_prep[iii].scratch[CDP_val].u_val);
1104 switch (current_cf) {
1106 rrd.cdp_prep[iii].scratch[CDP_val].u_val += pdp_temp[ii] *
1110 if (pdp_temp[ii] < rrd.cdp_prep[iii].scratch[CDP_val].u_val)
1111 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1114 if (pdp_temp[ii] > rrd.cdp_prep[iii].scratch[CDP_val].u_val)
1115 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1119 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1124 } else { /* rrd.rra_def[i].pdp_cnt == 1 */
1125 if (elapsed_pdp_st > 2)
1127 switch (current_cf) {
1130 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val=pdp_temp[ii];
1131 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val=pdp_temp[ii];
1134 case CF_DEVSEASONAL:
1135 /* need to update cached seasonal values, so they are consistent
1136 * with the bulk update */
1137 /* WARNING: code relies on the fact that CDP_hw_last_seasonal and
1138 * CDP_last_deviation are the same. */
1139 rrd.cdp_prep[iii].scratch[CDP_hw_last_seasonal].u_val =
1140 last_seasonal_coef[ii];
1141 rrd.cdp_prep[iii].scratch[CDP_hw_seasonal].u_val =
1145 /* need to update the null_count and last_null_count.
1146 * even do this for non-DNAN pdp_temp because the
1147 * algorithm is not learning from batch updates. */
1148 rrd.cdp_prep[iii].scratch[CDP_null_count].u_cnt +=
1150 rrd.cdp_prep[iii].scratch[CDP_last_null_count].u_cnt +=
1154 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = DNAN;
1155 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = DNAN;
1158 /* do not count missed bulk values as failures */
1159 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = 0;
1160 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = 0;
1161 /* need to reset violations buffer.
1162 * could do this more carefully, but for now, just
1163 * assume a bulk update wipes away all violations. */
1164 erase_violations(&rrd, iii, i);
1168 } /* endif rrd.rra_def[i].pdp_cnt == 1 */
1170 if (rrd_test_error()) break;
1172 } /* endif data sources loop */
1173 } /* end RRA Loop */
1175 /* this loop is only entered if elapsed_pdp_st < 3 */
1176 for (j = elapsed_pdp_st, scratch_idx = CDP_primary_val;
1177 j > 0 && j < 3; j--, scratch_idx = CDP_secondary_val)
1179 for(i = 0, rra_start = rra_begin;
1180 i < rrd.stat_head->rra_cnt;
1181 rra_start += rrd.rra_def[i].row_cnt * rrd.stat_head -> ds_cnt * sizeof(rrd_value_t),
1184 if (rrd.rra_def[i].pdp_cnt > 1) continue;
1186 current_cf = cf_conv(rrd.rra_def[i].cf_nam);
1187 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL)
1189 lookup_seasonal(&rrd,i,rra_start,rrd_file,
1190 elapsed_pdp_st + (scratch_idx == CDP_primary_val ? 1 : 2),
1192 rra_current = ftell(rrd_file);
1194 if (rrd_test_error()) break;
1195 /* loop over data soures within each RRA */
1197 ii < rrd.stat_head->ds_cnt;
1200 update_aberrant_CF(&rrd,pdp_temp[ii],current_cf,
1201 i*(rrd.stat_head->ds_cnt) + ii,i,ii,
1202 scratch_idx, seasonal_coef);
1204 } /* end RRA Loop */
1205 if (rrd_test_error()) break;
1206 } /* end elapsed_pdp_st loop */
1208 if (rrd_test_error()) break;
1210 /* Ready to write to disk */
1211 /* Move sequentially through the file, writing one RRA at a time.
1212 * Note this architecture divorces the computation of CDP with
1213 * flushing updated RRA entries to disk. */
1214 for(i = 0, rra_start = rra_begin;
1215 i < rrd.stat_head->rra_cnt;
1216 rra_start += rrd.rra_def[i].row_cnt * rrd.stat_head -> ds_cnt * sizeof(rrd_value_t),
1218 /* is there anything to write for this RRA? If not, continue. */
1219 if (rra_step_cnt[i] == 0) continue;
1221 /* write the first row */
1223 fprintf(stderr," -- RRA Preseek %ld\n",ftell(rrd_file));
1225 rrd.rra_ptr[i].cur_row++;
1226 if (rrd.rra_ptr[i].cur_row >= rrd.rra_def[i].row_cnt)
1227 rrd.rra_ptr[i].cur_row = 0; /* wrap around */
1228 /* positition on the first row */
1229 rra_pos_tmp = rra_start +
1230 (rrd.stat_head->ds_cnt)*(rrd.rra_ptr[i].cur_row)*sizeof(rrd_value_t);
1231 if(rra_pos_tmp != rra_current) {
1233 if(fseek(rrd_file, rra_pos_tmp, SEEK_SET) != 0){
1234 rrd_set_error("seek error in rrd");
1238 rra_current = rra_pos_tmp;
1242 fprintf(stderr," -- RRA Postseek %ld\n",ftell(rrd_file));
1244 scratch_idx = CDP_primary_val;
1245 if (pcdp_summary != NULL)
1247 rra_time = (current_time - current_time
1248 % (rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step))
1249 - ((rra_step_cnt[i]-1)*rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step);
1252 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1253 pcdp_summary, &rra_time, rrd_mmaped_file);
1255 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1256 pcdp_summary, &rra_time);
1258 if (rrd_test_error()) break;
1260 /* write other rows of the bulk update, if any */
1261 scratch_idx = CDP_secondary_val;
1262 for ( ; rra_step_cnt[i] > 1; rra_step_cnt[i]--)
1264 if (++rrd.rra_ptr[i].cur_row == rrd.rra_def[i].row_cnt)
1267 fprintf(stderr,"Wraparound for RRA %s, %lu updates left\n",
1268 rrd.rra_def[i].cf_nam, rra_step_cnt[i] - 1);
1271 rrd.rra_ptr[i].cur_row = 0;
1272 /* seek back to beginning of current rra */
1273 if (fseek(rrd_file, rra_start, SEEK_SET) != 0)
1275 rrd_set_error("seek error in rrd");
1279 fprintf(stderr," -- Wraparound Postseek %ld\n",ftell(rrd_file));
1281 rra_current = rra_start;
1283 if (pcdp_summary != NULL)
1285 rra_time = (current_time - current_time
1286 % (rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step))
1287 - ((rra_step_cnt[i]-2)*rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step);
1290 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1291 pcdp_summary, &rra_time, rrd_mmaped_file);
1293 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1294 pcdp_summary, &rra_time);
1298 if (rrd_test_error())
1302 /* break out of the argument parsing loop if error_string is set */
1303 if (rrd_test_error()){
1308 } /* endif a pdp_st has occurred */
1309 rrd.live_head->last_up = current_time;
1310 rrd.live_head->last_up_usec = current_time_usec;
1312 } /* function argument loop */
1314 if (seasonal_coef != NULL) free(seasonal_coef);
1315 if (last_seasonal_coef != NULL) free(last_seasonal_coef);
1316 if (rra_step_cnt != NULL) free(rra_step_cnt);
1317 rpnstack_free(&rpnstack);
1320 if (munmap(rrd_mmaped_file, rrd_filesize) == -1) {
1321 rrd_set_error("error writing(unmapping) file: %s", filename);
1324 /* if we got here and if there is an error and if the file has not been
1325 * written to, then close things up and return. */
1326 if (rrd_test_error()) {
1336 /* aargh ... that was tough ... so many loops ... anyway, its done.
1337 * we just need to write back the live header portion now*/
1339 if (fseek(rrd_file, (sizeof(stat_head_t)
1340 + sizeof(ds_def_t)*rrd.stat_head->ds_cnt
1341 + sizeof(rra_def_t)*rrd.stat_head->rra_cnt),
1343 rrd_set_error("seek rrd for live header writeback");
1354 if(fwrite( rrd.live_head,
1355 sizeof(live_head_t), 1, rrd_file) != 1){
1356 rrd_set_error("fwrite live_head to rrd");
1367 if(fwrite( &rrd.live_head->last_up,
1368 sizeof(time_t), 1, rrd_file) != 1){
1369 rrd_set_error("fwrite live_head to rrd");
1381 if(fwrite( rrd.pdp_prep,
1383 rrd.stat_head->ds_cnt, rrd_file) != rrd.stat_head->ds_cnt){
1384 rrd_set_error("ftwrite pdp_prep to rrd");
1394 if(fwrite( rrd.cdp_prep,
1396 rrd.stat_head->rra_cnt *rrd.stat_head->ds_cnt, rrd_file)
1397 != rrd.stat_head->rra_cnt *rrd.stat_head->ds_cnt){
1399 rrd_set_error("ftwrite cdp_prep to rrd");
1409 if(fwrite( rrd.rra_ptr,
1411 rrd.stat_head->rra_cnt,rrd_file) != rrd.stat_head->rra_cnt){
1412 rrd_set_error("fwrite rra_ptr to rrd");
1422 /* OK now close the files and free the memory */
1423 if(fclose(rrd_file) != 0){
1424 rrd_set_error("closing rrd");
1433 /* calling the smoothing code here guarantees at most
1434 * one smoothing operation per rrd_update call. Unfortunately,
1435 * it is possible with bulk updates, or a long-delayed update
1436 * for smoothing to occur off-schedule. This really isn't
1437 * critical except during the burning cycles. */
1438 if (schedule_smooth)
1440 rrd_file = fopen(filename,"rb+");
1441 rra_start = rra_begin;
1442 for (i = 0; i < rrd.stat_head -> rra_cnt; ++i)
1444 if (cf_conv(rrd.rra_def[i].cf_nam) == CF_DEVSEASONAL ||
1445 cf_conv(rrd.rra_def[i].cf_nam) == CF_SEASONAL)
1448 fprintf(stderr,"Running smoother for rra %ld\n",i);
1450 apply_smoother(&rrd,i,rra_start,rrd_file);
1451 if (rrd_test_error())
1454 rra_start += rrd.rra_def[i].row_cnt
1455 *rrd.stat_head->ds_cnt*sizeof(rrd_value_t);
1468 * get exclusive lock to whole file.
1469 * lock gets removed when we close the file
1471 * returns 0 on success
1474 LockRRD(FILE *rrdfile)
1476 int rrd_fd; /* File descriptor for RRD */
1479 rrd_fd = fileno(rrdfile);
1482 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
1485 if ( _fstat( rrd_fd, &st ) == 0 ) {
1486 rcstat = _locking ( rrd_fd, _LK_NBLCK, st.st_size );
1492 lock.l_type = F_WRLCK; /* exclusive write lock */
1493 lock.l_len = 0; /* whole file */
1494 lock.l_start = 0; /* start of file */
1495 lock.l_whence = SEEK_SET; /* end of file */
1497 rcstat = fcntl(rrd_fd, F_SETLK, &lock);
1507 *write_RRA_row (rrd_t *rrd, unsigned long rra_idx, unsigned long *rra_current,
1508 unsigned short CDP_scratch_idx,
1510 FILE UNUSED(*rrd_file),
1514 info_t *pcdp_summary, time_t *rra_time, void *rrd_mmaped_file)
1517 *write_RRA_row (rrd_t *rrd, unsigned long rra_idx, unsigned long *rra_current,
1518 unsigned short CDP_scratch_idx, FILE *rrd_file,
1519 info_t *pcdp_summary, time_t *rra_time)
1522 unsigned long ds_idx, cdp_idx;
1525 for (ds_idx = 0; ds_idx < rrd -> stat_head -> ds_cnt; ds_idx++)
1527 /* compute the cdp index */
1528 cdp_idx =rra_idx * (rrd -> stat_head->ds_cnt) + ds_idx;
1530 fprintf(stderr," -- RRA WRITE VALUE %e, at %ld CF:%s\n",
1531 rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val,ftell(rrd_file),
1532 rrd -> rra_def[rra_idx].cf_nam);
1534 if (pcdp_summary != NULL)
1536 iv.u_val = rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val;
1537 /* append info to the return hash */
1538 pcdp_summary = info_push(pcdp_summary,
1539 sprintf_alloc("[%d]RRA[%s][%lu]DS[%s]",
1540 *rra_time, rrd->rra_def[rra_idx].cf_nam,
1541 rrd->rra_def[rra_idx].pdp_cnt, rrd->ds_def[ds_idx].ds_nam),
1545 memcpy((char *)rrd_mmaped_file + *rra_current,
1546 &(rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val),
1547 sizeof(rrd_value_t));
1549 if(fwrite(&(rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val),
1550 sizeof(rrd_value_t),1,rrd_file) != 1)
1552 rrd_set_error("writing rrd");
1556 *rra_current += sizeof(rrd_value_t);
1558 return (pcdp_summary);