1 /*****************************************************************************
2 * RRDtool 1.2.11 Copyright by Tobi Oetiker, 1997-2005
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 */
711 /* make a copy of the command line argument for the next run */
719 rrd.pdp_prep[i].last_ds,
720 updvals[i+1], pdp_new[i]);
722 if(dst_idx == DST_COUNTER || dst_idx == DST_DERIVE){
723 strncpy(rrd.pdp_prep[i].last_ds,
724 updvals[i+1],LAST_DS_LEN-1);
725 rrd.pdp_prep[i].last_ds[LAST_DS_LEN-1]='\0';
728 /* break out of the argument parsing loop if the error_string is set */
729 if (rrd_test_error()){
733 /* has a pdp_st moment occurred since the last run ? */
735 if (proc_pdp_st == occu_pdp_st){
736 /* no we have not passed a pdp_st moment. therefore update is simple */
738 for(i=0;i<rrd.stat_head->ds_cnt;i++){
739 if(isnan(pdp_new[i]))
740 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt += floor(interval+0.5);
742 if (isnan( rrd.pdp_prep[i].scratch[PDP_val].u_val )){
743 rrd.pdp_prep[i].scratch[PDP_val].u_val= pdp_new[i];
745 rrd.pdp_prep[i].scratch[PDP_val].u_val+= pdp_new[i];
754 rrd.pdp_prep[i].scratch[PDP_val].u_val,
755 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
759 /* an pdp_st has occurred. */
761 /* in pdp_prep[].scratch[PDP_val].u_val we have collected rate*seconds which
762 * occurred up to the last run.
763 pdp_new[] contains rate*seconds from the latest run.
764 pdp_temp[] will contain the rate for cdp */
766 for(i=0;i<rrd.stat_head->ds_cnt;i++){
767 /* update pdp_prep to the current pdp_st. */
769 if(isnan(pdp_new[i]))
770 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt += floor(pre_int+0.5);
772 if (isnan( rrd.pdp_prep[i].scratch[PDP_val].u_val )){
773 rrd.pdp_prep[i].scratch[PDP_val].u_val= pdp_new[i]/interval*pre_int;
775 rrd.pdp_prep[i].scratch[PDP_val].u_val+= pdp_new[i]/interval*pre_int;
780 /* if too much of the pdp_prep is unknown we dump it */
782 /* removed because this does not agree with the definition
783 a heart beat can be unknown */
784 /* (rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt
785 > rrd.ds_def[i].par[DS_mrhb_cnt].u_cnt) || */
786 (occu_pdp_st-proc_pdp_st <=
787 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt)) {
790 pdp_temp[i] = rrd.pdp_prep[i].scratch[PDP_val].u_val
791 / (double)( occu_pdp_st
793 - rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
796 /* process CDEF data sources; remember each CDEF DS can
797 * only reference other DS with a lower index number */
798 if (dst_conv(rrd.ds_def[i].dst) == DST_CDEF) {
800 rpnp = rpn_expand((rpn_cdefds_t *) &(rrd.ds_def[i].par[DS_cdef]));
801 /* substitue data values for OP_VARIABLE nodes */
802 for (ii = 0; rpnp[ii].op != OP_END; ii++)
804 if (rpnp[ii].op == OP_VARIABLE) {
805 rpnp[ii].op = OP_NUMBER;
806 rpnp[ii].val = pdp_temp[rpnp[ii].ptr];
809 /* run the rpn calculator */
810 if (rpn_calc(rpnp,&rpnstack,0,pdp_temp,i) == -1) {
812 break; /* exits the data sources pdp_temp loop */
816 /* make pdp_prep ready for the next run */
817 if(isnan(pdp_new[i])){
818 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt = floor(post_int + 0.5);
819 rrd.pdp_prep[i].scratch[PDP_val].u_val = DNAN;
821 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt = 0;
822 rrd.pdp_prep[i].scratch[PDP_val].u_val =
823 pdp_new[i]/interval*post_int;
831 "new_unkn_sec %5lu\n",
833 rrd.pdp_prep[i].scratch[PDP_val].u_val,
834 rrd.pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
838 /* if there were errors during the last loop, bail out here */
839 if (rrd_test_error()){
844 /* compute the number of elapsed pdp_st moments */
845 elapsed_pdp_st = (occu_pdp_st - proc_pdp_st) / rrd.stat_head -> pdp_step;
847 fprintf(stderr,"elapsed PDP steps: %lu\n", elapsed_pdp_st);
849 if (rra_step_cnt == NULL)
851 rra_step_cnt = (unsigned long *)
852 malloc((rrd.stat_head->rra_cnt)* sizeof(unsigned long));
855 for(i = 0, rra_start = rra_begin;
856 i < rrd.stat_head->rra_cnt;
857 rra_start += rrd.rra_def[i].row_cnt * rrd.stat_head -> ds_cnt * sizeof(rrd_value_t),
860 current_cf = cf_conv(rrd.rra_def[i].cf_nam);
861 start_pdp_offset = rrd.rra_def[i].pdp_cnt -
862 (proc_pdp_st / rrd.stat_head -> pdp_step) % rrd.rra_def[i].pdp_cnt;
863 if (start_pdp_offset <= elapsed_pdp_st) {
864 rra_step_cnt[i] = (elapsed_pdp_st - start_pdp_offset) /
865 rrd.rra_def[i].pdp_cnt + 1;
870 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL)
872 /* If this is a bulk update, we need to skip ahead in the seasonal
873 * arrays so that they will be correct for the next observed value;
874 * note that for the bulk update itself, no update will occur to
875 * DEVSEASONAL or SEASONAL; futhermore, HWPREDICT and DEVPREDICT will
877 if (rra_step_cnt[i] > 2)
879 /* skip update by resetting rra_step_cnt[i],
880 * note that this is not data source specific; this is due
881 * to the bulk update, not a DNAN value for the specific data
884 lookup_seasonal(&rrd,i,rra_start,rrd_file,elapsed_pdp_st,
885 &last_seasonal_coef);
886 lookup_seasonal(&rrd,i,rra_start,rrd_file,elapsed_pdp_st + 1,
890 /* periodically run a smoother for seasonal effects */
891 /* Need to use first cdp parameter buffer to track
892 * burnin (burnin requires a specific smoothing schedule).
893 * The CDP_init_seasonal parameter is really an RRA level,
894 * not a data source within RRA level parameter, but the rra_def
895 * is read only for rrd_update (not flushed to disk). */
896 iii = i*(rrd.stat_head -> ds_cnt);
897 if (rrd.cdp_prep[iii].scratch[CDP_init_seasonal].u_cnt
900 if (rrd.rra_ptr[i].cur_row + elapsed_pdp_st
901 > rrd.rra_def[i].row_cnt - 1) {
902 /* mark off one of the burnin cycles */
903 ++(rrd.cdp_prep[iii].scratch[CDP_init_seasonal].u_cnt);
907 /* someone has no doubt invented a trick to deal with this
908 * wrap around, but at least this code is clear. */
909 if (rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt >
910 rrd.rra_ptr[i].cur_row)
912 /* here elapsed_pdp_st = rra_step_cnt[i] because of 1-1
913 * mapping between PDP and CDP */
914 if (rrd.rra_ptr[i].cur_row + elapsed_pdp_st
915 >= rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt)
919 "schedule_smooth 1: cur_row %lu, elapsed_pdp_st %lu, smooth idx %lu\n",
920 rrd.rra_ptr[i].cur_row, elapsed_pdp_st,
921 rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt);
926 /* can't rely on negative numbers because we are working with
928 /* Don't need modulus here. If we've wrapped more than once, only
929 * one smooth is executed at the end. */
930 if (rrd.rra_ptr[i].cur_row + elapsed_pdp_st >= rrd.rra_def[i].row_cnt
931 && rrd.rra_ptr[i].cur_row + elapsed_pdp_st - rrd.rra_def[i].row_cnt
932 >= rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt)
936 "schedule_smooth 2: cur_row %lu, elapsed_pdp_st %lu, smooth idx %lu\n",
937 rrd.rra_ptr[i].cur_row, elapsed_pdp_st,
938 rrd.rra_def[i].par[RRA_seasonal_smooth_idx].u_cnt);
945 rra_current = ftell(rrd_file);
946 } /* if cf is DEVSEASONAL or SEASONAL */
948 if (rrd_test_error()) break;
950 /* update CDP_PREP areas */
951 /* loop over data soures within each RRA */
953 ii < rrd.stat_head->ds_cnt;
957 /* iii indexes the CDP prep area for this data source within the RRA */
958 iii=i*rrd.stat_head->ds_cnt+ii;
960 if (rrd.rra_def[i].pdp_cnt > 1) {
962 if (rra_step_cnt[i] > 0) {
963 /* If we are in this block, as least 1 CDP value will be written to
964 * disk, this is the CDP_primary_val entry. If more than 1 value needs
965 * to be written, then the "fill in" value is the CDP_secondary_val
967 if (isnan(pdp_temp[ii]))
969 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt += start_pdp_offset;
970 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = DNAN;
972 /* CDP_secondary value is the RRA "fill in" value for intermediary
973 * CDP data entries. No matter the CF, the value is the same because
974 * the average, max, min, and last of a list of identical values is
975 * the same, namely, the value itself. */
976 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = pdp_temp[ii];
979 if (rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt
980 > rrd.rra_def[i].pdp_cnt*
981 rrd.rra_def[i].par[RRA_cdp_xff_val].u_val)
983 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = DNAN;
984 /* initialize carry over */
985 if (current_cf == CF_AVERAGE) {
986 if (isnan(pdp_temp[ii])) {
987 rrd.cdp_prep[iii].scratch[CDP_val].u_val = DNAN;
989 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii] *
990 ((elapsed_pdp_st - start_pdp_offset) % rrd.rra_def[i].pdp_cnt);
993 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
996 rrd_value_t cum_val, cur_val;
997 switch (current_cf) {
999 cum_val = IFDNAN(rrd.cdp_prep[iii].scratch[CDP_val].u_val, 0.0);
1000 cur_val = IFDNAN(pdp_temp[ii],0.0);
1001 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val =
1002 (cum_val + cur_val * start_pdp_offset) /
1003 (rrd.rra_def[i].pdp_cnt
1004 -rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt);
1005 /* initialize carry over value */
1006 if (isnan(pdp_temp[ii])) {
1007 rrd.cdp_prep[iii].scratch[CDP_val].u_val = DNAN;
1009 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii] *
1010 ((elapsed_pdp_st - start_pdp_offset) % rrd.rra_def[i].pdp_cnt);
1014 cum_val = IFDNAN(rrd.cdp_prep[iii].scratch[CDP_val].u_val, -DINF);
1015 cur_val = IFDNAN(pdp_temp[ii],-DINF);
1017 if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val) &&
1018 isnan(pdp_temp[ii])) {
1020 "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!",
1025 if (cur_val > cum_val)
1026 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cur_val;
1028 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cum_val;
1029 /* initialize carry over value */
1030 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1033 cum_val = IFDNAN(rrd.cdp_prep[iii].scratch[CDP_val].u_val, DINF);
1034 cur_val = IFDNAN(pdp_temp[ii],DINF);
1036 if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val) &&
1037 isnan(pdp_temp[ii])) {
1039 "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!",
1044 if (cur_val < cum_val)
1045 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cur_val;
1047 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = cum_val;
1048 /* initialize carry over value */
1049 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1053 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = pdp_temp[ii];
1054 /* initialize carry over value */
1055 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1058 } /* endif meets xff value requirement for a valid value */
1059 /* initialize carry over CDP_unkn_pdp_cnt, this must after CDP_primary_val
1060 * is set because CDP_unkn_pdp_cnt is required to compute that value. */
1061 if (isnan(pdp_temp[ii]))
1062 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt =
1063 (elapsed_pdp_st - start_pdp_offset) % rrd.rra_def[i].pdp_cnt;
1065 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt = 0;
1066 } else /* rra_step_cnt[i] == 0 */
1069 if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val)) {
1070 fprintf(stderr,"schedule CDP_val update, RRA %lu DS %lu, DNAN\n",
1073 fprintf(stderr,"schedule CDP_val update, RRA %lu DS %lu, %10.2f\n",
1074 i,ii,rrd.cdp_prep[iii].scratch[CDP_val].u_val);
1077 if (isnan(pdp_temp[ii])) {
1078 rrd.cdp_prep[iii].scratch[CDP_unkn_pdp_cnt].u_cnt += elapsed_pdp_st;
1079 } else if (isnan(rrd.cdp_prep[iii].scratch[CDP_val].u_val))
1081 if (current_cf == CF_AVERAGE) {
1082 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii] *
1085 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1088 fprintf(stderr,"Initialize CDP_val for RRA %lu DS %lu: %10.2f\n",
1089 i,ii,rrd.cdp_prep[iii].scratch[CDP_val].u_val);
1092 switch (current_cf) {
1094 rrd.cdp_prep[iii].scratch[CDP_val].u_val += pdp_temp[ii] *
1098 if (pdp_temp[ii] < rrd.cdp_prep[iii].scratch[CDP_val].u_val)
1099 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1102 if (pdp_temp[ii] > rrd.cdp_prep[iii].scratch[CDP_val].u_val)
1103 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1107 rrd.cdp_prep[iii].scratch[CDP_val].u_val = pdp_temp[ii];
1112 } else { /* rrd.rra_def[i].pdp_cnt == 1 */
1113 if (elapsed_pdp_st > 2)
1115 switch (current_cf) {
1118 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val=pdp_temp[ii];
1119 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val=pdp_temp[ii];
1122 case CF_DEVSEASONAL:
1123 /* need to update cached seasonal values, so they are consistent
1124 * with the bulk update */
1125 /* WARNING: code relies on the fact that CDP_hw_last_seasonal and
1126 * CDP_last_deviation are the same. */
1127 rrd.cdp_prep[iii].scratch[CDP_hw_last_seasonal].u_val =
1128 last_seasonal_coef[ii];
1129 rrd.cdp_prep[iii].scratch[CDP_hw_seasonal].u_val =
1133 /* need to update the null_count and last_null_count.
1134 * even do this for non-DNAN pdp_temp because the
1135 * algorithm is not learning from batch updates. */
1136 rrd.cdp_prep[iii].scratch[CDP_null_count].u_cnt +=
1138 rrd.cdp_prep[iii].scratch[CDP_last_null_count].u_cnt +=
1142 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = DNAN;
1143 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = DNAN;
1146 /* do not count missed bulk values as failures */
1147 rrd.cdp_prep[iii].scratch[CDP_primary_val].u_val = 0;
1148 rrd.cdp_prep[iii].scratch[CDP_secondary_val].u_val = 0;
1149 /* need to reset violations buffer.
1150 * could do this more carefully, but for now, just
1151 * assume a bulk update wipes away all violations. */
1152 erase_violations(&rrd, iii, i);
1156 } /* endif rrd.rra_def[i].pdp_cnt == 1 */
1158 if (rrd_test_error()) break;
1160 } /* endif data sources loop */
1161 } /* end RRA Loop */
1163 /* this loop is only entered if elapsed_pdp_st < 3 */
1164 for (j = elapsed_pdp_st, scratch_idx = CDP_primary_val;
1165 j > 0 && j < 3; j--, scratch_idx = CDP_secondary_val)
1167 for(i = 0, rra_start = rra_begin;
1168 i < rrd.stat_head->rra_cnt;
1169 rra_start += rrd.rra_def[i].row_cnt * rrd.stat_head -> ds_cnt * sizeof(rrd_value_t),
1172 if (rrd.rra_def[i].pdp_cnt > 1) continue;
1174 current_cf = cf_conv(rrd.rra_def[i].cf_nam);
1175 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL)
1177 lookup_seasonal(&rrd,i,rra_start,rrd_file,
1178 elapsed_pdp_st + (scratch_idx == CDP_primary_val ? 1 : 2),
1180 rra_current = ftell(rrd_file);
1182 if (rrd_test_error()) break;
1183 /* loop over data soures within each RRA */
1185 ii < rrd.stat_head->ds_cnt;
1188 update_aberrant_CF(&rrd,pdp_temp[ii],current_cf,
1189 i*(rrd.stat_head->ds_cnt) + ii,i,ii,
1190 scratch_idx, seasonal_coef);
1192 } /* end RRA Loop */
1193 if (rrd_test_error()) break;
1194 } /* end elapsed_pdp_st loop */
1196 if (rrd_test_error()) break;
1198 /* Ready to write to disk */
1199 /* Move sequentially through the file, writing one RRA at a time.
1200 * Note this architecture divorces the computation of CDP with
1201 * flushing updated RRA entries to disk. */
1202 for(i = 0, rra_start = rra_begin;
1203 i < rrd.stat_head->rra_cnt;
1204 rra_start += rrd.rra_def[i].row_cnt * rrd.stat_head -> ds_cnt * sizeof(rrd_value_t),
1206 /* is there anything to write for this RRA? If not, continue. */
1207 if (rra_step_cnt[i] == 0) continue;
1209 /* write the first row */
1211 fprintf(stderr," -- RRA Preseek %ld\n",ftell(rrd_file));
1213 rrd.rra_ptr[i].cur_row++;
1214 if (rrd.rra_ptr[i].cur_row >= rrd.rra_def[i].row_cnt)
1215 rrd.rra_ptr[i].cur_row = 0; /* wrap around */
1216 /* positition on the first row */
1217 rra_pos_tmp = rra_start +
1218 (rrd.stat_head->ds_cnt)*(rrd.rra_ptr[i].cur_row)*sizeof(rrd_value_t);
1219 if(rra_pos_tmp != rra_current) {
1221 if(fseek(rrd_file, rra_pos_tmp, SEEK_SET) != 0){
1222 rrd_set_error("seek error in rrd");
1226 rra_current = rra_pos_tmp;
1230 fprintf(stderr," -- RRA Postseek %ld\n",ftell(rrd_file));
1232 scratch_idx = CDP_primary_val;
1233 if (pcdp_summary != NULL)
1235 rra_time = (current_time - current_time
1236 % (rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step))
1237 - ((rra_step_cnt[i]-1)*rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step);
1240 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1241 pcdp_summary, &rra_time, rrd_mmaped_file);
1243 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1244 pcdp_summary, &rra_time);
1246 if (rrd_test_error()) break;
1248 /* write other rows of the bulk update, if any */
1249 scratch_idx = CDP_secondary_val;
1250 for ( ; rra_step_cnt[i] > 1; rra_step_cnt[i]--)
1252 if (++rrd.rra_ptr[i].cur_row == rrd.rra_def[i].row_cnt)
1255 fprintf(stderr,"Wraparound for RRA %s, %lu updates left\n",
1256 rrd.rra_def[i].cf_nam, rra_step_cnt[i] - 1);
1259 rrd.rra_ptr[i].cur_row = 0;
1260 /* seek back to beginning of current rra */
1261 if (fseek(rrd_file, rra_start, SEEK_SET) != 0)
1263 rrd_set_error("seek error in rrd");
1267 fprintf(stderr," -- Wraparound Postseek %ld\n",ftell(rrd_file));
1269 rra_current = rra_start;
1271 if (pcdp_summary != NULL)
1273 rra_time = (current_time - current_time
1274 % (rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step))
1275 - ((rra_step_cnt[i]-2)*rrd.rra_def[i].pdp_cnt*rrd.stat_head->pdp_step);
1278 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1279 pcdp_summary, &rra_time, rrd_mmaped_file);
1281 pcdp_summary = write_RRA_row(&rrd, i, &rra_current, scratch_idx, rrd_file,
1282 pcdp_summary, &rra_time);
1286 if (rrd_test_error())
1290 /* break out of the argument parsing loop if error_string is set */
1291 if (rrd_test_error()){
1296 } /* endif a pdp_st has occurred */
1297 rrd.live_head->last_up = current_time;
1298 rrd.live_head->last_up_usec = current_time_usec;
1300 } /* function argument loop */
1302 if (seasonal_coef != NULL) free(seasonal_coef);
1303 if (last_seasonal_coef != NULL) free(last_seasonal_coef);
1304 if (rra_step_cnt != NULL) free(rra_step_cnt);
1305 rpnstack_free(&rpnstack);
1308 if (munmap(rrd_mmaped_file, rrd_filesize) == -1) {
1309 rrd_set_error("error writing(unmapping) file: %s", filename);
1312 /* if we got here and if there is an error and if the file has not been
1313 * written to, then close things up and return. */
1314 if (rrd_test_error()) {
1324 /* aargh ... that was tough ... so many loops ... anyway, its done.
1325 * we just need to write back the live header portion now*/
1327 if (fseek(rrd_file, (sizeof(stat_head_t)
1328 + sizeof(ds_def_t)*rrd.stat_head->ds_cnt
1329 + sizeof(rra_def_t)*rrd.stat_head->rra_cnt),
1331 rrd_set_error("seek rrd for live header writeback");
1342 if(fwrite( rrd.live_head,
1343 sizeof(live_head_t), 1, rrd_file) != 1){
1344 rrd_set_error("fwrite live_head to rrd");
1355 if(fwrite( &rrd.live_head->last_up,
1356 sizeof(time_t), 1, rrd_file) != 1){
1357 rrd_set_error("fwrite live_head to rrd");
1369 if(fwrite( rrd.pdp_prep,
1371 rrd.stat_head->ds_cnt, rrd_file) != rrd.stat_head->ds_cnt){
1372 rrd_set_error("ftwrite pdp_prep to rrd");
1382 if(fwrite( rrd.cdp_prep,
1384 rrd.stat_head->rra_cnt *rrd.stat_head->ds_cnt, rrd_file)
1385 != rrd.stat_head->rra_cnt *rrd.stat_head->ds_cnt){
1387 rrd_set_error("ftwrite cdp_prep to rrd");
1397 if(fwrite( rrd.rra_ptr,
1399 rrd.stat_head->rra_cnt,rrd_file) != rrd.stat_head->rra_cnt){
1400 rrd_set_error("fwrite rra_ptr to rrd");
1410 /* OK now close the files and free the memory */
1411 if(fclose(rrd_file) != 0){
1412 rrd_set_error("closing rrd");
1421 /* calling the smoothing code here guarantees at most
1422 * one smoothing operation per rrd_update call. Unfortunately,
1423 * it is possible with bulk updates, or a long-delayed update
1424 * for smoothing to occur off-schedule. This really isn't
1425 * critical except during the burning cycles. */
1426 if (schedule_smooth)
1428 rrd_file = fopen(filename,"rb+");
1429 rra_start = rra_begin;
1430 for (i = 0; i < rrd.stat_head -> rra_cnt; ++i)
1432 if (cf_conv(rrd.rra_def[i].cf_nam) == CF_DEVSEASONAL ||
1433 cf_conv(rrd.rra_def[i].cf_nam) == CF_SEASONAL)
1436 fprintf(stderr,"Running smoother for rra %ld\n",i);
1438 apply_smoother(&rrd,i,rra_start,rrd_file);
1439 if (rrd_test_error())
1442 rra_start += rrd.rra_def[i].row_cnt
1443 *rrd.stat_head->ds_cnt*sizeof(rrd_value_t);
1456 * get exclusive lock to whole file.
1457 * lock gets removed when we close the file
1459 * returns 0 on success
1462 LockRRD(FILE *rrdfile)
1464 int rrd_fd; /* File descriptor for RRD */
1467 rrd_fd = fileno(rrdfile);
1470 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
1473 if ( _fstat( rrd_fd, &st ) == 0 ) {
1474 rcstat = _locking ( rrd_fd, _LK_NBLCK, st.st_size );
1480 lock.l_type = F_WRLCK; /* exclusive write lock */
1481 lock.l_len = 0; /* whole file */
1482 lock.l_start = 0; /* start of file */
1483 lock.l_whence = SEEK_SET; /* end of file */
1485 rcstat = fcntl(rrd_fd, F_SETLK, &lock);
1495 *write_RRA_row (rrd_t *rrd, unsigned long rra_idx, unsigned long *rra_current,
1496 unsigned short CDP_scratch_idx,
1498 FILE UNUSED(*rrd_file),
1502 info_t *pcdp_summary, time_t *rra_time, void *rrd_mmaped_file)
1505 *write_RRA_row (rrd_t *rrd, unsigned long rra_idx, unsigned long *rra_current,
1506 unsigned short CDP_scratch_idx, FILE *rrd_file,
1507 info_t *pcdp_summary, time_t *rra_time)
1510 unsigned long ds_idx, cdp_idx;
1513 for (ds_idx = 0; ds_idx < rrd -> stat_head -> ds_cnt; ds_idx++)
1515 /* compute the cdp index */
1516 cdp_idx =rra_idx * (rrd -> stat_head->ds_cnt) + ds_idx;
1518 fprintf(stderr," -- RRA WRITE VALUE %e, at %ld CF:%s\n",
1519 rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val,ftell(rrd_file),
1520 rrd -> rra_def[rra_idx].cf_nam);
1522 if (pcdp_summary != NULL)
1524 iv.u_val = rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val;
1525 /* append info to the return hash */
1526 pcdp_summary = info_push(pcdp_summary,
1527 sprintf_alloc("[%d]RRA[%s][%lu]DS[%s]",
1528 *rra_time, rrd->rra_def[rra_idx].cf_nam,
1529 rrd->rra_def[rra_idx].pdp_cnt, rrd->ds_def[ds_idx].ds_nam),
1533 memcpy((char *)rrd_mmaped_file + *rra_current,
1534 &(rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val),
1535 sizeof(rrd_value_t));
1537 if(fwrite(&(rrd -> cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val),
1538 sizeof(rrd_value_t),1,rrd_file) != 1)
1540 rrd_set_error("writing rrd");
1544 *rra_current += sizeof(rrd_value_t);
1546 return (pcdp_summary);