Change query label matching logic (#16827)

* Match multi labels

* Rework, add support for weights

* Fix function return value

* Cleanup function

2 files changed, 122 insertions, 7 deletions

diff --git a/database/contexts/query_target.c b/database/contexts/query_target.c
index 69f7efd0d0..31ed689bda 100644
--- a/database/contexts/query_target.c
+++ b/database/contexts/query_target.c
@@ -11,6 +11,7 @@ static void query_dimension_release(QUERY_DIMENSION *qd);
 static void query_instance_release(QUERY_INSTANCE *qi);
 static void query_context_release(QUERY_CONTEXT *qc);
 static void query_node_release(QUERY_NODE *qn);
+static void free_label_pattern_list(struct label_pattern_list *lpl);
 
 static __thread QUERY_TARGET *thread_qt = NULL;
 static struct {
@@ -83,6 +84,9 @@ void query_target_release(QUERY_TARGET *qt) {
     simple_pattern_free(qt->instances.labels_pattern);
     qt->instances.labels_pattern = NULL;
 
+    free_label_pattern_list(qt->instances.label_pattern_list);
+    qt->instances.label_pattern_list = NULL;
+
     simple_pattern_free(qt->query.pattern);
     qt->query.pattern = NULL;
 
@@ -725,11 +729,25 @@ static inline SIMPLE_PATTERN_RESULT query_instance_matches(QUERY_INSTANCE *qi,
     return ret;
 }
 
-static inline bool query_instance_matches_labels(RRDINSTANCE *ri, SIMPLE_PATTERN *chart_label_key_sp, SIMPLE_PATTERN *labels_sp) {
-    if ((chart_label_key_sp && !rrdlabels_match_simple_pattern_parsed(
-            ri->rrdlabels, chart_label_key_sp, '\0', NULL)) ||
-        (labels_sp && !rrdlabels_match_simple_pattern_parsed(
-                ri->rrdlabels, labels_sp, ':', NULL)))
+static inline bool query_instance_matches_labels(
+    RRDINSTANCE *ri,
+    SIMPLE_PATTERN *chart_label_key_sp,
+    SIMPLE_PATTERN *labels_sp,
+    struct label_pattern_list *lpl)
+{
+
+    if (chart_label_key_sp && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, chart_label_key_sp, '\0', NULL))
+        return false;
+
+    if (lpl) {
+        for (size_t i = 0; i < lpl->size; i++) {
+            if (!rrdlabels_match_simple_pattern_parsed(ri->rrdlabels,  lpl->labels_pattern[i], ':', NULL))
+                return false;
+        }
+        return true;
+    }
+
+    if (labels_sp && !rrdlabels_match_simple_pattern_parsed(ri->rrdlabels, labels_sp, ':', NULL))
         return false;
 
     return true;
@@ -752,7 +770,11 @@ static bool query_instance_add(QUERY_TARGET_LOCALS *qtl, QUERY_NODE *qn, QUERY_C
                 qi, ri, qt->instances.pattern, qtl->match_ids, qtl->match_names, qt->request.version, qtl->host_node_id_str));
 
     if(queryable_instance)
-        queryable_instance = query_instance_matches_labels(ri, qt->instances.chart_label_key_pattern, qt->instances.labels_pattern);
+        queryable_instance = query_instance_matches_labels(
+            ri,
+            qt->instances.chart_label_key_pattern,
+            qt->instances.labels_pattern,
+            qt->instances.label_pattern_list);
 
     if(queryable_instance) {
         if(qt->instances.alerts_pattern && !query_target_match_alert_pattern(ria, qt->instances.alerts_pattern))
@@ -1021,6 +1043,82 @@ void query_target_generate_name(QUERY_TARGET *qt) {
     json_fix_string(qt->id);
 }
 
+static void add_label_pattern(struct label_pattern_list *lpl, char *label_key_value)
+{
+    char *label_key;
+
+    if (unlikely(!label_key_value || !(label_key = strchr(label_key_value, ':'))))
+        return;
+
+    *label_key = '\0';
+    STRING *key_match = string_strdupz(label_key_value);
+    *label_key = ':';
+
+    size_t index;
+    bool need_to_add = true;
+
+    for (size_t i = 0; i < lpl->size; i++) {
+        if (lpl->key[i] == key_match) {
+            index = i;
+            need_to_add = false;
+            break;
+        }
+    }
+
+    if (need_to_add) {
+        index = lpl->size++;
+        lpl->buffer_list = reallocz(lpl->buffer_list, lpl->size * sizeof(BUFFER *));
+        lpl->key = reallocz(lpl->key, lpl->size * sizeof(STRING *));
+
+        lpl->buffer_list[index] = buffer_create(128, NULL);
+        lpl->key[index] = key_match;
+    } else
+        buffer_strncat(lpl->buffer_list[index], ",", 1);
+
+    buffer_strcat(lpl->buffer_list[index], label_key_value);
+}
+
+static struct label_pattern_list *build_pattern_list(SIMPLE_PATTERN *pattern)
+{
+    if (unlikely(!pattern))
+        return NULL;
+
+    char *label_key = NULL;
+
+    struct label_pattern_list *lpl = callocz(1, sizeof(*lpl));
+
+    while (pattern && (label_key = simple_pattern_iterate(&pattern)))
+        add_label_pattern(lpl, label_key);
+
+    lpl->labels_pattern = callocz(lpl->size, sizeof(SIMPLE_PATTERN *));
+
+    for (size_t i = 0; i < lpl->size; i++) {
+        lpl->labels_pattern[i] = string_to_simple_pattern(buffer_tostring(lpl->buffer_list[i]));
+        buffer_free(lpl->buffer_list[i]);
+        string_freez(lpl->key[i]);
+    }
+
+    freez(lpl->buffer_list);
+    lpl->buffer_list = NULL;
+
+    freez(lpl->key);
+    lpl->key = NULL;
+    return lpl;
+}
+
+static void free_label_pattern_list(struct label_pattern_list *lpl)
+{
+    if (unlikely(!lpl))
+        return;
+
+    for(size_t i = 0; i < lpl->size; i++)
+        simple_pattern_free(lpl->labels_pattern[i]);
+
+    freez(lpl->labels_pattern);
+    freez(lpl);
+}
+
+
 QUERY_TARGET *query_target_create(QUERY_TARGET_REQUEST *qtr) {
     if(!service_running(ABILITY_DATA_QUERIES))
         return NULL;
@@ -1085,6 +1183,10 @@ QUERY_TARGET *query_target_create(QUERY_TARGET_REQUEST *qtr) {
     qt->query.pattern = string_to_simple_pattern(qtl.dimensions);
     qt->instances.chart_label_key_pattern = string_to_simple_pattern(qtl.chart_label_key);
     qt->instances.labels_pattern = string_to_simple_pattern(qtl.labels);
+
+    if (qt->instances.labels_pattern)
+        qt->instances.label_pattern_list = build_pattern_list(qt->instances.labels_pattern);
+
     qt->instances.alerts_pattern = string_to_simple_pattern(qtl.alerts);
 
     qtl.match_ids = qt->request.options & RRDR_OPTION_MATCH_IDS;
@@ -1156,6 +1258,11 @@ ssize_t weights_foreach_rrdmetric_in_context(RRDCONTEXT_ACQUIRED *rca,
 
     ssize_t count = 0;
     RRDINSTANCE *ri;
+
+    struct label_pattern_list *lpl = NULL;
+    if (labels_sp)
+        lpl = build_pattern_list(labels_sp);
+
     dfe_start_read(rc->rrdinstances, ri) {
                 if(rrd_flag_is_deleted(ri))
                     continue;
@@ -1172,7 +1279,7 @@ ssize_t weights_foreach_rrdmetric_in_context(RRDCONTEXT_ACQUIRED *rca,
                         continue;
                 }
 
-                if(!query_instance_matches_labels(ri, chart_label_key_sp, labels_sp))
+                if(!query_instance_matches_labels(ri, chart_label_key_sp, labels_sp, lpl))
                     continue;
 
                 if(alerts_sp && !query_target_match_alert_pattern(ria, alerts_sp))
diff --git a/database/contexts/rrdcontext.h b/database/contexts/rrdcontext.h
index d576bb0162..0919435fa0 100644
--- a/database/contexts/rrdcontext.h
+++ b/database/contexts/rrdcontext.h
@@ -329,6 +329,13 @@ struct query_timings {
     usec_t finished_ut;
 };
 
+struct label_pattern_list {
+    BUFFER **buffer_list;
+    STRING **key;
+    SIMPLE_PATTERN **labels_pattern;
+    size_t size;
+};
+
 #define query_view_update_every(qt) ((qt)->window.group * (qt)->window.query_granularity)
 
 typedef struct query_target {
@@ -379,6 +386,7 @@ typedef struct query_target {
         uint32_t size;                      // the size of the array
         SIMPLE_PATTERN *pattern;
         SIMPLE_PATTERN *labels_pattern;
+        struct label_pattern_list *label_pattern_list;
         SIMPLE_PATTERN *alerts_pattern;
         SIMPLE_PATTERN *chart_label_key_pattern;
     } instances;