From c3d70ffcb43b62c95d71334ed49ad345ddf4360d Mon Sep 17 00:00:00 2001 From: Costa Tsaousis Date: Thu, 20 Apr 2023 20:49:06 +0300 Subject: WEBRTC for communication between agents and browsers (#14874) * initial webrtc setup * missing files * rewrite of webrtc integration * initialization and cleanup of webrtc connections * make it compile without libdatachannel * add missing webrtc_initialize() function when webrtc is not enabled * make c++17 optional * add build/m4/ax_compiler_vendor.m4 * add ax_cxx_compile_stdcxx.m4 * added new m4 files to makefile.am * id all webrtc connections * show warning when webrtc is disabled * fixed message * moved all webrtc error checking inside webrtc.cpp * working webrtc connection establishment and cleanup * remove obsolete code * rewrote webrtc code in C to remove dependency for c++17 * fixed left-over reference * detect binary and text messages * minor fix * naming of webrtc threads * added webrtc configuration * fix for thread_get_name_np() * smaller web_client memory footprint * universal web clients cache * free web clients every 100 uses * webrtc is now enabled by default only when compiled with internal checks * webrtc responses to /api/ requests, including LZ4 compression * fix for binary and text messages * web_client_cache is now global * unification of the internal web server API, for web requests, aclk request, webrtc requests * more cleanup and unification of web client timings * fixed compiler warnings * update sent and received bytes * eliminated of almost all big buffers in web client * registry now uses the new json generation * cookies are now an array; fixed redirects * fix redirects, again * write cookies directly to the header buffer, eliminating the need for cookie structures in web client * reset the has_cookies flag * gathered all web client cleanup to one function * fixes redirects * added summary.globals in /api/v2/data response * ars to arc in /api/v2/data * properly handle host impersonation * set the context of mem.numa_nodes --- aclk/aclk_query.c | 187 ++++++++++++------------------------------------ aclk/aclk_query_queue.c | 2 +- 2 files changed, 47 insertions(+), 142 deletions(-) (limited to 'aclk') diff --git a/aclk/aclk_query.c b/aclk/aclk_query.c index 1d00cdca3e..3908b8bad4 100644 --- a/aclk/aclk_query.c +++ b/aclk/aclk_query.c @@ -3,6 +3,7 @@ #include "aclk_query.h" #include "aclk_stats.h" #include "aclk_tx_msgs.h" +#include "../../web/server/web_client_cache.h" #define WEB_HDR_ACCEPT_ENC "Accept-Encoding:" #define ACLK_MAX_WEB_RESPONSE_SIZE (30 * 1024 * 1024) @@ -12,71 +13,11 @@ pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER; #define QUERY_THREAD_LOCK pthread_mutex_lock(&query_lock_wait) #define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait) -static usec_t aclk_web_api_request(RRDHOST *host, struct web_client *w, char *url, const size_t api_version) -{ - usec_t t; - - t = now_monotonic_high_precision_usec(); - - if(api_version == 2) - w->response.code = web_client_api_request_v2(host, w, url); - else - w->response.code = web_client_api_request_v1(host, w, url); - - if(buffer_strlen(w->response.data) > ACLK_MAX_WEB_RESPONSE_SIZE) { - buffer_flush(w->response.data); - buffer_strcat(w->response.data, "response is too big"); - w->response.data->content_type = CT_TEXT_PLAIN; - w->response.code = HTTP_RESP_CONTENT_TOO_LONG; - } - - t = now_monotonic_high_precision_usec() - t; - - if (aclk_stats_enabled) { - ACLK_STATS_LOCK; - aclk_metrics_per_sample.cloud_q_process_total += t; - aclk_metrics_per_sample.cloud_q_process_count++; - if (aclk_metrics_per_sample.cloud_q_process_max < t) - aclk_metrics_per_sample.cloud_q_process_max = t; - ACLK_STATS_UNLOCK; - } - - return t; -} - -static RRDHOST *node_id_2_rrdhost(const char *node_id) -{ - int res; - uuid_t node_id_bin, host_id_bin; - - RRDHOST *host = find_host_by_node_id((char *)node_id); - if (host) - return host; - - char host_id[UUID_STR_LEN]; - if (uuid_parse(node_id, node_id_bin)) { - error("Couldn't parse UUID %s", node_id); - return NULL; - } - if ((res = get_host_id(&node_id_bin, &host_id_bin))) { - error("node not found rc=%d", res); - return NULL; - } - uuid_unparse_lower(host_id_bin, host_id); - return rrdhost_find_by_guid(host_id); -} - -#define NODE_ID_QUERY "/node/" -// TODO this function should be quarantied and written nicely -// lots of skeletons from initial ACLK Legacy impl. -// quick and dirty from the start -static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) -{ +static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) { int retval = 0; - usec_t t; BUFFER *local_buffer = NULL; - BUFFER *log_buffer = buffer_create(NETDATA_WEB_REQUEST_URL_SIZE, &netdata_buffers_statistics.buffers_aclk); - RRDHOST *query_host = localhost; + size_t size = 0; + size_t sent = 0; #ifdef NETDATA_WITH_ZLIB int z_ret; @@ -84,83 +25,52 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) char *start, *end; #endif - struct web_client *w = (struct web_client *)callocz(1, sizeof(struct web_client)); - w->response.data = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE, &netdata_buffers_statistics.buffers_aclk); - w->response.header = buffer_create(NETDATA_WEB_RESPONSE_HEADER_INITIAL_SIZE, &netdata_buffers_statistics.buffers_aclk); - w->response.header_output = buffer_create(NETDATA_WEB_RESPONSE_HEADER_INITIAL_SIZE, &netdata_buffers_statistics.buffers_aclk); - strcpy(w->origin, "*"); // Simulate web_client_create_on_fd() - w->cookie1[0] = 0; // Simulate web_client_create_on_fd() - w->cookie2[0] = 0; // Simulate web_client_create_on_fd() + struct web_client *w = web_client_get_from_cache(); w->acl = WEB_CLIENT_ACL_ACLK; + w->mode = WEB_CLIENT_MODE_GET; + w->timings.tv_in = query->created_tv; - buffer_strcat(log_buffer, query->data.http_api_v2.query); - size_t size = 0; - size_t sent = 0; - w->tv_in = query->created_tv; - now_realtime_timeval(&w->tv_ready); - - if (query->timeout) { - int in_queue = (int) (dt_usec(&w->tv_in, &w->tv_ready) / 1000); - if (in_queue > query->timeout) { - log_access("QUERY CANCELED: QUEUE TIME EXCEEDED %d ms (LIMIT %d ms)", in_queue, query->timeout); - retval = 1; - w->response.code = HTTP_RESP_BACKEND_FETCH_FAILED; - aclk_http_msg_v2_err(query_thr->client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0); - goto cleanup; - } + usec_t t; + web_client_timeout_checkpoint_set(w, query->timeout); + if(web_client_timeout_checkpoint_and_check(w, &t)) { + log_access("QUERY CANCELED: QUEUE TIME EXCEEDED %llu ms (LIMIT %d ms)", t / USEC_PER_MS, query->timeout); + retval = 1; + w->response.code = HTTP_RESP_BACKEND_FETCH_FAILED; + aclk_http_msg_v2_err(query_thr->client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0); + goto cleanup; } - if (!strncmp(query->data.http_api_v2.query, NODE_ID_QUERY, strlen(NODE_ID_QUERY))) { - char *node_uuid = query->data.http_api_v2.query + strlen(NODE_ID_QUERY); - char nodeid[UUID_STR_LEN]; - if (strlen(node_uuid) < (UUID_STR_LEN - 1)) { - error_report(CLOUD_EMSG_MALFORMED_NODE_ID); - retval = 1; - w->response.code = 404; - aclk_http_msg_v2_err(query_thr->client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_MALFORMED_NODE_ID, CLOUD_EMSG_MALFORMED_NODE_ID, NULL, 0); - goto cleanup; - } - strncpyz(nodeid, node_uuid, UUID_STR_LEN - 1); - - query_host = node_id_2_rrdhost(nodeid); - if (!query_host) { - error_report("Host with node_id \"%s\" not found! Returning 404 to Cloud!", nodeid); - retval = 1; - w->response.code = 404; - aclk_http_msg_v2_err(query_thr->client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_NODE_NOT_FOUND, CLOUD_EMSG_NODE_NOT_FOUND, NULL, 0); - goto cleanup; - } - } + web_client_decode_path_and_query_string(w, query->data.http_api_v2.query); + char *path = (char *)buffer_tostring(w->url_path_decoded); - size_t api_version = 1; - { - char *s = strstr(query->data.http_api_v2.query, "/api/v"); - if(s && s[6]) { - api_version = str2u(&s[6]); - if(api_version != 1 && api_version != 2) - api_version = 1; - } + if (aclk_stats_enabled) { + char *url_path_endpoint = strrchr(path, '/'); + ACLK_STATS_LOCK; + int stat_idx = aclk_cloud_req_http_type_to_idx(url_path_endpoint ? url_path_endpoint + 1 : "other"); + aclk_metrics_per_sample.cloud_req_http_by_type[stat_idx]++; + ACLK_STATS_UNLOCK; } - char *mysep = strchr(query->data.http_api_v2.query, '?'); - if (mysep) { - url_decode_r(w->decoded_query_string, mysep, NETDATA_WEB_REQUEST_URL_SIZE + 1); - *mysep = '\0'; - } else - url_decode_r(w->decoded_query_string, query->data.http_api_v2.query, NETDATA_WEB_REQUEST_URL_SIZE + 1); + w->response.code = web_client_api_request_with_node_selection(localhost, w, path); + web_client_timeout_checkpoint_response_ready(w, &t); - mysep = strrchr(query->data.http_api_v2.query, '/'); + if(buffer_strlen(w->response.data) > ACLK_MAX_WEB_RESPONSE_SIZE) { + buffer_flush(w->response.data); + buffer_strcat(w->response.data, "response is too big"); + w->response.data->content_type = CT_TEXT_PLAIN; + w->response.code = HTTP_RESP_CONTENT_TOO_LONG; + } if (aclk_stats_enabled) { ACLK_STATS_LOCK; - int stat_idx = aclk_cloud_req_http_type_to_idx(mysep ? mysep + 1 : "other"); - aclk_metrics_per_sample.cloud_req_http_by_type[stat_idx]++; + aclk_metrics_per_sample.cloud_q_process_total += t; + aclk_metrics_per_sample.cloud_q_process_count++; + if (aclk_metrics_per_sample.cloud_q_process_max < t) + aclk_metrics_per_sample.cloud_q_process_max = t; ACLK_STATS_UNLOCK; } - // execute the query - t = aclk_web_api_request(query_host, w, mysep ? mysep + 1 : "noop", api_version); - size = (w->mode == WEB_CLIENT_MODE_FILECOPY) ? w->response.rlen : w->response.data->len; + size = w->response.data->len; sent = size; #ifdef NETDATA_WITH_ZLIB @@ -175,8 +85,8 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) w->response.zstream.zfree = Z_NULL; w->response.zstream.opaque = Z_NULL; if(deflateInit2(&w->response.zstream, web_gzip_level, Z_DEFLATED, 15 + 16, 8, web_gzip_strategy) == Z_OK) { - w->response.zinitialized = 1; - w->response.zoutput = 1; + w->response.zinitialized = true; + w->response.zoutput = true; } else error("Failed to initialize zlib. Proceeding without compression."); } @@ -212,7 +122,7 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) } #endif - w->response.data->date = w->tv_ready.tv_sec; + w->response.data->date = w->timings.tv_ready.tv_sec; web_client_build_http_header(w); local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE, &netdata_buffers_statistics.buffers_aclk); local_buffer->content_type = CT_APPLICATION_JSON; @@ -240,7 +150,7 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) struct timeval tv; cleanup: - now_realtime_timeval(&tv); + now_monotonic_high_precision_timeval(&tv); log_access("%llu: %d '[ACLK]:%d' '%s' (sent/all = %zu/%zu bytes %0.0f%%, prep/sent/total = %0.2f/%0.2f/%0.2f ms) %d '%s'", w->id , gettid() @@ -249,24 +159,19 @@ cleanup: , sent , size , size > sent ? -(((size - sent) / (double)size) * 100.0) : ((size > 0) ? (((sent - size ) / (double)size) * 100.0) : 0.0) - , dt_usec(&w->tv_ready, &w->tv_in) / 1000.0 - , dt_usec(&tv, &w->tv_ready) / 1000.0 - , dt_usec(&tv, &w->tv_in) / 1000.0 + , dt_usec(&w->timings.tv_ready, &w->timings.tv_in) / 1000.0 + , dt_usec(&tv, &w->timings.tv_ready) / 1000.0 + , dt_usec(&tv, &w->timings.tv_in) / 1000.0 , w->response.code - , strip_control_characters((char *)buffer_tostring(log_buffer)) + , strip_control_characters((char *)buffer_tostring(w->url_as_received)) ); + web_client_release_to_cache(w); + #ifdef NETDATA_WITH_ZLIB - if(w->response.zinitialized) - deflateEnd(&w->response.zstream); buffer_free(z_buffer); #endif - buffer_free(w->response.data); - buffer_free(w->response.header); - buffer_free(w->response.header_output); - freez(w); buffer_free(local_buffer); - buffer_free(log_buffer); return retval; } diff --git a/aclk/aclk_query_queue.c b/aclk/aclk_query_queue.c index e7cad5ded6..78a906d962 100644 --- a/aclk/aclk_query_queue.c +++ b/aclk/aclk_query_queue.c @@ -20,7 +20,7 @@ static struct aclk_query_queue { static inline int _aclk_queue_query(aclk_query_t query) { - now_realtime_timeval(&query->created_tv); + now_monotonic_high_precision_timeval(&query->created_tv); query->created = now_realtime_usec(); ACLK_QUEUE_LOCK; -- cgit v1.2.3