journal improvements part 3 (#16509)

all fields are now in the hashtable

5 files changed, 254 insertions, 131 deletions

diff --git a/collectors/log2journal/log2journal-params.c b/collectors/log2journal/log2journal-params.c
index c7ceda03fb..60d22f901d 100644
--- a/collectors/log2journal/log2journal-params.c
+++ b/collectors/log2journal/log2journal-params.c
@@ -4,7 +4,24 @@
 
 // ----------------------------------------------------------------------------
 
-void nd_log_cleanup(LOG_JOB *jb) {
+void log_job_init(LOG_JOB *jb) {
+    memset(jb, 0, sizeof(*jb));
+    simple_hashtable_init(&jb->hashtable, 32);
+}
+
+static void simple_hashtable_cleanup_allocated(SIMPLE_HASHTABLE *ht) {
+    for(size_t i = 0; i < ht->used ;i++) {
+        HASHED_KEY *k = ht->hashtable[i].data;
+        if(k && k->flags & HK_HASHTABLE_ALLOCATED) {
+            hashed_key_cleanup(k);
+            freez(k);
+            ht->hashtable[i].data = NULL;
+            ht->hashtable[i].hash = 0;
+        }
+    }
+}
+
+void log_job_cleanup(LOG_JOB *jb) {
     if(jb->prefix) {
         freez((void *) jb->prefix);
         jb->prefix = NULL;
@@ -30,6 +47,9 @@ void nd_log_cleanup(LOG_JOB *jb) {
     for(size_t i = 0; i < jb->rewrites.used; i++)
         rewrite_cleanup(&jb->rewrites.array[i]);
 
+    simple_hashtable_cleanup_allocated(&jb->hashtable);
+    simple_hashtable_free(&jb->hashtable);
+
     // remove references to everything else, to reveal them in valgrind
     memset(jb, 0, sizeof(*jb));
 }
@@ -42,10 +62,7 @@ bool log_job_filename_key_set(LOG_JOB *jb, const char *key, size_t key_len) {
         return false;
     }
 
-    if(jb->filename.key)
-        freez((char*)jb->filename.key);
-
-    jb->filename.key = strndupz(key, key_len);
+    hashed_key_len_set(&jb->filename.key, key, key_len);
 
     return true;
 }
@@ -295,7 +312,7 @@ bool log_job_command_line_parse_parameters(LOG_JOB *jb, int argc, char **argv) {
             }
 #endif
             else if (strcmp(param, "--unmatched-key") == 0)
-                jb->unmatched.key = value;
+                hashed_key_set(&jb->unmatched.key, value);
             else if (strcmp(param, "--duplicate") == 0) {
                 if (!parse_duplicate(jb, value))
                     return false;
diff --git a/collectors/log2journal/log2journal-rewrite.c b/collectors/log2journal/log2journal-rewrite.c
index 18eb8bfac1..4135143191 100644
--- a/collectors/log2journal/log2journal-rewrite.c
+++ b/collectors/log2journal/log2journal-rewrite.c
@@ -15,6 +15,8 @@ bool log_job_rewrite_add(LOG_JOB *jb, const char *key, const char *search_patter
     }
 
     REWRITE *rw = &jb->rewrites.array[jb->rewrites.used++];
+    rw->flags = RW_SEARCH_REPLACE | RW_MATCHED_ENTRIES;
+
     hashed_key_set(&rw->key, key);
 
     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
         return false;
     }
 
+    for(REPLACE_NODE *node = rw->replace.nodes; node; node = node->next) {
+        if(node->is_variable) {
+            rw->flags |= RW_HAS_VARIABLES;
+            break;
+        }
+    }
+
     return true;
 }
diff --git a/collectors/log2journal/log2journal-yaml.c b/collectors/log2journal/log2journal-yaml.c
index 1a208f3891..5a0ef4ae0f 100644
--- a/collectors/log2journal/log2journal-yaml.c
+++ b/collectors/log2journal/log2journal-yaml.c
@@ -520,7 +520,7 @@ static size_t yaml_parse_unmatched(yaml_parser_t *parser, LOG_JOB *jb) {
                         errors++;
                     } else {
                         if (sub_event.type == YAML_SCALAR_EVENT) {
-                            jb->unmatched.key = strndupz((char *)sub_event.data.scalar.value, sub_event.data.scalar.length);
+                            hashed_key_len_set(&jb->unmatched.key, (char *)sub_event.data.scalar.value, sub_event.data.scalar.length);
                         } else {
                             yaml_error(parser, &sub_event, "expected a scalar value for 'key'");
                             errors++;
@@ -952,10 +952,10 @@ void log_job_configuration_to_yaml(LOG_JOB *jb) {
         yaml_print_node("prefix", jb->prefix, 0, false);
     }
 
-    if(jb->filename.key) {
+    if(jb->filename.key.key) {
         fprintf(stderr, "\n");
         yaml_print_node("filename", NULL, 0, false);
-        yaml_print_node("key", jb->filename.key, 1, false);
+        yaml_print_node("key", jb->filename.key.key, 1, false);
     }
 
     if(jb->filter.include.pattern || jb->filter.exclude.pattern) {
@@ -1013,12 +1013,12 @@ void log_job_configuration_to_yaml(LOG_JOB *jb) {
         }
     }
 
-    if(jb->unmatched.key || jb->unmatched.injections.used) {
+    if(jb->unmatched.key.key || jb->unmatched.injections.used) {
         fprintf(stderr, "\n");
         yaml_print_node("unmatched", NULL, 0, false);
 
-        if(jb->unmatched.key)
-            yaml_print_node("key", jb->unmatched.key, 1, false);
+        if(jb->unmatched.key.key)
+            yaml_print_node("key", jb->unmatched.key.key, 1, false);
 
         if(jb->unmatched.injections.used) {
             fprintf(stderr, "\n");
diff --git a/collectors/log2journal/log2journal.c b/collectors/log2journal/log2journal.c
index 5f30ddf9f7..53cf9f7244 100644
--- a/collectors/log2journal/log2journal.c
+++ b/collectors/log2journal/log2journal.c
@@ -63,57 +63,152 @@ const char journal_key_characters_map[256] = {
 
 // ----------------------------------------------------------------------------
 
+static inline void validate_key(LOG_JOB *jb __maybe_unused, HASHED_KEY *k) {
+    if(k->len > JOURNAL_MAX_KEY_LEN)
+        log2stderr("WARNING: key '%s' has length %zu, which is more than %zu, the max systemd-journal allows",
+                k->key, k->len, (size_t)JOURNAL_MAX_KEY_LEN);
+
+    for(size_t i = 0; i < k->len ;i++) {
+        char c = k->key[i];
+
+        if((c < 'A' || c > 'Z') && !isdigit(c) && c != '_') {
+            log2stderr("WARNING: key '%s' contains characters that are not allowed by systemd-journal.", k->key);
+            break;
+        }
+    }
+
+    if(isdigit(k->key[0]))
+        log2stderr("WARNING: key '%s' starts with a digit and may not be accepted by systemd-journal.", k->key);
+
+    if(k->key[0] == '_')
+        log2stderr("WARNING: key '%s' starts with an underscore, which makes it a systemd-journal trusted field. "
+                   "Such fields are accepted by systemd-journal-remote, but not by systemd-journald.", k->key);
+}
+
+// ----------------------------------------------------------------------------
+
+static inline HASHED_KEY *get_key_from_hashtable_for_key(LOG_JOB *jb, HASHED_KEY *find) {
+    HASHED_KEY *k;
+    SIMPLE_HASHTABLE_SLOT *slot = simple_hashtable_get_slot(&jb->hashtable, find->hash, true);
+    if(slot->data) {
+        k = slot->data;
+
+        if(!(k->flags & HK_COLLISION_CHECKED)) {
+            k->flags |= HK_COLLISION_CHECKED;
+
+            if(strcmp(k->key, find->key) != 0)
+                log2stderr("Hashtable collision detected on key '%s' (hash %lx) and '%s' (hash %lx). "
+                           "Please file a bug report.",
+                        k->key, (unsigned long)k->hash, find->key, (unsigned long)find->hash);
+        }
+    }
+    else {
+        k = mallocz(sizeof(HASHED_KEY));
+        k->key = strdupz(find->key);
+        k->len = find->len;
+        k->hash = find->hash;
+        k->flags = HK_HASHTABLE_ALLOCATED;
+
+        slot->hash = k->hash;
+        slot->data = k;
+        jb->hashtable.used++;
+    }
+
+    return k;
+}
+
+static inline HASHED_KEY *get_key_from_hashtable(LOG_JOB *jb, const char *key) {
+    HASHED_KEY find = {
+            .key = key,
+            .len = strlen(key),
+    };
+    find.hash = XXH3_64bits(key, find.len);
+
+    return get_key_from_hashtable_for_key(jb, &find);
+}
+
+static inline HASHED_KEY *hashed_key_in_hashtable(LOG_JOB *jb, HASHED_KEY *k) {
+    if(k->flags & HK_HASHTABLE_ALLOCATED)
+        return k;
+
+    if(!k->hashtable_ptr)
+        k->hashtable_ptr = get_key_from_hashtable_for_key(jb, k);
+
+    return k->hashtable_ptr;
+}
+
+// ----------------------------------------------------------------------------
+
 static char *rewrite_value(LOG_JOB *jb, HASHED_KEY *k, const char *value, size_t value_len) {
     static __thread char rewritten_value[JOURNAL_MAX_VALUE_LEN + 1];
 
     if(!(k->flags & HK_REWRITES_CHECKED) || k->flags & HK_HAS_REWRITES) {
         k->flags |= HK_REWRITES_CHECKED;
 
+        char *copy_to = rewritten_value;
+        size_t remaining = sizeof(rewritten_value);
+
         for(size_t i = 0; i < jb->rewrites.used; i++) {
             REWRITE *rw = &jb->rewrites.array[i];
 
-            if(rw->key.hash == k->hash && strcmp(rw->key.key, k->key) == 0) {
+            if(!hashed_keys_match(&rw->key, k))
+                continue;
+
+            if(rw->flags & RW_SEARCH_REPLACE) {
                 if(!search_pattern_matches(&rw->search, value, value_len))
                     continue; // No match found, skip to next rewrite rule
 
                 PCRE2_SIZE *ovector = pcre2_get_ovector_pointer(rw->search.match_data);
 
-                char *buffer = rewritten_value;
-                size_t buffer_remaining = sizeof(rewritten_value);
-
                 // Iterate through the linked list of replacement nodes
                 for(REPLACE_NODE *node = rw->replace.nodes; node != NULL; node = node->next) {
                     if(node->is_variable) {
-                        int group_number = pcre2_substring_number_from_name(rw->search.re, (PCRE2_SPTR) node->name.key);
+                        int group_number = pcre2_substring_number_from_name(
+                                rw->search.re, (PCRE2_SPTR) node->name.key);
+
                         if(group_number >= 0) {
                             PCRE2_SIZE start_offset = ovector[2 * group_number];
                             PCRE2_SIZE end_offset = ovector[2 * group_number + 1];
                             PCRE2_SIZE length = end_offset - start_offset;
 
-                            size_t copied = copy_to_buffer(buffer, buffer_remaining, value + start_offset, length);
-                            buffer += copied;
-                            buffer_remaining -= copied;
+                            size_t copied = copy_to_buffer(copy_to, remaining, value + start_offset, length);
+                            copy_to += copied;
+                            remaining -= copied;
                         }
                         else {
                             // TODO: lookup in key names to get their values
 
                             if(!node->logged_error) {
-                                log2stderr("WARNING: variable '${%s}' in rewrite rule cannot be resolved.",
-                                           node->name.key);
+                                log2stderr("WARNING: variable '${%s}' in rewrite rule of key '%s' cannot be resolved.",
+                                           node->name.key, k->key);
+
                                 node->logged_error = true;
                             }
                         }
                     }
                     else {
-                        size_t copied = copy_to_buffer(buffer, buffer_remaining, node->name.key, node->name.len);
-                        buffer += copied;
-                        buffer_remaining -= copied;
+                        size_t copied = copy_to_buffer(copy_to, remaining, node->name.key, node->name.len);
+                        copy_to += copied;
+                        remaining -= copied;
                     }
                 }
-
-                k->flags |= HK_HAS_REWRITES;
-                return rewritten_value;
             }
+            else {
+                for(REPLACE_NODE *node = rw->replace.nodes; node != NULL; node = node->next) {
+                    if(node->is_variable) {
+                        // TODO: lookup in key names to get their values
+                        ;
+                    }
+                    else {
+                        size_t copied = copy_to_buffer(copy_to, remaining, node->name.key, node->name.len);
+                        copy_to += copied;
+                        remaining -= copied;
+                    }
+                }
+            }
+
+            k->flags |= HK_HAS_REWRITES;
+            return rewritten_value;
         }
     }
 
@@ -127,9 +222,10 @@ static inline HASHED_KEY *rename_key(LOG_JOB *jb, HASHED_KEY *k) {
         for(size_t i = 0; i < jb->renames.used; i++) {
             RENAME *rn = &jb->renames.array[i];
 
-            if(rn->old_key.hash == k->hash && strcmp(rn->old_key.key, k->key) == 0) {
+            if(hashed_keys_match(&rn->old_key, k)) {
                 k->flags |= HK_HAS_RENAMES;
-                return &rn->new_key;
+
+                return hashed_key_in_hashtable(jb, &rn->new_key);
             }
         }
     }
@@ -139,9 +235,17 @@ static inline HASHED_KEY *rename_key(LOG_JOB *jb, HASHED_KEY *k) {
 
 // ----------------------------------------------------------------------------
 
-static inline void send_key_value_error(const char *key, const char *format, ...) __attribute__ ((format(__printf__, 2, 3)));
-static inline void send_key_value_error(const char *key, const char *format, ...) {
-    printf("%s=", key);
+static inline void send_key_value_constant(LOG_JOB *jb __maybe_unused, HASHED_KEY *key, const char *value) {
+    HASHED_KEY *ht_key = hashed_key_in_hashtable(jb, key);
+
+    printf("%s=%s\n", ht_key->key, value);
+}
+
+static inline void send_key_value_error(LOG_JOB *jb, HASHED_KEY *key, const char *format, ...) __attribute__ ((format(__printf__, 3, 4)));
+static inline void send_key_value_error(LOG_JOB *jb, HASHED_KEY *key, const char *format, ...) {
+    HASHED_KEY *ht_key = hashed_key_in_hashtable(jb, key);
+
+    printf("%s=", ht_key->key);
     va_list args;
     va_start(args, format);
     vprintf(format, args);
@@ -149,65 +253,39 @@ static inline void send_key_value_error(const char *key, const char *format, ...
     printf("\n");
 }
 
-static inline void send_key_value_and_rewrite(LOG_JOB *jb, HASHED_KEY *k, const char *value, size_t len) {
-    char *rewritten = rewrite_value(jb, k, value, len);
-    if(!rewritten)
-        printf("%s=%.*s\n", k->key, (int)len, value);
-    else
-        printf("%s=%s\n", k->key, rewritten);
-}
-
-static inline HASHED_KEY *get_key_from_hashtable(LOG_JOB *jb, const char *key) {
-    size_t key_len = strlen(key);
-    XXH64_hash_t key_hash = XXH3_64bits(key, key_len);
+static inline void send_key_value_and_rewrite(LOG_JOB *jb, HASHED_KEY *key, const char *value, size_t len) {
+    HASHED_KEY *ht_key = hashed_key_in_hashtable(jb, key);
 
-    HASHED_KEY *k;
-    SIMPLE_HASHTABLE_SLOT *slot = simple_hashtable_get_slot(&jb->hashtable, key_hash, true);
-    if(slot->data) {
-        k = slot->data;
-
-        if(!(k->flags & HK_COLLISION_CHECKED)) {
-            k->flags |= HK_COLLISION_CHECKED;
-
-            if(strcmp(k->key, key) != 0)
-                log2stderr("Hashtable collision detected on key '%s' (hash %lx) and '%s' (hash %lx). "
-                           "Please report this to Netdata.",
-                        k->key, (unsigned long)k->hash, key, (unsigned long)key_hash);
-        }
+    if(!(ht_key->flags & HK_KEY_CHECKED)) {
+        ht_key->flags |= HK_KEY_CHECKED;
+        validate_key(jb, ht_key);
     }
-    else {
-        k = mallocz(sizeof(HASHED_KEY));
-        k->key = strdupz(key);
-        k->len = key_len;
-        k->hash = key_hash;
-        k->flags = HK_ALLOCATED;
 
-        slot->hash = k->hash;
-        slot->data = k;
-        jb->hashtable.used++;
-    }
-
-    return k;
+    char *rewritten = rewrite_value(jb, ht_key, value, len);
+    if(!rewritten)
+        printf("%s=%.*s\n", ht_key->key, (int)len, value);
+    else
+        printf("%s=%s\n", ht_key->key, rewritten);
 }
 
 inline void log_job_send_extracted_key_value(LOG_JOB *jb, const char *key, const char *value, size_t len) {
-    HASHED_KEY *k = get_key_from_hashtable(jb, key);
+    HASHED_KEY *ht_key = get_key_from_hashtable(jb, key);
 
-    if(!(k->flags & HK_FILTERED)) {
-        k->flags |= HK_FILTERED;
+    if(!(ht_key->flags & HK_FILTERED)) {
+        ht_key->flags |= HK_FILTERED;
 
-        bool included = jb->filter.include.re ? search_pattern_matches(&jb->filter.include, k->key, k->len) : true;
-        bool excluded = jb->filter.exclude.re ? search_pattern_matches(&jb->filter.exclude, k->key, k->len) : false;
+        bool included = jb->filter.include.re ? search_pattern_matches(&jb->filter.include, ht_key->key, ht_key->len) : true;
+        bool excluded = jb->filter.exclude.re ? search_pattern_matches(&jb->filter.exclude, ht_key->key, ht_key->len) : false;
 
         if(included && !excluded)
-            k->flags |= HK_FILTERED_INCLUDED;
+            ht_key->flags |= HK_FILTERED_INCLUDED;
         else
-            k->flags &= ~HK_FILTERED_INCLUDED;
+            ht_key->flags &= ~HK_FILTERED_INCLUDED;
     }
 
-    if(k->flags & HK_FILTERED_INCLUDED) {
+    if(ht_key->flags & HK_FILTERED_INCLUDED) {
         // process renames (changing the key)
-        HASHED_KEY *nk = rename_key(jb, k);
+        HASHED_KEY *nk = rename_key(jb, ht_key);
 
         // process rewrites (changing the value)
         // and send it to output
@@ -216,11 +294,7 @@ inline void log_job_send_extracted_key_value(LOG_JOB *jb, const char *key, const
 
     // process the duplications (using the original key)
     // and send them to output
-    send_duplications_for_key(jb, k, value, len);
-}
-
-static inline void send_key_value_constant(LOG_JOB *jb __maybe_unused, const char *key, const char *value) {
-    printf("%s=%s\n", key, value);
+    send_duplications_for_key(jb, ht_key, value, len);
 }
 
 // ----------------------------------------------------------------------------
@@ -252,7 +326,9 @@ static inline void jb_finalize_injections(LOG_JOB *jb, bool line_is_matched) {
         if(!line_is_matched && !jb->injections.keys[j].on_unmatched)
             continue;
 
-        send_key_value_constant(jb, jb->injections.keys[j].key.key, jb->injections.keys[j].value.txt);
+        INJECTION *inj = &jb->injections.keys[j];
+
+        send_key_value_constant(jb, &inj->key, inj->value.txt);
     }
 }
 
@@ -286,8 +362,9 @@ static inline void send_duplications_for_key(LOG_JOB *jb, HASHED_KEY *k, const c
 
             if(kd->used == 1) {
                 // just one key to be duplicated
-                if(kd->keys[0].hash == k->hash && strcmp(kd->keys[0].key, k->key) == 0) {
+                if(hashed_keys_match(&kd->keys[0], k)) {
                     k->flags |= HK_HAS_DUPS;
+
                     send_key_value_and_rewrite(jb, &kd->target, value, value_len);
                     kd->exposed = true;
                 }
@@ -295,7 +372,7 @@ static inline void send_duplications_for_key(LOG_JOB *jb, HASHED_KEY *k, const c
             else {
                 // multiple keys to be duplicated
                 for(size_t g = 0; g < kd->used; g++) {
-                    if(kd->keys[g].hash == k->hash && strcmp(kd->keys[g].key, k->key) == 0) {
+                    if(hashed_keys_match(&kd->keys[g], k)) {
                         k->flags |= HK_HAS_DUPS;
                         txt_replace(&kd->values[g], value, value_len);
                     }
@@ -345,6 +422,7 @@ static inline void jb_send_remaining_duplications(LOG_JOB *jb) {
                 break;
             }
         }
+
         send_key_value_and_rewrite(jb, &kd->target, buffer, s - buffer);
     }
 }
@@ -353,8 +431,8 @@ static inline void jb_send_remaining_duplications(LOG_JOB *jb) {
 // filename injection
 
 static inline void jb_inject_filename(LOG_JOB *jb) {
-    if (jb->filename.key && jb->filename.current[0])
-        send_key_value_constant(jb, jb->filename.key, jb->filename.current);
+    if (jb->filename.key.key && jb->filename.current[0])
+        send_key_value_constant(jb, &jb->filename.key, jb->filename.current);
 }
 
 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
     return false;
 }
 
-// ----------------------------------------------------------------------------
+static inline bool jb_send_unmatched_line(LOG_JOB *jb, const char *line) {
+    if (!jb->unmatched.key.key)
+        return false;
 
-static void simple_hashtable_cleanup_allocated(SIMPLE_HASHTABLE *ht) {
-    for(size_t i = 0; i < ht->used ;i++) {
-        HASHED_KEY *k = ht->hashtable[i].data;
-        if(k && k->flags & HK_ALLOCATED) {
-            hashed_key_cleanup(k);
-            freez(k);
-            ht->hashtable[i].data = NULL;
-            ht->hashtable[i].hash = 0;
-        }
+    // we are sending errors to systemd-journal
+    send_key_value_error(jb, &jb->unmatched.key, "Parsing error on: %s", line);
+
+    for (size_t j = 0; j < jb->unmatched.injections.used; j++) {
+        INJECTION *inj = &jb->unmatched.injections.keys[j];
+
+        send_key_value_constant(jb, &inj->key, inj->value.txt);
     }
+
+    return true;
 }
 
 // ----------------------------------------------------------------------------
@@ -429,8 +509,6 @@ static char *get_next_line(LOG_JOB *jb __maybe_unused, char *buffer, size_t size
 int log_job_run(LOG_JOB *jb) {
     select_which_injections_should_be_injected_on_unmatched(jb);
 
-    simple_hashtable_init(&jb->hashtable, 32);
-
     PCRE2_STATE *pcre2 = NULL;
     LOG_JSON_STATE *json = NULL;
     LOGFMT_STATE *logfmt = NULL;
@@ -477,18 +555,9 @@ int log_job_run(LOG_JOB *jb) {
             else
                 log2stderr("%s", pcre2_parser_error(pcre2));
 
-            if (jb->unmatched.key) {
-                // we are sending errors to systemd-journal
-                send_key_value_error(jb->unmatched.key, "Parsing error on: %s", line);
-
-                for (size_t j = 0; j < jb->unmatched.injections.used; j++)
-                    send_key_value_constant(jb, jb->unmatched.injections.keys[j].key.key,
-                            jb->unmatched.injections.keys[j].value.txt);
-            }
-            else {
-                // we are just logging errors to stderr
+            if(!jb_send_unmatched_line(jb, line))
+                // just logging to stderr, not sending unmatched lines
                 continue;
-            }
         }
         else {
             // print all non-exposed duplications
@@ -511,16 +580,15 @@ int log_job_run(LOG_JOB *jb) {
     else if(pcre2)
         pcre2_parser_destroy(pcre2);
 
-    simple_hashtable_cleanup_allocated(&jb->hashtable);
-    simple_hashtable_free(&jb->hashtable);
-
     return 0;
 }
 
 // ----------------------------------------------------------------------------
 
 int main(int argc, char *argv[]) {
-    LOG_JOB log_job = { 0 };
+    LOG_JOB log_job;
+
+    log_job_init(&log_job);
 
     if(!log_job_command_line_parse_parameters(&log_job, argc, argv))
         exit(1);
@@ -530,6 +598,6 @@ int main(int argc, char *argv[]) {
 
     int ret = log_job_run(&log_job);
 
-    nd_log_cleanup(&log_job);
+    log_job_cleanup(&log_job);
     return ret;
 }
diff --git a/collectors/log2journal/log2journal.h b/collectors/log2journal/log2journal.h
index 7ef805139b..0917430fd6 100644
--- a/collectors/log2journal/log2journal.h
+++ b/collectors/log2journal/log2journal.h
@@ -14,6 +14,7 @@
 #include <string.h>
 #include <ctype.h>
 #include <stdarg.h>
+#include <assert.h>
 
 // ----------------------------------------------------------------------------
 // logging
@@ -130,7 +131,7 @@ static inline size_t copy_to_buffer(char *dst, size_t dst_size, const char *src,
 
 typedef struct txt {
     char *txt;
-    size_t size;
+    uint32_t size;
 } TEXT;
 
 static inline void txt_cleanup(TEXT *t) {
@@ -170,17 +171,18 @@ static inline void txt_replace(TEXT *t, const char *s, size_t len) {
 // ----------------------------------------------------------------------------
 
 typedef enum __attribute__((__packed__)) {
-    HK_NONE              = 0,
-    HK_ALLOCATED         = (1 << 0),
-    HK_FILTERED          = (1 << 1),
-    HK_FILTERED_INCLUDED = (1 << 2),
-    HK_COLLISION_CHECKED = (1 << 3),
-    HK_RENAMES_CHECKED   = (1 << 4),
-    HK_HAS_RENAMES       = (1 << 5),
-    HK_DUPS_CHECKED      = (1 << 6),
-    HK_HAS_DUPS          = (1 << 7),
-    HK_REWRITES_CHECKED  = (1 << 8),
-    HK_HAS_REWRITES      = (1 << 9),
+    HK_NONE                 = 0,
+    HK_HASHTABLE_ALLOCATED  = (1 << 0),
+    HK_FILTERED             = (1 << 1),
+    HK_FILTERED_INCLUDED    = (1 << 2),
+    HK_COLLISION_CHECKED    = (1 << 3),
+    HK_RENAMES_CHECKED      = (1 << 4),
+    HK_HAS_RENAMES          = (1 << 5),
+    HK_DUPS_CHECKED         = (1 << 6),
+    HK_HAS_DUPS             = (1 << 7),
+    HK_REWRITES_CHECKED     = (1 << 8),
+    HK_HAS_REWRITES         = (1 << 9),
+    HK_KEY_CHECKED          = (1 << 10),
 } HASHED_KEY_FLAGS;
 
 typedef struct hashed_key {
@@ -188,6 +190,10 @@ typedef struct hashed_key {
     uint32_t len;
     HASHED_KEY_FLAGS flags;
     XXH64_hash_t hash;
+    union {
+        struct hashed_key *hashtable_ptr; // HK_HASHTABLE_ALLOCATED is not set
+        TEXT value;                       // HK_HASHTABLE_ALLOCATED is set
+    };
 } HASHED_KEY;
 
 static inline void hashed_key_cleanup(HASHED_KEY *k) {
@@ -195,6 +201,11 @@ static inline void hashed_key_cleanup(HASHED_KEY *k) {
         freez((void *)k->key);
         k->key = NULL;
     }
+
+    if(k->flags & HK_HASHTABLE_ALLOCATED)
+        txt_cleanup(&k->value);
+    else
+        k->hashtable_ptr = NULL;
 }
 
 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
     k->flags = HK_NONE;
 }
 
+static inline bool hashed_keys_match(HASHED_KEY *k1, HASHED_KEY *k2) {
+    return ((k1 == k2) || (k1->hash == k2->hash && strcmp(k1->key, k2->key) == 0));
+}
+
 // ----------------------------------------------------------------------------
 
 typedef struct search_pattern {
@@ -284,7 +299,18 @@ void rename_cleanup(RENAME *rn);
 
 // ----------------------------------------------------------------------------
 
+typedef enum __attribute__((__packed__)) {
+//    RW_NONE                 = 0,
+    RW_SEARCH_REPLACE       = (1 << 0), // a rewrite rule
+    RW_MATCHED_ENTRIES      = (1 << 1), // an injection on matched log entry
+//    RW_UNMATCHED_ENTRIES    = (1 << 2), // an injection on unmatched log entry
+//    RW_INJECT_ALWAYS        = (1 << 3), // an injection: inject always
+//    RW_INJECT_IF_SATISFIED  = (1 << 4), // a duplication: inject only if the variables are resolved
+    RW_HAS_VARIABLES        = (1 << 5), // the replacement has variables in it
+} RW_FLAGS;
+
 typedef struct key_rewrite {
+    RW_FLAGS flags;
     HASHED_KEY key;
     SEARCH_PATTERN search;
     REPLACE_PATTERN replace;
@@ -310,7 +336,7 @@ typedef struct log_job {
 
     struct {
         bool last_line_was_empty;
-        const char *key;
+        HASHED_KEY key;
         char current[FILENAME_MAX + 1];
     } filename;
 
@@ -320,7 +346,7 @@ typedef struct log_job {
     } injections;
 
     struct {
-        const char *key;
+        HASHED_KEY key;
         struct {
             uint32_t used;
             INJECTION keys[MAX_INJECTIONS];
@@ -343,8 +369,11 @@ typedef struct log_job {
     } renames;
 } LOG_JOB;
 
+// initialize a log job
+void log_job_init(LOG_JOB *jb);
+
 // free all resources consumed by the log job
-void nd_log_cleanup(LOG_JOB *jb);
+void log_job_cleanup(LOG_JOB *jb);
 
 // ----------------------------------------------------------------------------