1 files changed, 117 insertions, 68 deletions

diff --git a/database/rrdlabels.c b/database/rrdlabels.c
index 6505b4b2d..69ee55526 100644
--- a/database/rrdlabels.c
+++ b/database/rrdlabels.c
@@ -710,10 +710,13 @@ static void labels_add_already_sanitized(RRDLABELS *labels, const char *key, con
 
     if(*PValue) {
         new_ls |= RRDLABEL_FLAG_OLD;
+        *((RRDLABEL_SRC *)PValue) = new_ls;
+
         delete_label(new_label);
     }
     else {
         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);
         if (old_label_with_same_key) {
@@ -723,7 +726,6 @@ static void labels_add_already_sanitized(RRDLABELS *labels, const char *key, con
     }
 
     labels->version++;
-    *((RRDLABEL_SRC *)PValue) = new_ls;
 
     size_t mem_after_judyl = JudyLMemUsed(labels->JudyL);
     STATS_PLUS_MEMORY(&dictionary_stats_category_rrdlabels, 0, mem_after_judyl - mem_before_judyl, 0);
@@ -934,7 +936,7 @@ static void rrdlabels_remove_all_unmarked_unsafe(RRDLABELS *labels)
     bool first_then_next = true;
 
     while ((PValue = JudyLFirstThenNext(labels->JudyL, &Index, &first_then_next))) {
-        if (!((*((RRDLABEL_SRC *)PValue)) & (RRDLABEL_FLAG_OLD | RRDLABEL_FLAG_NEW | RRDLABEL_FLAG_PERMANENT))) {
+        if (!((*((RRDLABEL_SRC *)PValue)) & (RRDLABEL_FLAG_INTERNAL))) {
 
             size_t mem_before_judyl = JudyLMemUsed(labels->JudyL);
             (void)JudyLDel(&labels->JudyL, Index, PJE0);
@@ -1004,8 +1006,9 @@ void rrdlabels_migrate_to_these(RRDLABELS *dst, RRDLABELS *src) {
         if(unlikely(!PValue || PValue == PJERR))
             fatal("RRDLABELS migrate: corrupted labels array");
 
-        RRDLABEL_SRC flag = RRDLABEL_FLAG_NEW;
+        RRDLABEL_SRC flag;
         if (!*PValue) {
+            flag = (ls & ~(RRDLABEL_FLAG_OLD | RRDLABEL_FLAG_NEW)) | RRDLABEL_FLAG_NEW;
             dup_label(label);
             size_t mem_after_judyl = JudyLMemUsed(dst->JudyL);
             STATS_PLUS_MEMORY(&dictionary_stats_category_rrdlabels, 0, mem_after_judyl - mem_before_judyl, 0);
@@ -1040,14 +1043,13 @@ void rrdlabels_copy(RRDLABELS *dst, RRDLABELS *src)
     lfe_start_nolock(src, label, ls)
     {
         RRDLABEL *old_label_with_key = rrdlabels_find_label_with_key_unsafe(dst, label);
-
         Pvoid_t *PValue = JudyLIns(&dst->JudyL, (Word_t)label, PJE0);
         if(unlikely(!PValue || PValue == PJERR))
             fatal("RRDLABELS: corrupted labels array");
 
         if (!*PValue) {
             dup_label(label);
-            *((RRDLABEL_SRC *)PValue) = ls;
+            ls = (ls & ~(RRDLABEL_FLAG_OLD)) | RRDLABEL_FLAG_NEW;
             dst->version++;
             update_statistics = true;
             if (old_label_with_key) {
@@ -1055,6 +1057,10 @@ void rrdlabels_copy(RRDLABELS *dst, RRDLABELS *src)
                 delete_label((RRDLABEL *)old_label_with_key);
             }
         }
+        else
+            ls = (ls & ~(RRDLABEL_FLAG_NEW)) | RRDLABEL_FLAG_OLD;
+
+        *((RRDLABEL_SRC *)PValue) = ls;
     }
     lfe_done_nolock();
     if (update_statistics) {
@@ -1302,7 +1308,7 @@ void rrdset_update_rrdlabels(RRDSET *st, RRDLABELS *new_rrdlabels) {
 
     rrdset_flag_set(st, RRDSET_FLAG_METADATA_UPDATE);
     rrdhost_flag_set(st->rrdhost, RRDHOST_FLAG_METADATA_UPDATE);
-    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+    rrdset_metadata_updated(st);
 }
 
 
@@ -1318,7 +1324,30 @@ struct rrdlabels_unittest_add_a_pair {
     int errors;
 };
 
-int rrdlabels_unittest_add_a_pair_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
+RRDLABEL *rrdlabels_find_label_with_key(RRDLABELS *labels, const char *key, RRDLABEL_SRC *source)
+{
+    if (!labels || !key)
+        return NULL;
+
+    STRING *this_key = string_strdupz(key);
+
+    RRDLABEL *lb = NULL;
+    RRDLABEL_SRC ls;
+
+    lfe_start_read(labels, lb, ls)
+    {
+        if (lb->index.key == this_key) {
+            if (source)
+                *source = ls;
+            break;
+        }
+    }
+    lfe_done(labels);
+    string_freez(this_key);
+    return lb;
+}
+
+static int rrdlabels_unittest_add_a_pair_callback(const char *name, const char *value, RRDLABEL_SRC ls __maybe_unused, void *data) {
     struct rrdlabels_unittest_add_a_pair *t = (struct rrdlabels_unittest_add_a_pair *)data;
 
     t->name = name;
@@ -1344,7 +1373,7 @@ int rrdlabels_unittest_add_a_pair_callback(const char *name, const char *value,
     return 1;
 }
 
-int rrdlabels_unittest_add_a_pair(const char *pair, const char *name, const char *value) {
+static int rrdlabels_unittest_add_a_pair(const char *pair, const char *name, const char *value) {
     RRDLABELS *labels = rrdlabels_create();
     int errors;
 
@@ -1374,7 +1403,7 @@ int rrdlabels_unittest_add_a_pair(const char *pair, const char *name, const char
     return errors;
 }
 
-int rrdlabels_unittest_add_pairs() {
+static int rrdlabels_unittest_add_pairs() {
     fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
 
     int errors = 0;
@@ -1422,69 +1451,70 @@ int rrdlabels_unittest_add_pairs() {
     return errors;
 }
 
-int rrdlabels_unittest_double_check() {
+static int rrdlabels_unittest_expect_value(RRDLABELS *labels, const char *key, const char *value, RRDLABEL_SRC required_source)
+{
+    RRDLABEL_SRC source;
+    RRDLABEL *label = rrdlabels_find_label_with_key(labels, key, &source);
+    return (!label || strcmp(string2str(label->index.value), value) != 0 || (source != required_source));
+}
+
+static int rrdlabels_unittest_double_check()
+{
     fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
 
-    int errors = 1;
     int ret = 0;
     RRDLABELS *labels = rrdlabels_create();
 
-    const char *pair = "key1=value1";
-
-    struct rrdlabels_unittest_add_a_pair tmp = {
-        .pair = pair,
-        .expected_name = "key1",
-        .expected_value = NULL,
-        .errors = 0
-    };
+    rrdlabels_add(labels, "key1", "value1", RRDLABEL_SRC_CONFIG);
+    ret += rrdlabels_unittest_expect_value(labels, "key1", "value1", RRDLABEL_FLAG_NEW | RRDLABEL_SRC_CONFIG);
 
-    fprintf(stderr, "rrdlabels_add_pair(labels, %s) ...\n ", pair);
+    rrdlabels_add(labels, "key1", "value2", RRDLABEL_SRC_CONFIG);
+    ret += !rrdlabels_unittest_expect_value(labels, "key1", "value2", RRDLABEL_FLAG_OLD | RRDLABEL_SRC_CONFIG);
 
-    rrdlabels_add_pair(labels, pair, RRDLABEL_SRC_CONFIG);
-    size_t count = rrdlabels_entries(labels);
-    fprintf(stderr, "Added one key with \"value1\", entries found %zu\n", count);
-    tmp.expected_value = "value1";
-    ret = rrdlabels_walkthrough_read(labels, rrdlabels_unittest_add_a_pair_callback, &tmp);
+    rrdlabels_add(labels, "key2", "value1", RRDLABEL_SRC_ACLK|RRDLABEL_SRC_AUTO);
+    ret += !rrdlabels_unittest_expect_value(labels, "key1", "value3", RRDLABEL_FLAG_NEW | RRDLABEL_SRC_ACLK);
 
-    fprintf(stderr, "Adding key with same value \"value1\" (collision check)\n");
-    rrdlabels_add_pair(labels, pair, RRDLABEL_SRC_CONFIG);
-    count = rrdlabels_entries(labels);
-    fprintf(stderr, "Added same key again \"value1\", entries found %zu\n", count);
+    ret += (rrdlabels_entries(labels) != 2);
 
-    ret = rrdlabels_walkthrough_read(labels, rrdlabels_unittest_add_a_pair_callback, &tmp);
+    rrdlabels_destroy(labels);
 
-    // Add same key with different value
-    pair = "key1=value2";
-    rrdlabels_add_pair(labels, pair, RRDLABEL_SRC_CONFIG);
-    count = rrdlabels_entries(labels);
-    fprintf(stderr, "Added same key again with \"value2\", entries found %zu\n", count);
+    if (ret)
+        fprintf(stderr, "\n%s() tests failed\n", __FUNCTION__);
+    return ret;
+}
 
-    tmp.expected_value = "value2";
-    ret = rrdlabels_walkthrough_read(labels, rrdlabels_unittest_add_a_pair_callback, &tmp);
+static int rrdlabels_walkthrough_index_read(RRDLABELS *labels, int (*callback)(const char *name, const char *value, RRDLABEL_SRC ls, size_t index, void *data), void *data)
+{
+    int ret = 0;
 
-    fprintf(stderr, "Adding key with same value \"value2\" (collision check)\n");
-    rrdlabels_add_pair(labels, pair, RRDLABEL_SRC_CONFIG);
-    count = rrdlabels_entries(labels);
-    fprintf(stderr, "Added same key again with \"value2\", entries found %zu\n", count);
+    if(unlikely(!labels || !callback)) return 0;
 
-    ret = rrdlabels_walkthrough_read(labels, rrdlabels_unittest_add_a_pair_callback, &tmp);
-    errors = tmp.errors;
-    if(ret != 1) {
-        fprintf(stderr, "failed to get \"%s\" label", "key1");
-        errors++;
+    RRDLABEL *lb;
+    RRDLABEL_SRC ls;
+    size_t index = 0;
+    lfe_start_read(labels, lb, ls)
+    {
+        ret = callback(string2str(lb->index.key), string2str(lb->index.value), ls, index, data);
+        if (ret < 0)
+            break;
+        index++;
     }
+    lfe_done(labels);
 
-    if(!errors)
-        fprintf(stderr, " OK, name='%s' and value='%s'\n", tmp.name, tmp.value?tmp.value:"(null)");
-    else
-        fprintf(stderr, " FAILED\n");
-
-    rrdlabels_destroy(labels);
+    return ret;
+}
 
-    return errors;
+static int unittest_dump_labels(const char *name, const char *value, RRDLABEL_SRC ls, size_t index, void *data __maybe_unused)
+{
+    if (!index && data) {
+        fprintf(stderr, "%s\n", (char *) data);
+    }
+    fprintf(stderr, "LABEL \"%s\" = %d \"%s\"\n", name, ls & (~RRDLABEL_FLAG_INTERNAL), value);
+    return 1;
 }
 
-int rrdlabels_unittest_migrate_check() {
+static int rrdlabels_unittest_migrate_check()
+{
     fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
 
     RRDLABELS *labels1 = NULL;
@@ -1504,6 +1534,12 @@ int rrdlabels_unittest_migrate_check() {
     fprintf(stderr, "Labels2 entries found %zu  (should be 3)\n",  rrdlabels_entries(labels2));
 
     rrdlabels_migrate_to_these(labels1, labels2);
+
+    int rc = 0;
+    rc = rrdlabels_unittest_expect_value(labels1, "key1", "value2", RRDLABEL_FLAG_OLD | RRDLABEL_SRC_CONFIG);
+    if (rc)
+        return rc;
+
     fprintf(stderr, "labels1 (migrated) entries found %zu (should be 3)\n",  rrdlabels_entries(labels1));
     size_t entries = rrdlabels_entries(labels1);
 
@@ -1518,34 +1554,47 @@ int rrdlabels_unittest_migrate_check() {
     labels2 = rrdlabels_create();
 
     rrdlabels_add(labels1, "key1", "value1", RRDLABEL_SRC_CONFIG);
-    rrdlabels_add(labels1, "key2", "value1", RRDLABEL_SRC_CONFIG);
-    rrdlabels_add(labels1, "key3", "value1", RRDLABEL_SRC_CONFIG);
-    rrdlabels_add(labels1, "key4", "value1", RRDLABEL_SRC_CONFIG);  // 4 keys
+    rrdlabels_add(labels1, "key2", "value2", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels1, "key3", "value3", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels1, "key4", "value4", RRDLABEL_SRC_CONFIG);  // 4 keys
+    rrdlabels_walkthrough_index_read(labels1, unittest_dump_labels, "\nlabels1");
+
+    rrdlabels_add(labels2, "key0", "value0", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels2, "key1", "value1", RRDLABEL_SRC_CONFIG);
+    rrdlabels_add(labels2, "key2", "value2", RRDLABEL_SRC_CONFIG);
+
+    rc = rrdlabels_unittest_expect_value(labels1, "key1", "value1", RRDLABEL_FLAG_NEW | RRDLABEL_SRC_CONFIG);
+    if (rc)
+        return rc;
 
-    rrdlabels_add(labels2, "key1", "value10", RRDLABEL_SRC_CONFIG);
-    rrdlabels_add(labels2, "key2", "value1", RRDLABEL_SRC_CONFIG);
-    rrdlabels_add(labels2, "key0", "value1", RRDLABEL_SRC_CONFIG);
+    rrdlabels_walkthrough_index_read(labels2, unittest_dump_labels, "\nlabels2");
 
     rrdlabels_copy(labels1, labels2); // labels1 should have 5 keys
+    rc = rrdlabels_unittest_expect_value(labels1, "key1", "value1", RRDLABEL_FLAG_OLD  | RRDLABEL_SRC_CONFIG);
+    if (rc)
+        return rc;
 
+    rc = rrdlabels_unittest_expect_value(labels1, "key0", "value0", RRDLABEL_FLAG_NEW | RRDLABEL_SRC_CONFIG);
+    if (rc)
+        return rc;
+
+    rrdlabels_walkthrough_index_read(labels1, unittest_dump_labels, "\nlabels1 after copy from labels2");
     entries = rrdlabels_entries(labels1);
+
     fprintf(stderr, "labels1 (copied) entries found %zu (should be 5)\n",  rrdlabels_entries(labels1));
     if (entries != 5)
         return 1;
 
-    rrdlabels_add(labels1, "key100", "value1", RRDLABEL_SRC_CONFIG);
-    rrdlabels_copy(labels2, labels1); // labels2 should have 6 keys
-    entries = rrdlabels_entries(labels1);
-
-    fprintf(stderr, "labels2 (copied) entries found %zu (should be 6)\n",  rrdlabels_entries(labels1));
+    rrdlabels_add(labels1, "key0", "value0", RRDLABEL_SRC_CONFIG);
+    rc = rrdlabels_unittest_expect_value(labels1, "key0", "value0", RRDLABEL_FLAG_OLD | RRDLABEL_SRC_CONFIG);
 
     rrdlabels_destroy(labels1);
     rrdlabels_destroy(labels2);
 
-    return entries != 6;
+    return rc;
 }
 
-int rrdlabels_unittest_check_simple_pattern(RRDLABELS *labels, const char *pattern, bool expected) {
+static int rrdlabels_unittest_check_simple_pattern(RRDLABELS *labels, const char *pattern, bool expected) {
     fprintf(stderr, "rrdlabels_match_simple_pattern(labels, \"%s\") ... ", pattern);
 
     bool ret = rrdlabels_match_simple_pattern(labels, pattern);
@@ -1554,7 +1603,7 @@ int rrdlabels_unittest_check_simple_pattern(RRDLABELS *labels, const char *patte
     return (ret == expected)?0:1;
 }
 
-int rrdlabels_unittest_simple_pattern() {
+static int rrdlabels_unittest_simple_pattern() {
     fprintf(stderr, "\n%s() tests\n", __FUNCTION__);
 
     int errors = 0;