master
c 459 lines 19.4 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "database/rrd.h"
4
5 #ifdef ENABLE_DBENGINE
6
7 static RRDHOST *dbengine_rrdhost_find_or_create(char *name) {
8 /* We don't want to drop metrics when generating load,
9 * we prefer to block data generation itself */
10
11 SYSTEM_TZ tz = system_tz_get();
12 RRDHOST *host = rrdhost_find_or_create(
13 name,
14 name,
15 name,
16 os_type,
17 tz.timezone,
18 tz.abbrev_timezone,
19 tz.utc_offset,
20 program_name,
21 NETDATA_VERSION,
22 nd_profile.update_every,
23 default_rrd_history_entries,
24 RRD_DB_MODE_DBENGINE,
25 health_plugin_enabled(),
26 stream_send.enabled,
27 stream_send.parents.destination,
28 stream_send.api_key,
29 stream_send.send_charts_matching,
30 stream_receive.replication.enabled,
31 stream_receive.replication.period,
32 stream_receive.replication.step,
33 NULL,
34 0
35 );
36 system_tz_free(&tz);
37 return host;
38 }
39
40 static inline void rrddim_set_by_pointer_fake_time(RRDDIM *rd, collected_number value, time_t now) {
41 rd->collector.last_collected_time.tv_sec = now;
42 rd->collector.last_collected_time.tv_usec = 0;
43 rrddim_set_collected_int(rd, value);
44 rrddim_set_updated(rd);
45
46 rd->collector.counter++;
47
48 collected_number v = (value >= 0) ? value : -value;
49 if(unlikely(v > rd->collector.collected.i.collected_value_max)) rrddim_set_collected_max_int(rd, v);
50 }
51
52 struct dbengine_chart_thread {
53 uv_thread_t thread;
54 RRDHOST *host;
55 char *chartname; /* Will be prefixed by type, e.g. "example_local1.", "example_local2." etc */
56 unsigned dset_charts; /* number of charts */
57 unsigned dset_dims; /* dimensions per chart */
58 unsigned chart_i; /* current chart offset */
59 time_t time_present; /* current virtual time of the benchmark */
60 volatile time_t time_max; /* latest timestamp of stored values */
61 unsigned history_seconds; /* how far back in the past to go */
62
63 volatile long done; /* initialize to 0, set to 1 to stop thread */
64 struct completion charts_initialized;
65 unsigned long errors, stored_metrics_nr; /* statistics */
66
67 RRDSET *st;
68 RRDDIM *rd[]; /* dset_dims elements */
69 };
70
71 collected_number generate_dbengine_chart_value(int chart_i, int dim_i, time_t time_current)
72 {
73 collected_number value;
74
75 value = ((collected_number)time_current) * (chart_i + 1);
76 value += ((collected_number)time_current) * (dim_i + 1);
77 value %= 1024LLU;
78
79 return value;
80 }
81
82 static void generate_dbengine_chart(void *arg)
83 {
84 fprintf(stderr, "%s() running...\n", __FUNCTION__ );
85 struct dbengine_chart_thread *thread_info = (struct dbengine_chart_thread *)arg;
86 RRDHOST *host = thread_info->host;
87 char *chartname = thread_info->chartname;
88 const unsigned DSET_DIMS = thread_info->dset_dims;
89 unsigned history_seconds = thread_info->history_seconds;
90 time_t time_present = thread_info->time_present;
91
92 unsigned j, update_every = 1;
93 RRDSET *st;
94 RRDDIM *rd[DSET_DIMS];
95 char name[RRD_ID_LENGTH_MAX + 1];
96 time_t time_current;
97
98 // create the chart
99 snprintfz(name, RRD_ID_LENGTH_MAX, "example_local%u", thread_info->chart_i + 1);
100 thread_info->st = st = rrdset_create(host, name, chartname, chartname, "example", NULL, chartname, chartname,
101 chartname, NULL, 1, update_every, RRDSET_TYPE_LINE);
102 for (j = 0 ; j < DSET_DIMS ; ++j) {
103 snprintfz(name, RRD_ID_LENGTH_MAX, "%s%u", chartname, j + 1);
104
105 thread_info->rd[j] = rd[j] = rrddim_add(st, name, NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
106 }
107 completion_mark_complete(&thread_info->charts_initialized);
108
109 // feed it with the test data
110 time_current = time_present - history_seconds;
111 for (j = 0 ; j < DSET_DIMS ; ++j) {
112 rd[j]->collector.last_collected_time.tv_sec =
113 st->last_collected_time.tv_sec = st->last_updated.tv_sec = time_current - update_every;
114 rd[j]->collector.last_collected_time.tv_usec =
115 st->last_collected_time.tv_usec = st->last_updated.tv_usec = 0;
116 }
117 for( ; !thread_info->done && time_current < time_present ; time_current += update_every) {
118 st->usec_since_last_update = USEC_PER_SEC * update_every;
119
120 for (j = 0; j < DSET_DIMS; ++j) {
121 collected_number value;
122
123 value = generate_dbengine_chart_value(thread_info->chart_i, j, time_current);
124 rrddim_set_by_pointer_fake_time(rd[j], value, time_current);
125 ++thread_info->stored_metrics_nr;
126 }
127 rrdset_done(st);
128 thread_info->time_max = time_current;
129 }
130 for (j = 0; j < DSET_DIMS; ++j) {
131 rrdeng_store_metric_finalize((rd[j])->tiers[0].sch);
132 }
133 }
134
135 void generate_dbengine_dataset(unsigned history_seconds)
136 {
137 fprintf(stderr, "%s() running...\n", __FUNCTION__ );
138 const int DSET_CHARTS = 16;
139 const int DSET_DIMS = 128;
140 const uint64_t EXPECTED_COMPRESSION_RATIO = 20;
141 RRDHOST *host = NULL;
142 struct dbengine_chart_thread **thread_info;
143 int i;
144 time_t time_present;
145
146 default_rrd_memory_mode = RRD_DB_MODE_DBENGINE;
147 default_rrdeng_page_cache_mb = 128;
148 // Worst case for uncompressible data
149 default_rrdeng_disk_quota_mb = (((uint64_t)DSET_DIMS * DSET_CHARTS) * sizeof(storage_number) * history_seconds) /
150 (1024 * 1024);
151 default_rrdeng_disk_quota_mb -= default_rrdeng_disk_quota_mb * EXPECTED_COMPRESSION_RATIO / 100;
152
153 nd_log_limits_unlimited();
154 fprintf(stderr, "Initializing localhost with hostname 'dbengine-dataset'");
155
156 host = dbengine_rrdhost_find_or_create("dbengine-dataset");
157 if (NULL == host)
158 return;
159
160 thread_info = mallocz(sizeof(*thread_info) * DSET_CHARTS);
161 for (i = 0 ; i < DSET_CHARTS ; ++i) {
162 thread_info[i] = mallocz(sizeof(*thread_info[i]) + sizeof(RRDDIM *) * DSET_DIMS);
163 }
164 fprintf(stderr, "\nRunning DB-engine workload generator\n");
165
166 time_present = now_realtime_sec();
167 for (i = 0 ; i < DSET_CHARTS ; ++i) {
168 thread_info[i]->host = host;
169 thread_info[i]->chartname = "random";
170 thread_info[i]->dset_charts = DSET_CHARTS;
171 thread_info[i]->chart_i = i;
172 thread_info[i]->dset_dims = DSET_DIMS;
173 thread_info[i]->history_seconds = history_seconds;
174 thread_info[i]->time_present = time_present;
175 thread_info[i]->time_max = 0;
176 thread_info[i]->done = 0;
177 completion_init(&thread_info[i]->charts_initialized);
178 fatal_assert(0 == uv_thread_create(&thread_info[i]->thread, generate_dbengine_chart, thread_info[i]));
179 completion_wait_for(&thread_info[i]->charts_initialized);
180 completion_destroy(&thread_info[i]->charts_initialized);
181 }
182 for (i = 0 ; i < DSET_CHARTS ; ++i) {
183 fatal_assert(0 == uv_thread_join(&thread_info[i]->thread));
184 }
185
186 for (i = 0 ; i < DSET_CHARTS ; ++i) {
187 freez(thread_info[i]);
188 }
189 freez(thread_info);
190 rrd_wrlock();
191 rrdhost_free___while_having_rrd_wrlock(localhost);
192 rrd_wrunlock();
193 }
194
195 struct dbengine_query_thread {
196 uv_thread_t thread;
197 RRDHOST *host;
198 char *chartname; /* Will be prefixed by type, e.g. "example_local1.", "example_local2." etc */
199 unsigned dset_charts; /* number of charts */
200 unsigned dset_dims; /* dimensions per chart */
201 time_t time_present; /* current virtual time of the benchmark */
202 unsigned history_seconds; /* how far back in the past to go */
203 volatile long done; /* initialize to 0, set to 1 to stop thread */
204 unsigned long errors, queries_nr, queried_metrics_nr; /* statistics */
205 uint8_t delete_old_data; /* if non zero then data are deleted when disk space is exhausted */
206
207 struct dbengine_chart_thread *chart_threads[]; /* dset_charts elements */
208 };
209
210 static void query_dbengine_chart(void *arg)
211 {
212 fprintf(stderr, "%s() running...\n", __FUNCTION__ );
213 struct dbengine_query_thread *thread_info = (struct dbengine_query_thread *)arg;
214 const int DSET_CHARTS = thread_info->dset_charts;
215 const int DSET_DIMS = thread_info->dset_dims;
216 time_t time_after, time_before, time_min, time_approx_min, time_max, duration;
217 int i, j, update_every = 1;
218 RRDSET *st;
219 RRDDIM *rd;
220 uint8_t same;
221 time_t time_now, time_retrieved, end_time;
222 collected_number generatedv;
223 NETDATA_DOUBLE value, expected;
224 struct storage_engine_query_handle seqh;
225 size_t value_errors = 0, time_errors = 0;
226
227 do {
228 // pick a chart and dimension
229 i = random() % DSET_CHARTS;
230 st = thread_info->chart_threads[i]->st;
231 j = random() % DSET_DIMS;
232 rd = thread_info->chart_threads[i]->rd[j];
233
234 time_min = thread_info->time_present - thread_info->history_seconds + 1;
235 time_max = thread_info->chart_threads[i]->time_max;
236
237 if (thread_info->delete_old_data) {
238 /* A time window of twice the disk space is sufficient for compression space savings of up to 50% */
239 time_approx_min = time_max - (default_rrdeng_disk_quota_mb * 2 * 1024 * 1024) /
240 (((uint64_t) DSET_DIMS * DSET_CHARTS) * sizeof(storage_number));
241 time_min = MAX(time_min, time_approx_min);
242 }
243 if (!time_max) {
244 time_before = time_after = time_min;
245 } else {
246 time_after = time_min + random() % (MAX(time_max - time_min, 1));
247 duration = random() % 3600;
248 time_before = MIN(time_after + duration, time_max); /* up to 1 hour queries */
249 }
250
251 storage_engine_query_init(rd->tiers[0].seb, rd->tiers[0].smh, &seqh, time_after, time_before, STORAGE_PRIORITY_NORMAL);
252 ++thread_info->queries_nr;
253 for (time_now = time_after ; time_now <= time_before ; time_now += update_every) {
254 generatedv = generate_dbengine_chart_value(i, j, time_now);
255 expected = unpack_storage_number(pack_storage_number((NETDATA_DOUBLE) generatedv, SN_DEFAULT_FLAGS));
256
257 if (unlikely(storage_engine_query_is_finished(&seqh))) {
258 if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
259 fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value " NETDATA_DOUBLE_FORMAT
260 ", found data gap, ### ERROR 12 ###\n",
261 rrdset_name(st), rrddim_name(rd), (unsigned long) time_now, expected);
262 ++thread_info->errors;
263 }
264 break;
265 }
266
267 STORAGE_POINT sp = storage_engine_query_next_metric(&seqh);
268 value = sp.sum;
269 time_retrieved = sp.start_time_s;
270 end_time = sp.end_time_s;
271
272 if (!netdata_double_isnumber(value)) {
273 if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
274 fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value " NETDATA_DOUBLE_FORMAT
275 ", found data gap, ### ERROR 13 ###\n",
276 rrdset_name(st), rrddim_name(rd), (unsigned long) time_now, expected);
277 ++thread_info->errors;
278 }
279 break;
280 }
281 ++thread_info->queried_metrics_nr;
282
283 same = (roundndd(value) == roundndd(expected)) ? 1 : 0;
284 if (!same) {
285 if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
286 if(!value_errors)
287 fprintf(stderr, " DB-engine stresstest %s/%s: at %lu secs, expecting value " NETDATA_DOUBLE_FORMAT
288 ", found " NETDATA_DOUBLE_FORMAT ", ### ERROR 14 ###\n",
289 rrdset_name(st), rrddim_name(rd), (unsigned long) time_now, expected, value);
290 value_errors++;
291 thread_info->errors++;
292 }
293 }
294 if (end_time != time_now) {
295 if (!thread_info->delete_old_data) { /* data validation only when we don't delete */
296 if(!time_errors)
297 fprintf(stderr,
298 " DB-engine stresstest %s/%s: at %lu secs, found timestamp %lu ### ERROR 15 ###\n",
299 rrdset_name(st), rrddim_name(rd), (unsigned long) time_now, (unsigned long) time_retrieved);
300 time_errors++;
301 thread_info->errors++;
302 }
303 }
304 }
305 storage_engine_query_finalize(&seqh);
306 } while(!thread_info->done);
307
308 if(value_errors)
309 fprintf(stderr, "%zu value errors encountered\n", value_errors);
310
311 if(time_errors)
312 fprintf(stderr, "%zu time errors encountered\n", time_errors);
313 }
314
315 void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsigned QUERY_THREADS,
316 unsigned RAMP_UP_SECONDS, unsigned PAGE_CACHE_MB, unsigned DISK_SPACE_MB)
317 {
318 fprintf(stderr, "%s() running...\n", __FUNCTION__ );
319 const unsigned DSET_DIMS = 128;
320 const uint64_t EXPECTED_COMPRESSION_RATIO = 20;
321 const unsigned HISTORY_SECONDS = 3600 * 24 * 365 * 50; /* 50 year of history */
322 RRDHOST *host = NULL;
323 struct dbengine_chart_thread **chart_threads;
324 struct dbengine_query_thread **query_threads;
325 unsigned i, j;
326 time_t time_start, test_duration;
327
328 nd_log_limits_unlimited();
329
330 if (!TEST_DURATION_SEC)
331 TEST_DURATION_SEC = 10;
332 if (!DSET_CHARTS)
333 DSET_CHARTS = 1;
334 if (!QUERY_THREADS)
335 QUERY_THREADS = 1;
336 if (PAGE_CACHE_MB < RRDENG_MIN_PAGE_CACHE_SIZE_MB)
337 PAGE_CACHE_MB = RRDENG_MIN_PAGE_CACHE_SIZE_MB;
338
339 default_rrd_memory_mode = RRD_DB_MODE_DBENGINE;
340 default_rrdeng_page_cache_mb = PAGE_CACHE_MB;
341 if (DISK_SPACE_MB) {
342 fprintf(stderr, "By setting disk space limit data are allowed to be deleted. "
343 "Data validation is turned off for this run.\n");
344 default_rrdeng_disk_quota_mb = DISK_SPACE_MB;
345 } else {
346 // Worst case for uncompressible data
347 default_rrdeng_disk_quota_mb =
348 (((uint64_t) DSET_DIMS * DSET_CHARTS) * sizeof(storage_number) * HISTORY_SECONDS) / (1024 * 1024);
349 default_rrdeng_disk_quota_mb -= default_rrdeng_disk_quota_mb * EXPECTED_COMPRESSION_RATIO / 100;
350 }
351
352 fprintf(stderr, "Initializing localhost with hostname 'dbengine-stress-test'\n");
353
354 (void)sql_init_meta_database(DB_CHECK_NONE, 1);
355 host = dbengine_rrdhost_find_or_create("dbengine-stress-test");
356 if (NULL == host)
357 return;
358
359 chart_threads = mallocz(sizeof(*chart_threads) * DSET_CHARTS);
360 for (i = 0 ; i < DSET_CHARTS ; ++i) {
361 chart_threads[i] = mallocz(sizeof(*chart_threads[i]) + sizeof(RRDDIM *) * DSET_DIMS);
362 }
363 query_threads = mallocz(sizeof(*query_threads) * QUERY_THREADS);
364 for (i = 0 ; i < QUERY_THREADS ; ++i) {
365 query_threads[i] = mallocz(sizeof(*query_threads[i]) + sizeof(struct dbengine_chart_thread *) * DSET_CHARTS);
366 }
367 fprintf(stderr, "\nRunning DB-engine stress test, %u seconds writers ramp-up time,\n"
368 "%u seconds of concurrent readers and writers, %u writer threads, %u reader threads,\n"
369 "%u MiB of page cache.\n",
370 RAMP_UP_SECONDS, TEST_DURATION_SEC, DSET_CHARTS, QUERY_THREADS, PAGE_CACHE_MB);
371
372 time_start = now_realtime_sec() + HISTORY_SECONDS; /* move history to the future */
373 for (i = 0 ; i < DSET_CHARTS ; ++i) {
374 chart_threads[i]->host = host;
375 chart_threads[i]->chartname = "random";
376 chart_threads[i]->dset_charts = DSET_CHARTS;
377 chart_threads[i]->chart_i = i;
378 chart_threads[i]->dset_dims = DSET_DIMS;
379 chart_threads[i]->history_seconds = HISTORY_SECONDS;
380 chart_threads[i]->time_present = time_start;
381 chart_threads[i]->time_max = 0;
382 chart_threads[i]->done = 0;
383 chart_threads[i]->errors = chart_threads[i]->stored_metrics_nr = 0;
384 completion_init(&chart_threads[i]->charts_initialized);
385 fatal_assert(0 == uv_thread_create(&chart_threads[i]->thread, generate_dbengine_chart, chart_threads[i]));
386 }
387 /* barrier so that subsequent queries can access valid chart data */
388 for (i = 0 ; i < DSET_CHARTS ; ++i) {
389 completion_wait_for(&chart_threads[i]->charts_initialized);
390 completion_destroy(&chart_threads[i]->charts_initialized);
391 }
392 sleep(RAMP_UP_SECONDS);
393 /* at this point data have already began being written to the database */
394 for (i = 0 ; i < QUERY_THREADS ; ++i) {
395 query_threads[i]->host = host;
396 query_threads[i]->chartname = "random";
397 query_threads[i]->dset_charts = DSET_CHARTS;
398 query_threads[i]->dset_dims = DSET_DIMS;
399 query_threads[i]->history_seconds = HISTORY_SECONDS;
400 query_threads[i]->time_present = time_start;
401 query_threads[i]->done = 0;
402 query_threads[i]->errors = query_threads[i]->queries_nr = query_threads[i]->queried_metrics_nr = 0;
403 for (j = 0 ; j < DSET_CHARTS ; ++j) {
404 query_threads[i]->chart_threads[j] = chart_threads[j];
405 }
406 query_threads[i]->delete_old_data = DISK_SPACE_MB ? 1 : 0;
407 fatal_assert(0 == uv_thread_create(&query_threads[i]->thread, query_dbengine_chart, query_threads[i]));
408 }
409 sleep(TEST_DURATION_SEC);
410 /* stop workload */
411 for (i = 0 ; i < DSET_CHARTS ; ++i) {
412 chart_threads[i]->done = 1;
413 }
414 for (i = 0 ; i < QUERY_THREADS ; ++i) {
415 query_threads[i]->done = 1;
416 }
417 for (i = 0 ; i < DSET_CHARTS ; ++i) {
418 assert(0 == uv_thread_join(&chart_threads[i]->thread));
419 }
420 for (i = 0 ; i < QUERY_THREADS ; ++i) {
421 assert(0 == uv_thread_join(&query_threads[i]->thread));
422 }
423 test_duration = now_realtime_sec() - (time_start - HISTORY_SECONDS);
424 if (!test_duration)
425 test_duration = 1;
426 fprintf(stderr, "\nDB-engine stress test finished in %lld seconds.\n", (long long)test_duration);
427 unsigned long stored_metrics_nr = 0;
428 for (i = 0 ; i < DSET_CHARTS ; ++i) {
429 stored_metrics_nr += chart_threads[i]->stored_metrics_nr;
430 }
431 unsigned long queried_metrics_nr = 0;
432 for (i = 0 ; i < QUERY_THREADS ; ++i) {
433 queried_metrics_nr += query_threads[i]->queried_metrics_nr;
434 }
435 fprintf(stderr, "%u metrics were stored (dataset size of %lu MiB) in %u charts by 1 writer thread per chart.\n",
436 DSET_CHARTS * DSET_DIMS, stored_metrics_nr * sizeof(storage_number) / (1024 * 1024), DSET_CHARTS);
437 fprintf(stderr, "Metrics were being generated per 1 emulated second and time was accelerated.\n");
438 fprintf(stderr, "%lu metric data points were queried by %u reader threads.\n", queried_metrics_nr, QUERY_THREADS);
439 fprintf(stderr, "Query starting time is randomly chosen from the beginning of the time-series up to the time of\n"
440 "the latest data point, and ending time from 1 second up to 1 hour after the starting time.\n");
441 fprintf(stderr, "Performance is %lld written data points/sec and %lld read data points/sec.\n",
442 (long long)(stored_metrics_nr / test_duration), (long long)(queried_metrics_nr / test_duration));
443
444 for (i = 0 ; i < DSET_CHARTS ; ++i) {
445 freez(chart_threads[i]);
446 }
447 freez(chart_threads);
448 for (i = 0 ; i < QUERY_THREADS ; ++i) {
449 freez(query_threads[i]);
450 }
451 freez(query_threads);
452 rrd_wrlock();
453 rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);
454 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);
455 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);
456 rrd_wrunlock();
457 }
458
459 #endif