bearer authorization API (#15321)

* bearer authorization API - untested

* add machine guid to bearer token response

* removed REGISTRY_URL and replaced it with STRING

* eliminate url pointer from registry_machine_url

* remove registry_url counters from registry

* Revert "eliminate url pointer from registry_machine_url"

This reverts commit 79eff56f77d12a9f840f1b30040e4123e3d9a7a2.

* registry machine urls are now a double linked list

* registry machine urls are now using aral

* all registry objects now use aral

* strings now have 64 partitions and use R/W spinlock

* string to 128 partitions

* fix macro without internal checks

* registry now uses the bearer token when the cookie is not there

* api/v1/registry sends back all nodes on each host

* registry option to use mmap; optimization of registry structures

* do not index the terminator byte in strings; use 256 string partitions

* registry loading optimization

* convert person urls to double linked list to save memory

* re-organize items loading and make sure person urls are always available as machine urls too

* disable registry mmap by default

* keep track of all machine guids and their URLs, even if the cookie cannot be set

* fix bearer parsing

12 files changed, 530 insertions, 473 deletions

diff --git a/registry/registry.c b/registry/registry.c
index f8de470eb8..04edb95c0d 100644
--- a/registry/registry.c
+++ b/registry/registry.c
@@ -8,6 +8,10 @@
 #define REGISTRY_STATUS_FAILED "failed"
 #define REGISTRY_STATUS_DISABLED "disabled"
 
+bool registry_is_valid_url(const char *url) {
+    return url && (*url == 'h' || *url == '*');
+}
+
 // ----------------------------------------------------------------------------
 // REGISTRY concurrency locking
 
@@ -23,19 +27,19 @@ static inline void registry_unlock(void) {
 // COOKIES
 
 static void registry_set_cookie(struct web_client *w, const char *guid) {
-    char edate[100];
+    char e_date[100];
     time_t et = now_realtime_sec() + registry.persons_expiration;
-    struct tm etmbuf, *etm = gmtime_r(&et, &etmbuf);
-    strftime(edate, sizeof(edate), "%a, %d %b %Y %H:%M:%S %Z", etm);
+    struct tm e_tm_buf, *etm = gmtime_r(&et, &e_tm_buf);
+    strftime(e_date, sizeof(e_date), "%a, %d %b %Y %H:%M:%S %Z", etm);
 
-    buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s\r\n", guid, edate);
+    buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s\r\n", guid, e_date);
     if(registry.enable_cookies_samesite_secure)
-        buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s; SameSite=None; Secure\r\n", guid, edate);
+        buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s; SameSite=None; Secure\r\n", guid, e_date);
 
     if(registry.registry_domain && *registry.registry_domain) {
-        buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s; Domain=%s\r\n", guid, edate, registry.registry_domain);
+        buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s; Domain=%s\r\n", guid, e_date, registry.registry_domain);
         if(registry.enable_cookies_samesite_secure)
-            buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s; Domain=%s; SameSite=None; Secure\r\n", guid, edate, registry.registry_domain);
+            buffer_sprintf(w->response.header, "Set-Cookie: " NETDATA_REGISTRY_COOKIE_NAME "=%s; Expires=%s; Domain=%s; SameSite=None; Secure\r\n", guid, e_date, registry.registry_domain);
     }
 
     w->response.has_cookies = true;
@@ -84,37 +88,41 @@ struct registry_json_walk_person_urls_callback {
     int count;
 };
 
+static STRING *asterisks = NULL;
+
 // callback for rendering PERSON_URLs
-static int registry_json_person_url_callback(void *entry, void *data) {
-    REGISTRY_PERSON_URL *pu = (REGISTRY_PERSON_URL *)entry;
-    struct registry_json_walk_person_urls_callback *c = (struct registry_json_walk_person_urls_callback *)data;
+static int registry_json_person_url_callback(REGISTRY_PERSON_URL *pu, struct registry_json_walk_person_urls_callback *c) {
+    if(unlikely(!asterisks))
+        asterisks = string_strdupz("***");
+
     struct web_client *w = c->w;
 
-    if (!strcmp(pu->url->url,"***")) return 0;
+    if (pu->url == asterisks) return 0;
 
     buffer_json_add_array_item_array(w->response.data);
     buffer_json_add_array_item_string(w->response.data, pu->machine->guid);
-    buffer_json_add_array_item_string(w->response.data, pu->url->url);
+    buffer_json_add_array_item_string(w->response.data, string2str(pu->url));
     buffer_json_add_array_item_uint64(w->response.data, pu->last_t * (uint64_t) 1000);
     buffer_json_add_array_item_uint64(w->response.data, pu->usages);
-    buffer_json_add_array_item_string(w->response.data, pu->machine_name);
+    buffer_json_add_array_item_string(w->response.data, string2str(pu->machine_name));
     buffer_json_array_close(w->response.data);
 
     return 1;
 }
 
 // callback for rendering MACHINE_URLs
-static int registry_json_machine_url_callback(const DICTIONARY_ITEM *item __maybe_unused, void *entry, void *data) {
-    REGISTRY_MACHINE_URL *mu = (REGISTRY_MACHINE_URL *)entry;
-    struct registry_json_walk_person_urls_callback *c = (struct registry_json_walk_person_urls_callback *)data;
+static int registry_json_machine_url_callback(REGISTRY_MACHINE_URL *mu, struct registry_json_walk_person_urls_callback *c) {
+    if(unlikely(!asterisks))
+        asterisks = string_strdupz("***");
+
     struct web_client *w = c->w;
     REGISTRY_MACHINE *m = c->m;
 
-    if (!strcmp(mu->url->url,"***")) return 0;
+    if (mu->url == asterisks) return 0;
 
     buffer_json_add_array_item_array(w->response.data);
     buffer_json_add_array_item_string(w->response.data, m->guid);
-    buffer_json_add_array_item_string(w->response.data, mu->url->url);
+    buffer_json_add_array_item_string(w->response.data, string2str(mu->url));
     buffer_json_add_array_item_uint64(w->response.data, mu->last_t * (uint64_t) 1000);
     buffer_json_add_array_item_uint64(w->response.data, mu->usages);
     buffer_json_array_close(w->response.data);
@@ -130,9 +138,7 @@ struct registry_person_url_callback_verify_machine_exists_data {
     int count;
 };
 
-static inline int registry_person_url_callback_verify_machine_exists(void *entry, void *data) {
-    struct registry_person_url_callback_verify_machine_exists_data *d = (struct registry_person_url_callback_verify_machine_exists_data *)data;
-    REGISTRY_PERSON_URL *pu = (REGISTRY_PERSON_URL *)entry;
+static inline int registry_person_url_callback_verify_machine_exists(REGISTRY_PERSON_URL *pu, struct registry_person_url_callback_verify_machine_exists_data *d) {
     REGISTRY_MACHINE *m = d->m;
 
     if(pu->machine == m)
@@ -161,34 +167,54 @@ int registry_request_hello_json(RRDHOST *host, struct web_client *w) {
     buffer_json_member_add_string(w->response.data, "cloud_base_url", registry.cloud_base_url);
     buffer_json_member_add_boolean(w->response.data, "anonymous_statistics", netdata_anonymous_statistics_enabled);
 
+    buffer_json_member_add_array(w->response.data, "nodes");
+    RRDHOST *h;
+    dfe_start_read(rrdhost_root_index, h) {
+        buffer_json_add_array_item_object(w->response.data);
+        buffer_json_member_add_string(w->response.data, "machine_guid", h->machine_guid);
+        buffer_json_member_add_string(w->response.data, "hostname", rrdhost_registry_hostname(h));
+        buffer_json_object_close(w->response.data);
+    }
+    dfe_done(h);
+    buffer_json_array_close(w->response.data);
+
     registry_json_footer(w);
     return HTTP_RESP_OK;
 }
 
 // ----------------------------------------------------------------------------
-//public ACCESS request
-
-#define REGISTRY_VERIFY_COOKIES_GUID "give-me-back-this-cookie-now--please"
+// public ACCESS request
 
 // the main method for registering an access
 int registry_request_access_json(RRDHOST *host, struct web_client *w, char *person_guid, char *machine_guid, char *url, char *name, time_t when) {
     if(unlikely(!registry.enabled))
         return registry_json_disabled(host, w, "access");
 
+    if(!registry_is_valid_url(url)) {
+        buffer_flush(w->response.data);
+        buffer_strcat(w->response.data, "Invalid URL given in the request");
+        return HTTP_RESP_BAD_REQUEST;
+    }
+
     // ------------------------------------------------------------------------
-    // verify the browser supports cookies
+    // verify the browser supports cookies or the bearer
 
     if(registry.verify_cookies_redirects > 0 && !person_guid[0]) {
+        registry_request_access(REGISTRY_VERIFY_COOKIES_GUID, machine_guid, url, name, when);
+
         buffer_flush(w->response.data);
         registry_set_cookie(w, REGISTRY_VERIFY_COOKIES_GUID);
         w->response.data->content_type = CT_APPLICATION_JSON;
         registry_json_header(host, w, "access", REGISTRY_STATUS_REDIRECT);
+        buffer_json_member_add_string(w->response.data, "person_guid", REGISTRY_VERIFY_COOKIES_GUID);
         buffer_json_member_add_string(w->response.data, "registry", registry.registry_to_announce);
         registry_json_footer(w);
         return HTTP_RESP_OK;
     }
 
-    if(unlikely(person_guid[0] && !strcmp(person_guid, REGISTRY_VERIFY_COOKIES_GUID)))
+    if(unlikely(person_guid[0] && is_dummy_person(person_guid)))
+        // it passed the check - they gave us a different person_guid
+        // empty the dummy one, so that we will generate a new person_guid
         person_guid[0] = '\0';
 
     // ------------------------------------------------------------------------
@@ -212,7 +238,8 @@ int registry_request_access_json(RRDHOST *host, struct web_client *w, char *pers
     buffer_json_member_add_array(w->response.data, "urls");
 
     struct registry_json_walk_person_urls_callback c = { p, NULL, w, 0 };
-    avl_traverse(&p->person_urls, registry_json_person_url_callback, &c);
+    for(REGISTRY_PERSON_URL *pu = p->person_urls; pu ;pu = pu->next)
+        registry_json_person_url_callback(pu, &c);
     buffer_json_array_close(w->response.data); // urls
 
     registry_json_footer(w);
@@ -228,6 +255,12 @@ int registry_request_delete_json(RRDHOST *host, struct web_client *w, char *pers
     if(!registry.enabled)
         return registry_json_disabled(host, w, "delete");
 
+    if(!registry_is_valid_url(url)) {
+        buffer_flush(w->response.data);
+        buffer_strcat(w->response.data, "Invalid URL given in the request");
+        return HTTP_RESP_BAD_REQUEST;
+    }
+
     registry_lock();
 
     REGISTRY_PERSON *p = registry_request_delete(person_guid, machine_guid, url, delete_url, when);
@@ -253,6 +286,12 @@ int registry_request_search_json(RRDHOST *host, struct web_client *w, char *pers
     if(!registry.enabled)
         return registry_json_disabled(host, w, "search");
 
+    if(!registry_is_valid_url(url)) {
+        buffer_flush(w->response.data);
+        buffer_strcat(w->response.data, "Invalid URL given in the request");
+        return HTTP_RESP_BAD_REQUEST;
+    }
+
     registry_lock();
 
     REGISTRY_MACHINE *m = registry_request_machine(person_guid, machine_guid, url, request_machine, when);
@@ -267,7 +306,10 @@ int registry_request_search_json(RRDHOST *host, struct web_client *w, char *pers
 
     buffer_json_member_add_array(w->response.data, "urls");
     struct registry_json_walk_person_urls_callback c = { NULL, m, w, 0 };
-    dictionary_walkthrough_read(m->machine_urls, registry_json_machine_url_callback, &c);
+
+    for(REGISTRY_MACHINE_URL *mu = m->machine_urls; mu ; mu = mu->next)
+        registry_json_machine_url_callback(mu, &c);
+
     buffer_json_array_close(w->response.data);
 
     registry_json_footer(w);
@@ -279,12 +321,15 @@ int registry_request_search_json(RRDHOST *host, struct web_client *w, char *pers
 // SWITCH REQUEST
 
 // the main method for switching user identity
-int registry_request_switch_json(RRDHOST *host, struct web_client *w, char *person_guid, char *machine_guid, char *url, char *new_person_guid, time_t when) {
+int registry_request_switch_json(RRDHOST *host, struct web_client *w, char *person_guid, char *machine_guid, char *url __maybe_unused, char *new_person_guid, time_t when __maybe_unused) {
     if(!registry.enabled)
         return registry_json_disabled(host, w, "switch");
 
-    (void)url;
-    (void)when;
+    if(!registry_is_valid_url(url)) {
+        buffer_flush(w->response.data);
+        buffer_strcat(w->response.data, "Invalid URL given in the request");
+        return HTTP_RESP_BAD_REQUEST;
+    }
 
     registry_lock();
 
@@ -315,7 +360,9 @@ int registry_request_switch_json(RRDHOST *host, struct web_client *w, char *pers
     struct registry_person_url_callback_verify_machine_exists_data data = { m, 0 };
 
     // verify the old person has access to this machine
-    avl_traverse(&op->person_urls, registry_person_url_callback_verify_machine_exists, &data);
+    for(REGISTRY_PERSON_URL *pu = op->person_urls; pu ;pu = pu->next)
+        registry_person_url_callback_verify_machine_exists(pu, &data);
+
     if(!data.count) {
         registry_json_header(host, w, "switch", REGISTRY_STATUS_FAILED);
         registry_json_footer(w);
@@ -325,7 +372,9 @@ int registry_request_switch_json(RRDHOST *host, struct web_client *w, char *pers
 
     // verify the new person has access to this machine
     data.count = 0;
-    avl_traverse(&np->person_urls, registry_person_url_callback_verify_machine_exists, &data);
+    for(REGISTRY_PERSON_URL *pu = np->person_urls; pu ;pu = pu->next)
+        registry_person_url_callback_verify_machine_exists(pu, &data);
+
     if(!data.count) {
         registry_json_header(host, w, "switch", REGISTRY_STATUS_FAILED);
         registry_json_footer(w);
@@ -396,14 +445,12 @@ void registry_statistics(void) {
 
         rrddim_add(stc, "persons",        NULL,  1, 1, RRD_ALGORITHM_ABSOLUTE);
         rrddim_add(stc, "machines",       NULL,  1, 1, RRD_ALGORITHM_ABSOLUTE);
-        rrddim_add(stc, "urls",           NULL,  1, 1, RRD_ALGORITHM_ABSOLUTE);
         rrddim_add(stc, "persons_urls",   NULL,  1, 1, RRD_ALGORITHM_ABSOLUTE);
         rrddim_add(stc, "machines_urls",  NULL,  1, 1, RRD_ALGORITHM_ABSOLUTE);
     }
 
     rrddim_set(stc, "persons",       (collected_number)registry.persons_count);
     rrddim_set(stc, "machines",      (collected_number)registry.machines_count);
-    rrddim_set(stc, "urls",          (collected_number)registry.urls_count);
     rrddim_set(stc, "persons_urls",  (collected_number)registry.persons_urls_count);
     rrddim_set(stc, "machines_urls", (collected_number)registry.machines_urls_count);
     rrdset_done(stc);
@@ -428,15 +475,21 @@ void registry_statistics(void) {
 
         rrddim_add(stm, "persons",        NULL,  1, 1024, RRD_ALGORITHM_ABSOLUTE);
         rrddim_add(stm, "machines",       NULL,  1, 1024, RRD_ALGORITHM_ABSOLUTE);
-        rrddim_add(stm, "urls",           NULL,  1, 1024, RRD_ALGORITHM_ABSOLUTE);
         rrddim_add(stm, "persons_urls",   NULL,  1, 1024, RRD_ALGORITHM_ABSOLUTE);
         rrddim_add(stm, "machines_urls",  NULL,  1, 1024, RRD_ALGORITHM_ABSOLUTE);
     }
 
-    rrddim_set(stm, "persons",       (collected_number)registry.persons_memory + dictionary_stats_for_registry(registry.persons));
-    rrddim_set(stm, "machines",      (collected_number)registry.machines_memory + dictionary_stats_for_registry(registry.machines));
-    rrddim_set(stm, "urls",          (collected_number)registry.urls_memory);
-    rrddim_set(stm, "persons_urls",  (collected_number)registry.persons_urls_memory);
-    rrddim_set(stm, "machines_urls", (collected_number)registry.machines_urls_memory);
+    struct aral_statistics *p_aral_stats = aral_statistics(registry.persons_aral);
+    rrddim_set(stm, "persons",       (collected_number)p_aral_stats->structures.allocated_bytes + (collected_number)p_aral_stats->malloc.allocated_bytes);
+
+    struct aral_statistics *m_aral_stats = aral_statistics(registry.machines_aral);
+    rrddim_set(stm, "machines",      (collected_number)m_aral_stats->structures.allocated_bytes + (collected_number)m_aral_stats->malloc.allocated_bytes);
+
+    struct aral_statistics *pu_aral_stats = aral_statistics(registry.person_urls_aral);
+    rrddim_set(stm, "persons_urls",  (collected_number)pu_aral_stats->structures.allocated_bytes + (collected_number)pu_aral_stats->malloc.allocated_bytes);
+
+    struct aral_statistics *mu_aral_stats = aral_statistics(registry.machine_urls_aral);
+    rrddim_set(stm, "machines_urls", (collected_number)mu_aral_stats->structures.allocated_bytes + (collected_number)mu_aral_stats->malloc.allocated_bytes);
+
     rrdset_done(stm);
 }
diff --git a/registry/registry_db.c b/registry/registry_db.c
index f0586ab21a..903f39104d 100644
--- a/registry/registry_db.c
+++ b/registry/registry_db.c
@@ -11,18 +11,15 @@ int registry_db_should_be_saved(void) {
 // ----------------------------------------------------------------------------
 // INTERNAL FUNCTIONS FOR SAVING REGISTRY OBJECTS
 
-static int registry_machine_save_url(const DICTIONARY_ITEM *item __maybe_unused, void *entry, void *file) {
-    REGISTRY_MACHINE_URL *mu = entry;
-    FILE *fp = file;
-
-    debug(D_REGISTRY, "Registry: registry_machine_save_url('%s')", mu->url->url);
+static int registry_machine_save_url(REGISTRY_MACHINE_URL *mu, FILE *fp) {
+    debug(D_REGISTRY, "REGISTRY: registry_machine_save_url('%s')", string2str(mu->url));
 
     int ret = fprintf(fp, "V\t%08x\t%08x\t%08x\t%02x\t%s\n",
             mu->first_t,
             mu->last_t,
             mu->usages,
             mu->flags,
-            mu->url->url
+            string2str(mu->url)
     );
 
     // error handling is done at registry_db_save()
@@ -35,7 +32,7 @@ static int registry_machine_save(const DICTIONARY_ITEM *item __maybe_unused, voi
     REGISTRY_MACHINE *m = entry;
     FILE *fp = file;
 
-    debug(D_REGISTRY, "Registry: registry_machine_save('%s')", m->guid);
+    debug(D_REGISTRY, "REGISTRY: registry_machine_save('%s')", m->guid);
 
     int ret = fprintf(fp, "M\t%08x\t%08x\t%08x\t%s\n",
             m->first_t,
@@ -45,9 +42,13 @@ static int registry_machine_save(const DICTIONARY_ITEM *item __maybe_unused, voi
     );
 
     if(ret >= 0) {
-        int ret2 = dictionary_walkthrough_read(m->machine_urls, registry_machine_save_url, fp);
-        if(ret2 < 0) return ret2;
-        ret += ret2;
+        for(REGISTRY_MACHINE_URL *mu = m->machine_urls; mu ; mu = mu->next) {
+            int rc = registry_machine_save_url(mu, fp);
+            if(rc < 0)
+                return rc;
+
+            ret += rc;
+        }
     }
 
     // error handling is done at registry_db_save()
@@ -55,11 +56,8 @@ static int registry_machine_save(const DICTIONARY_ITEM *item __maybe_unused, voi
     return ret;
 }
 
-static inline int registry_person_save_url(void *entry, void *file) {
-    REGISTRY_PERSON_URL *pu = entry;
-    FILE *fp = file;
-
-    debug(D_REGISTRY, "Registry: registry_person_save_url('%s')", pu->url->url);
+static inline int registry_person_save_url(REGISTRY_PERSON_URL *pu, FILE *fp) {
+    debug(D_REGISTRY, "REGISTRY: registry_person_save_url('%s')", string2str(pu->url));
 
     int ret = fprintf(fp, "U\t%08x\t%08x\t%08x\t%02x\t%s\t%s\t%s\n",
             pu->first_t,
@@ -67,8 +65,8 @@ static inline int registry_person_save_url(void *entry, void *file) {
             pu->usages,
             pu->flags,
             pu->machine->guid,
-            pu->machine_name,
-            pu->url->url
+            string2str(pu->machine_name),
+            string2str(pu->url)
     );
 
     // error handling is done at registry_db_save()
@@ -80,7 +78,7 @@ static inline int registry_person_save(const DICTIONARY_ITEM *item __maybe_unuse
     REGISTRY_PERSON *p = entry;
     FILE *fp = file;
 
-    debug(D_REGISTRY, "Registry: registry_person_save('%s')", p->guid);
+    debug(D_REGISTRY, "REGISTRY: registry_person_save('%s')", p->guid);
 
     int ret = fprintf(fp, "P\t%08x\t%08x\t%08x\t%s\n",
             p->first_t,
@@ -90,10 +88,13 @@ static inline int registry_person_save(const DICTIONARY_ITEM *item __maybe_unuse
     );
 
     if(ret >= 0) {
-        //int ret2 = dictionary_walkthrough_read(p->person_urls, registry_person_save_url, fp);
-        int ret2 = avl_traverse(&p->person_urls, registry_person_save_url, fp);
-        if (ret2 < 0) return ret2;
-        ret += ret2;
+        for(REGISTRY_PERSON_URL *pu = p->person_urls; pu ;pu = pu->next) {
+            int rc = registry_person_save_url(pu, fp);
+            if(rc < 0)
+                return rc;
+            else
+                ret += rc;
+        }
     }
 
     // error handling is done at registry_db_save()
@@ -119,42 +120,42 @@ int registry_db_save(void) {
     snprintfz(old_filename, FILENAME_MAX, "%s.old", registry.db_filename);
     snprintfz(tmp_filename, FILENAME_MAX, "%s.tmp", registry.db_filename);
 
-    debug(D_REGISTRY, "Registry: Creating file '%s'", tmp_filename);
+    debug(D_REGISTRY, "REGISTRY: Creating file '%s'", tmp_filename);
     FILE *fp = fopen(tmp_filename, "w");
     if(!fp) {
-        netdata_log_error("Registry: Cannot create file: %s", tmp_filename);
+        netdata_log_error("REGISTRY: Cannot create file: %s", tmp_filename);
         error_log_limit_reset();
         return -1;
     }
 
     // dictionary_walkthrough_read() has its own locking, so this is safe to do
 
-    debug(D_REGISTRY, "Saving all machines");
+    debug(D_REGISTRY, "REGISTRY: saving all machines");
     int bytes1 = dictionary_walkthrough_read(registry.machines, registry_machine_save, fp);
     if(bytes1 < 0) {
-        netdata_log_error("Registry: Cannot save registry machines - return value %d", bytes1);
+        netdata_log_error("REGISTRY: Cannot save registry machines - return value %d", bytes1);
         fclose(fp);
         error_log_limit_reset();
         return bytes1;
     }
-    debug(D_REGISTRY, "Registry: saving machines took %d bytes", bytes1);
+    debug(D_REGISTRY, "REGISTRY: saving machines took %d bytes", bytes1);
 
     debug(D_REGISTRY, "Saving all persons");
     int bytes2 = dictionary_walkthrough_read(registry.persons, registry_person_save, fp);
     if(bytes2 < 0) {
-        netdata_log_error("Registry: Cannot save registry persons - return value %d", bytes2);
+        netdata_log_error("REGISTRY: Cannot save registry persons - return value %d", bytes2);
         fclose(fp);
         error_log_limit_reset();
         return bytes2;
     }
-    debug(D_REGISTRY, "Registry: saving persons took %d bytes", bytes2);
+    debug(D_REGISTRY, "REGISTRY: saving persons took %d bytes", bytes2);
 
     // save the totals
     fprintf(fp, "T\t%016llx\t%016llx\t%016llx\t%016llx\t%016llx\t%016llx\n",
             registry.persons_count,
             registry.machines_count,
             registry.usages_count + 1, // this is required - it is lost on db rotation
-            registry.urls_count,
+            0LLU, //registry.urls_count,
             registry.persons_urls_count,
             registry.machines_urls_count
     );
@@ -164,36 +165,36 @@ int registry_db_save(void) {
     errno = 0;
 
     // remove the .old db
-    debug(D_REGISTRY, "Registry: Removing old db '%s'", old_filename);
+    debug(D_REGISTRY, "REGISTRY: Removing old db '%s'", old_filename);
     if(unlink(old_filename) == -1 && errno != ENOENT)
-        netdata_log_error("Registry: cannot remove old registry file '%s'", old_filename);
+        netdata_log_error("REGISTRY: cannot remove old registry file '%s'", old_filename);
 
     // rename the db to .old
-    debug(D_REGISTRY, "Registry: Link current db '%s' to .old: '%s'", registry.db_filename, old_filename);
+    debug(D_REGISTRY, "REGISTRY: Link current db '%s' to .old: '%s'", registry.db_filename, old_filename);
     if(link(registry.db_filename, old_filename) == -1 && errno != ENOENT)
-        netdata_log_error("Registry: cannot move file '%s' to '%s'. Saving registry DB failed!", registry.db_filename, old_filename);
+        netdata_log_error("REGISTRY: cannot move file '%s' to '%s'. Saving registry DB failed!", registry.db_filename, old_filename);
 
     else {
         // remove the database (it is saved in .old)
-        debug(D_REGISTRY, "Registry: removing db '%s'", registry.db_filename);
+        debug(D_REGISTRY, "REGISTRY: removing db '%s'", registry.db_filename);
         if (unlink(registry.db_filename) == -1 && errno != ENOENT)
-            netdata_log_error("Registry: cannot remove old registry file '%s'", registry.db_filename);
+            netdata_log_error("REGISTRY: cannot remove old registry file '%s'", registry.db_filename);
 
         // move the .tmp to make it active
-        debug(D_REGISTRY, "Registry: linking tmp db '%s' to active db '%s'", tmp_filename, registry.db_filename);
+        debug(D_REGISTRY, "REGISTRY: linking tmp db '%s' to active db '%s'", tmp_filename, registry.db_filename);
         if (link(tmp_filename, registry.db_filename) == -1) {
-            netdata_log_error("Registry: cannot move file '%s' to '%s'. Saving registry DB failed!", tmp_filename,
+            netdata_log_error("REGISTRY: cannot move file '%s' to '%s'. Saving registry DB failed!", tmp_filename,
                     registry.db_filename);
 
             // move the .old back
-            debug(D_REGISTRY, "Registry: linking old db '%s' to active db '%s'", old_filename, registry.db_filename);
+            debug(D_REGISTRY, "REGISTRY: linking old db '%s' to active db '%s'", old_filename, registry.db_filename);
             if(link(old_filename, registry.db_filename) == -1)
-                netdata_log_error("Registry: cannot move file '%s' to '%s'. Recovering the old registry DB failed!", old_filename, registry.db_filename);
+                netdata_log_error("REGISTRY: cannot move file '%s' to '%s'. Recovering the old registry DB failed!", old_filename, registry.db_filename);
         }
         else {
-            debug(D_REGISTRY, "Registry: removing tmp db '%s'", tmp_filename);
+            debug(D_REGISTRY, "REGISTRY: removing tmp db '%s'", tmp_filename);
             if(unlink(tmp_filename) == -1)
-                netdata_log_error("Registry: cannot remove tmp registry file '%s'", tmp_filename);
+                netdata_log_error("REGISTRY: cannot remove tmp registry file '%s'", tmp_filename);
 
             // it has been moved successfully
             // discard the current registry log
@@ -215,75 +216,33 @@ size_t registry_db_load(void) {
     char *s, buf[4096 + 1];
     REGISTRY_PERSON *p = NULL;
     REGISTRY_MACHINE *m = NULL;
-    REGISTRY_URL *u = NULL;
+    STRING *u = NULL;
     size_t line = 0;
 
-    debug(D_REGISTRY, "Registry: loading active db from: '%s'", registry.db_filename);
+    debug(D_REGISTRY, "REGISTRY: loading active db from: '%s'", registry.db_filename);
     FILE *fp = fopen(registry.db_filename, "r");
     if(!fp) {
-        netdata_log_error("Registry: cannot open registry file: '%s'", registry.db_filename);
+        netdata_log_error("REGISTRY: cannot open registry file: '%s'", registry.db_filename);
         return 0;
     }
 
+    REGISTRY_MACHINE_URL *mu;
     size_t len = 0;
     buf[4096] = '\0';
     while((s = fgets_trim_len(buf, 4096, fp, &len))) {
         line++;
 
-        debug(D_REGISTRY, "Registry: read line %zu to length %zu: %s", line, len, s);
+        debug(D_REGISTRY, "REGISTRY: read line %zu to length %zu: %s", line, len, s);
         switch(*s) {
-            case 'T': // totals
-                if(unlikely(len != 103 || s[1] != '\t' || s[18] != '\t' || s[35] != '\t' || s[52] != '\t' || s[69] != '\t' || s[86] != '\t' || s[103] != '\0')) {
-                    netdata_log_error("Registry totals line %zu is wrong (len = %zu).", line, len);
-                    continue;
-                }
-                registry.persons_count = strtoull(&s[2], NULL, 16);
-                registry.machines_count = strtoull(&s[19], NULL, 16);
-                registry.usages_count = strtoull(&s[36], NULL, 16);
-                registry.urls_count = strtoull(&s[53], NULL, 16);
-                registry.persons_urls_count = strtoull(&s[70], NULL, 16);
-                registry.machines_urls_count = strtoull(&s[87], NULL, 16);
-                break;
-
-            case 'P': // person
-                m = NULL;
-                // verify it is valid
-                if(unlikely(len != 65 || s[1] != '\t' || s[10] != '\t' || s[19] != '\t' || s[28] != '\t' || s[65] != '\0')) {
-                    netdata_log_error("Registry person line %zu is wrong (len = %zu).", line, len);
-                    continue;
-                }
-
-                s[1] = s[10] = s[19] = s[28] = '\0';
-                p = registry_person_allocate(&s[29], strtoul(&s[2], NULL, 16));
-                p->last_t = (uint32_t)strtoul(&s[11], NULL, 16);
-                p->usages = (uint32_t)strtoul(&s[20], NULL, 16);
-                debug(D_REGISTRY, "Registry loaded person '%s', first: %u, last: %u, usages: %u", p->guid, p->first_t, p->last_t, p->usages);
-                break;
-
-            case 'M': // machine
-                p = NULL;
-                // verify it is valid
-                if(unlikely(len != 65 || s[1] != '\t' || s[10] != '\t' || s[19] != '\t' || s[28] != '\t' || s[65] != '\0')) {
-                    netdata_log_error("Registry person line %zu is wrong (len = %zu).", line, len);
-                    continue;
-                }
-
-                s[1] = s[10] = s[19] = s[28] = '\0';
-                m = registry_machine_allocate(&s[29], strtoul(&s[2], NULL, 16));
-                m->last_t = (uint32_t)strtoul(&s[11], NULL, 16);
-                m->usages = (uint32_t)strtoul(&s[20], NULL, 16);
-                debug(D_REGISTRY, "Registry loaded machine '%s', first: %u, last: %u, usages: %u", m->guid, m->first_t, m->last_t, m->usages);
-                break;
-
             case 'U': // person URL
                 if(unlikely(!p)) {
-                    netdata_log_error("Registry: ignoring line %zu, no person loaded: %s", line, s);
+                    netdata_log_error("REGISTRY: ignoring line %zu, no person loaded: %s", line, s);
                     continue;
                 }
 
                 // verify it is valid
                 if(len < 69 || s[1] != '\t' || s[10] != '\t' || s[19] != '\t' || s[28] != '\t' || s[31] != '\t' || s[68] != '\t') {
-                    netdata_log_error("Registry person URL line %zu is wrong (len = %zu).", line, len);
+                    netdata_log_error("REGISTRY: person URL line %zu is wrong (len = %zu).", line, len);
                     continue;
                 }
 
@@ -293,51 +252,125 @@ size_t registry_db_load(void) {
                 char *url = &s[69];
                 while(*url && *url != '\t') url++;
                 if(!*url) {
-                    netdata_log_error("Registry person URL line %zu does not have a url.", line);
+                    netdata_log_error("REGISTRY: person URL line %zu does not have a url.", line);
                     continue;
                 }
                 *url++ = '\0';
 
-                // u = registry_url_allocate_nolock(url, strlen(url));
-                u = registry_url_get(url, strlen(url));
+                if(*url != 'h' && *url != '*') {
+                    netdata_log_error("REGISTRY: person URL line %zu does not have a valid url: %s", line, url);
+                    continue;
+                }
+
+                u = string_strdupz(url);
 
-                time_t first_t = strtoul(&s[2], NULL, 16);
+                time_t first_t = (time_t)strtoul(&s[2], NULL, 16);
 
                 m = registry_machine_find(&s[32]);
                 if(!m) m = registry_machine_allocate(&s[32], first_t);
 
-                REGISTRY_PERSON_URL *pu = registry_person_url_allocate(p, m, u, &s[69], strlen(&s[69]), first_t);
+                mu = registry_machine_url_find(m, u);
+                if(!mu) {
+                    netdata_log_error("REGISTRY: person URL line %zu was not linked to the machine it refers to", line);
+                    mu = registry_machine_url_allocate(m, u, first_t);
+                }
+
+                REGISTRY_PERSON_URL *pu = registry_person_url_index_find(p, u);
+                if(!pu)
+                    pu = registry_person_url_allocate(p, m, u, &s[69], strlen(&s[69]), first_t);
+                else
+                    netdata_log_error("REGISTRY: person URL line %zu is duplicate, reusing the old one.", line);
+
                 pu->last_t = (uint32_t)strtoul(&s[11], NULL, 16);
                 pu->usages = (uint32_t)strtoul(&s[20], NULL, 16);
                 pu->flags = (uint8_t)strtoul(&s[29], NULL, 16);
-                debug(D_REGISTRY, "Registry loaded person URL '%s' with name '%s' of machine '%s', first: %u, last: %u, usages: %u, flags: %02x", u->url, pu->machine_name, m->guid, pu->first_t, pu->last_t, pu->usages, pu->flags);
+                debug(D_REGISTRY, "REGISTRY: loaded person URL '%s' with name '%s' of machine '%s', first: %u, last: %u, usages: %u, flags: %02x",
+                      string2str(u), string2str(pu->machine_name), m->guid, pu->first_t, pu->last_t, pu->usages, pu->flags);
+
+                string_freez(u);
+                break;
+
+            case 'P': // person
+                m = NULL;
+                // verify it is valid
+                if(unlikely(len != 65 || s[1] != '\t' || s[10] != '\t' || s[19] != '\t' || s[28] != '\t' || s[65] != '\0')) {
+                    netdata_log_error("REGISTRY: person line %zu is wrong (len = %zu).", line, len);
+                    continue;
+                }
+
+                s[1] = s[10] = s[19] = s[28] = '\0';
+                p = registry_person_allocate(&s[29], (time_t)strtoul(&s[2], NULL, 16));
+                p->last_t = (uint32_t)strtoul(&s[11], NULL, 16);
+                p->usages = (uint32_t)strtoul(&s[20], NULL, 16);
+                debug(D_REGISTRY, "REGISTRY: loaded person '%s', first: %u, last: %u, usages: %u", p->guid, p->first_t, p->last_t, p->usages);
                 break;
 
             case 'V': // machine URL
                 if(unlikely(!m)) {
-                    netdata_log_error("Registry: ignoring line %zu, no machine loaded: %s", line, s);
+                    netdata_log_error("REGISTRY: ignoring line %zu, no machine loaded: %s", line, s);
                     continue;
                 }
 
                 // verify it is valid
                 if(len < 32 || s[1] != '\t' || s[10] != '\t' || s[19] != '\t' || s[28] != '\t' || s[31] != '\t') {
-                    netdata_log_error("Registry person URL line %zu is wrong (len = %zu).", line, len);
+                    netdata_log_error("REGISTRY: person URL line %zu is wrong (len = %zu).", line, len);
                     continue;