@cryptotaxi247 / netdata-1 / commits / 4fffdfd12

journal improvements part 3 (#16509)

all fields are now in the hashtable

Costa Tsaousis committed Nov 30, 2023 at 13:25 UTC 4fffdfd1267458dfe087b2592ea0fb41ef9ec2d7
5 files changed +254 -131
collectors/log2journal/log2journal-params.c
+23 -6
@@ -4,7 +4,24 @@
4
5 // ----------------------------------------------------------------------------
6
7 -void nd_log_cleanup(LOG_JOB *jb) {
7 +void log_job_init(LOG_JOB *jb) {
8 + memset(jb, 0, sizeof(*jb));
9 + simple_hashtable_init(&jb->hashtable, 32);
10 +}
11 +
12 +static void simple_hashtable_cleanup_allocated(SIMPLE_HASHTABLE *ht) {
13 + for(size_t i = 0; i < ht->used ;i++) {
14 + HASHED_KEY *k = ht->hashtable[i].data;
15 + if(k && k->flags & HK_HASHTABLE_ALLOCATED) {
16 + hashed_key_cleanup(k);
17 + freez(k);
18 + ht->hashtable[i].data = NULL;
19 + ht->hashtable[i].hash = 0;
20 + }
21 + }
22 +}
23 +
24 +void log_job_cleanup(LOG_JOB *jb) {
25 if(jb->prefix) {
26 freez((void *) jb->prefix);
27 jb->prefix = NULL;
@@ -30,6 +47,9 @@ void nd_log_cleanup(LOG_JOB *jb) {
47 for(size_t i = 0; i < jb->rewrites.used; i++)
48 rewrite_cleanup(&jb->rewrites.array[i]);
49
50 + simple_hashtable_cleanup_allocated(&jb->hashtable);
51 + simple_hashtable_free(&jb->hashtable);
52 +
53 // remove references to everything else, to reveal them in valgrind
54 memset(jb, 0, sizeof(*jb));
55 }
@@ -42,10 +62,7 @@ bool log_job_filename_key_set(LOG_JOB *jb, const char *key, size_t key_len) {
62 return false;
63 }
64
45 - if(jb->filename.key)
46 - freez((char*)jb->filename.key);
47 -
48 - jb->filename.key = strndupz(key, key_len);
65 + hashed_key_len_set(&jb->filename.key, key, key_len);
66
67 return true;
68 }
@@ -295,7 +312,7 @@ bool log_job_command_line_parse_parameters(LOG_JOB *jb, int argc, char **argv) {
312 }
313 #endif
314 else if (strcmp(param, "--unmatched-key") == 0)
298 - jb->unmatched.key = value;
315 + hashed_key_set(&jb->unmatched.key, value);
316 else if (strcmp(param, "--duplicate") == 0) {
317 if (!parse_duplicate(jb, value))
318 return false;
collectors/log2journal/log2journal-rewrite.c
+9
@@ -15,6 +15,8 @@ bool log_job_rewrite_add(LOG_JOB *jb, const char *key, const char *search_patter
15 }
16
17 REWRITE *rw = &jb->rewrites.array[jb->rewrites.used++];
18 + rw->flags = RW_SEARCH_REPLACE | RW_MATCHED_ENTRIES;
19 +
20 hashed_key_set(&rw->key, key);
21
22 if(!search_pattern_set(&rw->search, search_pattern, strlen(search_pattern)) ||
@@ -24,5 +26,12 @@ bool log_job_rewrite_add(LOG_JOB *jb, const char *key, const char *search_patter
26 return false;
27 }
28
29 + for(REPLACE_NODE *node = rw->replace.nodes; node; node = node->next) {
30 + if(node->is_variable) {
31 + rw->flags |= RW_HAS_VARIABLES;
32 + break;
33 + }
34 + }
35 +
36 return true;
37 }
collectors/log2journal/log2journal-yaml.c
+6 -6
@@ -520,7 +520,7 @@ static size_t yaml_parse_unmatched(yaml_parser_t *parser, LOG_JOB *jb) {
520 errors++;
521 } else {
522 if (sub_event.type == YAML_SCALAR_EVENT) {
523 - jb->unmatched.key = strndupz((char *)sub_event.data.scalar.value, sub_event.data.scalar.length);
523 + hashed_key_len_set(&jb->unmatched.key, (char *)sub_event.data.scalar.value, sub_event.data.scalar.length);
524 } else {
525 yaml_error(parser, &sub_event, "expected a scalar value for 'key'");
526 errors++;
@@ -952,10 +952,10 @@ void log_job_configuration_to_yaml(LOG_JOB *jb) {
952 yaml_print_node("prefix", jb->prefix, 0, false);
953 }
954
955 - if(jb->filename.key) {
955 + if(jb->filename.key.key) {
956 fprintf(stderr, "\n");
957 yaml_print_node("filename", NULL, 0, false);
958 - yaml_print_node("key", jb->filename.key, 1, false);
958 + yaml_print_node("key", jb->filename.key.key, 1, false);
959 }
960
961 if(jb->filter.include.pattern || jb->filter.exclude.pattern) {
@@ -1013,12 +1013,12 @@ void log_job_configuration_to_yaml(LOG_JOB *jb) {
1013 }
1014 }
1015
1016 - if(jb->unmatched.key || jb->unmatched.injections.used) {
1016 + if(jb->unmatched.key.key || jb->unmatched.injections.used) {
1017 fprintf(stderr, "\n");
1018 yaml_print_node("unmatched", NULL, 0, false);
1019
1020 - if(jb->unmatched.key)
1021 - yaml_print_node("key", jb->unmatched.key, 1, false);
1020 + if(jb->unmatched.key.key)
1021 + yaml_print_node("key", jb->unmatched.key.key, 1, false);
1022
1023 if(jb->unmatched.injections.used) {
1024 fprintf(stderr, "\n");
collectors/log2journal/log2journal.c
+172 -104
@@ -63,57 +63,152 @@ const char journal_key_characters_map[256] = {
63
64 // ----------------------------------------------------------------------------
65
66 +static inline void validate_key(LOG_JOB *jb __maybe_unused, HASHED_KEY *k) {
67 + if(k->len > JOURNAL_MAX_KEY_LEN)
68 + log2stderr("WARNING: key '%s' has length %zu, which is more than %zu, the max systemd-journal allows",
69 + k->key, k->len, (size_t)JOURNAL_MAX_KEY_LEN);
70 +
71 + for(size_t i = 0; i < k->len ;i++) {
72 + char c = k->key[i];
73 +
74 + if((c < 'A' || c > 'Z') && !isdigit(c) && c != '_') {
75 + log2stderr("WARNING: key '%s' contains characters that are not allowed by systemd-journal.", k->key);
76 + break;
77 + }
78 + }
79 +
80 + if(isdigit(k->key[0]))
81 + log2stderr("WARNING: key '%s' starts with a digit and may not be accepted by systemd-journal.", k->key);
82 +
83 + if(k->key[0] == '_')
84 + log2stderr("WARNING: key '%s' starts with an underscore, which makes it a systemd-journal trusted field. "
85 + "Such fields are accepted by systemd-journal-remote, but not by systemd-journald.", k->key);
86 +}
87 +
88 +// ----------------------------------------------------------------------------
89 +
90 +static inline HASHED_KEY *get_key_from_hashtable_for_key(LOG_JOB *jb, HASHED_KEY *find) {
91 + HASHED_KEY *k;
92 + SIMPLE_HASHTABLE_SLOT *slot = simple_hashtable_get_slot(&jb->hashtable, find->hash, true);
93 + if(slot->data) {
94 + k = slot->data;
95 +
96 + if(!(k->flags & HK_COLLISION_CHECKED)) {
97 + k->flags |= HK_COLLISION_CHECKED;
98 +
99 + if(strcmp(k->key, find->key) != 0)
100 + log2stderr("Hashtable collision detected on key '%s' (hash %lx) and '%s' (hash %lx). "
101 + "Please file a bug report.",
102 + k->key, (unsigned long)k->hash, find->key, (unsigned long)find->hash);
103 + }
104 + }
105 + else {
106 + k = mallocz(sizeof(HASHED_KEY));
107 + k->key = strdupz(find->key);
108 + k->len = find->len;
109 + k->hash = find->hash;
110 + k->flags = HK_HASHTABLE_ALLOCATED;
111 +
112 + slot->hash = k->hash;
113 + slot->data = k;
114 + jb->hashtable.used++;
115 + }
116 +
117 + return k;
118 +}
119 +
120 +static inline HASHED_KEY *get_key_from_hashtable(LOG_JOB *jb, const char *key) {
121 + HASHED_KEY find = {
122 + .key = key,
123 + .len = strlen(key),
124 + };
125 + find.hash = XXH3_64bits(key, find.len);
126 +
127 + return get_key_from_hashtable_for_key(jb, &find);
128 +}
129 +
130 +static inline HASHED_KEY *hashed_key_in_hashtable(LOG_JOB *jb, HASHED_KEY *k) {
131 + if(k->flags & HK_HASHTABLE_ALLOCATED)
132 + return k;
133 +
134 + if(!k->hashtable_ptr)
135 + k->hashtable_ptr = get_key_from_hashtable_for_key(jb, k);
136 +
137 + return k->hashtable_ptr;
138 +}
139 +
140 +// ----------------------------------------------------------------------------
141 +
142 static char *rewrite_value(LOG_JOB *jb, HASHED_KEY *k, const char *value, size_t value_len) {
143 static __thread char rewritten_value[JOURNAL_MAX_VALUE_LEN + 1];
144
145 if(!(k->flags & HK_REWRITES_CHECKED) || k->flags & HK_HAS_REWRITES) {
146 k->flags |= HK_REWRITES_CHECKED;
147
148 + char *copy_to = rewritten_value;
149 + size_t remaining = sizeof(rewritten_value);
150 +
151 for(size_t i = 0; i < jb->rewrites.used; i++) {
152 REWRITE *rw = &jb->rewrites.array[i];
153
75 - if(rw->key.hash == k->hash && strcmp(rw->key.key, k->key) == 0) {
154 + if(!hashed_keys_match(&rw->key, k))
155 + continue;
156 +
157 + if(rw->flags & RW_SEARCH_REPLACE) {
158 if(!search_pattern_matches(&rw->search, value, value_len))
159 continue; // No match found, skip to next rewrite rule
160
161 PCRE2_SIZE *ovector = pcre2_get_ovector_pointer(rw->search.match_data);
162
81 - char *buffer = rewritten_value;
82 - size_t buffer_remaining = sizeof(rewritten_value);
83 -
163 // Iterate through the linked list of replacement nodes
164 for(REPLACE_NODE *node = rw->replace.nodes; node != NULL; node = node->next) {
165 if(node->is_variable) {
87 - int group_number = pcre2_substring_number_from_name(rw->search.re, (PCRE2_SPTR) node->name.key);
166 + int group_number = pcre2_substring_number_from_name(
167 + rw->search.re, (PCRE2_SPTR) node->name.key);
168 +
169 if(group_number >= 0) {
170 PCRE2_SIZE start_offset = ovector[2 * group_number];
171 PCRE2_SIZE end_offset = ovector[2 * group_number + 1];
172 PCRE2_SIZE length = end_offset - start_offset;
173
93 - size_t copied = copy_to_buffer(buffer, buffer_remaining, value + start_offset, length);
94 - buffer += copied;
95 - buffer_remaining -= copied;
174 + size_t copied = copy_to_buffer(copy_to, remaining, value + start_offset, length);
175 + copy_to += copied;
176 + remaining -= copied;
177 }
178 else {
179 // TODO: lookup in key names to get their values
180
181 if(!node->logged_error) {
101 - log2stderr("WARNING: variable '${%s}' in rewrite rule cannot be resolved.",
102 - node->name.key);
182 + log2stderr("WARNING: variable '${%s}' in rewrite rule of key '%s' cannot be resolved.",
183 + node->name.key, k->key);
184 +
185 node->logged_error = true;
186 }
187 }
188 }
189 else {
108 - size_t copied = copy_to_buffer(buffer, buffer_remaining, node->name.key, node->name.len);
109 - buffer += copied;
110 - buffer_remaining -= copied;
190 + size_t copied = copy_to_buffer(copy_to, remaining, node->name.key, node->name.len);
191 + copy_to += copied;
192 + remaining -= copied;
193 }
194 }
113 -
114 - k->flags |= HK_HAS_REWRITES;
115 - return rewritten_value;
195 }
196 + else {
197 + for(REPLACE_NODE *node = rw->replace.nodes; node != NULL; node = node->next) {
198 + if(node->is_variable) {
199 + // TODO: lookup in key names to get their values
200 + ;
201 + }
202 + else {
203 + size_t copied = copy_to_buffer(copy_to, remaining, node->name.key, node->name.len);
204 + copy_to += copied;
205 + remaining -= copied;
206 + }
207 + }
208 + }
209 +
210 + k->flags |= HK_HAS_REWRITES;
211 + return rewritten_value;
212 }
213 }
214
@@ -127,9 +222,10 @@ static inline HASHED_KEY *rename_key(LOG_JOB *jb, HASHED_KEY *k) {
222 for(size_t i = 0; i < jb->renames.used; i++) {
223 RENAME *rn = &jb->renames.array[i];
224
130 - if(rn->old_key.hash == k->hash && strcmp(rn->old_key.key, k->key) == 0) {
225 + if(hashed_keys_match(&rn->old_key, k)) {
226 k->flags |= HK_HAS_RENAMES;
132 - return &rn->new_key;
227 +
228 + return hashed_key_in_hashtable(jb, &rn->new_key);
229 }
230 }
231 }
@@ -139,9 +235,17 @@ static inline HASHED_KEY *rename_key(LOG_JOB *jb, HASHED_KEY *k) {
235
236 // ----------------------------------------------------------------------------
237
142 -static inline void send_key_value_error(const char *key, const char *format, ...) __attribute__ ((format(__printf__, 2, 3)));
143 -static inline void send_key_value_error(const char *key, const char *format, ...) {
144 - printf("%s=", key);
238 +static inline void send_key_value_constant(LOG_JOB *jb __maybe_unused, HASHED_KEY *key, const char *value) {
239 + HASHED_KEY *ht_key = hashed_key_in_hashtable(jb, key);
240 +
241 + printf("%s=%s\n", ht_key->key, value);
242 +}
243 +
244 +static inline void send_key_value_error(LOG_JOB *jb, HASHED_KEY *key, const char *format, ...) __attribute__ ((format(__printf__, 3, 4)));
245 +static inline void send_key_value_error(LOG_JOB *jb, HASHED_KEY *key, const char *format, ...) {
246 + HASHED_KEY *ht_key = hashed_key_in_hashtable(jb, key);
247 +
248 + printf("%s=", ht_key->key);
249 va_list args;
250 va_start(args, format);
251 vprintf(format, args);
@@ -149,65 +253,39 @@ static inline void send_key_value_error(const char *key, const char *format, ...
253 printf("\n");
254 }
255
152 -static inline void send_key_value_and_rewrite(LOG_JOB *jb, HASHED_KEY *k, const char *value, size_t len) {
153 - char *rewritten = rewrite_value(jb, k, value, len);
154 - if(!rewritten)
155 - printf("%s=%.*s\n", k->key, (int)len, value);
156 - else
157 - printf("%s=%s\n", k->key, rewritten);
158 -}
159 -
160 -static inline HASHED_KEY *get_key_from_hashtable(LOG_JOB *jb, const char *key) {
161 - size_t key_len = strlen(key);
162 - XXH64_hash_t key_hash = XXH3_64bits(key, key_len);
256 +static inline void send_key_value_and_rewrite(LOG_JOB *jb, HASHED_KEY *key, const char *value, size_t len) {
257 + HASHED_KEY *ht_key = hashed_key_in_hashtable(jb, key);
258
164 - HASHED_KEY *k;
165 - SIMPLE_HASHTABLE_SLOT *slot = simple_hashtable_get_slot(&jb->hashtable, key_hash, true);
166 - if(slot->data) {
167 - k = slot->data;
168 -
169 - if(!(k->flags & HK_COLLISION_CHECKED)) {
170 - k->flags |= HK_COLLISION_CHECKED;
171 -
172 - if(strcmp(k->key, key) != 0)
173 - log2stderr("Hashtable collision detected on key '%s' (hash %lx) and '%s' (hash %lx). "
174 - "Please report this to Netdata.",
175 - k->key, (unsigned long)k->hash, key, (unsigned long)key_hash);
176 - }
259 + if(!(ht_key->flags & HK_KEY_CHECKED)) {
260 + ht_key->flags |= HK_KEY_CHECKED;
261 + validate_key(jb, ht_key);
262 }
178 - else {
179 - k = mallocz(sizeof(HASHED_KEY));
180 - k->key = strdupz(key);
181 - k->len = key_len;
182 - k->hash = key_hash;
183 - k->flags = HK_ALLOCATED;
263
185 - slot->hash = k->hash;
186 - slot->data = k;
187 - jb->hashtable.used++;
188 - }
189 -
190 - return k;
264 + char *rewritten = rewrite_value(jb, ht_key, value, len);
265 + if(!rewritten)
266 + printf("%s=%.*s\n", ht_key->key, (int)len, value);
267 + else
268 + printf("%s=%s\n", ht_key->key, rewritten);
269 }
270
271 inline void log_job_send_extracted_key_value(LOG_JOB *jb, const char *key, const char *value, size_t len) {
194 - HASHED_KEY *k = get_key_from_hashtable(jb, key);
272 + HASHED_KEY *ht_key = get_key_from_hashtable(jb, key);
273
196 - if(!(k->flags & HK_FILTERED)) {
197 - k->flags |= HK_FILTERED;
274 + if(!(ht_key->flags & HK_FILTERED)) {
275 + ht_key->flags |= HK_FILTERED;
276
199 - bool included = jb->filter.include.re ? search_pattern_matches(&jb->filter.include, k->key, k->len) : true;
200 - bool excluded = jb->filter.exclude.re ? search_pattern_matches(&jb->filter.exclude, k->key, k->len) : false;
277 + bool included = jb->filter.include.re ? search_pattern_matches(&jb->filter.include, ht_key->key, ht_key->len) : true;
278 + bool excluded = jb->filter.exclude.re ? search_pattern_matches(&jb->filter.exclude, ht_key->key, ht_key->len) : false;
279
280 if(included && !excluded)
203 - k->flags |= HK_FILTERED_INCLUDED;
281 + ht_key->flags |= HK_FILTERED_INCLUDED;
282 else
205 - k->flags &= ~HK_FILTERED_INCLUDED;
283 + ht_key->flags &= ~HK_FILTERED_INCLUDED;
284 }
285
208 - if(k->flags & HK_FILTERED_INCLUDED) {
286 + if(ht_key->flags & HK_FILTERED_INCLUDED) {
287 // process renames (changing the key)
210 - HASHED_KEY *nk = rename_key(jb, k);
288 + HASHED_KEY *nk = rename_key(jb, ht_key);
289
290 // process rewrites (changing the value)
291 // and send it to output
@@ -216,11 +294,7 @@ inline void log_job_send_extracted_key_value(LOG_JOB *jb, const char *key, const
294
295 // process the duplications (using the original key)
296 // and send them to output
219 - send_duplications_for_key(jb, k, value, len);
220 -}
221 -
222 -static inline void send_key_value_constant(LOG_JOB *jb __maybe_unused, const char *key, const char *value) {
223 - printf("%s=%s\n", key, value);
297 + send_duplications_for_key(jb, ht_key, value, len);
298 }
299
300 // ----------------------------------------------------------------------------
@@ -252,7 +326,9 @@ static inline void jb_finalize_injections(LOG_JOB *jb, bool line_is_matched) {
326 if(!line_is_matched && !jb->injections.keys[j].on_unmatched)
327 continue;
328
255 - send_key_value_constant(jb, jb->injections.keys[j].key.key, jb->injections.keys[j].value.txt);
329 + INJECTION *inj = &jb->injections.keys[j];
330 +
331 + send_key_value_constant(jb, &inj->key, inj->value.txt);
332 }
333 }
334
@@ -286,8 +362,9 @@ static inline void send_duplications_for_key(LOG_JOB *jb, HASHED_KEY *k, const c
362
363 if(kd->used == 1) {
364 // just one key to be duplicated
289 - if(kd->keys[0].hash == k->hash && strcmp(kd->keys[0].key, k->key) == 0) {
365 + if(hashed_keys_match(&kd->keys[0], k)) {
366 k->flags |= HK_HAS_DUPS;
367 +
368 send_key_value_and_rewrite(jb, &kd->target, value, value_len);
369 kd->exposed = true;
370 }
@@ -295,7 +372,7 @@ static inline void send_duplications_for_key(LOG_JOB *jb, HASHED_KEY *k, const c
372 else {
373 // multiple keys to be duplicated
374 for(size_t g = 0; g < kd->used; g++) {
298 - if(kd->keys[g].hash == k->hash && strcmp(kd->keys[g].key, k->key) == 0) {
375 + if(hashed_keys_match(&kd->keys[g], k)) {
376 k->flags |= HK_HAS_DUPS;
377 txt_replace(&kd->values[g], value, value_len);
378 }
@@ -345,6 +422,7 @@ static inline void jb_send_remaining_duplications(LOG_JOB *jb) {
422 break;
423 }
424 }
425 +
426 send_key_value_and_rewrite(jb, &kd->target, buffer, s - buffer);
427 }
428 }
@@ -353,8 +431,8 @@ static inline void jb_send_remaining_duplications(LOG_JOB *jb) {
431 // filename injection
432
433 static inline void jb_inject_filename(LOG_JOB *jb) {
356 - if (jb->filename.key && jb->filename.current[0])
357 - send_key_value_constant(jb, jb->filename.key, jb->filename.current);
434 + if (jb->filename.key.key && jb->filename.current[0])
435 + send_key_value_constant(jb, &jb->filename.key, jb->filename.current);
436 }
437
438 static inline bool jb_switched_filename(LOG_JOB *jb, const char *line, size_t len) {
@@ -386,18 +464,20 @@ static inline bool jb_switched_filename(LOG_JOB *jb, const char *line, size_t le
464 return false;
465 }
466
389 -// ----------------------------------------------------------------------------
467 +static inline bool jb_send_unmatched_line(LOG_JOB *jb, const char *line) {
468 + if (!jb->unmatched.key.key)
469 + return false;
470
391 -static void simple_hashtable_cleanup_allocated(SIMPLE_HASHTABLE *ht) {
392 - for(size_t i = 0; i < ht->used ;i++) {
393 - HASHED_KEY *k = ht->hashtable[i].data;
394 - if(k && k->flags & HK_ALLOCATED) {
395 - hashed_key_cleanup(k);
396 - freez(k);
397 - ht->hashtable[i].data = NULL;
398 - ht->hashtable[i].hash = 0;
399 - }
471 + // we are sending errors to systemd-journal
472 + send_key_value_error(jb, &jb->unmatched.key, "Parsing error on: %s", line);
473 +
474 + for (size_t j = 0; j < jb->unmatched.injections.used; j++) {
475 + INJECTION *inj = &jb->unmatched.injections.keys[j];
476 +
477 + send_key_value_constant(jb, &inj->key, inj->value.txt);
478 }
479 +
480 + return true;
481 }
482
483 // ----------------------------------------------------------------------------
@@ -429,8 +509,6 @@ static char *get_next_line(LOG_JOB *jb __maybe_unused, char *buffer, size_t size
509 int log_job_run(LOG_JOB *jb) {
510 select_which_injections_should_be_injected_on_unmatched(jb);
511
432 - simple_hashtable_init(&jb->hashtable, 32);
433 -
512 PCRE2_STATE *pcre2 = NULL;
513 LOG_JSON_STATE *json = NULL;
514 LOGFMT_STATE *logfmt = NULL;
@@ -477,18 +555,9 @@ int log_job_run(LOG_JOB *jb) {
555 else
556 log2stderr("%s", pcre2_parser_error(pcre2));
557
480 - if (jb->unmatched.key) {
481 - // we are sending errors to systemd-journal
482 - send_key_value_error(jb->unmatched.key, "Parsing error on: %s", line);
483 -
484 - for (size_t j = 0; j < jb->unmatched.injections.used; j++)
485 - send_key_value_constant(jb, jb->unmatched.injections.keys[j].key.key,
486 - jb->unmatched.injections.keys[j].value.txt);
487 - }
488 - else {
489 - // we are just logging errors to stderr
558 + if(!jb_send_unmatched_line(jb, line))
559 + // just logging to stderr, not sending unmatched lines
560 continue;
491 - }
561 }
562 else {
563 // print all non-exposed duplications
@@ -511,16 +580,15 @@ int log_job_run(LOG_JOB *jb) {
580 else if(pcre2)
581 pcre2_parser_destroy(pcre2);
582
514 - simple_hashtable_cleanup_allocated(&jb->hashtable);
515 - simple_hashtable_free(&jb->hashtable);
516 -
583 return 0;
584 }
585
586 // ----------------------------------------------------------------------------
587
588 int main(int argc, char *argv[]) {
523 - LOG_JOB log_job = { 0 };
589 + LOG_JOB log_job;
590 +
591 + log_job_init(&log_job);
592
593 if(!log_job_command_line_parse_parameters(&log_job, argc, argv))
594 exit(1);
@@ -530,6 +598,6 @@ int main(int argc, char *argv[]) {
598
599 int ret = log_job_run(&log_job);
600
533 - nd_log_cleanup(&log_job);
601 + log_job_cleanup(&log_job);
602 return ret;
603 }
collectors/log2journal/log2journal.h
+44 -15
@@ -14,6 +14,7 @@
14 #include <string.h>
15 #include <ctype.h>
16 #include <stdarg.h>
17 +#include <assert.h>
18
19 // ----------------------------------------------------------------------------
20 // logging
@@ -130,7 +131,7 @@ static inline size_t copy_to_buffer(char *dst, size_t dst_size, const char *src,
131
132 typedef struct txt {
133 char *txt;
133 - size_t size;
134 + uint32_t size;
135 } TEXT;
136
137 static inline void txt_cleanup(TEXT *t) {
@@ -170,17 +171,18 @@ static inline void txt_replace(TEXT *t, const char *s, size_t len) {
171 // ----------------------------------------------------------------------------
172
173 typedef enum __attribute__((__packed__)) {
173 - HK_NONE = 0,
174 - HK_ALLOCATED = (1 << 0),
175 - HK_FILTERED = (1 << 1),
176 - HK_FILTERED_INCLUDED = (1 << 2),
177 - HK_COLLISION_CHECKED = (1 << 3),
178 - HK_RENAMES_CHECKED = (1 << 4),
179 - HK_HAS_RENAMES = (1 << 5),
180 - HK_DUPS_CHECKED = (1 << 6),
181 - HK_HAS_DUPS = (1 << 7),
182 - HK_REWRITES_CHECKED = (1 << 8),
183 - HK_HAS_REWRITES = (1 << 9),
174 + HK_NONE = 0,
175 + HK_HASHTABLE_ALLOCATED = (1 << 0),
176 + HK_FILTERED = (1 << 1),
177 + HK_FILTERED_INCLUDED = (1 << 2),
178 + HK_COLLISION_CHECKED = (1 << 3),
179 + HK_RENAMES_CHECKED = (1 << 4),
180 + HK_HAS_RENAMES = (1 << 5),
181 + HK_DUPS_CHECKED = (1 << 6),
182 + HK_HAS_DUPS = (1 << 7),
183 + HK_REWRITES_CHECKED = (1 << 8),
184 + HK_HAS_REWRITES = (1 << 9),
185 + HK_KEY_CHECKED = (1 << 10),
186 } HASHED_KEY_FLAGS;
187
188 typedef struct hashed_key {
@@ -188,6 +190,10 @@ typedef struct hashed_key {
190 uint32_t len;
191 HASHED_KEY_FLAGS flags;
192 XXH64_hash_t hash;
193 + union {
194 + struct hashed_key *hashtable_ptr; // HK_HASHTABLE_ALLOCATED is not set
195 + TEXT value; // HK_HASHTABLE_ALLOCATED is set
196 + };
197 } HASHED_KEY;
198
199 static inline void hashed_key_cleanup(HASHED_KEY *k) {
@@ -195,6 +201,11 @@ static inline void hashed_key_cleanup(HASHED_KEY *k) {
201 freez((void *)k->key);
202 k->key = NULL;
203 }
204 +
205 + if(k->flags & HK_HASHTABLE_ALLOCATED)
206 + txt_cleanup(&k->value);
207 + else
208 + k->hashtable_ptr = NULL;
209 }
210
211 static inline void hashed_key_set(HASHED_KEY *k, const char *name) {
@@ -215,6 +226,10 @@ static inline void hashed_key_len_set(HASHED_KEY *k, const char *name, size_t le
226 k->flags = HK_NONE;
227 }
228
229 +static inline bool hashed_keys_match(HASHED_KEY *k1, HASHED_KEY *k2) {
230 + return ((k1 == k2) || (k1->hash == k2->hash && strcmp(k1->key, k2->key) == 0));
231 +}
232 +
233 // ----------------------------------------------------------------------------
234
235 typedef struct search_pattern {
@@ -284,7 +299,18 @@ void rename_cleanup(RENAME *rn);
299
300 // ----------------------------------------------------------------------------
301
302 +typedef enum __attribute__((__packed__)) {
303 +// RW_NONE = 0,
304 + RW_SEARCH_REPLACE = (1 << 0), // a rewrite rule
305 + RW_MATCHED_ENTRIES = (1 << 1), // an injection on matched log entry
306 +// RW_UNMATCHED_ENTRIES = (1 << 2), // an injection on unmatched log entry
307 +// RW_INJECT_ALWAYS = (1 << 3), // an injection: inject always
308 +// RW_INJECT_IF_SATISFIED = (1 << 4), // a duplication: inject only if the variables are resolved
309 + RW_HAS_VARIABLES = (1 << 5), // the replacement has variables in it
310 +} RW_FLAGS;
311 +
312 typedef struct key_rewrite {
313 + RW_FLAGS flags;
314 HASHED_KEY key;
315 SEARCH_PATTERN search;
316 REPLACE_PATTERN replace;
@@ -310,7 +336,7 @@ typedef struct log_job {
336
337 struct {
338 bool last_line_was_empty;
313 - const char *key;
339 + HASHED_KEY key;
340 char current[FILENAME_MAX + 1];
341 } filename;
342
@@ -320,7 +346,7 @@ typedef struct log_job {
346 } injections;
347
348 struct {
323 - const char *key;
349 + HASHED_KEY key;
350 struct {
351 uint32_t used;
352 INJECTION keys[MAX_INJECTIONS];
@@ -343,8 +369,11 @@ typedef struct log_job {
369 } renames;
370 } LOG_JOB;
371
372 +// initialize a log job
373 +void log_job_init(LOG_JOB *jb);
374 +
375 // free all resources consumed by the log job
347 -void nd_log_cleanup(LOG_JOB *jb);
376 +void log_job_cleanup(LOG_JOB *jb);
377
378 // ----------------------------------------------------------------------------
379