summaryrefslogtreecommitdiffstats
path: root/aclk/aclk_rx_msgs.c
diff options
context:
space:
mode:
Diffstat (limited to 'aclk/aclk_rx_msgs.c')
-rw-r--r--aclk/aclk_rx_msgs.c39
1 files changed, 26 insertions, 13 deletions
diff --git a/aclk/aclk_rx_msgs.c b/aclk/aclk_rx_msgs.c
index 1f2cb27e..27f1bf2d 100644
--- a/aclk/aclk_rx_msgs.c
+++ b/aclk/aclk_rx_msgs.c
@@ -17,6 +17,7 @@ struct aclk_request {
char *callback_topic;
char *payload;
int version;
+ int timeout;
int min_version;
int max_version;
};
@@ -57,6 +58,10 @@ static int cloud_to_agent_parse(JSON_ENTRY *e)
data->version = e->data.number;
break;
}
+ if (!strcmp(e->name, "timeout")) {
+ data->timeout = e->data.number;
+ break;
+ }
if (!strcmp(e->name, "min-version")) {
data->min_version = e->data.number;
break;
@@ -160,6 +165,7 @@ static int aclk_handle_cloud_http_request_v2(struct aclk_request *cloud_to_agent
// aclk_queue_query takes ownership of data pointer
query->callback_topic = cloud_to_agent->callback_topic;
+ query->timeout = cloud_to_agent->timeout;
// 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;
@@ -271,32 +277,39 @@ int create_node_instance_result(const char *msg, size_t msg_len)
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);
+ node_instance_connection_t node_state_update = {
+ .hops = 1,
+ .live = 0,
+ .queryable = 1,
+ .session_id = aclk_session_newarch,
+ .node_id = res.node_id
+ };
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;
+ node_state_update.live = 1;
+ node_state_update.hops = 0;
} else {
netdata_mutex_lock(&host->receiver_lock);
- query->data.node_update.live = (host->receiver != NULL);
+ node_state_update.live = (host->receiver != NULL);
netdata_mutex_unlock(&host->receiver_lock);
- query->data.node_update.hops = host->system_info->hops;
+ node_state_update.hops = host->system_info->hops;
}
}
- 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;
+ rrdhost_aclk_state_lock(localhost);
+ node_state_update.claim_id = localhost->aclk_state.claimed_id;
+ query->data.bin_payload.payload = generate_node_instance_connection(&query->data.bin_payload.size, &node_state_update);
+ rrdhost_aclk_state_unlock(localhost);
+
+ query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
+ query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
+
aclk_queue_query(query);
+ freez(res.node_id);
freez(res.machine_guid);
return 0;
}