1 files changed, 350 insertions, 22 deletions

diff --git a/database/rrdlabels.c b/src/database/rrdlabels.c
index 69ee55526..9ea1d7c58 100644
--- a/database/rrdlabels.c
+++ b/src/database/rrdlabels.c
@@ -448,7 +448,7 @@ __attribute__((constructor)) void initialize_labels_keys_char_map(void) {
     label_names_char_map[' '] = '_';
     label_names_char_map['\\'] = '/';
 
-    // create the spaces map
+    // create the space map
     for(i = 0; i < 256 ;i++)
         label_spaces_char_map[i] = (isspace(i) || iscntrl(i) || !isprint(i))?1:0;
 
@@ -460,8 +460,8 @@ __attribute__((constructor)) void initialize_label_stats(void) {
     dictionary_stats_category_rrdlabels.memory.values = 0;
 }
 
-size_t text_sanitize(unsigned char *dst, const unsigned char *src, size_t dst_size, unsigned char *char_map, bool utf, const char *empty, size_t *multibyte_length) {
-    if(unlikely(!dst_size)) return 0;
+size_t text_sanitize(unsigned char *dst, const unsigned char *src, size_t dst_size, const unsigned char *char_map, bool utf, const char *empty, size_t *multibyte_length) {
+    if(unlikely(!src || !dst_size)) return 0;
 
     if(unlikely(!src || !*src)) {
         strncpyz((char *)dst, empty, dst_size);
@@ -476,7 +476,7 @@ size_t text_sanitize(unsigned char *dst, const unsigned char *src, size_t dst_si
     // make room for the final string termination
     unsigned char *end = &d[dst_size - 1];
 
-    // copy while converting, but keep only one white space
+    // copy while converting, but keep only one space
     // we start wil last_is_space = 1 to skip leading spaces
     int last_is_space = 1;
 
@@ -671,8 +671,11 @@ void rrdlabels_destroy(RRDLABELS *labels)
     freez(labels);
 }
 
+//
 // Check in labels to see if we have the key specified in label
-static RRDLABEL *rrdlabels_find_label_with_key_unsafe(RRDLABELS *labels, RRDLABEL *label)
+// same_value indicates if the value should also be matched
+//
+static RRDLABEL *rrdlabels_find_label_with_key_unsafe(RRDLABELS *labels, RRDLABEL *label, bool same_value)
 {
     if (unlikely(!labels))
         return NULL;
@@ -683,7 +686,7 @@ static RRDLABEL *rrdlabels_find_label_with_key_unsafe(RRDLABELS *labels, RRDLABE
     RRDLABEL *found = NULL;
     while ((PValue = JudyLFirstThenNext(labels->JudyL, &Index, &first_then_next))) {
         RRDLABEL *lb = (RRDLABEL *)Index;
-        if (lb->index.key == label->index.key && lb != label) {
+        if (lb->index.key == label->index.key && ((lb == label) == same_value)) {
             found = (RRDLABEL *)Index;
             break;
         }
@@ -718,7 +721,7 @@ static void labels_add_already_sanitized(RRDLABELS *labels, const char *key, con
         new_ls |= RRDLABEL_FLAG_NEW;
         *((RRDLABEL_SRC *)PValue) = new_ls;
 
-        RRDLABEL *old_label_with_same_key = rrdlabels_find_label_with_key_unsafe(labels, new_label);
+        RRDLABEL *old_label_with_same_key = rrdlabels_find_label_with_key_unsafe(labels, new_label, false);
         if (old_label_with_same_key) {
             (void) JudyLDel(&labels->JudyL, (Word_t) old_label_with_same_key, PJE0);
             delete_label(old_label_with_same_key);
@@ -982,6 +985,25 @@ int rrdlabels_walkthrough_read(RRDLABELS *labels, int (*callback)(const char *na
     return ret;
 }
 
+static SIMPLE_PATTERN_RESULT rrdlabels_walkthrough_read_sp(RRDLABELS *labels, SIMPLE_PATTERN_RESULT (*callback)(const char *name, const char *value, RRDLABEL_SRC ls, void *data), void *data)
+{
+    SIMPLE_PATTERN_RESULT ret = SP_NOT_MATCHED;
+
+    if(unlikely(!labels || !callback)) return 0;
+
+    RRDLABEL *lb;
+    RRDLABEL_SRC ls;
+    lfe_start_read(labels, lb, ls)
+    {
+        ret = callback(string2str(lb->index.key), string2str(lb->index.value), ls, data);
+        if (ret != SP_NOT_MATCHED)
+            break;
+    }
+    lfe_done(labels);
+
+    return ret;
+}
+
 // ----------------------------------------------------------------------------
 // rrdlabels_migrate_to_these()
 // migrate an existing label list to a new list
@@ -1027,6 +1049,39 @@ void rrdlabels_migrate_to_these(RRDLABELS *dst, RRDLABELS *src) {
     spinlock_unlock(&dst->spinlock);
 }
 
+//
+//
+// Return the common labels count in labels1, labels2
+//
+size_t rrdlabels_common_count(RRDLABELS *labels1, RRDLABELS *labels2)
+{
+    if (!labels1 || !labels2)
+        return 0;
+
+    if (labels1 == labels2)
+        return rrdlabels_entries(labels1);
+
+    RRDLABEL *label;
+    RRDLABEL_SRC ls;
+
+    spinlock_lock(&labels1->spinlock);
+    spinlock_lock(&labels2->spinlock);
+
+    size_t count = 0;
+    lfe_start_nolock(labels2, label, ls)
+    {
+        RRDLABEL *old_label_with_key = rrdlabels_find_label_with_key_unsafe(labels1, label, true);
+        if (old_label_with_key)
+            count++;
+    }
+    lfe_done_nolock();
+
+    spinlock_unlock(&labels2->spinlock);
+    spinlock_unlock(&labels1->spinlock);
+    return count;
+}
+
+
 void rrdlabels_copy(RRDLABELS *dst, RRDLABELS *src)
 {
     if (!dst || !src || (dst == src))
@@ -1042,7 +1097,7 @@ void rrdlabels_copy(RRDLABELS *dst, RRDLABELS *src)
     bool update_statistics = false;
     lfe_start_nolock(src, label, ls)
     {
-        RRDLABEL *old_label_with_key = rrdlabels_find_label_with_key_unsafe(dst, label);
+        RRDLABEL *old_label_with_key = rrdlabels_find_label_with_key_unsafe(dst, label, false);
         Pvoid_t *PValue = JudyLIns(&dst->JudyL, (Word_t)label, PJE0);
         if(unlikely(!PValue || PValue == PJERR))
             fatal("RRDLABELS: corrupted labels array");
@@ -1083,18 +1138,16 @@ struct simple_pattern_match_name_value {
     char equal;
 };
 
-static int simple_pattern_match_name_only_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
+static SIMPLE_PATTERN_RESULT simple_pattern_match_name_only_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
     struct simple_pattern_match_name_value *t = (struct simple_pattern_match_name_value *)data;
     (void)value;
 
     // we return -1 to stop the walkthrough on first match
     t->searches++;
-    if(simple_pattern_matches(t->pattern, name)) return -1;
-
-    return 0;
+    return simple_pattern_matches_extract(t->pattern, name, NULL, 0);
 }
 
-static int simple_pattern_match_name_and_value_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
+static SIMPLE_PATTERN_RESULT simple_pattern_match_name_and_value_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
     struct simple_pattern_match_name_value *t = (struct simple_pattern_match_name_value *)data;
 
     // we return -1 to stop the walkthrough on first match
@@ -1118,13 +1171,10 @@ static int simple_pattern_match_name_and_value_callback(const char *name, const
     *dst = '\0';
 
     t->searches++;
-    if(simple_pattern_matches_length_extract(t->pattern, tmp, dst - tmp, NULL, 0) == SP_MATCHED_POSITIVE)
-        return -1;
-
-    return 0;
+    return simple_pattern_matches_length_extract(t->pattern, tmp, dst - tmp, NULL, 0);
 }
 
-bool rrdlabels_match_simple_pattern_parsed(RRDLABELS *labels, SIMPLE_PATTERN *pattern, char equal, size_t *searches) {
+SIMPLE_PATTERN_RESULT rrdlabels_match_simple_pattern_parsed(RRDLABELS *labels, SIMPLE_PATTERN *pattern, char equal, size_t *searches) {
     if (!labels) return false;
 
     struct simple_pattern_match_name_value t = {
@@ -1133,12 +1183,12 @@ bool rrdlabels_match_simple_pattern_parsed(RRDLABELS *labels, SIMPLE_PATTERN *pa
         .equal = equal
     };
 
-    int ret = rrdlabels_walkthrough_read(labels, equal?simple_pattern_match_name_and_value_callback:simple_pattern_match_name_only_callback, &t);
+    SIMPLE_PATTERN_RESULT ret = rrdlabels_walkthrough_read_sp(labels, equal?simple_pattern_match_name_and_value_callback:simple_pattern_match_name_only_callback, &t);
 
     if(searches)
         *searches = t.searches;
 
-    return (ret == -1)?true:false;
+    return ret;
 }
 
 bool rrdlabels_match_simple_pattern(RRDLABELS *labels, const char *simple_pattern_txt) {
@@ -1155,11 +1205,11 @@ bool rrdlabels_match_simple_pattern(RRDLABELS *labels, const char *simple_patter
         }
     }
 
-    bool ret = rrdlabels_match_simple_pattern_parsed(labels, pattern, equal, NULL);
+    SIMPLE_PATTERN_RESULT ret = rrdlabels_match_simple_pattern_parsed(labels, pattern, equal, NULL);
 
     simple_pattern_free(pattern);
 
-    return ret;
+    return ret == SP_MATCHED_POSITIVE;
 }
 
 
@@ -1311,6 +1361,153 @@ void rrdset_update_rrdlabels(RRDSET *st, RRDLABELS *new_rrdlabels) {
     rrdset_metadata_updated(st);
 }
 
+struct pattern_array *pattern_array_allocate()
+{
+    struct pattern_array *pa = callocz(1, sizeof(*pa));
+    return pa;
+}
+
+static void pattern_array_add_lblkey_with_sp(struct pattern_array *pa, const char *key, SIMPLE_PATTERN *sp)
+{
+    if (!pa || !key || !sp)
+        return;
+
+    STRING *string_key = string_strdupz(key);
+    Pvoid_t *Pvalue = JudyLIns(&pa->JudyL, (Word_t) string_key, PJE0);
+    if (!Pvalue) {
+        string_freez(string_key);
+        simple_pattern_free(sp);
+        return;
+    }
+
+    struct pattern_array_item *pai;
+    if (*Pvalue) {
+        pai = *Pvalue;
+    } else {
+        *Pvalue = pai = callocz(1, sizeof(*pai));
+        pa->key_count++;
+    }
+
+    pai->size++;
+    Pvalue = JudyLIns(&pai->JudyL, (Word_t) pai->size, PJE0);
+    if (!Pvalue) {
+        simple_pattern_free(sp);
+        return;
+    }
+
+    *Pvalue = sp;
+}
+
+bool pattern_array_label_match(
+    struct pattern_array *pa,
+    RRDLABELS *labels,
+    char eq,
+    size_t *searches)
+{
+    if (!pa || !labels)
+        return true;
+
+    Pvoid_t *Pvalue;
+    Word_t Index = 0;
+    bool first_then_next = true;
+    while ((Pvalue = JudyLFirstThenNext(pa->JudyL, &Index, &first_then_next))) {
+        // for each label key in the patterns array
+
+        struct pattern_array_item *pai = *Pvalue;
+        SIMPLE_PATTERN_RESULT match = SP_NOT_MATCHED ;
+        for (Word_t i = 1; i <= pai->size; i++) {
+            // for each pattern in the label key pattern list
+
+            if (!(Pvalue = JudyLGet(pai->JudyL, i, PJE0)) || !*Pvalue)
+                continue;
+
+            match = rrdlabels_match_simple_pattern_parsed(labels, (SIMPLE_PATTERN *)(*Pvalue), eq, searches);
+
+            if(match != SP_NOT_MATCHED)
+                break;
+        }
+
+        if (match != SP_MATCHED_POSITIVE)
+            return false;
+    }
+    return true;
+}
+
+struct pattern_array *pattern_array_add_key_simple_pattern(struct pattern_array *pa, const char *key, SIMPLE_PATTERN *pattern)
+{
+    if (unlikely(!pattern || !key))
+        return pa;
+
+    if (!pa)
+        pa = pattern_array_allocate();
+
+    pattern_array_add_lblkey_with_sp(pa, key, pattern);
+    return pa;
+}
+
+struct pattern_array *pattern_array_add_simple_pattern(struct pattern_array *pa, SIMPLE_PATTERN *pattern, char sep)
+{
+    if (unlikely(!pattern))
+        return pa;
+
+    if (!pa)
+        pa = pattern_array_allocate();
+
+    char *label_key;
+    while (pattern && (label_key = simple_pattern_iterate(&pattern))) {
+        char key[RRDLABELS_MAX_NAME_LENGTH + 1], *key_sep;
+
+        if (unlikely(!label_key || !(key_sep = strchr(label_key, sep))))
+            return pa;
+
+        *key_sep = '\0';
+        strncpyz(key, label_key, RRDLABELS_MAX_NAME_LENGTH);
+        *key_sep = sep;
+
+        pattern_array_add_lblkey_with_sp(pa, key, string_to_simple_pattern(label_key));
+    }
+    return pa;
+}
+
+struct pattern_array *pattern_array_add_key_value(struct pattern_array *pa, const char *key, const char *value, char sep)
+{
+    if (unlikely(!key || !value))
+        return pa;
+
+    if (!pa)
+        pa = pattern_array_allocate();
+
+    char label_key[RRDLABELS_MAX_NAME_LENGTH + RRDLABELS_MAX_VALUE_LENGTH + 2];
+    snprintfz(label_key, sizeof(label_key) - 1, "%s%c%s", key, sep, value);
+    pattern_array_add_lblkey_with_sp(
+        pa, key, simple_pattern_create(label_key, SIMPLE_PATTERN_DEFAULT_WEB_SEPARATORS, SIMPLE_PATTERN_EXACT, true));
+    return pa;
+}
+
+void pattern_array_free(struct pattern_array *pa)
+{
+    if (!pa)
+        return;
+
+    Pvoid_t *Pvalue;
+    Word_t Index = 0;
+    while ((Pvalue = JudyLFirst(pa->JudyL, &Index, PJE0))) {
+        struct pattern_array_item *pai = *Pvalue;
+
+        for (Word_t i = 1; i <= pai->size; i++) {
+            if (!(Pvalue = JudyLGet(pai->JudyL, i, PJE0)))
+                continue;
+            simple_pattern_free((SIMPLE_PATTERN *) (*Pvalue));
+        }
+        JudyLFreeArray(&(pai->JudyL), PJE0);
+
+        string_freez((STRING *)Index);
+        (void) JudyLDel(&(pa->JudyL), Index, PJE0);
+        freez(pai);
+        Index = 0;
+    }
+    freez(pa);
+}
 
 // ----------------------------------------------------------------------------
 // rrdlabels unit test
@@ -1497,6 +1694,7 @@ static int rrdlabels_walkthrough_index_read(RRDLABELS *labels, int (*callback)(c
         ret = callback(string2str(lb->index.key), string2str(lb->index.value), ls, index, data);
         if (ret < 0)
             break;
+
         index++;
     }
     lfe_done(labels);
@@ -1513,6 +1711,70 @@ static int unittest_dump_labels(const char *name, const char *value, RRDLABEL_SR
     return 1;
 }
 
+static int rrdlabels_unittest_pattern_check()
+{
+    fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
+    int rc = 0;
+
+    RRDLABELS *labels = NULL;
+
+    labels = rrdlabels_create();
+
+    rrdlabels_add(labels, "_module", "disk_detection", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "_plugin", "super_plugin", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "key1", "value1", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "key2", "caterpillar", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "key3", "elephant", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "key4", "value4", RRDLABEL_SRC_CONFIG);
+
+    bool match;
+    struct pattern_array *pa = pattern_array_add_key_value(NULL, "_module", "wrong_module", '=');
+    match = pattern_array_label_match(pa, labels, '=', NULL);
+    // This should not match:  _module in ("wrong_module")
+    if (match)
+        rc++;
+
+    pattern_array_add_key_value(pa, "_module", "disk_detection", '=');
+    match = pattern_array_label_match(pa, labels, '=', NULL);
+    // This should match: _module in ("wrong_module","disk_detection")
+    if (!match)
+        rc++;
+
+    pattern_array_add_key_value(pa, "key1", "wrong_key1_value", '=');
+    match = pattern_array_label_match(pa, labels, '=', NULL);
+    // This should not match: _module in ("wrong_module","disk_detection") AND key1 in ("wrong_key1_value")
+    if (match)
+        rc++;
+
+    pattern_array_add_key_value(pa, "key1", "value1", '=');
+    match = pattern_array_label_match(pa, labels, '=', NULL);
+    // This should match: _module in ("wrong_module","disk_detection") AND key1 in ("wrong_key1_value", "value1")
+    if (!match)
+        rc++;
+
+    SIMPLE_PATTERN *sp = simple_pattern_create("key2=cat*,!d*", SIMPLE_PATTERN_DEFAULT_WEB_SEPARATORS, SIMPLE_PATTERN_EXACT, true);
+    pattern_array_add_lblkey_with_sp(pa, "key2", sp);
+
+    sp = simple_pattern_create("key3=*phant", SIMPLE_PATTERN_DEFAULT_WEB_SEPARATORS, SIMPLE_PATTERN_EXACT, true);
+    pattern_array_add_lblkey_with_sp(pa, "key3", sp);
+
+    match = pattern_array_label_match(pa, labels, '=', NULL);
+    // This should match: _module in ("wrong_module","disk_detection") AND key1 in ("wrong_key1_value", "value1") AND key2 in ("cat* !d*") AND key3 in ("*phant")
+    if (!match)
+        rc++;
+
+    rrdlabels_add(labels, "key3", "now_fail", RRDLABEL_SRC_CONFIG);
+    match = pattern_array_label_match(pa, labels, '=', NULL);
+    // This should not match: _module in ("wrong_module","disk_detection") AND key1 in ("wrong_key1_value", "value1") AND key2 in ("cat* !d*") AND key3 in ("*phant")
+    if (match)
+        rc++;
+
+    pattern_array_free(pa);
+    rrdlabels_destroy(labels);
+
+    return rc;
+}
+
 static int rrdlabels_unittest_migrate_check()
 {
     fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
@@ -1594,6 +1856,69 @@ static int rrdlabels_unittest_migrate_check()
     return rc;
 }
 
+struct pattern_array *trim_and_add_key_to_values(struct pattern_array *pa, const char *key, STRING *input);
+static int rrdlabels_unittest_check_pattern_list(RRDLABELS *labels, const char *pattern, bool expected) {
+    fprintf(stderr, "rrdlabels_match_simple_pattern(labels, \"%s\") ... ", pattern);
+
+    STRING *str = string_strdupz(pattern);
+    struct pattern_array *pa = trim_and_add_key_to_values(NULL, NULL, str);
+
+    bool ret = pattern_array_label_match(pa, labels, '=', NULL);
+
+    fprintf(stderr, "%s, got %s expected %s\n", (ret == expected)?"OK":"FAILED", ret?"true":"false", expected?"true":"false");
+
+    string_freez(str);
+    pattern_array_free(pa);
+
+    return (ret == expected)?0:1;
+}
+
+static int rrdlabels_unittest_host_chart_labels() {
+    fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
+
+    int errors = 0;
+
+    RRDLABELS *labels = rrdlabels_create();
+    rrdlabels_add(labels, "_hostname", "hostname1", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "_os", "linux", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels, "_distro", "ubuntu", RRDLABEL_SRC_CONFIG);
+
+    // match a single key
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=*", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=!*", false);
+
+    // conflicting keys (some positive, some negative)
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=* _os=!*", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=!* _os=*", false);
+
+    // the user uses a key that is not there
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_not_a_key=*", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_not_a_key=!*", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_not_a_key=* _hostname=* _os=*", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_not_a_key=!* _hostname=* _os=*", false);
+
+    // positive and negative matches on the same key
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=!*invalid* !*bad* *name*", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=*name* !*invalid* !*bad*", true);
+
+    // positive and negative matches on the same key with catch all
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=!*invalid* !*bad* *", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=* !*invalid* !*bad*", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=!*invalid* !*name* *", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=* !*invalid* !*name*", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=*name* !*", true);
+
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=!*name* _os=l*", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_os=l* hostname=!*name*", false);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=*name* _hostname=*", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_hostname=*name* _os=l*", true);
+    errors += rrdlabels_unittest_check_pattern_list(labels, "_os=l* _hostname=*name*", true);
+
+    rrdlabels_destroy(labels);
+
+    return errors;
+}
+
 static int rrdlabels_unittest_check_simple_pattern(RRDLABELS *labels, const char *pattern, bool expected) {
     fprintf(stderr, "rrdlabels_match_simple_pattern(labels, \"%s\") ... ", pattern);
 
@@ -1686,9 +2011,12 @@ int rrdlabels_unittest(void) {
     errors += rrdlabels_unittest_sanitization();
     errors += rrdlabels_unittest_add_pairs();
     errors += rrdlabels_unittest_simple_pattern();
+    errors += rrdlabels_unittest_host_chart_labels();
     errors += rrdlabels_unittest_double_check();
     errors += rrdlabels_unittest_migrate_check();
+    errors += rrdlabels_unittest_pattern_check();
 
     fprintf(stderr, "%d errors found\n", errors);
     return errors;
 }
+