faster facets and journal fixes (#15737)
Costa Tsaousis committed
Aug 4, 2023 at 02:58 UTC
48fc717d7384ae7c305a311f110ae9bdf83b5b5f
3 files changed
+61
-36
collectors/systemd-journal.plugin/systemd-journal.c
+4
-4
@@ -247,7 +247,7 @@ static void systemd_journal_transform_syslog_facility(FACETS *facets __maybe_unu
247
const char *name = syslog_facility_to_name(facility);
248
if (name) {
249
buffer_flush(wb);
250
- buffer_json_add_array_item_string(wb, name);
250
+ buffer_strcat(wb, name);
251
}
252
}
253
}
@@ -259,7 +259,7 @@ static void systemd_journal_transform_priority(FACETS *facets __maybe_unused, BU
259
const char *name = syslog_priority_to_name(priority);
260
if (name) {
261
buffer_flush(wb);
262
- buffer_json_add_array_item_string(wb, name);
262
+ buffer_strcat(wb, name);
263
}
264
}
265
}
@@ -304,7 +304,7 @@ static void systemd_journal_transform_gid(FACETS *facets __maybe_unused, BUFFER
304
}
305
}
306
307
-static void systemd_journal_dynamic_row_id(FACETS *facets __maybe_unused, BUFFER *wb, FACET_ROW_KEY_VALUE *rkv, FACET_ROW *row, void *data __maybe_unused) {
307
+static void systemd_journal_dynamic_row_id(FACETS *facets __maybe_unused, BUFFER *json_array, FACET_ROW_KEY_VALUE *rkv, FACET_ROW *row, void *data __maybe_unused) {
308
FACET_ROW_KEY_VALUE *syslog_identifier_rkv = dictionary_get(row->dict, "SYSLOG_IDENTIFIER");
309
FACET_ROW_KEY_VALUE *pid_rkv = dictionary_get(row->dict, "_PID");
310
@@ -314,7 +314,7 @@ static void systemd_journal_dynamic_row_id(FACETS *facets __maybe_unused, BUFFER
314
buffer_flush(rkv->wb);
315
buffer_sprintf(rkv->wb, "%s[%s]", identifier, pid);
316
317
- buffer_json_add_array_item_string(wb, buffer_tostring(rkv->wb));
317
+ buffer_json_add_array_item_string(json_array, buffer_tostring(rkv->wb));
318
}
319
320
static void function_systemd_journal(const char *transaction, char *function, char *line_buffer __maybe_unused, int line_max __maybe_unused, int timeout __maybe_unused) {
libnetdata/facets/facets.c
+56
-31
@@ -81,7 +81,12 @@ struct facet_key {
81
// members about the current row
82
uint32_t key_found_in_row;
83
uint32_t key_values_selected_in_row;
84
- BUFFER *current_value;
84
+
85
+ struct {
86
+ char hash[FACET_STRING_HASH_SIZE];
87
+ bool updated;
88
+ BUFFER *b;
89
+ } current_value;
90
91
uint32_t order;
92
@@ -93,8 +98,9 @@ struct facet_key {
98
struct {
99
facets_key_transformer_t cb;
100
void *data;
96
-
101
} transform;
102
+
103
+ struct facet_key *prev, *next;
104
};
105
106
struct facets {
@@ -111,6 +117,7 @@ struct facets {
117
118
DICTIONARY *accepted_params;
119
120
+ FACET_KEY *keys_ll;
121
DICTIONARY *keys;
122
FACET_ROW *base; // double linked list of the selected facets rows
123
@@ -253,7 +260,9 @@ static void facet_key_insert_callback(const DICTIONARY_ITEM *item __maybe_unused
260
facet_key_late_init(facets, k);
261
}
262
256
- k->current_value = buffer_create(0, NULL);
263
+ k->current_value.b = buffer_create(0, NULL);
264
+
265
+ DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(facets->keys_ll, k, prev, next);
266
}
267
268
static bool facet_key_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *data) {
@@ -277,8 +286,12 @@ static bool facet_key_conflict_callback(const DICTIONARY_ITEM *item __maybe_unus
286
287
static void facet_key_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) {
288
FACET_KEY *k = value;
289
+ FACETS *facets = data;
290
+
291
+ DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(facets->keys_ll, k, prev, next);
292
+
293
dictionary_destroy(k->values);
281
- buffer_free(k->current_value);
294
+ buffer_free(k->current_value.b);
295
freez((char *)k->name);
296
}
297
@@ -392,28 +405,32 @@ void facets_register_facet_filter(FACETS *facets, const char *key_id, char *valu
405
// ----------------------------------------------------------------------------
406
407
static inline void facets_check_value(FACETS *facets __maybe_unused, FACET_KEY *k) {
408
+ if(!k->current_value.updated)
409
+ buffer_flush(k->current_value.b);
410
+
411
if(k->transform.cb)
396
- k->transform.cb(facets, k->current_value, k->transform.data);
412
+ k->transform.cb(facets, k->current_value.b, k->transform.data);
413
398
- if(buffer_strlen(k->current_value) == 0)
399
- buffer_strcat(k->current_value, FACET_VALUE_UNSET);
414
+ if(!k->current_value.updated) {
415
+ buffer_strcat(k->current_value.b, FACET_VALUE_UNSET);
416
+ k->current_value.updated = true;
417
+ }
418
419
// bool found = false;
420
// if(strstr(buffer_tostring(k->current_value), "fprintd") != NULL)
421
// found = true;
422
423
if(facets->query && ((k->options & FACET_KEY_OPTION_FTS) || facets->options & FACETS_OPTION_ALL_KEYS_FTS)) {
406
- if(simple_pattern_matches(facets->query, buffer_tostring(k->current_value)))
424
+ if(simple_pattern_matches(facets->query, buffer_tostring(k->current_value.b)))
425
facets->keys_matched_by_query++;
426
}
427
428
if(k->values) {
429
FACET_VALUE tk = {
412
- .name = buffer_tostring(k->current_value),
430
+ .name = buffer_tostring(k->current_value.b),
431
};
414
- char hash[FACET_STRING_HASH_SIZE];
415
- facets_string_hash(tk.name, hash);
416
- dictionary_set(k->values, hash, &tk, sizeof(tk));
432
+ facets_string_hash(tk.name, k->current_value.hash);
433
+ dictionary_set(k->values, k->current_value.hash, &tk, sizeof(tk));
434
}
435
else {
436
k->key_found_in_row++;
@@ -423,16 +440,18 @@ static inline void facets_check_value(FACETS *facets __maybe_unused, FACET_KEY *
440
441
void facets_add_key_value(FACETS *facets, const char *key, const char *value) {
442
FACET_KEY *k = facets_register_key(facets, key, 0);
426
- buffer_flush(k->current_value);
427
- buffer_strcat(k->current_value, value);
443
+ buffer_flush(k->current_value.b);
444
+ buffer_strcat(k->current_value.b, value);
445
+ k->current_value.updated = true;
446
447
facets_check_value(facets, k);
448
}
449
450
void facets_add_key_value_length(FACETS *facets, const char *key, const char *value, size_t value_len) {
451
FACET_KEY *k = facets_register_key(facets, key, 0);
434
- buffer_flush(k->current_value);
435
- buffer_strncat(k->current_value, value, value_len);
452
+ buffer_flush(k->current_value.b);
453
+ buffer_strncat(k->current_value.b, value, value_len);
454
+ k->current_value.updated = true;
455
456
facets_check_value(facets, k);
457
}
@@ -492,7 +511,8 @@ static FACET_ROW *facets_row_create(FACETS *facets, usec_t usec, FACET_ROW *into
511
FACET_KEY *k;
512
dfe_start_read(facets->keys, k) {
513
FACET_ROW_KEY_VALUE t = {
495
- .tmp = buffer_strlen(k->current_value) ? buffer_tostring(k->current_value) : FACET_VALUE_UNSET,
514
+ .tmp = (k->current_value.updated && buffer_strlen(k->current_value.b)) ?
515
+ buffer_tostring(k->current_value.b) : FACET_VALUE_UNSET,
516
.wb = NULL,
517
};
518
dictionary_set(row->dict, k->name, &t, sizeof(t));
@@ -577,12 +597,14 @@ static void facets_row_keep(FACETS *facets, usec_t usec) {
597
598
void facets_rows_begin(FACETS *facets) {
599
FACET_KEY *k;
580
- dfe_start_read(facets->keys, k) {
581
- k->key_found_in_row = 0;
582
- k->key_values_selected_in_row = 0;
583
- buffer_flush(k->current_value);
584
- }
585
- dfe_done(k);
600
+ // dfe_start_read(facets->keys, k) {
601
+ for(k = facets->keys_ll ; k ; k = k->next) {
602
+ k->key_found_in_row = 0;
603
+ k->key_values_selected_in_row = 0;
604
+ k->current_value.updated = false;
605
+ k->current_value.hash[0] = '\0';
606
+ }
607
+ // dfe_done(k);
608
609
facets->keys_matched_by_query = 0;
610
}
@@ -597,9 +619,10 @@ void facets_row_finished(FACETS *facets, usec_t usec) {
619
uint32_t selected_by = 0;
620
621
FACET_KEY *k;
600
- dfe_start_read(facets->keys, k) {
622
+ // dfe_start_read(facets->keys, k) {
623
+ for(k = facets->keys_ll ; k ; k = k->next) {
624
if(!k->key_found_in_row) {
602
- internal_fatal(buffer_strlen(k->current_value), "key is not found in row but it has a current value");
625
+ internal_fatal(buffer_strlen(k->current_value.b), "key is not found in row but it has a current value");
626
// put the FACET_VALUE_UNSET value into it
627
facets_check_value(facets, k);
628
}
@@ -613,12 +636,13 @@ void facets_row_finished(FACETS *facets, usec_t usec) {
636
total_keys += k->key_found_in_row;
637
selected_by += (k->key_values_selected_in_row) ? 1 : 0;
638
}
616
- dfe_done(k);
639
+ // dfe_done(k);
640
641
if(selected_by >= total_keys - 1) {
642
uint32_t found = 0;
643
621
- dfe_start_read(facets->keys, k){
644
+ // dfe_start_read(facets->keys, k){
645
+ for(k = facets->keys_ll ; k ; k = k->next) {
646
uint32_t counted_by = selected_by;
647
648
if (counted_by != total_keys && !k->key_values_selected_in_row)
@@ -626,16 +650,17 @@ void facets_row_finished(FACETS *facets, usec_t usec) {
650
651
if(counted_by == total_keys) {
652
if(k->values) {
629
- char hash[FACET_STRING_HASH_SIZE];
630
- facets_string_hash(buffer_tostring(k->current_value), hash);
631
- FACET_VALUE *v = dictionary_get(k->values, hash);
653
+ if(!k->current_value.hash[0])
654
+ facets_string_hash(buffer_tostring(k->current_value.b), k->current_value.hash);
655
+
656
+ FACET_VALUE *v = dictionary_get(k->values, k->current_value.hash);
657
v->final_facet_value_counter++;
658
}
659
660
found++;
661
}
662
}
638
- dfe_done(k);
663
+ // dfe_done(k);
664
665
internal_fatal(!found, "We should find at least one facet to count this row");
666
(void)found;
libnetdata/facets/facets.h
+1
-1
@@ -31,7 +31,7 @@ typedef struct facet_key FACET_KEY;
31
void facets_string_hash(const char *src, char *out);
32
33
typedef void (*facets_key_transformer_t)(FACETS *facets __maybe_unused, BUFFER *wb, void *data);
34
-typedef void (*facet_dynamic_row_t)(FACETS *facets, BUFFER *wb, FACET_ROW_KEY_VALUE *rkv, FACET_ROW *row, void *data);
34
+typedef void (*facet_dynamic_row_t)(FACETS *facets, BUFFER *json_array, FACET_ROW_KEY_VALUE *rkv, FACET_ROW *row, void *data);
35
FACET_KEY *facets_register_dynamic_key(FACETS *facets, const char *key, FACET_KEY_OPTIONS options, facet_dynamic_row_t cb, void *data);
36
FACET_KEY *facets_register_key_transformation(FACETS *facets, const char *key, FACET_KEY_OPTIONS options, facets_key_transformer_t cb, void *data);
37