@cryptotaxi247 / netdata-1 / commits / 54916258a

Implements ACLK v2 http message with compression (#9895)

Allows cloud to use v2 queries which support compression.

Timotej S committed Sep 10, 2020 at 13:48 UTC 54916258aa5b687766b8bcf7bb5a058fada3f855
14 files changed +538 -227
CMakeLists.txt
+2
@@ -703,6 +703,8 @@ set(ACLK_PLUGIN_FILES
703 aclk/mqtt.h
704 aclk/aclk_stats.c
705 aclk/aclk_stats.h
706 + aclk/aclk_rx_msgs.c
707 + aclk/aclk_rx_msgs.h
708 )
709
710 set(SPAWN_PLUGIN_FILES
Makefile.am
+2
@@ -535,6 +535,8 @@ ACLK_FILES += \
535 aclk/aclk_query.h \
536 aclk/mqtt.c \
537 aclk/mqtt.h \
538 + aclk/aclk_rx_msgs.c \
539 + aclk/aclk_rx_msgs.h \
540 aclk/aclk_lws_wss_client.c \
541 aclk/aclk_lws_wss_client.h \
542 aclk/aclk_lws_https_client.c \
aclk/aclk_common.c
+1
@@ -5,6 +5,7 @@
5 netdata_mutex_t aclk_shared_state_mutex = NETDATA_MUTEX_INITIALIZER;
6
7 int aclk_disable_runtime = 0;
8 +int aclk_kill_link = 0;
9
10 struct aclk_shared_state aclk_shared_state = {
11 .metadata_submitted = ACLK_METADATA_REQUIRED,
aclk/aclk_common.h
+11 -5
@@ -9,8 +9,8 @@ extern netdata_mutex_t aclk_shared_state_mutex;
9
10 // minimum and maximum supported version of ACLK
11 // in this version of agent
12 -#define ACLK_VERSION_MIN 1
13 -#define ACLK_VERSION_MAX 1
12 +#define ACLK_VERSION_MIN 2
13 +#define ACLK_VERSION_MAX 2
14
15 // Version negotiation messages have they own versioning
16 // this is also used for LWT message as we set that up
@@ -25,6 +25,9 @@ extern netdata_mutex_t aclk_shared_state_mutex;
25 #error "ACLK_VERSION_MAX must be >= than ACLK_VERSION_MIN"
26 #endif
27
28 +// Define ACLK Feature Version Boundaries Here
29 +#define ACLK_V_COMPRESSION 2
30 +
31 typedef enum aclk_cmd {
32 ACLK_CMD_CLOUD,
33 ACLK_CMD_ONCONNECT,
@@ -32,7 +35,7 @@ typedef enum aclk_cmd {
35 ACLK_CMD_CHART,
36 ACLK_CMD_CHARTDEL,
37 ACLK_CMD_ALARM,
35 - ACLK_CMD_MAX
38 + ACLK_CMD_CLOUD_QUERY_2
39 } ACLK_CMD;
40
41 typedef enum aclk_metadata_state {
@@ -64,18 +67,21 @@ typedef enum aclk_proxy_type {
67 PROXY_NOT_SET,
68 } ACLK_PROXY_TYPE;
69
70 +extern int aclk_kill_link; // Tells the agent to tear down the link
71 +extern int aclk_disable_runtime;
72 +
73 const char *aclk_proxy_type_to_s(ACLK_PROXY_TYPE *type);
74
75 #define ACLK_PROXY_PROTO_ADDR_SEPARATOR "://"
76 #define ACLK_PROXY_ENV "env"
77 #define ACLK_PROXY_CONFIG_VAR "proxy"
78
79 +#define ACLK_CLOUD_REQ_V2_PREFIX "GET /api/v1/"
80 +
81 ACLK_PROXY_TYPE aclk_verify_proxy(const char *string);
82 const char *aclk_lws_wss_get_proxy_setting(ACLK_PROXY_TYPE *type);
83 void safe_log_proxy_censor(char *proxy);
84 int aclk_decode_base_url(char *url, char **aclk_hostname, char **aclk_port);
85 const char *aclk_get_proxy(ACLK_PROXY_TYPE *type);
86
79 -extern int aclk_disable_runtime;
80 -
87 #endif //ACLK_COMMON_H
aclk/aclk_query.c
+168 -25
@@ -1,6 +1,9 @@
1 #include "aclk_common.h"
2 #include "aclk_query.h"
3 #include "aclk_stats.h"
4 +#include "aclk_rx_msgs.h"
5 +
6 +#define WEB_HDR_ACCEPT_ENC "Accept-Encoding:"
7
8 pthread_cond_t query_cond_wait = PTHREAD_COND_INITIALIZER;
9 pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
@@ -18,7 +21,7 @@ static netdata_mutex_t queue_mutex = NETDATA_MUTEX_INITIALIZER;
21 #define ACLK_QUEUE_UNLOCK netdata_mutex_unlock(&queue_mutex)
22
23 struct aclk_query {
21 - time_t created;
24 + usec_t created;
25 time_t run_after; // Delay run until after this time
26 ACLK_CMD cmd; // What command is this
27 char *topic; // Topic to respond to
@@ -56,7 +59,7 @@ static void aclk_query_free(struct aclk_query *this_query)
59 freez(this_query->topic);
60 if (likely(this_query->query))
61 freez(this_query->query);
59 - if (likely(this_query->data))
62 + if(this_query->data && this_query->cmd == ACLK_CMD_CLOUD_QUERY_2)
63 freez(this_query->data);
64 if (likely(this_query->msg_id))
65 freez(this_query->msg_id);
@@ -150,7 +153,7 @@ static struct aclk_query *aclk_query_find_position(time_t time_to_run)
153
154 // Need to have a QUERY lock before calling this
155 static struct aclk_query *
153 -aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query)
156 +aclk_query_find(char *topic, void *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query)
157 {
158 struct aclk_query *tmp_query, *prev_query;
159 UNUSED(cmd);
@@ -160,7 +163,7 @@ aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd
163 while (tmp_query) {
164 if (likely(!tmp_query->deleted)) {
165 if (strcmp(tmp_query->topic, topic) == 0 && (!query || strcmp(tmp_query->query, query) == 0)) {
163 - if ((!data || (data && strcmp(data, tmp_query->data) == 0)) &&
166 + if ((!data || data == tmp_query->data) &&
167 (!msg_id || (msg_id && strcmp(msg_id, tmp_query->msg_id) == 0))) {
168 if (likely(last_query))
169 *last_query = prev_query;
@@ -178,7 +181,7 @@ aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd
181 * Add a query to execute, the result will be send to the specified topic
182 */
183
181 -int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run_after, int internal, ACLK_CMD aclk_cmd)
184 +int aclk_queue_query(char *topic, void *data, char *msg_id, char *query, int run_after, int internal, ACLK_CMD aclk_cmd)
185 {
186 struct aclk_query *new_query, *tmp_query;
187
@@ -227,11 +230,9 @@ int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run
230 new_query->msg_id = msg_id;
231 }
232
230 - if (data)
231 - new_query->data = strdupz(data);
232 -
233 + new_query->data = data;
234 new_query->next = NULL;
234 - new_query->created = now_realtime_sec();
235 + new_query->created = now_realtime_usec();
236 new_query->run_after = run_after;
237
238 debug(D_ACLK, "Added query (%s) (%s)", topic, query ? query : "");
@@ -318,6 +319,26 @@ static char *aclk_encode_response(char *src, size_t content_size, int keep_newli
319 #pragma region ACLK_QUERY
320 #endif
321
322 +static usec_t aclk_web_api_request_v1(RRDHOST *host, struct web_client *w, char *url)
323 +{
324 + usec_t t;
325 +
326 + t = now_monotonic_high_precision_usec();
327 + w->response.code = web_client_api_request_v1(host, w, url);
328 + t = now_monotonic_high_precision_usec() - t;
329 +
330 + if (aclk_stats_enabled) {
331 + ACLK_STATS_LOCK;
332 + aclk_metrics_per_sample.cloud_q_process_total += t;
333 + aclk_metrics_per_sample.cloud_q_process_count++;
334 + if (aclk_metrics_per_sample.cloud_q_process_max < t)
335 + aclk_metrics_per_sample.cloud_q_process_max = t;
336 + ACLK_STATS_UNLOCK;
337 + }
338 +
339 + return t;
340 +}
341 +
342 static int aclk_execute_query(struct aclk_query *this_query)
343 {
344 if (strncmp(this_query->query, "/api/v1/", 8) == 0) {
@@ -340,7 +361,7 @@ static int aclk_execute_query(struct aclk_query *this_query)
361 mysep = strrchr(this_query->query, '/');
362
363 // TODO: handle bad response perhaps in a different way. For now it does to the payload
343 - w->response.code = web_client_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
364 + aclk_web_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
365 now_realtime_timeval(&w->tv_ready);
366 w->response.data->date = w->tv_ready.tv_sec;
367 web_client_build_http_header(w); // TODO: this function should offset from date, not tv_ready
@@ -375,16 +396,137 @@ static int aclk_execute_query(struct aclk_query *this_query)
396 return 1;
397 }
398
399 +static int aclk_execute_query_v2(struct aclk_query *this_query)
400 +{
401 + int retval = 0;
402 + usec_t t;
403 + BUFFER *local_buffer = NULL;
404 +
405 +#ifdef NETDATA_WITH_ZLIB
406 + int z_ret;
407 + BUFFER *z_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
408 + char *start, *end;
409 +#endif
410 +
411 + struct web_client *w = (struct web_client *)callocz(1, sizeof(struct web_client));
412 + w->response.data = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
413 + w->response.header = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
414 + w->response.header_output = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
415 + strcpy(w->origin, "*"); // Simulate web_client_create_on_fd()
416 + w->cookie1[0] = 0; // Simulate web_client_create_on_fd()
417 + w->cookie2[0] = 0; // Simulate web_client_create_on_fd()
418 + w->acl = 0x1f;
419 +
420 + char *mysep = strchr(this_query->query, '?');
421 + if (mysep) {
422 + url_decode_r(w->decoded_query_string, mysep, NETDATA_WEB_REQUEST_URL_SIZE + 1);
423 + *mysep = '\0';
424 + } else
425 + url_decode_r(w->decoded_query_string, this_query->query, NETDATA_WEB_REQUEST_URL_SIZE + 1);
426 +
427 + mysep = strrchr(this_query->query, '/');
428 +
429 + // execute the query
430 + t = aclk_web_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
431 +
432 +#ifdef NETDATA_WITH_ZLIB
433 + // check if gzip encoding can and should be used
434 + if ((start = strstr((char *)this_query->data, WEB_HDR_ACCEPT_ENC))) {
435 + start += strlen(WEB_HDR_ACCEPT_ENC);
436 + end = strstr(start, "\x0D\x0A");
437 + start = strstr(start, "gzip");
438 +
439 + if (start && start < end) {
440 + w->response.zstream.zalloc = Z_NULL;
441 + w->response.zstream.zfree = Z_NULL;
442 + w->response.zstream.opaque = Z_NULL;
443 + if(deflateInit2(&w->response.zstream, web_gzip_level, Z_DEFLATED, 15 + 16, 8, web_gzip_strategy) == Z_OK) {
444 + w->response.zinitialized = 1;
445 + w->response.zoutput = 1;
446 + } else
447 + error("Failed to initialize zlib. Proceeding without compression.");
448 + }
449 + }
450 +
451 + if (w->response.data->len && w->response.zinitialized) {
452 + w->response.zstream.next_in = (Bytef *)w->response.data->buffer;
453 + w->response.zstream.avail_in = w->response.data->len;
454 + do {
455 + w->response.zstream.avail_out = NETDATA_WEB_RESPONSE_ZLIB_CHUNK_SIZE;
456 + w->response.zstream.next_out = w->response.zbuffer;
457 + z_ret = deflate(&w->response.zstream, Z_FINISH);
458 + if(z_ret < 0) {
459 + if(w->response.zstream.msg)
460 + error("Error compressing body. ZLIB error: \"%s\"", w->response.zstream.msg);
461 + else
462 + error("Unknown error during zlib compression.");
463 + retval = 1;
464 + goto cleanup;
465 + }
466 + int bytes_to_cpy = NETDATA_WEB_RESPONSE_ZLIB_CHUNK_SIZE - w->response.zstream.avail_out;
467 + buffer_need_bytes(z_buffer, bytes_to_cpy);
468 + memcpy(&z_buffer->buffer[z_buffer->len], w->response.zbuffer, bytes_to_cpy);
469 + z_buffer->len += bytes_to_cpy;
470 + } while(z_ret != Z_STREAM_END);
471 + // so that web_client_build_http_header
472 + // puts correct content lenght into header
473 + buffer_free(w->response.data);
474 + w->response.data = z_buffer;
475 + z_buffer = NULL;
476 + }
477 +#endif
478 +
479 + now_realtime_timeval(&w->tv_ready);
480 + w->response.data->date = w->tv_ready.tv_sec;
481 + web_client_build_http_header(w);
482 + local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
483 + local_buffer->contenttype = CT_APPLICATION_JSON;
484 +
485 + aclk_create_header(local_buffer, "http", this_query->msg_id, 0, 0, aclk_shared_state.version_neg);
486 + buffer_sprintf(local_buffer, ",\"t-exec\": %llu,\"t-rx\": %llu,\"http-code\": %d", t, this_query->created, w->response.code);
487 + buffer_strcat(local_buffer, "}\x0D\x0A\x0D\x0A");
488 + buffer_strcat(local_buffer, w->response.header_output->buffer);
489 +
490 + if (w->response.data->len) {
491 +#ifdef NETDATA_WITH_ZLIB
492 + if (w->response.zinitialized) {
493 + buffer_need_bytes(local_buffer, w->response.data->len);
494 + memcpy(&local_buffer->buffer[local_buffer->len], w->response.data->buffer, w->response.data->len);
495 + local_buffer->len += w->response.data->len;
496 + } else {
497 +#endif
498 + buffer_strcat(local_buffer, w->response.data->buffer);
499 +#ifdef NETDATA_WITH_ZLIB
500 + }
501 +#endif
502 + }
503 +
504 + aclk_send_message_bin(this_query->topic, local_buffer->buffer, local_buffer->len, this_query->msg_id);
505 +
506 +cleanup:
507 +#ifdef NETDATA_WITH_ZLIB
508 + if(w->response.zinitialized)
509 + deflateEnd(&w->response.zstream);
510 + buffer_free(z_buffer);
511 +#endif
512 + buffer_free(w->response.data);
513 + buffer_free(w->response.header);
514 + buffer_free(w->response.header_output);
515 + freez(w);
516 + buffer_free(local_buffer);
517 + return retval;
518 +}
519 +
520 /*
521 * This function will fetch the next pending command and process it
522 *
523 */
382 -static int aclk_process_query(int t_idx)
524 +static int aclk_process_query(struct aclk_query_thread *t_info)
525 {
526 struct aclk_query *this_query;
527 static long int query_count = 0;
528 ACLK_METADATA_STATE meta_state;
387 - usec_t t = 0;
529 + RRDHOST *host;
530
531 if (!aclk_connected)
532 return 0;
@@ -401,9 +543,11 @@ static int aclk_process_query(int t_idx)
543 }
544 query_count++;
545
546 + host = (RRDHOST*)this_query->data;
547 +
548 debug(
405 - D_ACLK, "Query #%ld (%s) size=%zu in queue %d seconds", query_count, this_query->topic,
406 - this_query->query ? strlen(this_query->query) : 0, (int)(now_realtime_sec() - this_query->created));
549 + D_ACLK, "Query #%ld (%s) size=%zu in queue %llu ms", query_count, this_query->topic,
550 + this_query->query ? strlen(this_query->query) : 0, (now_realtime_usec() - this_query->created)/USEC_PER_MS);
551
552 switch (this_query->cmd) {
553 case ACLK_CMD_ONCONNECT:
@@ -417,7 +561,9 @@ static int aclk_process_query(int t_idx)
561
562 case ACLK_CMD_CHART:
563 debug(D_ACLK, "EXECUTING a chart update command");
420 - aclk_send_single_chart(this_query->data, this_query->query);
564 + if (!host)
565 + fatal("Pointer to host compulsory");
566 + aclk_send_single_chart(host->hostname, this_query->query);
567 break;
568
569 case ACLK_CMD_CHARTDEL:
@@ -432,10 +578,12 @@ static int aclk_process_query(int t_idx)
578 break;
579
580 case ACLK_CMD_CLOUD:
435 - t = now_monotonic_high_precision_usec();
581 debug(D_ACLK, "EXECUTING a cloud command");
582 aclk_execute_query(this_query);
438 - t = now_monotonic_high_precision_usec() - t;
583 + break;
584 + case ACLK_CMD_CLOUD_QUERY_2:
585 + debug(D_ACLK, "EXECUTING Cloud Query v2");
586 + aclk_execute_query_v2(this_query);
587 break;
588
589 default:
@@ -446,13 +594,7 @@ static int aclk_process_query(int t_idx)
594 if (aclk_stats_enabled) {
595 ACLK_STATS_LOCK;
596 aclk_metrics_per_sample.queries_dispatched++;
449 - aclk_queries_per_thread[t_idx]++;
450 - if(this_query->cmd == ACLK_CMD_CLOUD) {
451 - aclk_metrics_per_sample.cloud_q_process_total += t;
452 - aclk_metrics_per_sample.cloud_q_process_count++;
453 - if(aclk_metrics_per_sample.cloud_q_process_max < t)
454 - aclk_metrics_per_sample.cloud_q_process_max = t;
455 - }
597 + aclk_queries_per_thread[t_info->idx]++;
598 ACLK_STATS_UNLOCK;
599 }
600
@@ -553,6 +695,7 @@ void *aclk_query_main_thread(void *ptr)
695 error("ACLK version negotiation failed. No reply to \"hello\" with \"version\" from cloud in time of %ds."
696 " Reverting to default ACLK version of %d.", VERSION_NEG_TIMEOUT, ACLK_VERSION_MIN);
697 aclk_shared_state.version_neg = ACLK_VERSION_MIN;
698 + aclk_set_rx_handlers(aclk_shared_state.version_neg);
699 }
700 if (unlikely(aclk_shared_state.metadata_submitted == ACLK_METADATA_REQUIRED)) {
701 if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
@@ -566,7 +709,7 @@ void *aclk_query_main_thread(void *ptr)
709 }
710 ACLK_SHARED_STATE_UNLOCK;
711
569 - while (aclk_process_query(info->idx)) {
712 + while (aclk_process_query(info)) {
713 // Process all commands
714 };
715
aclk/aclk_query.h
+2 -1
@@ -4,6 +4,7 @@
4 #define NETDATA_ACLK_QUERY_H
5
6 #include "libnetdata/libnetdata.h"
7 +#include "web/server/web_client.h"
8
9 #define ACLK_STABLE_TIMEOUT 3 // Minimum delay to mark AGENT as stable
10
@@ -25,7 +26,7 @@ struct aclk_query_threads {
26 };
27
28 void *aclk_query_main_thread(void *ptr);
28 -int aclk_queue_query(char *token, char *data, char *msg_type, char *query, int run_after, int internal, ACLK_CMD cmd);
29 +int aclk_queue_query(char *token, void *data, char *msg_type, char *query, int run_after, int internal, ACLK_CMD cmd);
30
31 void aclk_query_threads_start(struct aclk_query_threads *query_threads);
32 void aclk_query_threads_cleanup(struct aclk_query_threads *query_threads);
aclk/aclk_rx_msgs.c new
+314
@@ -0,0 +1,314 @@
1 +
2 +#include "aclk_rx_msgs.h"
3 +
4 +#include "aclk_common.h"
5 +#include "aclk_stats.h"
6 +#include "aclk_query.h"
7 +
8 +static inline int aclk_extract_v2_data(char *payload, char **data)
9 +{
10 + char* ptr = strstr(payload, ACLK_V2_PAYLOAD_SEPARATOR);
11 + if(!ptr)
12 + return 1;
13 + ptr += strlen(ACLK_V2_PAYLOAD_SEPARATOR);
14 + *data = strdupz(ptr);
15 + return 0;
16 +}
17 +
18 +static inline int aclk_v2_payload_get_query(const char *payload, struct aclk_request *req)
19 +{
20 + const char *start, *end;
21 +
22 + if(strncmp(payload, ACLK_CLOUD_REQ_V2_PREFIX, strlen(ACLK_CLOUD_REQ_V2_PREFIX))) {
23 + errno = 0;
24 + error("Only accepting requests that start with \"%s\" from CLOUD.", ACLK_CLOUD_REQ_V2_PREFIX);
25 + return 1;
26 + }
27 + start = payload + 4;
28 +
29 + if(!(end = strstr(payload, " HTTP/1.1\x0D\x0A"))) {
30 + errno = 0;
31 + error("Doesn't look like HTTP GET request.");
32 + return 1;
33 + }
34 +
35 + req->payload = mallocz((end - start) + 1);
36 + strncpyz(req->payload, start, end - start);
37 +
38 + return 0;
39 +}
40 +
41 +#define HTTP_CHECK_AGENT_INITIALIZED() ACLK_SHARED_STATE_LOCK;\
42 + if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {\
43 + debug(D_ACLK, "Ignoring \"http\" cloud request; agent not in stable state");\
44 + ACLK_SHARED_STATE_UNLOCK;\
45 + return 1;\
46 + }\
47 + ACLK_SHARED_STATE_UNLOCK;
48 +
49 +/*
50 + * Parse the incoming payload and queue a command if valid
51 + */
52 +static int aclk_handle_cloud_request_v1(struct aclk_request *cloud_to_agent, char *raw_payload)
53 +{
54 + UNUSED(raw_payload);
55 + HTTP_CHECK_AGENT_INITIALIZED();
56 +
57 + errno = 0;
58 + if (unlikely(cloud_to_agent->version != 1)) {
59 + error(
60 + "Received \"http\" message from Cloud with version %d, but ACLK version %d is used",
61 + cloud_to_agent->version,
62 + aclk_shared_state.version_neg);
63 + return 1;
64 + }
65 +
66 + if (unlikely(!cloud_to_agent->payload)) {
67 + error("payload missing");
68 + return 1;
69 + }
70 +
71 + if (unlikely(!cloud_to_agent->callback_topic)) {
72 + error("callback_topic missing");
73 + return 1;
74 + }
75 +
76 + if (unlikely(!cloud_to_agent->msg_id)) {
77 + error("msg_id missing");
78 + return 1;
79 + }
80 +
81 + if (unlikely(aclk_queue_query(cloud_to_agent->callback_topic, NULL, cloud_to_agent->msg_id, cloud_to_agent->payload, 0, 0, ACLK_CMD_CLOUD)))
82 + debug(D_ACLK, "ACLK failed to queue incoming \"http\" message");
83 +
84 + return 0;
85 +}
86 +
87 +static int aclk_handle_cloud_request_v2(struct aclk_request *cloud_to_agent, char *raw_payload)
88 +{
89 + HTTP_CHECK_AGENT_INITIALIZED();
90 +
91 + char *data;
92 +
93 + errno = 0;
94 + if (cloud_to_agent->version < ACLK_V_COMPRESSION) {
95 + error(
96 + "This handler cannot reply to request with version older than %d, received %d.",
97 + ACLK_V_COMPRESSION,
98 + cloud_to_agent->version);
99 + return 1;
100 + }
101 +
102 + if (unlikely(aclk_extract_v2_data(raw_payload, &data))) {
103 + error("Error extracting payload expected after the JSON dictionary.");
104 + return 1;
105 + }
106 +
107 + if (unlikely(aclk_v2_payload_get_query(data, cloud_to_agent)))
108 + return 1;
109 +
110 + if (unlikely(!cloud_to_agent->callback_topic)) {
111 + error("Missing callback_topic");
112 + freez(data);
113 + return 1;
114 + }
115 +
116 + if (unlikely(!cloud_to_agent->msg_id)) {
117 + error("Missing msg_id");
118 + freez(data);
119 + return 1;
120 + }
121 +
122 + // aclk_queue_query takes ownership of data pointer
123 + if (unlikely(aclk_queue_query(
124 + cloud_to_agent->callback_topic, data, cloud_to_agent->msg_id, cloud_to_agent->payload, 0, 0,
125 + ACLK_CMD_CLOUD_QUERY_2)))
126 + debug(D_ACLK, "ACLK failed to queue incoming \"http\" message");
127 +
128 + UNUSED(cloud_to_agent);
129 + return 0;
130 +}
131 +
132 +// This handles `version` message from cloud used to negotiate
133 +// protocol version we will use
134 +static int aclk_handle_version_response(struct aclk_request *cloud_to_agent, char *raw_payload)
135 +{
136 + UNUSED(raw_payload);
137 + int version = -1;
138 + errno = 0;
139 +
140 + if (unlikely(cloud_to_agent->version != ACLK_VERSION_NEG_VERSION)) {
141 + error(
142 + "Unsuported version of \"version\" message from cloud. Expected %d, Got %d",
143 + ACLK_VERSION_NEG_VERSION,
144 + cloud_to_agent->version);
145 + return 1;
146 + }
147 + if (unlikely(!cloud_to_agent->min_version)) {
148 + error("Min version missing or 0");
149 + return 1;
150 + }
151 + if (unlikely(!cloud_to_agent->max_version)) {
152 + error("Max version missing or 0");
153 + return 1;
154 + }
155 + if (unlikely(cloud_to_agent->max_version < cloud_to_agent->min_version)) {
156 + error(
157 + "Max version (%d) must be >= than min version (%d)", cloud_to_agent->max_version,
158 + cloud_to_agent->min_version);
159 + return 1;
160 + }
161 +
162 + if (unlikely(cloud_to_agent->min_version > ACLK_VERSION_MAX)) {
163 + error(
164 + "Agent too old for this cloud. Minimum version required by cloud %d."
165 + " Maximum version supported by this agent %d.",
166 + cloud_to_agent->min_version, ACLK_VERSION_MAX);
167 + aclk_kill_link = 1;
168 + aclk_disable_runtime = 1;
169 + return 1;
170 + }
171 + if (unlikely(cloud_to_agent->max_version < ACLK_VERSION_MIN)) {
172 + error(
173 + "Cloud version is too old for this agent. Maximum version supported by cloud %d."
174 + " Minimum (oldest) version supported by this agent %d.",
175 + cloud_to_agent->max_version, ACLK_VERSION_MIN);
176 + aclk_kill_link = 1;
177 + return 1;
178 + }
179 +
180 + version = MIN(cloud_to_agent->max_version, ACLK_VERSION_MAX);
181 +
182 + ACLK_SHARED_STATE_LOCK;
183 + if (unlikely(now_monotonic_usec() > aclk_shared_state.version_neg_wait_till)) {
184 + errno = 0;
185 + error("The \"version\" message came too late ignoring.");
186 + goto err_cleanup;
187 + }
188 + if (unlikely(aclk_shared_state.version_neg)) {
189 + errno = 0;
190 + error("Version has already been set to %d", aclk_shared_state.version_neg);
191 + goto err_cleanup;
192 + }
193 + aclk_shared_state.version_neg = version;
194 + ACLK_SHARED_STATE_UNLOCK;
195 +
196 + info("Choosing version %d of ACLK", version);
197 +
198 + aclk_set_rx_handlers(version);
199 +
200 + return 0;
201 +
202 +err_cleanup:
203 + ACLK_SHARED_STATE_UNLOCK;
204 + return 1;
205 +}
206 +
207 +typedef struct aclk_incoming_msg_type{
208 + char *name;
209 + int(*fnc)(struct aclk_request *, char *);
210 +}aclk_incoming_msg_type;
211 +
212 +aclk_incoming_msg_type aclk_incoming_msg_types_v1[] = {
213 + { .name = "http", .fnc = aclk_handle_cloud_request_v1 },
214 + { .name = "version", .fnc = aclk_handle_version_response },
215 + { .name = NULL, .fnc = NULL }
216 +};
217 +
218 +aclk_incoming_msg_type aclk_incoming_msg_types_compression[] = {
219 + { .name = "http", .fnc = aclk_handle_cloud_request_v2 },
220 + { .name = "version", .fnc = aclk_handle_version_response },
221 + { .name = NULL, .fnc = NULL }
222 +};
223 +
224 +struct aclk_incoming_msg_type *aclk_incoming_msg_types = aclk_incoming_msg_types_v1;
225 +
226 +void aclk_set_rx_handlers(int version)
227 +{
228 + if(version >= ACLK_V_COMPRESSION) {
229 + aclk_incoming_msg_types = aclk_incoming_msg_types_compression;
230 + return;
231 + }
232 +
233 + aclk_incoming_msg_types = aclk_incoming_msg_types_v1;
234 +}
235 +
236 +int aclk_handle_cloud_message(char *payload)
237 +{
238 + struct aclk_request cloud_to_agent;
239 + memset(&cloud_to_agent, 0, sizeof(struct aclk_request));
240 +
241 + if (aclk_stats_enabled) {
242 + ACLK_STATS_LOCK;
243 + aclk_metrics_per_sample.cloud_req_recvd++;
244 + ACLK_STATS_UNLOCK;
245 + }
246 +
247 + if (unlikely(!payload)) {
248 + errno = 0;
249 + error("ACLK incoming message is empty");
250 + goto err_cleanup_nojson;
251 + }
252 +
253 + debug(D_ACLK, "ACLK incoming message (%s)", payload);
254 +
255 + int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
256 +
257 + if (unlikely(rc != JSON_OK)) {
258 + errno = 0;
259 + error("Malformed json request (%s)", payload);
260 + goto err_cleanup;
261 + }
262 +
263 + if (!cloud_to_agent.type_id) {
264 + errno = 0;
265 + error("Cloud message is missing compulsory key \"type\"");
266 + goto err_cleanup;
267 + }
268 +
269 + if (!aclk_shared_state.version_neg && strcmp(cloud_to_agent.type_id, "version")) {
270 + error("Only \"version\" message is allowed before popcorning and version negotiation is finished. Ignoring");
271 + goto err_cleanup;
272 + }
273 +
274 + for (int i = 0; aclk_incoming_msg_types[i].name; i++) {
275 + if (strcmp(cloud_to_agent.type_id, aclk_incoming_msg_types[i].name) == 0) {
276 + if (likely(!aclk_incoming_msg_types[i].fnc(&cloud_to_agent, payload))) {
277 + // in case of success handler is supposed to clean up after itself
278 + // or as in the case of aclk_handle_cloud_request take
279 + // ownership of the pointers (done to avoid copying)
280 + // see what `aclk_queue_query` parameter `internal` does
281 +
282 + // NEVER CONTINUE THIS LOOP AFTER CALLING FUNCTION!!!
283 + // msg handlers (namely aclk_handle_version_responce)
284 + // can freely change what aclk_incoming_msg_types points to
285 + // so either exit or restart this for loop
286 + freez(cloud_to_agent.type_id);
287 + return 0;
288 + }
289 + goto err_cleanup;
290 + }
291 + }
292 +
293 + errno = 0;
294 + error("Unknown message type from Cloud \"%s\"", cloud_to_agent.type_id);
295 +
296 +err_cleanup:
297 + if (cloud_to_agent.payload)
298 + freez(cloud_to_agent.payload);
299 + if (cloud_to_agent.type_id)
300 + freez(cloud_to_agent.type_id);
301 + if (cloud_to_agent.msg_id)
302 + freez(cloud_to_agent.msg_id);
303 + if (cloud_to_agent.callback_topic)
304 + freez(cloud_to_agent.callback_topic);
305 +
306 +err_cleanup_nojson:
307 + if (aclk_stats_enabled) {
308 + ACLK_STATS_LOCK;
309 + aclk_metrics_per_sample.cloud_req_err++;
310 + ACLK_STATS_UNLOCK;
311 + }
312 +
313 + return 1;
314 +}
aclk/aclk_rx_msgs.h new
+13
@@ -0,0 +1,13 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef NETDATA_ACLK_RX_MSGS_H
4 +#define NETDATA_ACLK_RX_MSGS_H
5 +
6 +#include "../daemon/common.h"
7 +#include "libnetdata/libnetdata.h"
8 +
9 +int aclk_handle_cloud_message(char *payload);
10 +void aclk_set_rx_handlers(int version);
11 +
12 +
13 +#endif /* NETDATA_ACLK_RX_MSGS_H */
aclk/agent_cloud_link.c
+8 -185
@@ -18,7 +18,6 @@ static char *aclk_password = NULL;
18 static char *global_base_topic = NULL;
19 static int aclk_connecting = 0;
20 int aclk_force_reconnect = 0; // Indication from lower layers
21 -int aclk_kill_link = 0; // Tell the agent to tear down the link
21 usec_t aclk_session_us = 0; // Used by the mqtt layer
22 time_t aclk_session_sec = 0; // Used by the mqtt layer
23
@@ -1074,7 +1073,7 @@ exited:
1073 * If base_topic is missing then the global_base_topic will be used (if available)
1074 *
1075 */
1077 -int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1076 +int aclk_send_message_bin(char *sub_topic, const void *message, size_t len, char *msg_id)
1077 {
1078 int rc;
1079 int mid;
@@ -1098,7 +1097,7 @@ int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1097 }
1098
1099 ACLK_LOCK;
1101 - rc = _link_send_message(final_topic, (unsigned char *)message, &mid);
1100 + rc = _link_send_message(final_topic, message, len, &mid);
1101 // TODO: link the msg_id with the mid so we can trace it
1102 ACLK_UNLOCK;
1103
@@ -1110,6 +1109,11 @@ int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1109 return rc;
1110 }
1111
1112 +int aclk_send_message(char *sub_topic, char *message, char *msg_id)
1113 +{
1114 + return aclk_send_message_bin(sub_topic, message, strlen(message), msg_id);
1115 +}
1116 +
1117 /*
1118 * Subscribe to a topic in the cloud
1119 * The final subscription will be in the form
@@ -1415,7 +1419,7 @@ int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1419 if (aclk_popcorn_check_bump())
1420 return 0;
1421
1418 - if (unlikely(aclk_queue_query("_chart", host->hostname, NULL, chart_name, 0, 1, aclk_cmd))) {
1422 + if (unlikely(aclk_queue_query("_chart", host, NULL, chart_name, 0, 1, aclk_cmd))) {
1423 if (likely(aclk_connected)) {
1424 errno = 0;
1425 error("ACLK failed to queue chart_update command");
@@ -1478,184 +1482,3 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1482
1483 return 0;
1484 }
1481 -
1482 -/*
1483 - * Parse the incoming payload and queue a command if valid
1484 - */
1485 -static int aclk_handle_cloud_request(struct aclk_request *cloud_to_agent)
1486 -{
1487 - errno = 0;
1488 - ACLK_SHARED_STATE_LOCK;
1489 - if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1490 - debug(D_ACLK, "Ignoring \"http\" cloud request; agent not in stable state");
1491 - ACLK_SHARED_STATE_UNLOCK;
1492 - return 1;
1493 - }
1494 - ACLK_SHARED_STATE_UNLOCK;
1495 -
1496 - if (unlikely(cloud_to_agent->version != aclk_shared_state.version_neg)) {
1497 - error("Received \"http\" message from Cloud with version %d, but ACLK version %d is used", cloud_to_agent->version, aclk_shared_state.version_neg);
1498 - return 1;
1499 - }
1500 -
1501 - if (unlikely(!cloud_to_agent->payload)) {
1502 - error("payload missing");
1503 - return 1;
1504 - }
1505 -
1506 - if (unlikely(!cloud_to_agent->callback_topic)) {
1507 - error("callback_topic missing");
1508 - return 1;
1509 - }
1510 -
1511 - if (unlikely(!cloud_to_agent->msg_id)) {
1512 - error("msg_id missing");
1513 - return 1;
1514 - }
1515 -
1516 - if (unlikely(aclk_queue_query(cloud_to_agent->callback_topic, NULL, cloud_to_agent->msg_id, cloud_to_agent->payload, 0, 0, ACLK_CMD_CLOUD)))
1517 - debug(D_ACLK, "ACLK failed to queue incoming \"http\" message");
1518 -
1519 - // Note: the payload comes from the callback and it will be automatically freed
1520 - return 0;
1521 -}
1522 -
1523 -// This handles `version` message from cloud used to negotiate
1524 -// protocol version we will use
1525 -static int aclk_handle_version_response(struct aclk_request *cloud_to_agent)
1526 -{
1527 - int version = -1;
1528 - errno = 0;
1529 -
1530 - if(unlikely(cloud_to_agent->version != ACLK_VERSION_NEG_VERSION)) {
1531 - error("Unsuported version of \"version\" message from cloud. Expected %d, Got %d", ACLK_VERSION_NEG_VERSION, cloud_to_agent->version);
1532 - return 1;
1533 - }
1534 - if(unlikely(!cloud_to_agent->min_version)) {
1535 - error("Min version missing or 0");
1536 - return 1;
1537 - }
1538 - if(unlikely(!cloud_to_agent->max_version)) {
1539 - error("Max version missing or 0");
1540 - return 1;
1541 - }
1542 - if(unlikely(cloud_to_agent->max_version < cloud_to_agent->min_version)) {
1543 - error("Max version (%d) must be >= than min version (%d)", cloud_to_agent->max_version, cloud_to_agent->min_version);
1544 - return 1;
1545 - }
1546 -
1547 - if(unlikely(cloud_to_agent->min_version > ACLK_VERSION_MAX)) {
1548 - error("Agent too old for this cloud. Minimum version required by cloud %d. Maximum version supported by this agent %d.", cloud_to_agent->min_version, ACLK_VERSION_MAX);
1549 - aclk_kill_link = 1;
1550 - aclk_disable_runtime = 1;
1551 - return 1;
1552 - }
1553 - if(unlikely(cloud_to_agent->max_version < ACLK_VERSION_MIN)) {
1554 - error("Cloud version is too old for this agent. Maximum version supported by cloud %d. Minimum (oldest) version supported by this agent %d.", cloud_to_agent->max_version, ACLK_VERSION_MIN);
1555 - aclk_kill_link = 1;
1556 - return 1;
1557 - }
1558 -
1559 - version = MIN(cloud_to_agent->max_version, ACLK_VERSION_MAX);
1560 -
1561 - ACLK_SHARED_STATE_LOCK;
1562 - if (unlikely(now_monotonic_usec() > aclk_shared_state.version_neg_wait_till)) {
1563 - errno = 0;
1564 - error("The \"version\" message came too late ignoring.");
1565 - goto err_cleanup;
1566 - }
1567 - if (unlikely(aclk_shared_state.version_neg)) {
1568 - errno = 0;
1569 - error("Version has already been set to %d", aclk_shared_state.version_neg);
1570 - goto err_cleanup;
1571 - }
1572 - aclk_shared_state.version_neg = version;
1573 - ACLK_SHARED_STATE_UNLOCK;
1574 -
1575 - info("Choosing version %d of ACLK", version);
1576 -
1577 - return 0;
1578 -
1579 -err_cleanup:
1580 - ACLK_SHARED_STATE_UNLOCK;
1581 - return 1;
1582 -}
1583 -
1584 -struct {
1585 - char *name;
1586 - int(*fnc)(struct aclk_request *cloud_to_agent);
1587 -} aclk_incoming_msg_types[] = {
1588 - { .name = "http", .fnc = aclk_handle_cloud_request },
1589 - { .name = "version", .fnc = aclk_handle_version_response },
1590 - { .name = NULL, .fnc = NULL }
1591 -};
1592 -
1593 -int aclk_handle_cloud_message(char *payload)
1594 -{
1595 - struct aclk_request cloud_to_agent;
1596 - memset(&cloud_to_agent, 0, sizeof(struct aclk_request));
1597 -
1598 - if (aclk_stats_enabled) {
1599 - ACLK_STATS_LOCK;
1600 - aclk_metrics_per_sample.cloud_req_recvd++;
1601 - ACLK_STATS_UNLOCK;
1602 - }
1603 -
1604 - if (unlikely(!payload)) {
1605 - errno = 0;
1606 - error("ACLK incoming message is empty");
1607 - goto err_cleanup_nojson;
1608 - }
1609 -
1610 - debug(D_ACLK, "ACLK incoming message (%s)", payload);
1611 -
1612 - int rc = json_parse(payload, &cloud_to_agent, cloud_to_agent_parse);
1613 -
1614 - if (unlikely(rc != JSON_OK)) {
1615 - errno = 0;
1616 - error("Malformed json request (%s)", payload);
1617 - goto err_cleanup;
1618 - }
1619 -
1620 - if (!cloud_to_agent.type_id) {
1621 - errno = 0;
1622 - error("Cloud message is missing compulsory key \"type\"");
1623 - goto err_cleanup;
1624 - }
1625 -
1626 - for (int i = 0; aclk_incoming_msg_types[i].name; i++) {
1627 - if (strcmp(cloud_to_agent.type_id, aclk_incoming_msg_types[i].name) == 0) {
1628 - if (likely(!aclk_incoming_msg_types[i].fnc(&cloud_to_agent))) {
1629 - // in case of success handler is supposed to clean up after itself
1630 - // or as in the case of aclk_handle_cloud_request take
1631 - // ownership of the pointers (done to avoid copying)
1632 - // see what `aclk_queue_query` parameter `internal` does
1633 - freez(cloud_to_agent.type_id);
1634 - return 0;
1635 - }
1636 - goto err_cleanup;
1637 - }
1638 - }
1639 -
1640 - errno = 0;
1641 - error("Unknown message type from Cloud \"%s\"", cloud_to_agent.type_id);
1642 -
1643 -err_cleanup:
1644 - if (cloud_to_agent.payload)
1645 - freez(cloud_to_agent.payload);
1646 - if (cloud_to_agent.type_id)
1647 - freez(cloud_to_agent.type_id);
1648 - if (cloud_to_agent.msg_id)
1649 - freez(cloud_to_agent.msg_id);
1650 - if (cloud_to_agent.callback_topic)
1651 - freez(cloud_to_agent.callback_topic);
1652 -
1653 -err_cleanup_nojson:
1654 - if (aclk_stats_enabled) {
1655 - ACLK_STATS_LOCK;
1656 - aclk_metrics_per_sample.cloud_req_err++;
1657 - ACLK_STATS_UNLOCK;
1658 - }
1659 -
1660 - return 1;
1661 -}
aclk/agent_cloud_link.h
+3
@@ -28,6 +28,8 @@
28 #define ACLK_DEFAULT_PORT 9002
29 #define ACLK_DEFAULT_HOST "localhost"
30
31 +#define ACLK_V2_PAYLOAD_SEPARATOR "\x0D\x0A\x0D\x0A"
32 +
33 struct aclk_request {
34 char *type_id;
35 char *msg_id;
@@ -52,6 +54,7 @@ void *aclk_main(void *ptr);
54 .start_routine = aclk_main },
55
56 extern int aclk_send_message(char *sub_topic, char *message, char *msg_id);
57 +extern int aclk_send_message_bin(char *sub_topic, const void *message, size_t len, char *msg_id);
58
59 extern char *is_agent_claimed(void);
60 extern void aclk_lws_wss_mqtt_layer_disconect_notif();
aclk/mqtt.c
+4 -4
@@ -5,6 +5,7 @@
5 #include "mqtt.h"
6 #include "aclk_lws_wss_client.h"
7 #include "aclk_stats.h"
8 +#include "aclk_rx_msgs.h"
9
10 extern usec_t aclk_session_us;
11 extern time_t aclk_session_sec;
@@ -339,7 +340,7 @@ int _link_subscribe(char *topic, int qos)
340 *
341 */
342
342 -int _link_send_message(char *topic, unsigned char *message, int *mid)
343 +int _link_send_message(char *topic, const void *message, size_t len, int *mid)
344 {
345 int rc;
346 size_t write_q, write_q_bytes, read_q;
@@ -349,9 +350,8 @@ int _link_send_message(char *topic, unsigned char *message, int *mid)
350 if (unlikely(rc != MOSQ_ERR_SUCCESS))
351 return rc;
352
352 - int msg_len = strlen((char*)message);
353 lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
354 - rc = mosquitto_publish(mosq, mid, topic, msg_len, message, ACLK_QOS, 0);
354 + rc = mosquitto_publish(mosq, mid, topic, len, message, ACLK_QOS, 0);
355
356 #ifdef NETDATA_INTERNAL_CHECKS
357 char msg_head[64];
@@ -359,7 +359,7 @@ int _link_send_message(char *topic, unsigned char *message, int *mid)
359 strncpy(msg_head, (char*)message, 60);
360 for (size_t i = 0; i < sizeof(msg_head); i++)
361 if(msg_head[i] == '\n') msg_head[i] = ' ';
362 - info("Sending MQTT len=%d mid=%d wq=%zu (%zu-bytes) readq=%zu: %s", msg_len,
362 + info("Sending MQTT len=%d mid=%d wq=%zu (%zu-bytes) readq=%zu: %s", (int)len,
363 *mid, write_q, write_q_bytes, read_q, msg_head);
364 now_realtime_timeval(&sendTimes[ *mid & 0x3ff ]);
365 #endif
aclk/mqtt.h
+1 -1
@@ -14,7 +14,7 @@ int mqtt_attempt_connection(char *aclk_hostname, int aclk_port, char *username,
14 //int _link_lib_init();
15 int _mqtt_lib_init();
16 int _link_subscribe(char *topic, int qos);
17 -int _link_send_message(char *topic, unsigned char *message, int *mid);
17 +int _link_send_message(char *topic, const void *message, size_t len, int *mid);
18 const char *_link_strerror(int rc);
19 int _link_set_lwt(char *topic, int qos);
20
web/server/web_client.c
+7 -6
@@ -199,6 +199,7 @@ void web_client_request_done(struct web_client *w) {
199 w->response.zstream.total_in = 0;
200 w->response.zstream.total_out = 0;
201 w->response.zinitialized = 0;
202 + w->flags &= ~WEB_CLIENT_CHUNKED_TRANSFER;
203 }
204 #endif // NETDATA_WITH_ZLIB
205 }
@@ -501,6 +502,7 @@ void web_client_enable_deflate(struct web_client *w, int gzip) {
502 w->response.zsent = 0;
503 w->response.zoutput = 1;
504 w->response.zinitialized = 1;
505 + w->flags |= WEB_CLIENT_CHUNKED_TRANSFER;
506
507 debug(D_DEFLATE, "%llu: Initialized compression.", w->id);
508 }
@@ -1238,12 +1240,11 @@ void web_client_build_http_header(struct web_client *w) {
1240 buffer_strcat(w->response.header_output, buffer_tostring(w->response.header));
1241
1242 // headers related to the transfer method
1241 - if(likely(w->response.zoutput)) {
1242 - buffer_strcat(w->response.header_output,
1243 - "Content-Encoding: gzip\r\n"
1244 - "Transfer-Encoding: chunked\r\n"
1245 - );
1246 - }
1243 + if(likely(w->response.zoutput))
1244 + buffer_strcat(w->response.header_output, "Content-Encoding: gzip\r\n");
1245 +
1246 + if(likely(w->flags & WEB_CLIENT_CHUNKED_TRANSFER))
1247 + buffer_strcat(w->response.header_output, "Transfer-Encoding: chunked\r\n");
1248 else {
1249 if(likely((w->response.data->len || w->response.rlen))) {
1250 // we know the content length, put it
web/server/web_client.h
+2
@@ -64,6 +64,8 @@ typedef enum web_client_flags {
64 WEB_CLIENT_FLAG_UNIX_CLIENT = 1 << 8, // if set, the client is using a UNIX socket
65
66 WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET = 1 << 9, // don't close the socket when cleaning up (static-threaded web server)
67 +
68 + WEB_CLIENT_CHUNKED_TRANSFER = 1 << 10, // chunked transfer (used with zlib compression)
69 } WEB_CLIENT_FLAGS;
70
71 //#ifdef HAVE_C___ATOMIC