@cryptotaxi247 / netdata-1 / commits / 70e868f5e

added mmap() support for all data, so that netdata can be restarted and continue with all data

Costa Tsaousis (ktsaou) committed May 11, 2014 at 18:16 UTC 70e868f5eda569ec13dda1cf88a6cf8c8e79e9d0
2 files changed +278 -60
netdata.c
+273 -60
@@ -35,6 +35,7 @@
35 #include <malloc.h>
36 #include <inttypes.h>
37 #include <dirent.h>
38 +#include <sys/mman.h>
39
40 // enabling this will detach the plugins from netdata
41 // each plugin will have its own process group
@@ -402,7 +403,7 @@ FILE *mypopen(const char *command, pid_t *pidptr)
403 // ignore all signals
404 for (i = 1 ; i < 65 ;i++) if(i != SIGSEGV) signal(i, SIG_DFL);
405
405 - fprintf(stderr, "executing command: '%s'\n", command);
406 + fprintf(stderr, "executing command: '%s' on pid %d.\n", command, getpid());
407 execl("/bin/sh", "sh", "-c", command, NULL);
408 exit(1);
409 }
@@ -526,8 +527,8 @@ int create_listen_socket(int port)
527 // ----------------------------------------------------------------------------
528 // CONFIG
529
529 -#define CONFIG_MAX_NAME 1024
530 -#define CONFIG_MAX_VALUE 4096
530 +#define CONFIG_MAX_NAME 100
531 +#define CONFIG_MAX_VALUE 1024
532 #define CONFIG_FILENAME "netdata.conf"
533 #define CONFIG_FILE_LINE_MAX 4096
534
@@ -923,6 +924,50 @@ static void strreverse(char* begin, char* end)
924 }
925
926
927 +// ----------------------------------------------------------------------------
928 +// mmap() wrapper
929 +
930 +void *mymmap(const char *filename, unsigned long size)
931 +{
932 + static int enabled = -1;
933 +
934 + if(enabled == -1) {
935 + char *cache_dir = config_get("global", "cache data in directory", "cache");
936 + if(strcmp(cache_dir, "none") == 0) enabled = 0;
937 + else enabled = 1;
938 + }
939 + if(!enabled) return NULL;
940 +
941 + int fd;
942 + void *mem = NULL;
943 +
944 + errno = 0;
945 + fd = open(filename, O_RDWR|O_CREAT|O_NOATIME, 0664);
946 + if(fd != -1) {
947 + if(lseek(fd, size, SEEK_SET) == (long)size) {
948 + if(write(fd, "", 1) == 1) {
949 +
950 + if(ftruncate(fd, size))
951 + error("Cannot truncate file '%s' to size %ld. Will use the larger file.", filename, size);
952 +
953 + mem = mmap(NULL, size, PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);
954 + if(mem) {
955 + if(madvise(mem, size, MADV_SEQUENTIAL|MADV_DONTFORK) != 0)
956 + error("Cannot advise the kernel about the memory usage of file '%s'.", filename);
957 + }
958 + }
959 + else error("Cannot write to file '%s' at position %ld.", filename, size);
960 + }
961 + else error("Cannot seek file '%s' to size %ld.", filename, size);
962 +
963 + close(fd);
964 + }
965 + else error("Cannot create/open file '%s'.", filename);
966 +
967 + return mem;
968 +}
969 +
970 +
971 // ----------------------------------------------------------------------------
972 // RRD STATS
973
@@ -943,8 +988,11 @@ typedef int32_t storage_number;
988 typedef uint32_t ustorage_number;
989 #define STORAGE_NUMBER_FORMAT "%d"
990
991 +#define RRD_STATS_MAGIC "NETDATA CACHE STATS FILE V001"
992 +#define RRD_DIMENSION_MAGIC "NETDATA CACHE DIMENSION FILE V002"
993
994 struct rrd_dimension {
995 + char magic[sizeof(RRD_DIMENSION_MAGIC) + 1];// our magic
996 char id[RRD_STATS_NAME_MAX + 1]; // the id of this dimension (for internal identification)
997 char *name; // the name of this dimension (as presented to user)
998
@@ -953,15 +1001,16 @@ struct rrd_dimension {
1001
1002 long entries; // how many entries this dimension has
1003 // this should be the same to the entries of the data set
1004 + long current_entry; // the entry that is currently being updated
1005
1006 int hidden; // if set to non zero, this dimension will not be sent to the client
1007 + int mapped; // 1 if the file is mapped
1008 + unsigned long memsize; // the memory allocated for this dimension
1009
1010 int algorithm;
1011 long multiplier;
1012 long divisor;
1013
963 - storage_number *values; // the array of values
964 -
1014 struct timeval last_updated; // when was this dimension last updated
1015
1016 calculated_number calculated_value;
@@ -971,27 +1020,33 @@ struct rrd_dimension {
1020 collected_number last_collected_value;
1021
1022 struct rrd_dimension *next; // linking of dimensions within the same data set
1023 +
1024 + storage_number values[]; // the array of values - THIS HAS TO BE THE LAST MEMBER
1025 };
1026 typedef struct rrd_dimension RRD_DIMENSION;
1027
1028 struct rrd_stats {
978 - pthread_mutex_t mutex;
979 -
980 - unsigned long counter; // the number of times we added values to this rrd
1029 + char magic[sizeof(RRD_STATS_MAGIC) + 1];// our magic
1030
1031 char id[RRD_STATS_NAME_MAX + 1]; // id of the data set
1032 char *name; // name of the data set
984 -
985 - unsigned long hash; // a simple hash on the id, to speed up searching
986 - // we first compare hashes, and only if the hashes are equal we do string comparisons
987 -
988 - unsigned long hash_name; // a simple hash on the name
1033 + char *cache_dir; // the directory to store dimension maps
1034
1035 char *type; // the type of graph RRD_TYPE_* (a category, for determining graphing options)
1036 char *family; // the family of this data set (for grouping them together)
1037 char *title; // title shown to user
1038 char *units; // units of measurement
1039
1040 + pthread_mutex_t mutex;
1041 + unsigned long counter; // the number of times we added values to this rrd
1042 +
1043 + int mapped; // if set to 1, this is memory mapped
1044 + unsigned long memsize; // how much mem we have allocated for this (without dimensions)
1045 +
1046 + unsigned long hash_name; // a simple hash on the name
1047 + unsigned long hash; // a simple hash on the id, to speed up searching
1048 + // we first compare hashes, and only if the hashes are equal we do string comparisons
1049 +
1050 long priority;
1051
1052 long entries; // total number of entries in the data set
@@ -1012,10 +1067,11 @@ struct rrd_stats {
1067 int isdetail; // if set, the data set should be considered as a detail of another
1068 // (the master data set should be the one that has the same family and is not detail)
1069
1015 - uint32_t *timediff; // the time in microseconds between each data entry
1070 RRD_DIMENSION *dimensions; // the actual data for every dimension
1071
1072 struct rrd_stats *next; // linking of rrd stats
1073 +
1074 + uint32_t timediff[]; // the time in microseconds between each data entry - HAS TO BE LAST
1075 };
1076 typedef struct rrd_stats RRD_STATS;
1077
@@ -1046,26 +1102,113 @@ void rrd_stats_set_name(RRD_STATS *st, const char *name)
1102 st->hash_name = simple_hash(st->name);
1103 }
1104
1049 -RRD_STATS *rrd_stats_create(const char *type, const char *id, const char *name, const char *family, const char *title, const char *units, long priority, int update_every, int chart_type)
1105 +char *rrd_stats_cache_dir(const char *id)
1106 {
1051 - RRD_STATS *st = NULL;
1107 + char *ret = NULL;
1108 +
1109 + static int enabled = 1;
1110 + static char *cache_dir = NULL;
1111 + if(!cache_dir) {
1112 + cache_dir = config_get("global", "cache data in directory", "cache");
1113 + if(strcmp(cache_dir, "none") == 0) enabled = 0;
1114 + }
1115 +
1116 + char b[FILENAME_MAX + 1];
1117 + char n[FILENAME_MAX + 1];
1118 + rrd_stats_strncpy_name(b, id, FILENAME_MAX);
1119 +
1120 + snprintf(n, FILENAME_MAX, "%s/%s", cache_dir, b);
1121 + ret = config_get(id, "cache_dir", n);
1122
1123 + if(enabled) {
1124 + int r = mkdir(ret, 0775);
1125 + if(r != 0 && errno != EEXIST)
1126 + error("Cannot create directory '%s'", ret);
1127 + }
1128 +
1129 + return ret;
1130 +}
1131 +
1132 +RRD_STATS *rrd_stats_create(const char *type, const char *id, const char *name, const char *family, const char *title, const char *units, long priority, int update_every, int chart_type)
1133 +{
1134 if(!id || !id[0]) {
1135 fatal("Cannot create rrd stats without an id.");
1136 return NULL;
1137 }
1138
1139 + char fullid[RRD_STATS_NAME_MAX + 1];
1140 + char fullfilename[FILENAME_MAX + 1];
1141 + RRD_STATS *st;
1142 +
1143 + snprintf(fullid, RRD_STATS_NAME_MAX, "%s.%s", type, id);
1144 +
1145 + long entries = config_get_number(fullid, "history", save_history);
1146 + if(entries < 5) entries = config_set_number(fullid, "history", 5);
1147 + if(entries > HISTORY_MAX) entries = config_set_number(fullid, "history", HISTORY_MAX);
1148 +
1149 + int enabled = config_get_boolean(fullid, "enabled", 1);
1150 + if(!enabled) entries = 5;
1151 +
1152 + unsigned long size = sizeof(RRD_STATS) + (entries * sizeof(uint32_t));
1153 + char *cache_dir = rrd_stats_cache_dir(fullid);
1154 +
1155 debug(D_RRD_STATS, "Creating RRD_STATS for '%s.%s'.", type, id);
1156
1060 - st = calloc(1, sizeof(RRD_STATS));
1061 - if(!st) {
1062 - fatal("Cannot allocate memory for RRD_STATS %s.%s", type, id);
1063 - return NULL;
1157 + snprintf(fullfilename, FILENAME_MAX, "%s/main.db", cache_dir);
1158 + st = (RRD_STATS *)mymmap(fullfilename, size);
1159 + if(st) {
1160 + if(strcmp(st->magic, RRD_STATS_MAGIC) != 0) {
1161 + errno = 0;
1162 + error("File %s does not have our version. Clearing it.", fullfilename);
1163 + bzero(st, size);
1164 + }
1165 + else if(strcmp(st->id, fullid) != 0) {
1166 + errno = 0;
1167 + error("File %s does not have our id. Unmapping it.", fullfilename);
1168 + munmap(st, size);
1169 + st = NULL;
1170 + }
1171 + else if(st->memsize != size || st->entries != entries) {
1172 + errno = 0;
1173 + error("File %s does not have the desired size. Clearing it.", fullfilename);
1174 + bzero(st, size);
1175 + }
1176 + else if(st->update_every != update_every) {
1177 + errno = 0;
1178 + error("File %s does not have the desired update frequency. Clearing it.", fullfilename);
1179 + bzero(st, size);
1180 + }
1181 }
1182
1066 - snprintf(st->id, RRD_STATS_NAME_MAX, "%s.%s", type, id);
1183 + if(st) {
1184 + st->name = NULL;
1185 + st->type = NULL;
1186 + st->family = NULL;
1187 + st->title = NULL;
1188 + st->units = NULL;
1189 + st->dimensions = NULL;
1190 + st->next = NULL;
1191 + st->mapped = 1;
1192 + }
1193 + else {
1194 + st = calloc(1, size);
1195 + if(!st) {
1196 + fatal("Cannot allocate memory for RRD_STATS %s.%s", type, id);
1197 + return NULL;
1198 + }
1199 + st->mapped = 0;
1200 + }
1201 + st->memsize = size;
1202 + st->entries = entries;
1203 + st->update_every = update_every;
1204 +
1205 + strcpy(st->magic, RRD_STATS_MAGIC);
1206 +
1207 + strcpy(st->id, fullid);
1208 st->hash = simple_hash(st->id);
1209
1210 + st->cache_dir = cache_dir;
1211 +
1212 st->family = config_get(st->id, "family", family?family:st->id);
1213 st->units = config_get(st->id, "units", units?units:"");
1214 st->type = config_get(st->id, "type", type);
@@ -1080,22 +1223,8 @@ RRD_STATS *rrd_stats_create(const char *type, const char *id, const char *name,
1223 st->title = config_get(st->id, "title", varvalue);
1224 }
1225
1083 - st->entries = config_get_number(st->id, "history", save_history);
1084 - if(st->entries < 5) st->entries = config_set_number(st->id, "history", 5);
1085 - if(st->entries > HISTORY_MAX) st->entries = config_set_number(st->id, "history", HISTORY_MAX);
1086 -
1226 st->priority = config_get_number(st->id, "priority", priority);
1088 - st->enabled = config_get_boolean(st->id, "enabled", 1);
1089 - if(!st->enabled) st->entries = 5;
1090 -
1091 - st->timediff = calloc(st->entries, sizeof(uint32_t));
1092 - if(!st->timediff) {
1093 - free(st);
1094 - fatal("Cannot allocate %lu entries of %lu bytes each for RRD_STATS.", st->entries, sizeof(struct timeval));
1095 - return NULL;
1096 - }
1097 -
1098 - st->update_every = update_every;
1227 + st->enabled = enabled;
1228
1229 pthread_mutex_init(&st->mutex, NULL);
1230 pthread_mutex_lock(&root_mutex);
@@ -1110,17 +1239,73 @@ RRD_STATS *rrd_stats_create(const char *type, const char *id, const char *name,
1239
1240 RRD_DIMENSION *rrd_stats_dimension_add(RRD_STATS *st, const char *id, const char *name, long multiplier, long divisor, int algorithm)
1241 {
1242 + char filename[FILENAME_MAX + 1];
1243 + char fullfilename[FILENAME_MAX + 1];
1244 +
1245 char varname[CONFIG_MAX_NAME + 1];
1114 - RRD_DIMENSION *rd = NULL;
1246 + RRD_DIMENSION *rd;
1247 + unsigned long size = sizeof(RRD_DIMENSION) + (st->entries * sizeof(storage_number));
1248
1249 debug(D_RRD_STATS, "Adding dimension '%s/%s'.", st->id, id);
1250
1118 - rd = calloc(1, sizeof(RRD_DIMENSION));
1119 - if(!rd) {
1120 - fatal("Cannot allocate RRD_DIMENSION %s/%s.", st->id, id);
1121 - return NULL;
1251 + rrd_stats_strncpy_name(filename, id, FILENAME_MAX);
1252 + snprintf(fullfilename, FILENAME_MAX, "%s/%s.db", st->cache_dir, filename);
1253 + rd = (RRD_DIMENSION *)mymmap(fullfilename, size);
1254 + if(rd) {
1255 + if(strcmp(rd->magic, RRD_DIMENSION_MAGIC) != 0) {
1256 + errno = 0;
1257 + error("File %s does not have our version. Clearing it.", fullfilename);
1258 + bzero(rd, size);
1259 + }
1260 + else if(rd->memsize != size) {
1261 + errno = 0;
1262 + error("File %s does not have the desired size. Clearing it.", fullfilename);
1263 + bzero(rd, size);
1264 + }
1265 + else if(rd->multiplier != multiplier) {
1266 + errno = 0;
1267 + error("File %s does not have the same multiplier. Clearing it.", fullfilename);
1268 + bzero(rd, size);
1269 + }
1270 + else if(rd->divisor != divisor) {
1271 + errno = 0;
1272 + error("File %s does not have the same divisor. Clearing it.", fullfilename);
1273 + bzero(rd, size);
1274 + }
1275 + else if(rd->algorithm != algorithm) {
1276 + errno = 0;
1277 + error("File %s does not have the same algorithm. Clearing it.", fullfilename);
1278 + bzero(rd, size);
1279 + }
1280 + else if(strcmp(rd->id, id) != 0) {
1281 + errno = 0;
1282 + error("File %s does not have our dimension id. Unmapping it.", fullfilename);
1283 + munmap(rd, size);
1284 + rd = NULL;
1285 + }
1286 + }
1287 +
1288 + if(rd) {
1289 + // we have a file mapped for rd
1290 + rd->mapped = 1;
1291 + rd->hidden = 0;
1292 + rd->next = NULL;
1293 + rd->name = NULL;
1294 + }
1295 + else {
1296 + // if we didn't manage to get a mmap'd dimension, just create one
1297 +
1298 + rd = calloc(1, size);
1299 + if(!rd) {
1300 + fatal("Cannot allocate RRD_DIMENSION %s/%s.", st->id, id);
1301 + return NULL;
1302 + }
1303 +
1304 + rd->mapped = 0;
1305 }
1306 + rd->memsize = size;
1307
1308 + strcpy(rd->magic, RRD_DIMENSION_MAGIC);
1309 strncpy(rd->id, id, RRD_STATS_NAME_MAX);
1310 rd->hash = simple_hash(rd->id);
1311
@@ -1139,13 +1324,6 @@ RRD_DIMENSION *rrd_stats_dimension_add(RRD_STATS *st, const char *id, const char
1324
1325 rd->entries = st->entries;
1326
1142 - rd->values = calloc(rd->entries, sizeof(storage_number));
1143 - if(!rd->values) {
1144 - free(rd);
1145 - fatal("Cannot allocate %lu entries for RRD_DIMENSION values.", rd->entries);
1146 - return NULL;
1147 - }
1148 -
1327 // append this dimension
1328 if(!st->dimensions)
1329 st->dimensions = rd;
@@ -1168,10 +1346,40 @@ void rrd_stats_dimension_set_name(RRD_STATS *st, RRD_DIMENSION *rd, const char *
1346 void rrd_stats_dimension_free(RRD_DIMENSION *rd)
1347 {
1348 if(rd->next) rrd_stats_dimension_free(rd->next);
1171 - debug(D_RRD_STATS, "Removing dimension '%s'.", rd->name);
1349 // free(rd->annotations);
1173 - free(rd->values);
1174 - free(rd);
1350 + if(rd->mapped) {
1351 + debug(D_RRD_STATS, "Unmapping dimension '%s'.", rd->name);
1352 + munmap(rd, rd->memsize);
1353 + }
1354 + else {
1355 + debug(D_RRD_STATS, "Removing dimension '%s'.", rd->name);
1356 + free(rd);
1357 + }
1358 +}
1359 +
1360 +void rrd_stats_free_all(void)
1361 +{
1362 + RRD_STATS *st;
1363 + pthread_mutex_lock(&root_mutex);
1364 +
1365 + for(st = root; st ;) {
1366 + RRD_STATS *next = st->next;
1367 +
1368 + pthread_mutex_lock(&st->mutex);
1369 +
1370 + if(st->dimensions) rrd_stats_dimension_free(st->dimensions);
1371 + st->dimensions = NULL;
1372 +
1373 + pthread_mutex_unlock(&st->mutex);
1374 +
1375 + if(st->mapped) munmap(st, st->memsize);
1376 + else free(st);
1377 +
1378 + st = next;
1379 + }
1380 + root = NULL;
1381 +
1382 + pthread_mutex_unlock(&root_mutex);
1383 }
1384
1385 RRD_STATS *rrd_stats_find(const char *id)
@@ -1842,12 +2050,11 @@ unsigned long rrd_stats_one_json(RRD_STATS *st, char *options, struct web_buffer
2050 , st->last_absolute_total
2051 );
2052
1845 - unsigned long memory = sizeof(RRD_STATS) + (sizeof(uint32_t) * st->entries);
2053 + unsigned long memory = st->memsize;
2054
2055 RRD_DIMENSION *rd;
2056 for(rd = st->dimensions; rd ; rd = rd->next) {
1849 - unsigned long rdmem = sizeof(RRD_DIMENSION) + (sizeof(storage_number) * rd->entries);
1850 - memory += rdmem;
2057 + memory += rd->memsize;
2058
2059 web_buffer_printf(wb,
2060 "\t\t\t\t{\n"
@@ -1877,7 +2084,7 @@ unsigned long rrd_stats_one_json(RRD_STATS *st, char *options, struct web_buffer
2084 , rd->calculated_value
2085 , rd->last_collected_value
2086 , rd->last_calculated_value
1880 - , rdmem
2087 + , rd->memsize
2088 , rd->next?",":""
2089 );
2090 }
@@ -5809,6 +6016,8 @@ struct plugind {
6016 int obsolete;
6017 int enabled;
6018
6019 + time_t started_t;
6020 +
6021 struct plugind *next;
6022 } *pluginsd_root = NULL;
6023
@@ -5933,11 +6142,13 @@ void *pluginsd_worker_thread(void *arg)
6142 if(st->debug) debug(D_PLUGINSD, "PLUGINSD: script %s is requesting a END on chart %s", cd->fullfilename, st->id);
6143
6144 unsigned long long usec_since_last_update = rrd_stats_done(st);
5936 - if(usec_since_last_update < (cd->update_every * 1000000ULL / 10)) {
5937 - error("PLUGINSD: script '%s' updates charts too frequently. Chart %s updated after %llu microseconds, expected %llu microseconds. Disabling it.", cd->fullfilename, st->id, usec_since_last_update, cd->update_every * 1000000ULL);
5938 - cd->enabled = 0;
5939 - kill(cd->pid, SIGTERM);
5940 - break;
6145 + if((time(NULL) - cd->started_t) > 10) {
6146 + if(usec_since_last_update < (cd->update_every * 1000000ULL / 10)) {
6147 + error("PLUGINSD: script '%s' (up for %lu secs) updates charts too frequently. Chart %s updated after %llu microseconds, expected %llu microseconds.", cd->fullfilename, time(NULL) - cd->started_t, st->id, usec_since_last_update, cd->update_every * 1000000ULL);
6148 + //cd->enabled = 0;
6149 + //kill(cd->pid, SIGTERM);
6150 + //break;
6151 + }
6152 }
6153
6154 st = NULL;
@@ -6176,6 +6387,7 @@ void *pluginsd_main(void *ptr)
6387
6388 cd->enabled = enabled;
6389 cd->update_every = config_get_number(cd->id, "update every", update_every);
6390 + cd->started_t = time(NULL);
6391
6392 char *def = "";
6393 snprintf(cd->cmd, PLUGINSD_CMD_MAX, "exec %s %d %s", cd->fullfilename, cd->update_every, config_get(cd->id, "command options", def));
@@ -6232,6 +6444,7 @@ void sig_handler(int signo)
6444 case SIGSEGV:
6445 error("Signaled exit (signal %d). Errno: %d (%s)", signo, errno, strerror(errno));
6446 kill_childs();
6447 + rrd_stats_free_all();
6448 exit(1);
6449 break;
6450
netdata.start
+5
@@ -37,6 +37,11 @@ then
37 mkdir conf.d || exit 1
38 fi
39
40 +if [ ! -d cache ]
41 +then
42 + mkdir cache || exit 1
43 +fi
44 +
45 if [ -f conf.d/netdata.conf ]
46 then
47 # find if debug is enabled in the config