fix positive and negative matches on labels (#17290)

* fix positive and negative matches

* fix typo

* fix unit tests

* unittest for rrdlabels pattern lists

6 files changed, 115 insertions, 35 deletions

diff --git a/src/database/contexts/api_v1.c b/src/database/contexts/api_v1.c
index f144e6f7b8..355aaf91a3 100644
--- a/src/database/contexts/api_v1.c
+++ b/src/database/contexts/api_v1.c
@@ -131,13 +131,13 @@ static inline int rrdinstance_to_json_callback(const DICTIONARY_ITEM *item, void
     if(before && (!ri->first_time_s || before < ri->first_time_s))
         return 0;
 
-    if(t_parent->chart_label_key && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, t_parent->chart_label_key,
-                                                                           '\0', NULL))
+    if(t_parent->chart_label_key && rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, t_parent->chart_label_key,
+                                                                           '\0', NULL) != SP_MATCHED_POSITIVE)
         return 0;
 
-    if(t_parent->chart_labels_filter && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels,
+    if(t_parent->chart_labels_filter && rrdlabels_match_simple_pattern_parsed(ri->rrdlabels,
                                                                                t_parent->chart_labels_filter, ':',
-                                                                               NULL))
+                                                                               NULL) != SP_MATCHED_POSITIVE)
         return 0;
 
     time_t first_time_s = ri->first_time_s;
diff --git a/src/database/contexts/api_v2.c b/src/database/contexts/api_v2.c
index 24163e703c..edfeab1d63 100644
--- a/src/database/contexts/api_v2.c
+++ b/src/database/contexts/api_v2.c
@@ -473,7 +473,7 @@ static FTS_MATCH rrdcontext_to_json_v2_full_text_search(struct rrdcontext_to_jso
 
         size_t label_searches = 0;
         if(unlikely(ri->rrdlabels && rrdlabels_entries(ri->rrdlabels) &&
-                    rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, q, ':', &label_searches))) {
+                    rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, q, ':', &label_searches) == SP_MATCHED_POSITIVE)) {
             ctl->q.fts.searches += label_searches;
             ctl->q.fts.char_searches += label_searches;
             matched = FTS_MATCHED_LABEL;
diff --git a/src/database/contexts/query_target.c b/src/database/contexts/query_target.c
index 1bc3014aa8..29a9c3e591 100644
--- a/src/database/contexts/query_target.c
+++ b/src/database/contexts/query_target.c
@@ -732,12 +732,12 @@ static inline bool query_instance_matches_labels(
     SIMPLE_PATTERN *labels_sp)
 {
 
-    if (chart_label_key_sp && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, chart_label_key_sp, '\0', NULL))
+    if (chart_label_key_sp && rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, chart_label_key_sp, '\0', NULL) != SP_MATCHED_POSITIVE)
         return false;
 
     if (labels_sp) {
         struct pattern_array *pa = pattern_array_add_simple_pattern(NULL, labels_sp, ':');
-        bool found = pattern_array_label_match(pa, ri->rrdlabels, ':', NULL, rrdlabels_match_simple_pattern_parsed);
+        bool found = pattern_array_label_match(pa, ri->rrdlabels, ':', NULL);
         pattern_array_free(pa);
         return found;
     }
diff --git a/src/database/rrdlabels.c b/src/database/rrdlabels.c
index 633287cba9..929d0064e9 100644
--- a/src/database/rrdlabels.c
+++ b/src/database/rrdlabels.c
@@ -977,7 +977,7 @@ int rrdlabels_walkthrough_read(RRDLABELS *labels, int (*callback)(const char *na
     lfe_start_read(labels, lb, ls)
     {
         ret = callback(string2str(lb->index.key), string2str(lb->index.value), ls, data);
-        if (ret < 0)
+        if (ret != 0)
             break;
     }
     lfe_done(labels);
@@ -1125,9 +1125,12 @@ static int simple_pattern_match_name_only_callback(const char *name, const char
 
     // we return -1 to stop the walkthrough on first match
     t->searches++;
-    if(simple_pattern_matches(t->pattern, name)) return -1;
+    SIMPLE_PATTERN_RESULT rc = simple_pattern_matches_extract(t->pattern, name, NULL, 0);
 
-    return 0;
+    if(rc == SP_MATCHED_NEGATIVE)
+        return -1;
+
+    return rc == SP_MATCHED_POSITIVE;
 }
 
 static int simple_pattern_match_name_and_value_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
@@ -1154,13 +1157,15 @@ 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)
+    SIMPLE_PATTERN_RESULT rc = simple_pattern_matches_length_extract(t->pattern, tmp, dst - tmp, NULL, 0);
+
+    if(rc == SP_MATCHED_NEGATIVE)
         return -1;
 
-    return 0;
+    return rc == SP_MATCHED_POSITIVE ? 1 : 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 = {
@@ -1174,7 +1179,10 @@ bool rrdlabels_match_simple_pattern_parsed(RRDLABELS *labels, SIMPLE_PATTERN *pa
     if(searches)
         *searches = t.searches;
 
-    return (ret == -1)?true:false;
+    if(ret < 0)
+        return SP_MATCHED_NEGATIVE;
+
+    return (ret > 0)?SP_MATCHED_POSITIVE:SP_NOT_MATCHED;
 }
 
 bool rrdlabels_match_simple_pattern(RRDLABELS *labels, const char *simple_pattern_txt) {
@@ -1191,11 +1199,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;
 }
 
 
@@ -1388,8 +1396,7 @@ bool pattern_array_label_match(
     struct pattern_array *pa,
     RRDLABELS *labels,
     char eq,
-    size_t *searches,
-    bool (*callback_function)(RRDLABELS *, SIMPLE_PATTERN *, char, size_t *))
+    size_t *searches)
 {
     if (!pa || !labels)
         return true;
@@ -1398,14 +1405,23 @@ bool pattern_array_label_match(
     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;
-        bool match = false;
-        for (Word_t i = 1; !match && i <= pai->size; i++) {
+        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 = callback_function(labels, (SIMPLE_PATTERN *)(*Pvalue), eq, searches);
+
+            match = rrdlabels_match_simple_pattern_parsed(labels, (SIMPLE_PATTERN *)(*Pvalue), eq, searches);
+
+            if(match != SP_NOT_MATCHED)
+                break;
         }
-        if (!match)
+
+        if (match != SP_MATCHED_POSITIVE)
             return false;
     }
     return true;
@@ -1481,6 +1497,7 @@ void pattern_array_free(struct pattern_array *pa)
 
         string_freez((STRING *)Index);
         (void) JudyLDel(&(pa->JudyL), Index, PJE0);
+        freez(pai);
         Index = 0;
     }
     freez(pa);
@@ -1669,7 +1686,7 @@ static int rrdlabels_walkthrough_index_read(RRDLABELS *labels, int (*callback)(c
     lfe_start_read(labels, lb, ls)
     {
         ret = callback(string2str(lb->index.key), string2str(lb->index.value), ls, index, data);
-        if (ret < 0)
+        if (ret != 0)
             break;
         index++;
     }
@@ -1705,25 +1722,25 @@ static int rrdlabels_unittest_pattern_check()
 
     bool match;
     struct pattern_array *pa = pattern_array_add_key_value(NULL, "_module", "wrong_module", '=');
-    match = pattern_array_label_match(pa, labels, '=', NULL, rrdlabels_match_simple_pattern_parsed);
+    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, rrdlabels_match_simple_pattern_parsed);
+    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, rrdlabels_match_simple_pattern_parsed);
+    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, rrdlabels_match_simple_pattern_parsed);
+    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++;
@@ -1734,13 +1751,13 @@ static int rrdlabels_unittest_pattern_check()
     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, rrdlabels_match_simple_pattern_parsed);
