Viewing file:  aclk_rx_msgs.c (13.28 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
// SPDX-License-Identifier: GPL-3.0-or-later
#include "aclk_rx_msgs.h"
#include "aclk_stats.h"
#include "aclk_query_queue.h"
#include "aclk.h"
#define ACLK_V2_PAYLOAD_SEPARATOR "\x0D\x0A\x0D\x0A"
#define ACLK_CLOUD_REQ_V2_PREFIX "GET /"
#define ACLK_V_COMPRESSION 2
struct aclk_request {
char *type_id;
char *msg_id;
char *callback_topic;
char *payload;
int version;
int min_version;
int max_version;
};
static int cloud_to_agent_parse(JSON_ENTRY *e)
{
struct aclk_request *data = e->callback_data;
switch (e->type) {
case JSON_OBJECT:
case JSON_ARRAY:
break;
case JSON_STRING:
if (!strcmp(e->name, "msg-id")) {
data->msg_id = strdupz(e->data.string);
break;
}
if (!strcmp(e->name, "type")) {
data->type_id = strdupz(e->data.string);
break;
}
if (!strcmp(e->name, "callback-topic")) {
data->callback_topic = strdupz(e->data.string);
break;
}
if (!strcmp(e->name, "payload")) {
if (likely(e->data.string)) {
size_t len = strlen(e->data.string);
data->payload = mallocz(len+1);
if (!url_decode_r(data->payload, e->data.string, len + 1))
strcpy(data->payload, e->data.string);
}
break;
}
break;
case JSON_NUMBER:
if (!strcmp(e->name, "version")) {
data->version = e->data.number;
break;
}
if (!strcmp(e->name, "min-version")) {
data->min_version = e->data.number;
break;
}
if (!strcmp(e->name, "max-version")) {
data->max_version = e->data.number;
break;
}
break;
case JSON_BOOLEAN:
break;
case JSON_NULL:
break;
}
return 0;
}
static inline int aclk_extract_v2_data(char *payload, char **data)
{
char* ptr = strstr(payload, ACLK_V2_PAYLOAD_SEPARATOR);
if(!ptr)
return 1;
ptr += strlen(ACLK_V2_PAYLOAD_SEPARATOR);
*data = strdupz(ptr);
return 0;
}
static inline int aclk_v2_payload_get_query(const char *payload, char **query_url)
{
const char *start, *end;
// TODO better check of URL
if(strncmp(payload, ACLK_CLOUD_REQ_V2_PREFIX, strlen(ACLK_CLOUD_REQ_V2_PREFIX))) {
errno = 0;
error("Only accepting requests that start with \"%s\" from CLOUD.", ACLK_CLOUD_REQ_V2_PREFIX);
return 1;
}
start = payload + 4;
if(!(end = strstr(payload, " HTTP/1.1\x0D\x0A"))) {
errno = 0;
error("Doesn't look like HTTP GET request.");
return 1;
}
*query_url = mallocz((end - start) + 1);
strncpyz(*query_url, start, end - start);
return 0;
}
#define HTTP_CHECK_AGENT_INITIALIZED() ACLK_SHARED_STATE_LOCK;\
if (unlikely(aclk_shared_state.agent_state == ACLK_HOST_INITIALIZING)) {\
debug(D_ACLK, "Ignoring \"http\" cloud request; agent not in stable state");\
ACLK_SHARED_STATE_UNLOCK;\
return 1;\
}\
ACLK_SHARED_STATE_UNLOCK;
static int aclk_handle_cloud_request_v2(struct aclk_request *cloud_to_agent, char *raw_payload)
{
if (!aclk_use_new_cloud_arch) {
HTTP_CHECK_AGENT_INITIALIZED();
}
aclk_query_t query;
errno = 0;
if (cloud_to_agent->version < ACLK_V_COMPRESSION) {
error(
"This handler cannot reply to request with version older than %d, received %d.",
ACLK_V_COMPRESSION,
cloud_to_agent->version);
return 1;
}
query = aclk_query_new(HTTP_API_V2);
if (unlikely(aclk_extract_v2_data(raw_payload, &query->data.http_api_v2.payload))) {
error("Error extracting payload expected after the JSON dictionary.");
goto error;
}
if (unlikely(aclk_v2_payload_get_query(query->data.http_api_v2.payload, &query->dedup_id))) {
error("Could not extract payload from query");
goto error;
}
if (unlikely(!cloud_to_agent->callback_topic)) {
error("Missing callback_topic");
goto error;
}
if (unlikely(!cloud_to_agent->msg_id)) {
error("Missing msg_id");
goto error;
}
// aclk_queue_query takes ownership of data pointer
query->callback_topic = cloud_to_agent->callback_topic;
// for clarity and code readability as when we process the request
// it would be strange to get URL from `dedup_id`
query->data.http_api_v2.query = query->dedup_id;
query->msg_id = cloud_to_agent->msg_id;
aclk_queue_query(query);
return 0;
error:
aclk_query_free(query);
return 1;
}
typedef struct aclk_incoming_msg_type{
char *name;
int(*fnc)(struct aclk_request *, char *);
}aclk_incoming_msg_type;
aclk_incoming_msg_type aclk_incoming_msg_types_compression[] = {
{ .name = "http", .fnc = aclk_handle_cloud_request_v2 },
{ .name = NULL, .fnc = NULL }
};
struct aclk_incoming_msg_type *aclk_incoming_msg_types = aclk_incoming_msg_types_compression;
int aclk_handle_cloud_message(char *payload)
{
struct aclk_request cloud_to_agent;
memset(&cloud_to_agent, 0, sizeof(struct aclk_request));
if (aclk_stats_enabled) {
ACLK_STATS_LOCK;
aclk_metrics_per_sample.cloud_req_recvd++;
ACLK_STATS_UNLOCK;
}
if (unlikely(!payload)) {
errno = 0;
error("ACLK incoming message is empty");
goto err_cleanup_nojson;
}
debug(D_ACLK, "ACLK incoming message (%s)", payload);
int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
if (unlikely(rc != JSON_OK)) {
errno = 0;
error("Malformed json request (%s)", payload);
goto err_cleanup;
}
if (!cloud_to_agent.type_id) {
errno = 0;
error("Cloud message is missing compulsory key \"type\"");
goto err_cleanup;
}
for (int i = 0; aclk_incoming_msg_types[i].name; i++) {
if (strcmp(cloud_to_agent.type_id, aclk_incoming_msg_types[i].name) == 0) {
if (likely(!aclk_incoming_msg_types[i].fnc(&cloud_to_agent, payload))) {
// in case of success handler is supposed to clean up after itself
// or as in the case of aclk_handle_cloud_request take
// ownership of the pointers (done to avoid copying)
// see what `aclk_queue_query` parameter `internal` does
// NEVER CONTINUE THIS LOOP AFTER CALLING FUNCTION!!!
// msg handlers (namely aclk_handle_version_responce)
// can freely change what aclk_incoming_msg_types points to
// so either exit or restart this for loop
freez(cloud_to_agent.type_id);
return 0;
}
goto err_cleanup;
}
}
errno = 0;
error("Unknown message type from Cloud \"%s\"", cloud_to_agent.type_id);
err_cleanup:
if (cloud_to_agent.payload)
freez(cloud_to_agent.payload);
if (cloud_to_agent.type_id)
freez(cloud_to_agent.type_id);
if (cloud_to_agent.msg_id)
freez(cloud_to_agent.msg_id);
if (cloud_to_agent.callback_topic)
freez(cloud_to_agent.callback_topic);
err_cleanup_nojson:
if (aclk_stats_enabled) {
ACLK_STATS_LOCK;
aclk_metrics_per_sample.cloud_req_err++;
ACLK_STATS_UNLOCK;
}
return 1;
}
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
{
// TODO do the look up table with hashes to optimize when there are more
// than few
if (!strcmp(message_type, "cmd")) {
aclk_handle_cloud_message((char *)msg);
return;
}
if (!strcmp(message_type, "CreateNodeInstanceResult")) {
node_instance_creation_result_t res = parse_create_node_instance_result(msg, msg_len);
if (!res.machine_guid || !res.node_id) {
error_report("Error parsing CreateNodeInstanceResult");
freez(res.machine_guid);
freez(res.node_id);
return;
}
debug(D_ACLK, "CreateNodeInstanceResult: guid:%s nodeid:%s", res.machine_guid, res.node_id);
uuid_t host_id, node_id;
if (uuid_parse(res.machine_guid, host_id)) {
error("Error parsing machine_guid provided by CreateNodeInstanceResult");
freez(res.machine_guid);
freez(res.node_id);
return;
}
if (uuid_parse(res.node_id, node_id)) {
error("Error parsing node_id provided by CreateNodeInstanceResult");
freez(res.machine_guid);
freez(res.node_id);
return;
}
update_node_id(&host_id, &node_id);
aclk_query_t query = aclk_query_new(NODE_STATE_UPDATE);
query->data.node_update.hops = 1; //TODO - real hop count instead of hardcoded
rrdhost_aclk_state_lock(localhost);
query->data.node_update.claim_id = strdupz(localhost->aclk_state.claimed_id);
rrdhost_aclk_state_unlock(localhost);
RRDHOST *host = rrdhost_find_by_guid(res.machine_guid, 0);
query->data.node_update.live = 0;
if (host) {
// not all host must have RRDHOST struct created for them
// if they never connected during runtime of agent
if (host == localhost) {
query->data.node_update.live = 1;
query->data.node_update.hops = 0;
} else {
netdata_mutex_lock(&host->receiver_lock);
query->data.node_update.live = (host->receiver != NULL);
netdata_mutex_unlock(&host->receiver_lock);
}
}
query->data.node_update.node_id = res.node_id; // aclk_query_free will free it
query->data.node_update.queryable = 1;
query->data.node_update.session_id = aclk_session_newarch;
aclk_queue_query(query);
freez(res.machine_guid);
return;
}
if (!strcmp(message_type, "SendNodeInstances")) {
debug(D_ACLK, "Got SendNodeInstances");
aclk_send_node_instances();
return;
}
if (!strcmp(message_type, "StreamChartsAndDimensions")) {
stream_charts_and_dims_t res = parse_stream_charts_and_dims(msg, msg_len);
if (!res.claim_id || !res.node_id) {
error("Error parsing StreamChartsAndDimensions msg");
freez(res.claim_id);
freez(res.node_id);
return;
}
chart_batch_id = res.batch_id;
aclk_start_streaming(res.node_id, res.seq_id, res.seq_id_created_at.tv_sec, res.batch_id);
freez(res.claim_id);
freez(res.node_id);
return;
}
if (!strcmp(message_type, "ChartsAndDimensionsAck")) {
chart_and_dim_ack_t res = parse_chart_and_dimensions_ack(msg, msg_len);
if (!res.claim_id || !res.node_id) {
error("Error parsing StreamChartsAndDimensions msg");
freez(res.claim_id);
freez(res.node_id);
return;
}
aclk_ack_chart_sequence_id(res.node_id, res.last_seq_id);
freez(res.claim_id);
freez(res.node_id);
return;
}
if (!strcmp(message_type, "UpdateChartConfigs")) {
struct update_chart_config res = parse_update_chart_config(msg, msg_len);
if (!res.claim_id || !res.node_id || !res.hashes)
error("Error parsing UpdateChartConfigs msg");
else
aclk_get_chart_config(res.hashes);
destroy_update_chart_config(&res);
return;
}
if (!strcmp(message_type, "StartAlarmStreaming")) {
struct start_alarm_streaming res = parse_start_alarm_streaming(msg, msg_len);
if (!res.node_id || !res.batch_id) {
error("Error parsing StartAlarmStreaming");
freez(res.node_id);
return;
}
aclk_start_alert_streaming(res.node_id, res.batch_id, res.start_seq_id);
freez(res.node_id);
return;
}
if (!strcmp(message_type, "SendAlarmLogHealth")) {
char *node_id = parse_send_alarm_log_health(msg, msg_len);
if (!node_id) {
error("Error parsing SendAlarmLogHealth");
return;
}
aclk_send_alarm_health_log(node_id);
freez(node_id);
return;
}
if (!strcmp(message_type, "SendAlarmConfiguration")) {
char *config_hash = parse_send_alarm_configuration(msg, msg_len);
if (!config_hash || !*config_hash) {
error("Error parsing SendAlarmConfiguration");
freez(config_hash);
return;
}
aclk_send_alarm_configuration(config_hash);
freez(config_hash);
return;
}
if (!strcmp(message_type, "SendAlarmSnapshot")) {
struct send_alarm_snapshot *sas = parse_send_alarm_snapshot(msg, msg_len);
if (!sas->node_id || !sas->claim_id) {
error("Error parsing SendAlarmSnapshot");
destroy_send_alarm_snapshot(sas);
return;
}
aclk_process_send_alarm_snapshot(sas->node_id, sas->claim_id, sas->snapshot_id, sas->sequence_id);
destroy_send_alarm_snapshot(sas);
return;
}
error ("Unknown new cloud arch message type received \"%s\"", message_type);
}
#endif
|