add support for the info parameter to all external plugin functions (#16915)

* add support for the info parameter to all functions

* add SO_CLOEXEC to inbound connections

* network-connections aggregated view for servers with dozens of thousands of connections

* optimized /proc/net files parsing; aggregated view of network-connections returns sorted sockets to avoid dancing visualization

* local-sockets is now using aral

* lower the payload returned by network-connections

9 files changed, 131 insertions, 54 deletions

diff --git a/src/database/rrdfunctions-inflight.c b/src/database/rrdfunctions-inflight.c
index 048aae9a3b..585a6d2699 100644
--- a/src/database/rrdfunctions-inflight.c
+++ b/src/database/rrdfunctions-inflight.c
@@ -427,7 +427,6 @@ int rrd_function_run(RRDHOST *host, BUFFER *result_wb, int timeout_s,
 
     code = rrd_functions_find_by_name(host, result_wb, sanitized_cmd, sanitized_cmd_length, &host_function_acquired);
     if(code != HTTP_RESP_OK) {
-        rrd_call_function_error(result_wb, "not found", code);
 
         if(result_cb)
             result_cb(result_wb, code, result_cb_data);
diff --git a/src/database/rrdfunctions-streaming.c b/src/database/rrdfunctions-streaming.c
index 4a6c55876a..baf3ebc388 100644
--- a/src/database/rrdfunctions-streaming.c
+++ b/src/database/rrdfunctions-streaming.c
@@ -14,6 +14,7 @@ int rrdhost_function_streaming(BUFFER *wb, const char *function __maybe_unused)
     buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
     buffer_json_member_add_string(wb, "type", "table");
     buffer_json_member_add_time_t(wb, "update_every", 1);
+    buffer_json_member_add_boolean(wb, "has_history", false);
     buffer_json_member_add_string(wb, "help", RRDFUNCTIONS_STREAMING_HELP);
     buffer_json_member_add_array(wb, "data");
 
diff --git a/src/libnetdata/aral/aral.c b/src/libnetdata/aral/aral.c
index 7223ee3597..b8ed47f04a 100644
--- a/src/libnetdata/aral/aral.c
+++ b/src/libnetdata/aral/aral.c
@@ -464,6 +464,12 @@ static inline ARAL_PAGE *aral_acquire_a_free_slot(ARAL *ar TRACE_ALLOCATIONS_FUN
     return page;
 }
 
+void *aral_callocz_internal(ARAL *ar TRACE_ALLOCATIONS_FUNCTION_DEFINITION_PARAMS) {
+    void *r = aral_mallocz_internal(ar TRACE_ALLOCATIONS_FUNCTION_CALL_PARAMS);
+    memset(r, 0, ar->config.requested_element_size);
+    return r;
+}
+
 void *aral_mallocz_internal(ARAL *ar TRACE_ALLOCATIONS_FUNCTION_DEFINITION_PARAMS) {
 #ifdef FSANITIZE_ADDRESS
     return mallocz(ar->config.requested_element_size);
diff --git a/src/libnetdata/aral/aral.h b/src/libnetdata/aral/aral.h
index 96f5a9c44f..2e749bc4c2 100644
--- a/src/libnetdata/aral/aral.h
+++ b/src/libnetdata/aral/aral.h
@@ -46,10 +46,12 @@ int aral_unittest(size_t elements);
 
 #ifdef NETDATA_TRACE_ALLOCATIONS
 
+#define aral_callocz(ar) aral_callocz_internal(ar, __FILE__, __FUNCTION__, __LINE__)
 #define aral_mallocz(ar) aral_mallocz_internal(ar, __FILE__, __FUNCTION__, __LINE__)
 #define aral_freez(ar, ptr) aral_freez_internal(ar, ptr, __FILE__, __FUNCTION__, __LINE__)
 #define aral_destroy(ar) aral_destroy_internal(ar, __FILE__, __FUNCTION__, __LINE__)
 
+void *aral_callocz_internal(ARAL *ar, const char *file, const char *function, size_t line);
 void *aral_mallocz_internal(ARAL *ar, const char *file, const char *function, size_t line);
 void aral_freez_internal(ARAL *ar, void *ptr, const char *file, const char *function, size_t line);
 void aral_destroy_internal(ARAL *ar, const char *file, const char *function, size_t line);
@@ -57,9 +59,11 @@ void aral_destroy_internal(ARAL *ar, const char *file, const char *function, siz
 #else // NETDATA_TRACE_ALLOCATIONS
 
 #define aral_mallocz(ar) aral_mallocz_internal(ar)
+#define aral_callocz(ar) aral_callocz_internal(ar)
 #define aral_freez(ar, ptr) aral_freez_internal(ar, ptr)
 #define aral_destroy(ar) aral_destroy_internal(ar)
 
+void *aral_callocz_internal(ARAL *ar);
 void *aral_mallocz_internal(ARAL *ar);
 void aral_freez_internal(ARAL *ar, void *ptr);
 void aral_destroy_internal(ARAL *ar);
diff --git a/src/libnetdata/buffer/buffer.c b/src/libnetdata/buffer/buffer.c
index 21902dd4d7..119216dd93 100644
--- a/src/libnetdata/buffer/buffer.c
+++ b/src/libnetdata/buffer/buffer.c
@@ -341,8 +341,10 @@ __attribute__((constructor)) void initialize_ascii_maps(void) {
         base64_value_from_ascii[i] = 255;
     }
 
-    for(size_t i = 0; i < 16 ; i++)
-        hex_value_from_ascii[(int)hex_digits[i]] = i;
+    for(size_t i = 0; i < 16 ; i++) {
+        hex_value_from_ascii[(int)toupper(hex_digits[i])] = i;
+        hex_value_from_ascii[(int)tolower(hex_digits[i])] = i;
+    }
 
     for(size_t i = 0; i < 64 ; i++)
         base64_value_from_ascii[(int)base64_digits[i]] = i;
diff --git a/src/libnetdata/inlined.h b/src/libnetdata/inlined.h
index 6fb845dce9..9c0d2dd0b6 100644
--- a/src/libnetdata/inlined.h
+++ b/src/libnetdata/inlined.h
@@ -204,12 +204,28 @@ static inline long long str2ll(const char *s, char **endptr) {
     }
 }
 
+static inline uint32_t str2uint32_hex(const char *src, char **endptr) {
+    uint32_t num = 0;
+    const unsigned char *s = (const unsigned char *)src;
+    unsigned char c;
+
+    while ((c = hex_value_from_ascii[(uint8_t)*s]) != 255) {
+        num = (num << 4) | c;
+        s++;
+    }
+
+    if(endptr)
+        *endptr = (char *)s;
+
+    return num;
+}
+
 static inline uint64_t str2uint64_hex(const char *src, char **endptr) {
     uint64_t num = 0;
     const unsigned char *s = (const unsigned char *)src;
     unsigned char c;
 
-    while ((c = hex_value_from_ascii[toupper(*s)]) != 255) {
+    while ((c = hex_value_from_ascii[(uint8_t)*s]) != 255) {
         num = (num << 4) | c;
         s++;
     }
diff --git a/src/libnetdata/maps/local-sockets.h b/src/libnetdata/maps/local-sockets.h
index 70282259f2..ce52012423 100644
--- a/src/libnetdata/maps/local-sockets.h
+++ b/src/libnetdata/maps/local-sockets.h
@@ -104,6 +104,9 @@ typedef struct local_socket_state {
     uint16_t tmp_protocol;
 #endif
 
+    ARAL *local_socket_aral;
+    ARAL *pid_socket_aral;
+
     uint64_t proc_self_net_ns_inode;
 
     SIMPLE_HASHTABLE_NET_NS ns_hashtable;
@@ -162,7 +165,7 @@ static inline void ipv6_to_in6_addr(const char *ipv6_str, struct in6_addr *d) {
     for (size_t k = 0; k < 4; ++k) {
         memcpy(buf, ipv6_str + (k * 8), 8);
         buf[sizeof(buf) - 1] = '\0';
-        d->s6_addr32[k] = strtoul(buf, NULL, 16);
+        d->s6_addr32[k] = str2uint32_hex(buf, NULL);
     }
 }
 
@@ -185,13 +188,17 @@ typedef struct local_socket {
     uid_t uid;
 
     char comm[TASK_COMM_LEN];
-    char *cmdline;
+    STRING *cmdline;
 
     struct local_port local_port_key;
 
     XXH64_hash_t local_ip_hash;
     XXH64_hash_t remote_ip_hash;
     XXH64_hash_t local_port_hash;
+
+#ifdef LOCAL_SOCKETS_EXTENDED_MEMBERS
+    LOCAL_SOCKETS_EXTENDED_MEMBERS
+#endif
 } LOCAL_SOCKET;
 
 // --------------------------------------------------------------------------------------------------------------------
@@ -393,7 +400,7 @@ static inline bool local_sockets_find_all_sockets_in_proc(LS_STATE *ls, const ch
                 }
 
                 if(!ps)
-                    ps = callocz(1, sizeof(*ps));
+                    ps = aral_callocz(ls->pid_socket_aral);
 
                 ps->inode = inode;
                 ps->pid = pid;
@@ -559,7 +566,7 @@ static inline bool local_sockets_add_socket(LS_STATE *ls, LOCAL_SOCKET *tmp) {
         return false;
     }
 
-    n = (LOCAL_SOCKET *)callocz(1, sizeof(LOCAL_SOCKET));
+    n = aral_mallocz(ls->local_socket_aral);
     *n = *tmp; // copy all contents
 
     // fix the key
@@ -584,7 +591,8 @@ static inline bool local_sockets_add_socket(LS_STATE *ls, LOCAL_SOCKET *tmp) {
             n->uid = ps->uid;
 
         if(ps->cmdline)
-            n->cmdline = strdupz(ps->cmdline);
+            n->cmdline = string_strdupz(ps->cmdline);
+
         strncpyz(n->comm, ps->comm, sizeof(n->comm) - 1);
     }
 
@@ -731,18 +739,27 @@ static inline bool local_sockets_netlink_get_sockets(LS_STATE *ls, uint16_t fami
 #endif // HAVE_LIBMNL
 
 static inline bool local_sockets_read_proc_net_x(LS_STATE *ls, const char *filename, uint16_t family, uint16_t protocol) {
+    static bool is_space[256] = {
+        [':'] = true,
+        [' '] = true,
+    };
+
     if(family != AF_INET && family != AF_INET6)
         return false;
 
-    FILE *fp;
-    char *line = NULL;
-    size_t len = 0;
-    ssize_t read;
-
-    fp = fopen(filename, "r");
+    FILE *fp = fopen(filename, "r");
     if (fp == NULL)
         return false;
 
+    char *line = malloc(1024);  // no mallocz() here because getline() may resize
+    if(!line) {
+        fclose(fp);
+        return false;
+    }
+
+    size_t len = 1024;
+    ssize_t read;
+
     ssize_t min_line_length = (family == AF_INET) ? 105 : 155;
     size_t counter = 0;
 
@@ -755,49 +772,60 @@ static inline bool local_sockets_read_proc_net_x(LS_STATE *ls, const char *filen
             continue;
         }
 
-        unsigned int local_address, local_port, state, remote_address, remote_port;
-        uint64_t  inode = 0;
-        char local_address6[33], remote_address6[33];
-
-        if(family == AF_INET) {
-            if (sscanf(line, "%*d: %X:%X %X:%X %X %*X:%*X %*X:%*X %*X %*d %*d %"PRIu64,
-                       &local_address, &local_port, &remote_address, &remote_port, &state, &inode) != 6) {
-                local_sockets_log(ls, "cannot parse ipv4 line No %zu of filename '%s': %s", counter, filename, line);
-                continue;
-            }
-        }
-        else if(family == AF_INET6) {
-            if(sscanf(line, "%*d: %32[0-9A-Fa-f]:%X %32[0-9A-Fa-f]:%X %X %*X:%*X %*X:%*X %*X %*d %*d %"PRIu64,
-                       local_address6, &local_port, remote_address6, &remote_port, &state, &inode) != 6) {
-                local_sockets_log(ls, "cannot parse ipv6 line No %zu of filename '%s': %s", counter, filename, line);
-                continue;
-            }
-        }
-
         LOCAL_SOCKET n = {
-            .inode = inode,
             .direction = SOCKET_DIRECTION_NONE,
-            .state = (int)state,
             .local = {
                 .family = family,
                 .protocol = protocol,
-                .port = local_port,
             },
             .remote = {
                 .family = family,
                 .protocol = protocol,
-                .port = remote_port,
             },
             .uid = UID_UNSET,
         };
 
+        char *words[32];
+        size_t num_words = quoted_strings_splitter(line, words, 32, is_space);
+        // char *sl_txt = get_word(words, num_words, 0);
+        char *local_ip_txt = get_word(words, num_words, 1);
+        char *local_port_txt = get_word(words, num_words, 2);
+        char *remote_ip_txt = get_word(words, num_words, 3);
+        char *remote_port_txt = get_word(words, num_words, 4);
+        char *state_txt = get_word(words, num_words, 5);
+        char *tx_queue_txt = get_word(words, num_words, 6);
+        char *rx_queue_txt = get_word(words, num_words, 7);
+        char *tr_txt = get_word(words, num_words, 8);
+        char *tm_when_txt = get_word(words, num_words, 9);
+        char *retrans_txt = get_word(words, num_words, 10);
+        char *uid_txt = get_word(words, num_words, 11);
+        // char *timeout_txt = get_word(words, num_words, 12);
+        char *inode_txt = get_word(words, num_words, 13);
+
+        if(!local_ip_txt || !local_port_txt || !remote_ip_txt || !remote_port_txt || !state_txt ||
+            !tx_queue_txt || !rx_queue_txt || !tr_txt || !tm_when_txt || !retrans_txt || !uid_txt || !inode_txt) {
+            local_sockets_log(ls, "cannot parse ipv4 line No %zu of filename '%s'", counter, filename);
+            continue;
+        }
+
+        n.local.port = str2uint32_hex(local_port_txt, NULL);
+        n.remote.port = str2uint32_hex(remote_port_txt, NULL);
+        n.state = str2uint32_hex(state_txt, NULL);
+        n.wqueue = str2uint32_hex(tx_queue_txt, NULL);
+        n.rqueue = str2uint32_hex(rx_queue_txt, NULL);
+        n.timer = str2uint32_hex(tr_txt, NULL);
+        n.expires = str2uint32_hex(tm_when_txt, NULL);
+        n.retransmits = str2uint32_hex(retrans_txt, NULL);
+        n.uid = str2uint32_t(uid_txt, NULL);
+        n.inode = str2uint64_t(inode_txt, NULL);
+
         if(family == AF_INET) {
-            n.local.ip.ipv4 = local_address;
-            n.remote.ip.ipv4 = remote_address;
+            n.local.ip.ipv4 = str2uint32_hex(local_ip_txt, NULL);
+            n.remote.ip.ipv4 = str2uint32_hex(remote_ip_txt, NULL);
         }
         else if(family == AF_INET6) {
-            ipv6_to_in6_addr(local_address6, &n.local.ip.ipv6);
-            ipv6_to_in6_addr(remote_address6, &n.remote.ip.ipv6);
+            ipv6_to_in6_addr(local_ip_txt, &n.local.ip.ipv6);
+            ipv6_to_in6_addr(remote_ip_txt, &n.remote.ip.ipv6);
         }
 
         local_sockets_add_socket(ls, &n);
@@ -806,7 +834,7 @@ static inline bool local_sockets_read_proc_net_x(LS_STATE *ls, const char *filen
     fclose(fp);
 
     if (line)
-        freez(line);
+        free(line); // no freez() here because getline() may resize
 
     return true;
 }
@@ -881,6 +909,20 @@ static inline void local_sockets_init(LS_STATE *ls) {
     simple_hashtable_init_LOCAL_SOCKET(&ls->sockets_hashtable, 65535);
     simple_hashtable_init_LOCAL_IP(&ls->local_ips_hashtable, 4096);
     simple_hashtable_init_LISTENING_PORT(&ls->listening_ports_hashtable, 4096);
+
+    ls->local_socket_aral = aral_create(
+        "local-sockets",
+        sizeof(LOCAL_SOCKET),
+        65536,
+        65536,
+        NULL, NULL, NULL, false, true);
+
+    ls->pid_socket_aral = aral_create(
+        "pid-sockets",
+        sizeof(struct pid_socket),
+        65536,
+        65536,
+        NULL, NULL, NULL, false, true);
 }
 
 static inline void local_sockets_cleanup(LS_STATE *ls) {
@@ -891,8 +933,8 @@ static inline void local_sockets_cleanup(LS_STATE *ls) {
         LOCAL_SOCKET *n = SIMPLE_HASHTABLE_SLOT_DATA(sl);
         if(!n) continue;
 
-        freez(n->cmdline);
-        freez(n);
+        string_freez(n->cmdline);
+        aral_freez(ls->local_socket_aral, n);
     }
 
     // free the pid_socket hashtable data
@@ -903,7 +945,7 @@ static inline void local_sockets_cleanup(LS_STATE *ls) {
         if(!ps) continue;
 
         freez(ps->cmdline);
-        freez(ps);
+        aral_freez(ls->pid_socket_aral, ps);
     }
 
     // free the hashtable
@@ -912,6 +954,9 @@ static inline void local_sockets_cleanup(LS_STATE *ls) {
     simple_hashtable_destroy_LISTENING_PORT(&ls->listening_ports_hashtable);
     simple_hashtable_destroy_LOCAL_IP(&ls->local_ips_hashtable);
     simple_hashtable_destroy_LOCAL_SOCKET(&ls->sockets_hashtable);
+
+    aral_destroy(ls->local_socket_aral);
+    aral_destroy(ls->pid_socket_aral);
 }
 
 // --------------------------------------------------------------------------------------------------------------------
@@ -982,12 +1027,12 @@ static inline void local_sockets_send_to_parent(struct local_socket_state *ls __
     if(write(fd, n, sizeof(*n)) != sizeof(*n))
         local_sockets_log(ls, "failed to write local socket to pipe");
 
-    size_t len = n->cmdline ? strlen(n->cmdline) + 1 : 0;
+    size_t len = n->cmdline ? string_strlen(n->cmdline) + 1 : 0;
     if(write(fd, &len, sizeof(len)) != sizeof(len))
         local_sockets_log(ls, "failed to write cmdline length to pipe");
 
     if(len)
-        if(write(fd, n->cmdline, len) != (ssize_t)len)
+        if(write(fd, string2str(n->cmdline), len) != (ssize_t)len)
             local_sockets_log(ls, "failed to write cmdline to pipe");
 }
 
@@ -1087,9 +1132,13 @@ static inline bool local_sockets_get_namespace_sockets(LS_STATE *ls, struct pid_
             local_sockets_log(ls, "failed to read cmdline length from pipe");
 
         if(len) {
-            buf.cmdline = mallocz(len);
-            if(read(pipefd[0], buf.cmdline, len) != (ssize_t)len)
+            char cmdline[len + 1];
+            if(read(pipefd[0], cmdline, len) != (ssize_t)len)
                 local_sockets_log(ls, "failed to read cmdline from pipe");
+            else {
+                cmdline[len] = '\0';
+                buf.cmdline = string_strdupz(cmdline);
+            }
         }
         else
             buf.cmdline = NULL;
@@ -1107,8 +1156,7 @@ static inline bool local_sockets_get_namespace_sockets(LS_STATE *ls, struct pid_
         SIMPLE_HASHTABLE_SLOT_LOCAL_SOCKET *sl = simple_hashtable_get_slot_LOCAL_SOCKET(&ls->sockets_hashtable, buf.inode, &buf, true);
         LOCAL_SOCKET *n = SIMPLE_HASHTABLE_SLOT_DATA(sl);
         if(n) {
-            if(buf.cmdline)
-                freez(buf.cmdline);
+            string_freez(buf.cmdline);
 
 //            local_sockets_log(ls,
 //                              "ns inode %" PRIu64" (comm: '%s', pid: %u, ns: %"PRIu64") already exists in hashtable (comm: '%s', pid: %u, ns: %"PRIu64") - ignoring duplicate",
@@ -1116,7 +1164,7 @@ static inline bool local_sockets_get_namespace_sockets(LS_STATE *ls, struct pid_
             continue;
         }
         else {
-            n = mallocz(sizeof(*n));
+            n = aral_mallocz(ls->local_socket_aral);
             memcpy(n, &buf, sizeof(*n));
             simple_hashtable_set_slot_LOCAL_SOCKET(&ls->sockets_hashtable, sl, n->inode, n);
 
diff --git a/src/libnetdata/query_progress/progress.c b/src/libnetdata/query_progress/progress.c
index 1c772f87df..4ddb451350 100644
--- a/src/libnetdata/query_progress/progress.c
+++ b/src/libnetdata/query_progress/progress.c
@@ -408,6 +408,7 @@ int progress_function_result(BUFFER *wb, const char *hostname) {
     buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
     buffer_json_member_add_string(wb, "type", "table");
     buffer_json_member_add_time_t(wb, "update_every", 1);
+    buffer_json_member_add_boolean(wb, "has_history", false);
     buffer_json_member_add_string(wb, "help", RRDFUNCTIONS_PROGRESS_HELP);
     buffer_json_member_add_array(wb, "data");
 
diff --git a/src/libnetdata/socket/socket.c b/src/libnetdata/socket/socket.c
index 7dd731f857..17e9f6fb49 100644
--- a/src/libnetdata/socket/socket.c
+++ b/src/libnetdata/socket/socket.c
@@ -1387,7 +1387,7 @@ int accept_socket(int fd, int flags, char *client_ip, size_t ipsize, char *clien
     struct sockaddr_storage sadr;
     socklen_t addrlen = sizeof(sadr);
 
-    int nfd = accept4(fd, (struct sockaddr *)&sadr, &addrlen, flags);
+    int nfd = accept4(fd, (struct sockaddr *)&sadr, &addrlen, flags | SOCK_CLOEXEC);
     if (likely(nfd >= 0)) {
         if (getnameinfo((struct sockaddr *)&sadr, addrlen, client_ip, (socklen_t)ipsize,
                         client_port, (socklen_t)portsize, NI_NUMERICHOST | NI_NUMERICSERV) != 0) {