+    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, rrdlabels_match_simple_pattern_parsed);
+    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++;
@@ -1832,6 +1849,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);
 
@@ -1924,6 +2004,7 @@ 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();
diff --git a/src/database/rrdlabels.h b/src/database/rrdlabels.h
index a8d7838430..88b35cf926 100644
--- a/src/database/rrdlabels.h
+++ b/src/database/rrdlabels.h
@@ -51,7 +51,7 @@ int rrdlabels_walkthrough_read(RRDLABELS *labels, int (*callback)(const char *na
 void rrdlabels_log_to_buffer(RRDLABELS *labels, BUFFER *wb);
 bool rrdlabels_match_simple_pattern(RRDLABELS *labels, const char *simple_pattern_txt);
 
-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);
 int rrdlabels_to_buffer(RRDLABELS *labels, BUFFER *wb, const char *before_each, const char *equal, const char *quote, const char *between_them,
                         bool (*filter_callback)(const char *name, const char *value, RRDLABEL_SRC ls, void *data), void *filter_data,
                         void (*name_sanitizer)(char *dst, const char *src, size_t dst_size),
@@ -69,8 +69,7 @@ bool pattern_array_label_match(
     struct pattern_array *pa,
     RRDLABELS *labels,
     char eq,
-    size_t *searches,
-    bool (*callback_function)(RRDLABELS *, SIMPLE_PATTERN *, char, size_t *));
+    size_t *searches);
 struct pattern_array *pattern_array_add_simple_pattern(struct pattern_array *pa, SIMPLE_PATTERN *pattern, char sep);
 struct pattern_array *
 pattern_array_add_key_simple_pattern(struct pattern_array *pa, const char *key, SIMPLE_PATTERN *pattern);
diff --git a/src/health/health_prototypes.c b/src/health/health_prototypes.c
index 8d585ac1b4..8e343bfa1a 100644
--- a/src/health/health_prototypes.c
+++ b/src/health/health_prototypes.c
@@ -350,7 +350,7 @@ static char *simple_pattern_trim_around_equal(const char *src) {
     return store;
 }
 
-static struct pattern_array *trim_and_add_key_to_values(struct pattern_array *pa, const char *key, STRING *input) {
+struct pattern_array *trim_and_add_key_to_values(struct pattern_array *pa, const char *key, STRING *input) {
     char *tmp = simple_pattern_trim_around_equal(string2str(input));
     pa = health_config_add_key_to_values(pa, key, tmp);
     freez(tmp);
@@ -483,7 +483,7 @@ static bool prototype_matches_host(RRDHOST *host, RRD_ALERT_PROTOTYPE *ap) {
         return false;
 
     if (host->rrdlabels && ap->match.host_labels_pattern &&
-        !pattern_array_label_match(ap->match.host_labels_pattern, host->rrdlabels, '=', NULL, rrdlabels_match_simple_pattern_parsed))
+        !pattern_array_label_match(ap->match.host_labels_pattern, host->rrdlabels, '=', NULL))
         return false;
 
     return true;
@@ -501,7 +501,7 @@ static bool prototype_matches_rrdset(RRDSET *st, RRD_ALERT_PROTOTYPE *ap) {
         return false;
 
     if (st->rrdlabels && ap->match.chart_labels_pattern &&
-        !pattern_array_label_match(ap->match.chart_labels_pattern, st->rrdlabels, '=', NULL, rrdlabels_match_simple_pattern_parsed))
+        !pattern_array_label_match(ap->match.chart_labels_pattern, st->rrdlabels, '=', NULL))
         return false;
 
     return true;