1
// SPDX-License-Identifier: GPL-3.0-or-later
2
#include "rrdengine.h"
3
+#include "../storage_engine.h"
4
5
/* Default global database instance */
6
struct rrdengine_instance multidb_ctx_storage_tier0;
36
/* Default behaviour is to unblock data collection if the page cache is full of dirty pages by dropping metrics */
37
uint8_t rrdeng_drop_metrics_under_page_cache_pressure = 1;
38
39
+
40
+// ----------------------------------------------------------------------------
41
+// helpers
42
+
43
static inline struct rrdengine_instance *get_rrdeng_ctx_from_host(RRDHOST *host, int tier) {
44
if(tier < 0 || tier >= RRD_STORAGE_TIERS) tier = 0;
45
if(!host->storage_instance[tier]) tier = 0;
46
return (struct rrdengine_instance *)host->storage_instance[tier];
47
}
48
49
+// ----------------------------------------------------------------------------
50
+// metrics groups
51
+
52
+STORAGE_METRICS_GROUP *rrdeng_metrics_group_get(STORAGE_INSTANCE *db_instance __maybe_unused, uuid_t *uuid __maybe_unused) {
53
+ return callocz(1, sizeof(struct pg_alignment));
54
+}
55
+
56
+void rrdeng_metrics_group_release(STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg) {
57
+ if(!smg) return;
58
+
59
+ struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
60
+ struct pg_alignment *pa = (struct pg_alignment *)smg;
61
+ struct page_cache *pg_cache = &ctx->pg_cache;
62
+
63
+ uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
64
+ if(pa->refcount == 0)
65
+ freez(pa);
66
+ uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
67
+}
68
+
69
+// ----------------------------------------------------------------------------
70
+// metric handle for legacy dbs
71
+
72
/* This UUID is not unique across hosts */
45
-void rrdeng_generate_legacy_uuid(const char *dim_id, char *chart_id, uuid_t *ret_uuid)
73
+void rrdeng_generate_legacy_uuid(const char *dim_id, const char *chart_id, uuid_t *ret_uuid)
74
{
75
EVP_MD_CTX *evpctx;
76
unsigned char hash_value[EVP_MAX_MD_SIZE];
103
memcpy(ret_uuid, hash_value, sizeof(uuid_t));
104
}
105
78
-struct rrdeng_metric_handle {
79
- RRDDIM *rd;
80
- struct rrdengine_instance *ctx;
81
- uuid_t *rrdeng_uuid; // database engine metric UUID
82
- struct pg_cache_page_index *page_index;
83
-};
106
+STORAGE_METRIC_HANDLE *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const char *rd_id, const char *st_id, STORAGE_METRICS_GROUP *smg) {
107
+ uuid_t legacy_uuid;
108
+ rrdeng_generate_legacy_uuid(rd_id, st_id, &legacy_uuid);
109
+ return rrdeng_metric_get(db_instance, &legacy_uuid, smg);
110
+}
111
+
112
+// ----------------------------------------------------------------------------
113
+// metric handle
114
+
115
+void rrdeng_metric_release(STORAGE_METRIC_HANDLE *db_metric_handle) {
116
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
117
+ struct rrdengine_instance *ctx = page_index->ctx;
118
+ struct page_cache *pg_cache = &ctx->pg_cache;
119
+
120
+ uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
121
+ page_index->refcount--;
122
+ if(page_index->alignment && page_index->refcount == 0) {
123
+ page_index->alignment->refcount--;
124
+ page_index->alignment = NULL;
125
+ }
126
+ uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
127
85
-void rrdeng_metric_free(STORAGE_METRIC_HANDLE *db_metric_handle) {
128
freez(db_metric_handle);
129
}
130
89
-STORAGE_METRIC_HANDLE *rrdeng_metric_init(RRDDIM *rd, STORAGE_INSTANCE *db_instance) {
131
+STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg) {
132
struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
91
- struct page_cache *pg_cache;
92
- uuid_t legacy_uuid;
93
- uuid_t multihost_legacy_uuid;
94
- Pvoid_t *PValue;
133
+ struct pg_alignment *pa = (struct pg_alignment *)smg;
134
+ struct page_cache *pg_cache = &ctx->pg_cache;
135
struct pg_cache_page_index *page_index = NULL;
96
- int is_multihost_child = 0;
97
- RRDHOST *host = rd->rrdset->rrdhost;
98
-
99
- pg_cache = &ctx->pg_cache;
100
-
101
- rrdeng_generate_legacy_uuid(rrddim_id(rd), (char *)rrdset_id(rd->rrdset), &legacy_uuid);
102
- if (host != localhost && is_storage_engine_shared((STORAGE_INSTANCE *)ctx))
103
- is_multihost_child = 1;
136
137
uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
106
- PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, &legacy_uuid, sizeof(uuid_t));
138
+ Pvoid_t *PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, uuid, sizeof(uuid_t));
139
if (likely(NULL != PValue)) {
140
page_index = *PValue;
141
+ page_index->refcount++;
142
+
143
+ if(pa) {
144
+ if(page_index->alignment && page_index->alignment != pa)
145
+ fatal("DBENGINE: page_index has a different alignment.");
146
+
147
+ if(!page_index->alignment) {
148
+ page_index->alignment = pa;
149
+ pa->refcount++;
150
+ }
151
+ }
152
}
153
uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
111
- if (is_multihost_child || NULL == PValue) {
112
- /* First time we see the legacy UUID or metric belongs to child host in multi-host DB.
113
- * Drop legacy support, normal path */
114
-
115
- uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
116
- PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, &rd->metric_uuid, sizeof(uuid_t));
117
- if (likely(NULL != PValue)) {
118
- page_index = *PValue;
119
- }
120
- uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
121
- if (NULL == PValue) {
122
- uv_rwlock_wrlock(&pg_cache->metrics_index.lock);
123
- PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, &rd->metric_uuid, sizeof(uuid_t), PJE0);
124
- fatal_assert(NULL == *PValue); /* TODO: figure out concurrency model */
125
- *PValue = page_index = create_page_index(&rd->metric_uuid);
126
- page_index->prev = pg_cache->metrics_index.last_page_index;
127
- pg_cache->metrics_index.last_page_index = page_index;
128
- uv_rwlock_wrunlock(&pg_cache->metrics_index.lock);
129
- }
130
- } else {
131
- /* There are legacy UUIDs in the database, implement backward compatibility */
154
133
- rrdeng_convert_legacy_uuid_to_multihost(rd->rrdset->rrdhost->machine_guid, &legacy_uuid,
134
- &multihost_legacy_uuid);
155
+ return (STORAGE_METRIC_HANDLE *)page_index;
156
+}
157
+
158
+STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg) {
159
+ internal_fatal(!db_instance, "DBENGINE: db_instance is NULL");
160
+
161
+ struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
162
+ struct pg_alignment *pa = (struct pg_alignment *)smg;
163
+ struct pg_cache_page_index *page_index;
164
+ struct page_cache *pg_cache = &ctx->pg_cache;
165
+
166
+ uv_rwlock_wrlock(&pg_cache->metrics_index.lock);
167
+ Pvoid_t *PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, uuid, sizeof(uuid_t), PJE0);
168
+ fatal_assert(NULL == *PValue); /* TODO: figure out concurrency model */
169
+ *PValue = page_index = create_page_index(uuid, ctx);
170
+ page_index->prev = pg_cache->metrics_index.last_page_index;
171
+ pg_cache->metrics_index.last_page_index = page_index;
172
+ page_index->alignment = pa;
173
+ if(pa)
174
+ pa->refcount++;
175
+ uv_rwlock_wrunlock(&pg_cache->metrics_index.lock);
176
+
177
+ return (STORAGE_METRIC_HANDLE *)page_index;
178
+}
179
136
- int need_to_store = uuid_compare(rd->metric_uuid, multihost_legacy_uuid);
180
+STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg) {
181
+ STORAGE_METRIC_HANDLE *db_metric_handle;
182
138
- uuid_copy(rd->metric_uuid, multihost_legacy_uuid);
183
+ db_metric_handle = rrdeng_metric_get(db_instance, &rd->metric_uuid, smg);
184
+ if(!db_metric_handle) {
185
+ db_metric_handle = rrdeng_metric_get_legacy(db_instance, rrddim_id(rd), rrdset_id(rd->rrdset), smg);
186
+ if(db_metric_handle) {
187
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
188
+ uuid_copy(rd->metric_uuid, page_index->id);
189
+ }
190
+ }
191
+ if(!db_metric_handle)
192
+ db_metric_handle = rrdeng_metric_create(db_instance, &rd->metric_uuid, smg);
193
140
- if (unlikely(need_to_store && !ctx->tier))
141
- (void)sql_store_dimension(&rd->metric_uuid, &rd->rrdset->chart_uuid, rrddim_id(rd), rrddim_name(rd), rd->multiplier, rd->divisor, rd->algorithm);
194
+#ifdef NETDATA_INTERNAL_CHECKS
195
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
196
+ if(uuid_compare(rd->metric_uuid, page_index->id) != 0) {
197
+ char uuid1[UUID_STR_LEN + 1];
198
+ char uuid2[UUID_STR_LEN + 1];
199
+
200
+ uuid_unparse(rd->metric_uuid, uuid1);
201
+ uuid_unparse(page_index->id, uuid2);
202
+ fatal("DBENGINE: uuids do not match, asked for metric '%s', but got page_index of metric '%s'", uuid1, uuid2);
203
}
204
144
- struct rrdeng_metric_handle *mh = mallocz(sizeof(struct rrdeng_metric_handle));
145
- mh->rd = rd;
146
- mh->ctx = ctx;
147
- mh->rrdeng_uuid = &page_index->id;
148
- mh->page_index = page_index;
149
- return (STORAGE_METRIC_HANDLE *)mh;
205
+ struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
206
+ if(page_index->ctx != ctx)
207
+ fatal("DBENGINE: mixed up rrdengine instances, asked for metric from %p, got from %p", ctx, page_index->ctx);
208
+#endif
209
+
210
+ return db_metric_handle;
211
}
212
213
+
214
+// ----------------------------------------------------------------------------
215
+// collect ops
216
+
217
/*
218
* Gets a handle for storing metrics to the database.
219
* The handle must be released with rrdeng_store_metric_final().
220
*/
156
-STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle) {
157
- struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)db_metric_handle;
158
-
221
+STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every) {
222
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
223
struct rrdeng_collect_handle *handle;
160
- struct pg_cache_page_index *page_index;
224
+
225
+ if(!page_index->alignment)
226
+ fatal("DBENGINE: metric group is required for collect operations");
227
228
handle = callocz(1, sizeof(struct rrdeng_collect_handle));
163
- handle->metric_handle = metric_handle;
164
- handle->ctx = metric_handle->ctx;
229
+ handle->page_index = page_index;
230
handle->descr = NULL;
231
handle->unaligned_page = 0;
232
+ page_index->latest_update_every_s = update_every;
233
168
- page_index = metric_handle->page_index;
234
uv_rwlock_wrlock(&page_index->lock);
235
++page_index->writers;
236
uv_rwlock_wrunlock(&page_index->lock);
278
void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_handle) {
279
struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
280
// struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)handle->metric_handle;
216
- struct rrdengine_instance *ctx = handle->ctx;
281
+ struct rrdengine_instance *ctx = handle->page_index->ctx;
282
struct rrdeng_page_descr *descr = handle->descr;
283
284
if (unlikely(!ctx)) return;
291
292
page_is_empty = page_has_only_empty_metrics(descr);
293
if (page_is_empty) {
229
- debug(D_RRDENGINE, "Page has empty metrics only, deleting:");
230
- if (unlikely(debug_flags & D_RRDENGINE))
231
- print_page_cache_descr(descr);
294
+ print_page_cache_descr(descr, "Page has empty metrics only, deleting", true);
295
pg_cache_put(ctx, descr);
296
pg_cache_punch_hole(ctx, descr, 1, 0, NULL);
297
} else
304
handle->descr = NULL;
305
}
306
244
-void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle,
245
- usec_t point_in_time,
307
+static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection_handle,
308
+ usec_t point_in_time_ut,
309
NETDATA_DOUBLE n,
310
NETDATA_DOUBLE min_value,
311
NETDATA_DOUBLE max_value,
314
SN_FLAGS flags)
315
{
316
struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
254
- struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)handle->metric_handle;
255
- struct rrdengine_instance *ctx = handle->ctx;
317
+ struct pg_cache_page_index *page_index = handle->page_index;
318
+ struct rrdengine_instance *ctx = handle->page_index->ctx;
319
struct page_cache *pg_cache = &ctx->pg_cache;
320
struct rrdeng_page_descr *descr = handle->descr;
258
- RRDDIM *rd = metric_handle->rd;
321
322
void *page;
323
uint8_t must_flush_unaligned_page = 0, perfect_page_alignment = 0;
325
if (descr) {
326
/* Make alignment decisions */
327
266
- if (descr->page_length == rd->rrdset->rrddim_page_alignment) {
328
+#ifdef NETDATA_INTERNAL_CHECKS
329
+ if(descr->end_time_ut + page_index->latest_update_every_s * USEC_PER_SEC != point_in_time_ut) {
330
+ char buffer[200 + 1];
331
+ snprintfz(buffer, 200,
332
+ "metrics collected are %s, end_time_ut = %llu, point_in_time_ut = %llu, update_every = %u, delta = %llu",
333
+ (point_in_time_ut / USEC_PER_SEC - descr->end_time_ut / USEC_PER_SEC > page_index->latest_update_every_s)?"far apart":"not aligned",
334
+ descr->end_time_ut / USEC_PER_SEC,
335
+ point_in_time_ut / USEC_PER_SEC,
336
+ page_index->latest_update_every_s,
337
+ point_in_time_ut / USEC_PER_SEC - descr->end_time_ut / USEC_PER_SEC);
338
+ print_page_cache_descr(descr, buffer, false);
339
+ }
340
+#endif
341
+
342
+ if (descr->page_length == page_index->alignment->page_length) {
343
/* this is the leading dimension that defines chart alignment */
344
perfect_page_alignment = 1;
345
}
346
/* is the metric far enough out of alignment with the others? */
271
- if (unlikely(descr->page_length + PAGE_POINT_SIZE_BYTES(descr) < rd->rrdset->rrddim_page_alignment)) {
347
+ if (unlikely(descr->page_length + PAGE_POINT_SIZE_BYTES(descr) < page_index->alignment->page_length)) {
348
handle->unaligned_page = 1;
273
- debug(D_RRDENGINE, "Metric page is not aligned with chart:");
274
- if (unlikely(debug_flags & D_RRDENGINE))
275
- print_page_cache_descr(descr);
349
+ print_page_cache_descr(descr, "Metric page is not aligned with chart", true);
350
}
351
if (unlikely(handle->unaligned_page &&
352
/* did the other metrics change page? */
279
- rd->rrdset->rrddim_page_alignment <= PAGE_POINT_SIZE_BYTES(descr))) {
280
- debug(D_RRDENGINE, "Flushing unaligned metric page.");
353
+ page_index->alignment->page_length <= PAGE_POINT_SIZE_BYTES(descr))) {
354
+ print_page_cache_descr(descr, "must_flush_unaligned_page = 1", true);
355
must_flush_unaligned_page = 1;
356
handle->unaligned_page = 0;
357
}
359
if (unlikely(NULL == descr ||
360
descr->page_length + PAGE_POINT_SIZE_BYTES(descr) > RRDENG_BLOCK_SIZE ||
361
must_flush_unaligned_page)) {
288
- rrdeng_store_metric_flush_current_page(collection_handle);
362
290
- page = rrdeng_create_page(ctx, &metric_handle->page_index->id, &descr);
363
+ if(descr) {
364
+ print_page_cache_descr(descr, "flushing metric", true);
365
+ rrdeng_store_metric_flush_current_page(collection_handle);
366
+ }
367
+
368
+ page = rrdeng_create_page(ctx, &page_index->id, &descr);
369
fatal_assert(page);
370
371
+ descr->update_every_s = page_index->latest_update_every_s;
372
handle->descr = descr;
373
374
handle->page_correlation_id = rrd_atomic_fetch_add(&pg_cache->committed_page_index.latest_corr_id, 1);
375
297
- if (0 == rd->rrdset->rrddim_page_alignment) {
376
+ if (0 == page_index->alignment->page_length) {
377
/* this is the leading dimension that defines chart alignment */
378
perfect_page_alignment = 1;
379
}
408
break;
409
}
410
332
- pg_cache_atomic_set_pg_info(descr, point_in_time, descr->page_length + PAGE_POINT_SIZE_BYTES(descr));
411
+ pg_cache_atomic_set_pg_info(descr, point_in_time_ut, descr->page_length + PAGE_POINT_SIZE_BYTES(descr));
412
413
if (perfect_page_alignment)
335
- rd->rrdset->rrddim_page_alignment = descr->page_length;
336
- if (unlikely(INVALID_TIME == descr->start_time)) {
414
+ page_index->alignment->page_length = descr->page_length;
415
+ if (unlikely(INVALID_TIME == descr->start_time_ut)) {
416
unsigned long new_metric_API_producers, old_metric_API_max_producers, ret_metric_API_max_producers;
338
- descr->start_time = point_in_time;
417
+ descr->start_time_ut = point_in_time_ut;
418
419
new_metric_API_producers = rrd_atomic_add_fetch(&ctx->stats.metric_API_producers, 1);
420
while (unlikely(new_metric_API_producers > (old_metric_API_max_producers = ctx->metric_API_max_producers))) {
428
}
429
}
430
352
- pg_cache_insert(ctx, metric_handle->page_index, descr);
431
+ pg_cache_insert(ctx, page_index, descr);
432
} else {
354
- pg_cache_add_new_metric_time(metric_handle->page_index, descr);
433
+ pg_cache_add_new_metric_time(page_index, descr);
434
+ }
435
+
436
+// {
437
+// unsigned char u[16] = { 0x0C, 0x0A, 0x40, 0xD6, 0x2A, 0x43, 0x4A, 0x7C, 0x95, 0xF7, 0xD1, 0x1E, 0x0C, 0x9E, 0x8A, 0xE7 };
438
+// if(uuid_compare(u, page_index->id) == 0) {
439
+// char buffer[100];
440
+// snprintfz(buffer, 100, "store system.cpu, collect:%u, page_index first:%u, last:%u",
441
+// (uint32_t)(point_in_time / USEC_PER_SEC),
442
+// (uint32_t)(page_index->oldest_time / USEC_PER_SEC),
443
+// (uint32_t)(page_index->latest_time / USEC_PER_SEC));
444
+//
445
+// print_page_cache_descr(descr, buffer, false);
446
+// }
447
+// }
448
+}
449
+
450
+void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle,
451
+ usec_t point_in_time_ut,
452
+ NETDATA_DOUBLE n,
453
+ NETDATA_DOUBLE min_value,
454
+ NETDATA_DOUBLE max_value,
455
+ uint16_t count,
456
+ uint16_t anomaly_count,
457
+ SN_FLAGS flags)
458
+{
459
+ struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
460
+ struct pg_cache_page_index *page_index = handle->page_index;
461
+ struct rrdeng_page_descr *descr = handle->descr;
462
+
463
+ if(likely(descr)) {
464
+ usec_t last_point_in_time_ut = descr->end_time_ut;
465
+ usec_t update_every_ut = page_index->latest_update_every_s * USEC_PER_SEC;
466
+ size_t points_gap = (point_in_time_ut <= last_point_in_time_ut) ?
467
+ (size_t)0 :
468
+ (size_t)((point_in_time_ut - last_point_in_time_ut) / update_every_ut);
469
+
470
+ if(unlikely(points_gap != 1)) {
471
+ if (unlikely(points_gap <= 0)) {
472
+ time_t now = now_realtime_sec();
473
+ static __thread size_t counter = 0;
474
+ static __thread time_t last_time_logged = 0;
475
+ counter++;
476
+
477
+ if(now - last_time_logged > 600) {
478
+ error("DBENGINE: collected point is in the past (repeated %zu times in the last %zu secs). Ignoring these data collection points.",
479
+ counter, (size_t)(last_time_logged?(now - last_time_logged):0));
480
+
481
+ last_time_logged = now;
482
+ counter = 0;
483
+ }
484
+ return;
485
+ }
486
+
487
+ size_t point_size = PAGE_POINT_SIZE_BYTES(descr);
488
+ size_t page_size_in_points = RRDENG_BLOCK_SIZE / point_size;
489
+ size_t used_points = descr->page_length / point_size;
490
+ size_t remaining_points_in_page = page_size_in_points - used_points;
491
+
492
+ bool new_point_is_aligned = true;
493
+ if(unlikely((point_in_time_ut - last_point_in_time_ut) / points_gap != update_every_ut))
494
+ new_point_is_aligned = false;
495
+
496
+ if(unlikely(points_gap > remaining_points_in_page || !new_point_is_aligned)) {
497
+// char buffer[200];
498
+// snprintfz(buffer, 200, "data collection skipped %zu points, last stored point %llu, new point %llu, update every %d. Cutting page.",
499
+// points_gap, last_point_in_time_ut / USEC_PER_SEC, point_in_time_ut / USEC_PER_SEC, page_index->latest_update_every_s);
500
+// print_page_cache_descr(descr, buffer, false);
501
+
502
+ rrdeng_store_metric_flush_current_page(collection_handle);
503
+ }
504
+ else {
505
+// char buffer[200];
506
+// snprintfz(buffer, 200, "data collection skipped %zu points, last stored point %llu, new point %llu, update every %d. Filling the gap.",
507
+// points_gap, last_point_in_time_ut / USEC_PER_SEC, point_in_time_ut / USEC_PER_SEC, page_index->latest_update_every_s);
508
+// print_page_cache_descr(descr, buffer, false);
509
+
510
+ // loop to fill the gap
511
+ usec_t step_ut = page_index->latest_update_every_s * USEC_PER_SEC;
512
+ usec_t last_point_filled_ut = last_point_in_time_ut + step_ut;
513
+
514
+ while (last_point_filled_ut < point_in_time_ut) {
515
+ rrdeng_store_metric_next_internal(
516
+ collection_handle, last_point_filled_ut, NAN, NAN, NAN,
517
+ 1, 0, SN_EMPTY_SLOT);
518
+
519
+ last_point_filled_ut += step_ut;
520
+ }
521
+ }
522
+ }
523
}
524
+
525
+ rrdeng_store_metric_next_internal(collection_handle, point_in_time_ut, n, min_value, max_value, count, anomaly_count, flags);
526
}
527
528
+
529
/*
530
* Releases the database reference from the handle for storing metrics.
531
* Returns 1 if it's safe to delete the dimension.
532
*/
533
int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) {
534
struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
364
- struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)handle->metric_handle;
365
- struct pg_cache_page_index *page_index = metric_handle->page_index;
535
+ struct pg_cache_page_index *page_index = handle->page_index;
536
537
uint8_t can_delete_metric = 0;
538
547
return can_delete_metric;
548
}
549
550
+void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *collection_handle, int update_every) {
551
+ struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
552
+ struct pg_cache_page_index *page_index = handle->page_index;
553
+ rrdeng_store_metric_flush_current_page(collection_handle);
554
+ uv_rwlock_rdlock(&page_index->lock);
555
+ page_index->latest_update_every_s = update_every;
556
+ uv_rwlock_rdunlock(&page_index->lock);
557
+}
558
+
559
+// ----------------------------------------------------------------------------
560
+// query ops
561
+
562
//static inline uint32_t *pginfo_to_dt(struct rrdeng_page_info *page_info)
563
//{
564
// return (uint32_t *)&page_info->scratch[0];
573
* Gets a handle for loading metrics from the database.
574
* The handle must be released with rrdeng_load_metric_final().
575
*/
394
-void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct rrddim_query_handle *rrdimm_handle, time_t start_time, time_t end_time, TIER_QUERY_FETCH tier_query_fetch_type)
576
+void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct rrddim_query_handle *rrdimm_handle, time_t start_time_s, time_t end_time_s)
577
{
396
- struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)db_metric_handle;
397
- struct rrdengine_instance *ctx = metric_handle->ctx;
398
- RRDDIM *rd = metric_handle->rd;
578
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
579
+ struct rrdengine_instance *ctx = page_index->ctx;
580
581
// fprintf(stderr, "%s: %s/%s start time %ld, end time %ld\n", __FUNCTION__ , rd->rrdset->name, rd->name, start_time, end_time);
582
583
struct rrdeng_query_handle *handle;
584
unsigned pages_nr;
585
405
- rrdimm_handle->start_time = start_time;
406
- rrdimm_handle->end_time = end_time;
586
+ if(!page_index->latest_update_every_s)
587
+ page_index->latest_update_every_s = default_rrd_update_every;
588
+
589
+ rrdimm_handle->start_time_s = start_time_s;
590
+ rrdimm_handle->end_time_s = end_time_s;
591
592
handle = callocz(1, sizeof(struct rrdeng_query_handle));
409
- handle->next_page_time = start_time;
410
- handle->now = start_time;
411
- handle->tier_query_fetch_type = tier_query_fetch_type;
412
- // TODO we should store the dt of each page in each page
413
- // this will produce wrong values for dt in case the user changes
414
- // the update every of the charts or the tier grouping iterations
415
- handle->dt_sec = get_tier_grouping(ctx->tier) * (time_t)rd->update_every;
416
- handle->dt = handle->dt_sec * USEC_PER_SEC;
593
+ handle->wanted_start_time_s = start_time_s;
594
+ handle->now_s = start_time_s;
595
handle->position = 0;
596
handle->ctx = ctx;
419
- handle->metric_handle = metric_handle;
597
handle->descr = NULL;
598
+ handle->dt_s = page_index->latest_update_every_s;
599
rrdimm_handle->handle = (STORAGE_QUERY_HANDLE *)handle;
422
- pages_nr = pg_cache_preload(ctx, metric_handle->rrdeng_uuid, start_time * USEC_PER_SEC, end_time * USEC_PER_SEC,
600
+ pages_nr = pg_cache_preload(ctx, &page_index->id, start_time_s * USEC_PER_SEC, end_time_s * USEC_PER_SEC,
601
NULL, &handle->page_index);
602
if (unlikely(NULL == handle->page_index || 0 == pages_nr))
603
// there are no metrics to load
426
- handle->next_page_time = INVALID_TIME;
604
+ handle->wanted_start_time_s = INVALID_TIME;
605
}
606
429
-static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle) {
607
+static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle, bool debug_this __maybe_unused) {
608
struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle;
609
610
struct rrdengine_instance *ctx = handle->ctx;
611
struct rrdeng_page_descr *descr = handle->descr;
612
613
uint32_t page_length;
436
- usec_t page_end_time;
614
+ usec_t page_end_time_ut;
615
unsigned position;
616
617
if (likely(descr)) {
623
624
pg_cache_put(ctx, descr);
625
handle->descr = NULL;
448
- handle->next_page_time = (handle->page_end_time / USEC_PER_SEC) + 1;
626
+ handle->wanted_start_time_s = (time_t)((handle->page_end_time_ut / USEC_PER_SEC) + handle->dt_s);
627
450
- if (unlikely(handle->next_page_time > rrdimm_handle->end_time))
628
+ if (unlikely(handle->wanted_start_time_s > rrdimm_handle->end_time_s))
629
return 1;
630
}
631
454
- usec_t next_page_time = handle->next_page_time * USEC_PER_SEC;
455
- descr = pg_cache_lookup_next(ctx, handle->page_index, &handle->page_index->id, next_page_time, rrdimm_handle->end_time * USEC_PER_SEC);
632
+ usec_t wanted_start_time_ut = handle->wanted_start_time_s * USEC_PER_SEC;
633
+ descr = pg_cache_lookup_next(ctx, handle->page_index, &handle->page_index->id,
634
+ wanted_start_time_ut, rrdimm_handle->end_time_s * USEC_PER_SEC);
635
if (NULL == descr)
636
return 1;
637
640
#endif
641
642
handle->descr = descr;
464
- pg_cache_atomic_get_pg_info(descr, &page_end_time, &page_length);
465
- if (unlikely(INVALID_TIME == descr->start_time || INVALID_TIME == page_end_time))
643
+ pg_cache_atomic_get_pg_info(descr, &page_end_time_ut, &page_length);
644
+ if (unlikely(INVALID_TIME == descr->start_time_ut || INVALID_TIME == page_end_time_ut || 0 == descr->update_every_s)) {
645
+ error("DBENGINE: discarding invalid page descriptor (start_time = %llu, end_time = %llu, update_every_s = %d)",
646
+ descr->start_time_ut, page_end_time_ut, descr->update_every_s);
647
return 1;
648
+ }
649
468
- if (unlikely(descr->start_time != page_end_time && next_page_time > descr->start_time)) {
650
+ if (unlikely(descr->start_time_ut != page_end_time_ut && wanted_start_time_ut > descr->start_time_ut)) {
651
// we're in the middle of the page somewhere
652
unsigned entries = page_length / PAGE_POINT_SIZE_BYTES(descr);
471
- position = ((uint64_t)(next_page_time - descr->start_time)) * (entries - 1) /
472
- (page_end_time - descr->start_time);
653
+ position = ((uint64_t)(wanted_start_time_ut - descr->start_time_ut)) * (entries - 1) /
654
+ (page_end_time_ut - descr->start_time_ut);
655
}
656
else
657
position = 0;
658
477
- handle->page_end_time = page_end_time;
659
+ handle->page_end_time_ut = page_end_time_ut;
660
handle->page_length = page_length;
661
+ handle->entries = page_length / PAGE_POINT_SIZE_BYTES(descr);
662
handle->page = descr->pg_cache_descr->page;
480
- usec_t entries = handle->entries = page_length / PAGE_POINT_SIZE_BYTES(descr);
481
- if (likely(entries > 1))
482
- handle->dt = (page_end_time - descr->start_time) / (entries - 1);
483
- else {
484
- // TODO we should store the dt of each page in each page
485
- // now we keep the dt of whatever was before
486
- ;
487
- }
488
-
489
- handle->dt_sec = (time_t)(handle->dt / USEC_PER_SEC);
663
+ handle->dt_s = descr->update_every_s;
664
handle->position = position;
665
666
+// if(debug_this)
667
+// info("DBENGINE: rrdeng_load_page_next(), "
668
+// "position:%d, "
669
+// "start_time_ut:%llu, "
670
+// "page_end_time_ut:%llu, "
671
+// "next_page_time_ut:%llu, "
672
+// "in_out:%s"
673
+// , position
674
+// , descr->start_time_ut
675
+// , page_end_time_ut
676
+// ,
677
+// wanted_start_time_ut, in_out?"true":"false"
678
+// );
679
+
680
return 0;
681
}
682
683
// Returns the metric and sets its timestamp into current_time
684
// IT IS REQUIRED TO **ALWAYS** SET ALL RETURN VALUES (current_time, end_time, flags)
685
// IT IS REQUIRED TO **ALWAYS** KEEP TRACK OF TIME, EVEN OUTSIDE THE DATABASE BOUNDARIES
498
-STORAGE_POINT rrdeng_load_metric_next(struct rrddim_query_handle *rrdimm_handle) {
499
- struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle;
686
+STORAGE_POINT rrdeng_load_metric_next(struct rrddim_query_handle *rrddim_handle) {
687
+ struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle;
688
// struct rrdeng_metric_handle *metric_handle = handle->metric_handle;
689
502
- STORAGE_POINT sp;
690
struct rrdeng_page_descr *descr = handle->descr;
691
+ time_t now = handle->now_s + handle->dt_s;
692
+
693
+// bool debug_this = false;
694
+// {
695
+// unsigned char u[16] = { 0x0C, 0x0A, 0x40, 0xD6, 0x2A, 0x43, 0x4A, 0x7C, 0x95, 0xF7, 0xD1, 0x1E, 0x0C, 0x9E, 0x8A, 0xE7 };
696
+// if(uuid_compare(u, handle->page_index->id) == 0) {
697
+// char buffer[100];
698
+// snprintfz(buffer, 100, "load system.cpu, now:%u, dt:%u, position:%u page_index first:%u, last:%u",
699
+// (uint32_t)(now),
700
+// (uint32_t)(handle->dt_s),
701
+// (uint32_t)(handle->position),
702
+// (uint32_t)(handle->page_index->oldest_time / USEC_PER_SEC),
703
+// (uint32_t)(handle->page_index->latest_time / USEC_PER_SEC));
704
+//
705
+// print_page_cache_descr(descr, buffer, false);
706
+// debug_this = true;
707
+// }
708
+// }
709
+
710
+ STORAGE_POINT sp;
711
unsigned position = handle->position + 1;
505
- time_t now = handle->now + handle->dt_sec;
712
storage_number_tier1_t tier1_value;
713
508
- if (unlikely(INVALID_TIME == handle->next_page_time)) {
509
- handle->next_page_time = INVALID_TIME;
510
- handle->now = now;
511
- storage_point_empty(sp, now - handle->dt_sec, now);
714
+ if (unlikely(INVALID_TIME == handle->wanted_start_time_s)) {
715
+ handle->wanted_start_time_s = INVALID_TIME;
716
+ handle->now_s = now;
717
+ storage_point_empty(sp, now - handle->dt_s, now);
718
return sp;
719
}
720
721
if (unlikely(!descr || position >= handle->entries)) {
722
// We need to get a new page
517
- if(rrdeng_load_page_next(rrdimm_handle)) {
723
+ if(rrdeng_load_page_next(rrddim_handle, false)) {
724
// next calls will not load any more metrics
519
- handle->next_page_time = INVALID_TIME;
520
- handle->now = now;
521
- storage_point_empty(sp, now - handle->dt_sec, now);
725
+ handle->wanted_start_time_s = INVALID_TIME;
726
+ handle->now_s = now;
727
+ storage_point_empty(sp, now - handle->dt_s, now);
728
return sp;
729
}
730
731
descr = handle->descr;
732
position = handle->position;
527
- now = (time_t)((descr->start_time + position * handle->dt) / USEC_PER_SEC);
733
+ now = (time_t)((descr->start_time_ut / USEC_PER_SEC) + position * descr->update_every_s);
734
+
735
+// if(debug_this) {
736
+// char buffer[100];
737
+// snprintfz(buffer, 100, "NEW PAGE system.cpu, now:%u, dt:%u, position:%u page_index first:%u, last:%u",
738
+// (uint32_t)(now),
739
+// (uint32_t)(handle->dt_s),
740
+// (uint32_t)(handle->position),
741
+// (uint32_t)(handle->page_index->oldest_time / USEC_PER_SEC),
742
+// (uint32_t)(handle->page_index->latest_time / USEC_PER_SEC));
743
+//
744
+// print_page_cache_descr(descr, buffer, false);
745
+// }
746
}
747
530
- sp.start_time = now - handle->dt_sec;
748
+ sp.start_time = now - handle->dt_s;
749
sp.end_time = now;
750
751
handle->position = position;
534
- handle->now = now;
752
+ handle->now_s = now;
753
754
switch(descr->type) {
755
case PAGE_METRICS: {
784
break;
785
}
786
569
- if (unlikely(now >= rrdimm_handle->end_time)) {
787
+ if (unlikely(now >= rrddim_handle->end_time_s)) {
788
// next calls will not load any more metrics
571
- handle->next_page_time = INVALID_TIME;
789
+ handle->wanted_start_time_s = INVALID_TIME;
790
}
791
792
+// if(debug_this)
793
+// info("DBENGINE: returning point: "
794
+// "time from %ld to %ld // query from %ld to %ld // wanted_start_time_s %ld"
795
+// , sp.start_time, sp.end_time
796
+// , rrddim_handle->start_time_s, rrddim_handle->end_time_s
797
+// , handle->wanted_start_time_s
798
+// );
799
+
800
return sp;
801
}
802
803
int rrdeng_load_metric_is_finished(struct rrddim_query_handle *rrdimm_handle)
804
{
805
struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle;
580
- return (INVALID_TIME == handle->next_page_time);
806
+ return (INVALID_TIME == handle->wanted_start_time_s);
807
}
808
809
/*
828
}
829
830
time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *db_metric_handle) {
605
- struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)db_metric_handle;
606
-
607
- struct pg_cache_page_index *page_index = metric_handle->page_index;
608
- return page_index->latest_time / USEC_PER_SEC;
831
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
832
+ return (time_t)(page_index->latest_time_ut / USEC_PER_SEC);
833
}
834
time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *db_metric_handle) {
611
- struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)db_metric_handle;
612
-
613
- struct pg_cache_page_index *page_index = metric_handle->page_index;
614
- return page_index->oldest_time / USEC_PER_SEC;
615
-}
616
-
617
-int rrdeng_metric_latest_time_by_uuid(uuid_t *dim_uuid, time_t *first_entry_t, time_t *last_entry_t, int tier)
618
-{
619
- struct page_cache *pg_cache;
620
- struct rrdengine_instance *ctx;
621
- Pvoid_t *PValue;
622
- struct pg_cache_page_index *page_index = NULL;
623
-
624
- ctx = get_rrdeng_ctx_from_host(localhost, tier);
625
- if (unlikely(!ctx)) {
626
- error("Failed to fetch multidb context");
627
- return 1;
628
- }
629
- pg_cache = &ctx->pg_cache;
630
-
631
- uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
632
- PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, dim_uuid, sizeof(uuid_t));
633
- if (likely(NULL != PValue)) {
634
- page_index = *PValue;
635
- }
636
- uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
637
-
638
- if (likely(page_index)) {
639
- *first_entry_t = page_index->oldest_time / USEC_PER_SEC;
640
- *last_entry_t = page_index->latest_time / USEC_PER_SEC;
641
- return 0;
642
- }
643
-
644
- return 1;
835
+ struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
836
+ return (time_t)(page_index->oldest_time_ut / USEC_PER_SEC);
837
}
838
839
int rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *si, uuid_t *dim_uuid, time_t *first_entry_t, time_t *last_entry_t)
858
uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
859
860
if (likely(page_index)) {
669
- *first_entry_t = page_index->oldest_time / USEC_PER_SEC;
670
- *last_entry_t = page_index->latest_time / USEC_PER_SEC;
861
+ *first_entry_t = page_index->oldest_time_ut / USEC_PER_SEC;
862
+ *last_entry_t = page_index->latest_time_ut / USEC_PER_SEC;
863
return 0;
864
}
865
886
887
debug(D_RRDENGINE, "Created new page:");
888
if (unlikely(debug_flags & D_RRDENGINE))
697
- print_page_cache_descr(descr);
889
+ print_page_cache_descr(descr, "", true);
890
rrdeng_page_descr_mutex_unlock(ctx, descr);
891
*ret_descr = descr;
892
return page;
958
}
959
960
/* Gets a reference for the page */
769
-void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time, void **handle)
961
+void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time_ut, void **handle)
962
{
963
struct rrdeng_page_descr *descr;
964
struct page_cache_descr *pg_cache_descr;
965
966
debug(D_RRDENGINE, "Reading existing page:");
775
- descr = pg_cache_lookup(ctx, NULL, id, point_in_time);
967
+ descr = pg_cache_lookup(ctx, NULL, id, point_in_time_ut);
968
if (NULL == descr) {
969
*handle = NULL;
970
1201
size_t points = descr->page_length / PAGE_POINT_SIZE_BYTES(descr);
1202
1203
if(likely(points > 1))
1012
- update_every_usec = (descr->end_time - descr->start_time) / (points - 1);
1204
+ update_every_usec = (descr->end_time_ut - descr->start_time_ut) / (points - 1);
1205
else {
1206
update_every_usec = default_rrd_update_every * get_tier_grouping(ctx->tier) * USEC_PER_SEC;
1207
stats.single_point_pages++;
1208
}
1209
1018
- time_t duration_secs = (time_t)((descr->end_time - descr->start_time + update_every_usec)/USEC_PER_SEC);
1210
+ time_t duration_secs = (time_t)((descr->end_time_ut - descr->start_time_ut + update_every_usec)/USEC_PER_SEC);
1211
1212
stats.extents_pages++;
1213
stats.pages_uncompressed_bytes += descr->page_length;
1219
stats.page_types[descr->type].pages_duration_secs += duration_secs;
1220
stats.page_types[descr->type].points += points;
1221
1030
- if(!stats.first_t || (descr->start_time - update_every_usec) < stats.first_t)
1031
- stats.first_t = (descr->start_time - update_every_usec) / USEC_PER_SEC;
1222
+ if(!stats.first_t || (descr->start_time_ut - update_every_usec) < stats.first_t)
1223
+ stats.first_t = (descr->start_time_ut - update_every_usec) / USEC_PER_SEC;
1224
1033
- if(!stats.last_t || descr->end_time > stats.last_t)
1034
- stats.last_t = descr->end_time / USEC_PER_SEC;
1225
+ if(!stats.last_t || descr->end_time_ut > stats.last_t)
1226
+ stats.last_t = descr->end_time_ut / USEC_PER_SEC;
1227
}
1228
}
1229
}
1263
}
1264
}
1265
1074
- stats.sizeof_metric = struct_natural_alignment(sizeof(struct pg_cache_page_index));
1266
+ stats.sizeof_metric = struct_natural_alignment(sizeof(struct pg_cache_page_index) + sizeof(struct pg_alignment));
1267
stats.sizeof_page = struct_natural_alignment(sizeof(struct rrdeng_page_descr));
1268
stats.sizeof_datafile = struct_natural_alignment(sizeof(struct rrdengine_datafile)) + struct_natural_alignment(sizeof(struct rrdengine_journalfile));
1269
stats.sizeof_page_in_cache = struct_natural_alignment(sizeof(struct page_cache_descr));