summaryrefslogtreecommitdiffstats
path: root/aclk/aclk_query.c
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2021-12-01 06:15:11 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2021-12-01 06:15:11 +0000
commit483926a283e118590da3f9ecfa75a8a4d62143ce (patch)
treecb77052778df9a128a8cd3ff5bf7645322a13bc5 /aclk/aclk_query.c
parentReleasing debian version 1.31.0-4. (diff)
downloadnetdata-483926a283e118590da3f9ecfa75a8a4d62143ce.tar.xz
netdata-483926a283e118590da3f9ecfa75a8a4d62143ce.zip
Merging upstream version 1.32.0.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'aclk/aclk_query.c')
-rw-r--r--aclk/aclk_query.c174
1 files changed, 142 insertions, 32 deletions
diff --git a/aclk/aclk_query.c b/aclk/aclk_query.c
index 3e2f88e46..001c1ba02 100644
--- a/aclk/aclk_query.c
+++ b/aclk/aclk_query.c
@@ -17,20 +17,20 @@ pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
typedef struct aclk_query_handler {
aclk_query_type_t type;
char *name; // for logging purposes
- int(*fnc)(mqtt_wss_client client, aclk_query_t query);
+ int(*fnc)(struct aclk_query_thread *query_thr, aclk_query_t query);
} aclk_query_handler;
-static int info_metadata(mqtt_wss_client client, aclk_query_t query)
+static int info_metadata(struct aclk_query_thread *query_thr, aclk_query_t query)
{
- aclk_send_info_metadata(client,
+ aclk_send_info_metadata(query_thr->client,
!query->data.metadata_info.initial_on_connect,
query->data.metadata_info.host);
return 0;
}
-static int alarms_metadata(mqtt_wss_client client, aclk_query_t query)
+static int alarms_metadata(struct aclk_query_thread *query_thr, aclk_query_t query)
{
- aclk_send_alarm_metadata(client,
+ aclk_send_alarm_metadata(query_thr->client,
!query->data.metadata_info.initial_on_connect);
return 0;
}
@@ -55,11 +55,34 @@ static usec_t aclk_web_api_v1_request(RRDHOST *host, struct web_client *w, char
return t;
}
-static int http_api_v2(mqtt_wss_client client, aclk_query_t query)
+static RRDHOST *node_id_2_rrdhost(const char *node_id)
+{
+ int res;
+ uuid_t node_id_bin, host_id_bin;
+ 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, 0);
+}
+
+#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)
{
int retval = 0;
usec_t t;
BUFFER *local_buffer = NULL;
+ BUFFER *log_buffer = buffer_create(NETDATA_WEB_REQUEST_URL_SIZE);
+ RRDHOST *query_host = localhost;
#ifdef NETDATA_WITH_ZLIB
int z_ret;
@@ -76,6 +99,26 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)
w->cookie2[0] = 0; // Simulate web_client_create_on_fd()
w->acl = 0x1f;
+ 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("URL requests node_id but there is not enough chars following");
+ retval = 1;
+ goto cleanup;
+ }
+ strncpyz(nodeid, node_uuid, UUID_STR_LEN - 1);
+
+ query_host = node_id_2_rrdhost(nodeid);
+ if (!query_host) {
+ error("Host with node_id \"%s\" not found! Query Ignored!", node_uuid);
+ retval = 1;
+ goto cleanup;
+ }
+ }
+
+ buffer_strcat(log_buffer, query->data.http_api_v2.query);
+
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);
@@ -85,8 +128,19 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)
mysep = strrchr(query->data.http_api_v2.query, '/');
+ 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_STATS_UNLOCK;
+ }
+
// execute the query
- t = aclk_web_api_v1_request(localhost, w, mysep ? mysep + 1 : "noop");
+ w->tv_in = query->created_tv;
+ now_realtime_timeval(&w->tv_ready);
+ t = aclk_web_api_v1_request(query_host, w, mysep ? mysep + 1 : "noop");
+ size_t size = (w->mode == WEB_CLIENT_MODE_FILECOPY) ? w->response.rlen : w->response.data->len;
+ size_t sent = size;
#ifdef NETDATA_WITH_ZLIB
// check if gzip encoding can and should be used
@@ -128,14 +182,13 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)
z_buffer->len += bytes_to_cpy;
} while(z_ret != Z_STREAM_END);
// so that web_client_build_http_header
- // puts correct content lenght into header
+ // puts correct content length into header
buffer_free(w->response.data);
w->response.data = z_buffer;
z_buffer = NULL;
}
#endif
- now_realtime_timeval(&w->tv_ready);
w->response.data->date = w->tv_ready.tv_sec;
web_client_build_http_header(w);
local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
@@ -149,6 +202,7 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)
buffer_need_bytes(local_buffer, w->response.data->len);
memcpy(&local_buffer->buffer[local_buffer->len], w->response.data->buffer, w->response.data->len);
local_buffer->len += w->response.data->len;
+ sent = sent - size + w->response.data->len;
} else {
#endif
buffer_strcat(local_buffer, w->response.data->buffer);
@@ -157,7 +211,26 @@ static int http_api_v2(mqtt_wss_client client, aclk_query_t query)
#endif
}
- aclk_http_msg_v2(client, query->callback_topic, query->msg_id, t, query->created, w->response.code, local_buffer->buffer, local_buffer->len);
+ // send msg.
+ aclk_http_msg_v2(query_thr->client, query->callback_topic, query->msg_id, t, query->created, w->response.code, local_buffer->buffer, local_buffer->len);
+
+ // log.
+ struct timeval tv;
+ now_realtime_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()
+ , query_thr->idx
+ , "DATA"
+ , 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
+ , w->response.code
+ , strip_control_characters((char *)buffer_tostring(log_buffer))
+ );
cleanup:
#ifdef NETDATA_WITH_ZLIB
@@ -170,45 +243,83 @@ cleanup:
buffer_free(w->response.header_output);
freez(w);
buffer_free(local_buffer);
+ buffer_free(log_buffer);
return retval;
}
-static int chart_query(mqtt_wss_client client, aclk_query_t query)
+static int chart_query(struct aclk_query_thread *query_thr, aclk_query_t query)
{
- aclk_chart_msg(client, query->data.chart_add_del.host, query->data.chart_add_del.chart_name);
+ aclk_chart_msg(query_thr->client, query->data.chart_add_del.host, query->data.chart_add_del.chart_name);
return 0;
}
-static int alarm_state_update_query(mqtt_wss_client client, aclk_query_t query)
+static int alarm_state_update_query(struct aclk_query_thread *query_thr, aclk_query_t query)
{
- aclk_alarm_state_msg(client, query->data.alarm_update);
+ aclk_alarm_state_msg(query_thr->client, query->data.alarm_update);
// aclk_alarm_state_msg frees the json object including the header it generates
query->data.alarm_update = NULL;
return 0;
}
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
+static int register_node(struct aclk_query_thread *query_thr, aclk_query_t query) {
+ // TODO create a pending registrations list
+ // with some timeouts to detect registration requests that
+ // go unanswered from the cloud
+ aclk_generate_node_registration(query_thr->client, &query->data.node_creation);
+ return 0;
+}
+
+static int node_state_update(struct aclk_query_thread *query_thr, aclk_query_t query) {
+ // TODO create a pending registrations list
+ // with some timeouts to detect registration requests that
+ // go unanswered from the cloud
+ aclk_generate_node_state_update(query_thr->client, &query->data.node_update);
+ return 0;
+}
+
+static int send_bin_msg(struct aclk_query_thread *query_thr, aclk_query_t query)
+{
+ // this will be simplified when legacy support is removed
+ aclk_send_bin_message_subtopic_pid(query_thr->client, query->data.bin_payload.payload, query->data.bin_payload.size, query->data.bin_payload.topic, query->data.bin_payload.msg_name);
+ return 0;
+}
+#endif
+
aclk_query_handler aclk_query_handlers[] = {
- { .type = HTTP_API_V2, .name = "http api request v2", .fnc = http_api_v2 },
- { .type = ALARM_STATE_UPDATE, .name = "alarm state update", .fnc = alarm_state_update_query },
- { .type = METADATA_INFO, .name = "info metadata", .fnc = info_metadata },
- { .type = METADATA_ALARMS, .name = "alarms metadata", .fnc = alarms_metadata },
- { .type = CHART_NEW, .name = "chart new", .fnc = chart_query },
- { .type = CHART_DEL, .name = "chart delete", .fnc = info_metadata },
- { .type = UNKNOWN, .name = NULL, .fnc = NULL }
+ { .type = HTTP_API_V2, .name = "http api request v2", .fnc = http_api_v2 },
+ { .type = ALARM_STATE_UPDATE, .name = "alarm state update", .fnc = alarm_state_update_query },
+ { .type = METADATA_INFO, .name = "info metadata", .fnc = info_metadata },
+ { .type = METADATA_ALARMS, .name = "alarms metadata", .fnc = alarms_metadata },
+ { .type = CHART_NEW, .name = "chart new", .fnc = chart_query },
+ { .type = CHART_DEL, .name = "chart delete", .fnc = info_metadata },
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
+ { .type = REGISTER_NODE, .name = "register node", .fnc = register_node },
+ { .type = NODE_STATE_UPDATE, .name = "node state update", .fnc = node_state_update },
+ { .type = CHART_DIMS_UPDATE, .name = "chart and dim update bin", .fnc = send_bin_msg },
+ { .type = CHART_CONFIG_UPDATED, .name = "chart config updated", .fnc = send_bin_msg },
+ { .type = CHART_RESET, .name = "reset chart messages", .fnc = send_bin_msg },
+ { .type = RETENTION_UPDATED, .name = "update retention info", .fnc = send_bin_msg },
+ { .type = UPDATE_NODE_INFO, .name = "update node info", .fnc = send_bin_msg },
+ { .type = ALARM_LOG_HEALTH, .name = "alarm log health", .fnc = send_bin_msg },
+ { .type = ALARM_PROVIDE_CFG, .name = "provide alarm config", .fnc = send_bin_msg },
+ { .type = ALARM_SNAPSHOT, .name = "alarm snapshot", .fnc = send_bin_msg },
+#endif
+ { .type = UNKNOWN, .name = NULL, .fnc = NULL }
};
-static void aclk_query_process_msg(struct aclk_query_thread *info, aclk_query_t query)
+static void aclk_query_process_msg(struct aclk_query_thread *query_thr, aclk_query_t query)
{
for (int i = 0; aclk_query_handlers[i].type != UNKNOWN; i++) {
if (aclk_query_handlers[i].type == query->type) {
debug(D_ACLK, "Processing Queued Message of type: \"%s\"", aclk_query_handlers[i].name);
- aclk_query_handlers[i].fnc(info->client, query);
+ aclk_query_handlers[i].fnc(query_thr, query);
aclk_query_free(query);
if (aclk_stats_enabled) {
ACLK_STATS_LOCK;
aclk_metrics_per_sample.queries_dispatched++;
- aclk_queries_per_thread[info->idx]++;
+ aclk_queries_per_thread[query_thr->idx]++;
ACLK_STATS_UNLOCK;
}
return;
@@ -219,11 +330,11 @@ static void aclk_query_process_msg(struct aclk_query_thread *info, aclk_query_t
/* Processes messages from queue. Compete for work with other threads
*/
-int aclk_query_process_msgs(struct aclk_query_thread *info)
+int aclk_query_process_msgs(struct aclk_query_thread *query_thr)
{
aclk_query_t query;
while ((query = aclk_queue_pop()))
- aclk_query_process_msg(info, query);
+ aclk_query_process_msg(query_thr, query);
return 0;
}
@@ -233,21 +344,20 @@ int aclk_query_process_msgs(struct aclk_query_thread *info)
*/
void *aclk_query_main_thread(void *ptr)
{
- struct aclk_query_thread *info = ptr;
+ struct aclk_query_thread *query_thr = ptr;
+
while (!netdata_exit) {
- aclk_query_process_msgs(info);
+ aclk_query_process_msgs(query_thr);
QUERY_THREAD_LOCK;
-
if (unlikely(pthread_cond_wait(&query_cond_wait, &query_lock_wait)))
sleep_usec(USEC_PER_SEC * 1);
-
QUERY_THREAD_UNLOCK;
}
return NULL;
}
-#define TASK_LEN_MAX 16
+#define TASK_LEN_MAX 22
void aclk_query_threads_start(struct aclk_query_threads *query_threads, mqtt_wss_client client)
{
info("Starting %d query threads.", query_threads->count);
@@ -257,7 +367,7 @@ void aclk_query_threads_start(struct aclk_query_threads *query_threads, mqtt_wss
for (int i = 0; i < query_threads->count; i++) {
query_threads->thread_list[i].idx = i; //thread needs to know its index for statistics
- if(unlikely(snprintf(thread_name, TASK_LEN_MAX, "%s_%d", ACLK_QUERY_THREAD_NAME, i) < 0))
+ if(unlikely(snprintfz(thread_name, TASK_LEN_MAX, "%s_%d", ACLK_QUERY_THREAD_NAME, i) < 0))
error("snprintf encoding error");
netdata_thread_create(
&query_threads->thread_list[i].thread, thread_name, NETDATA_THREAD_OPTION_JOINABLE, aclk_query_main_thread,