@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 226 lines 7.41 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "aclk_query.h"
4 #include "aclk_tx_msgs.h"
5 #include "web/server/web_client_cache.h"
6
7 static HTTP_ACL default_aclk_http_acl = HTTP_ACL_ALL_FEATURES;
8
9 struct pending_req_list {
10 const char *msg_id;
11 uint32_t hash;
12
13 int canceled;
14
15 struct pending_req_list *next;
16 };
17
18 static struct pending_req_list *pending_req_list_head = NULL;
19 static SPINLOCK pending_req_list_lock = SPINLOCK_INITIALIZER;
20
21 void aclk_config_get_query_scope(void) {
22 const char *s = inicfg_get(&netdata_config, CONFIG_SECTION_CLOUD, "scope", "full");
23 if(strcmp(s, "license manager") == 0)
24 default_aclk_http_acl = HTTP_ACL_ACLK_LICENSE_MANAGER;
25 }
26
27 bool aclk_query_scope_has(HTTP_ACL acl) {
28 return (default_aclk_http_acl & acl) == acl;
29 }
30
31 static struct pending_req_list *pending_req_list_add(const char *msg_id)
32 {
33 struct pending_req_list *new = callocz(1, sizeof(struct pending_req_list));
34 new->msg_id = msg_id;
35 new->hash = simple_hash(msg_id);
36
37 spinlock_lock(&pending_req_list_lock);
38 new->next = pending_req_list_head;
39 pending_req_list_head = new;
40 spinlock_unlock(&pending_req_list_lock);
41 return new;
42 }
43
44 void pending_req_list_rm(const char *msg_id)
45 {
46 uint32_t hash = simple_hash(msg_id);
47 struct pending_req_list *prev = NULL;
48
49 spinlock_lock(&pending_req_list_lock);
50 struct pending_req_list *curr = pending_req_list_head;
51
52 while (curr) {
53 if (curr->hash == hash && strcmp(curr->msg_id, msg_id) == 0) {
54 if (prev)
55 prev->next = curr->next;
56 else
57 pending_req_list_head = curr->next;
58
59 freez(curr);
60 break;
61 }
62
63 prev = curr;
64 curr = curr->next;
65 }
66 spinlock_unlock(&pending_req_list_lock);
67 }
68
69 void mark_pending_req_cancel_all()
70 {
71 spinlock_lock(&pending_req_list_lock);
72 struct pending_req_list *curr = pending_req_list_head;
73 while (curr) {
74 __atomic_store_n(&curr->canceled, 1, __ATOMIC_RELAXED);
75 curr = curr->next;
76 }
77 spinlock_unlock(&pending_req_list_lock);
78 }
79
80 int mark_pending_req_cancelled(const char *msg_id)
81 {
82 uint32_t hash = simple_hash(msg_id);
83
84 spinlock_lock(&pending_req_list_lock);
85 struct pending_req_list *curr = pending_req_list_head;
86
87 while (curr) {
88 if (curr->hash == hash && strcmp(curr->msg_id, msg_id) == 0) {
89 __atomic_store_n(&curr->canceled, 1, __ATOMIC_RELAXED);
90 spinlock_unlock(&pending_req_list_lock);
91 return 0;
92 }
93
94 curr = curr->next;
95 }
96 spinlock_unlock(&pending_req_list_lock);
97 return 1;
98 }
99
100 static bool aclk_web_client_interrupt_cb(struct web_client *w __maybe_unused, void *data)
101 {
102 struct pending_req_list *req = (struct pending_req_list *)data;
103 return __atomic_load_n(&req->canceled, __ATOMIC_RELAXED);
104 }
105
106 int http_api_v2(mqtt_wss_client client, aclk_query_t *query)
107 {
108 ND_LOG_STACK lgs[] = {
109 ND_LOG_FIELD_TXT(NDF_SRC_TRANSPORT, "aclk"),
110 ND_LOG_FIELD_END(),
111 };
112 ND_LOG_STACK_PUSH(lgs);
113
114 int retval = 0;
115 usec_t dt_ut = 0;
116
117 int z_ret;
118 BUFFER *z_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE, &netdata_buffers_statistics.buffers_aclk);
119
120 struct web_client *w = web_client_get_from_cache();
121 web_client_set_conn_cloud(w);
122 w->port_acl = HTTP_ACL_ACLK | default_aclk_http_acl;
123 w->acl = w->port_acl;
124 web_client_set_permissions(w, HTTP_ACCESS_MAP_OLD_MEMBER, HTTP_USER_ROLE_MEMBER, USER_AUTH_METHOD_CLOUD);
125
126 w->mode = HTTP_REQUEST_MODE_GET;
127 w->timings.tv_in = query->created_tv;
128
129 w->interrupt.callback = aclk_web_client_interrupt_cb;
130 w->interrupt.callback_data = pending_req_list_add(query->msg_id);
131
132 buffer_flush(w->response.data);
133 buffer_strcat(w->response.data, query->data.http_api_v2.payload);
134
135 HTTP_VALIDATION validation = http_request_validate(w);
136 if(validation != HTTP_VALIDATION_OK) {
137 nd_log(NDLS_ACCESS, NDLP_ERR, "ACLK received request is not valid, code %d", validation);
138 retval = 1;
139 w->response.code = HTTP_RESP_BAD_REQUEST;
140 w->response.code = (short)aclk_http_msg_v2(client, query->callback_topic, query->msg_id,
141 dt_ut, query->created, w->response.code,
142 NULL, 0);
143 goto cleanup;
144 }
145
146 web_client_timeout_checkpoint_set(w, query->timeout);
147 if(web_client_timeout_checkpoint_and_check(w, &dt_ut)) {
148 nd_log(NDLS_ACCESS, NDLP_ERR,
149 "QUERY CANCELED: QUEUE TIME EXCEEDED %llu ms (LIMIT %d ms)",
150 dt_ut / USEC_PER_MS, query->timeout);
151 retval = 1;
152 w->response.code = HTTP_RESP_SERVICE_UNAVAILABLE;
153 aclk_http_msg_v2_err(client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0);
154 goto cleanup;
155 }
156
157 char *path = (char *)buffer_tostring(w->url_path_decoded);
158
159 w->response.code = (short)web_client_api_request_with_node_selection(localhost, w, path);
160 web_client_timeout_checkpoint_response_ready(w, &dt_ut);
161
162 if (w->response.data->len && w->response.zinitialized) {
163 w->response.zstream.next_in = (Bytef *)w->response.data->buffer;
164 w->response.zstream.avail_in = w->response.data->len;
165 do {
166 w->response.zstream.avail_out = NETDATA_WEB_RESPONSE_ZLIB_CHUNK_SIZE;
167 w->response.zstream.next_out = w->response.zbuffer;
168 z_ret = deflate(&w->response.zstream, Z_FINISH);
169 if(z_ret < 0) {
170 if(w->response.zstream.msg)
171 netdata_log_error("Error compressing body. ZLIB error: \"%s\"", w->response.zstream.msg);
172 else
173 netdata_log_error("Unknown error during zlib compression.");
174 retval = 1;
175 w->response.code = 500;
176 aclk_http_msg_v2_err(client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_ZLIB_ERROR, CLOUD_EMSG_ZLIB_ERROR, NULL, 0);
177 goto cleanup;
178 }
179 int bytes_to_cpy = NETDATA_WEB_RESPONSE_ZLIB_CHUNK_SIZE - w->response.zstream.avail_out;
180 buffer_need_bytes(z_buffer, bytes_to_cpy);
181 memcpy(&z_buffer->buffer[z_buffer->len], w->response.zbuffer, bytes_to_cpy);
182 z_buffer->len += bytes_to_cpy;
183 } while(z_ret != Z_STREAM_END);
184
185 // so that web_client_build_http_header
186 // puts correct content length into header
187 buffer_free(w->response.data);
188 w->response.data = z_buffer;
189 z_buffer = NULL;
190 }
191
192 web_client_build_http_header(w);
193
194 w->response.code = (short)aclk_http_msg_v2_direct(
195 client,
196 query->callback_topic,
197 query->msg_id,
198 dt_ut,
199 query->created,
200 w->response.code,
201 w->response.header_output->buffer,
202 w->response.header_output->len,
203 w->response.data->buffer,
204 w->response.data->len);
205
206 cleanup:
207 web_client_log_completed_request(w, false);
208 web_client_release_to_cache(w);
209
210 pending_req_list_rm(query->msg_id);
211
212 buffer_free(z_buffer);
213 return retval;
214 }
215
216 int send_bin_msg(mqtt_wss_client client, aclk_query_t *query)
217 {
218 // this will be simplified when legacy support is removed
219 aclk_send_bin_message_subtopic_pid(
220 client,
221 query->data.bin_payload.payload,
222 query->data.bin_payload.size,
223 query->data.bin_payload.topic,
224 query->data.bin_payload.msg_name);
225 return 0;
226 }