@cryptotaxi247 / netdata-1 / commits / db8bde8e5

h2o evloop netdata stream support (#14868)

* implement streaming trough h2o managed port

Timotej S committed Nov 3, 2023 at 17:58 UTC db8bde8e5af3b8fa856f4263bc9c368c8aea57d6
26 files changed +1031 -90
CMakeLists.txt
+1
@@ -884,6 +884,7 @@ set(STREAMING_PLUGIN_FILES
884 streaming/sender.c
885 streaming/replication.c
886 streaming/replication.h
887 + streaming/common.h
888 )
889
890 set(CLAIM_PLUGIN_FILES
Makefile.am
+24 -11
@@ -690,6 +690,7 @@ STREAMING_PLUGIN_FILES = \
690 streaming/replication.h \
691 streaming/replication.c \
692 streaming/rrdpush.h \
693 + streaming/common.h \
694 $(NULL)
695
696 REGISTRY_PLUGIN_FILES = \
@@ -730,8 +731,6 @@ CLAIM_FILES = \
731
732 if ENABLE_ACLK
733 ACLK_FILES = \
733 - aclk/aclk_util.c \
734 - aclk/aclk_util.h \
734 aclk/aclk_stats.c \
735 aclk/aclk_stats.h \
736 aclk/aclk_query.c \
@@ -744,8 +743,6 @@ ACLK_FILES = \
743 aclk/aclk_tx_msgs.h \
744 aclk/aclk_rx_msgs.c \
745 aclk/aclk_rx_msgs.h \
747 - aclk/https_client.c \
748 - aclk/https_client.h \
746 aclk/aclk_alarm_api.c \
747 aclk/aclk_alarm_api.h \
748 aclk/aclk_contexts_api.c \
@@ -793,15 +790,9 @@ libmqttwebsockets_a_SOURCES = \
790 mqtt_websockets/src/common_public.c \
791 mqtt_websockets/src/include/common_public.h \
792 mqtt_websockets/src/include/common_internal.h \
796 - mqtt_websockets/c-rbuf/src/ringbuffer.c \
797 - mqtt_websockets/c-rbuf/include/ringbuffer.h \
798 - mqtt_websockets/c-rbuf/src/ringbuffer_internal.h \
799 - mqtt_websockets/c_rhash/src/c_rhash.c \
800 - mqtt_websockets/c_rhash/include/c_rhash.h \
801 - mqtt_websockets/c_rhash/src/c_rhash_internal.h \
793 $(NULL)
794
804 -libmqttwebsockets_a_CFLAGS = $(CFLAGS) -DMQTT_WSS_CUSTOM_ALLOC -DRBUF_CUSTOM_MALLOC -DMQTT_WSS_CPUSTATS -I$(srcdir)/aclk/helpers -I$(srcdir)/mqtt_websockets/c_rhash/include
795 +libmqttwebsockets_a_CFLAGS = $(CFLAGS) -DMQTT_WSS_CUSTOM_ALLOC -DMQTT_WSS_CPUSTATS -I$(srcdir)/aclk/helpers -I$(srcdir)/mqtt_websockets/c_rhash/include
796
797 if MQTT_WSS_DEBUG
798 libmqttwebsockets_a_CFLAGS += -DMQTT_WSS_DEBUG
@@ -907,8 +898,25 @@ ACLK_ALWAYS_BUILD_FILES = \
898 aclk/aclk.h \
899 aclk/aclk_capas.c \
900 aclk/aclk_capas.h \
901 + aclk/aclk_util.c \
902 + aclk/aclk_util.h \
903 + aclk/https_client.c \
904 + aclk/https_client.h \
905 + $(NULL)
906 +
907 +noinst_LIBRARIES += libcrutils.a
908 +
909 +libcrutils_a_SOURCES = \
910 + mqtt_websockets/c-rbuf/src/ringbuffer.c \
911 + mqtt_websockets/c-rbuf/include/ringbuffer.h \
912 + mqtt_websockets/c-rbuf/src/ringbuffer_internal.h \
913 + mqtt_websockets/c_rhash/src/c_rhash.c \
914 + mqtt_websockets/c_rhash/include/c_rhash.h \
915 + mqtt_websockets/c_rhash/src/c_rhash_internal.h \
916 $(NULL)
917
918 +libcrutils_a_CFLAGS = $(CFLAGS) -DRBUF_CUSTOM_MALLOC -I$(srcdir)/aclk/helpers -I$(abs_top_srcdir)/mqtt_websockets/c-rbuf/include -I$(srcdir)/mqtt_websockets/c_rhash/include
919 +
920 SPAWN_PLUGIN_FILES = \
921 spawn/spawn.c \
922 spawn/spawn_server.c \
@@ -996,6 +1004,10 @@ H2O_FILES = \
1004 web/server/h2o/http_server.h \
1005 web/server/h2o/h2o_utils.c \
1006 web/server/h2o/h2o_utils.h \
1007 + web/server/h2o/streaming.c \
1008 + web/server/h2o/streaming.h \
1009 + web/server/h2o/connlist.c \
1010 + web/server/h2o/connlist.h \
1011 $(NULL)
1012
1013 libh2o_a_SOURCES = \
@@ -1167,6 +1179,7 @@ NETDATA_COMMON_LIBS = \
1179 $(OPTIONAL_BROTLIDEC_LIBS) \
1180 $(OPTIONAL_DATACHANNEL_LIBS) \
1181 libjudy.a \
1182 + libcrutils.a \
1183 $(OPTIONAL_SSL_LIBS) \
1184 $(OPTIONAL_JSONC_LIBS) \
1185 $(OPTIONAL_YAML_LIBS) \
aclk/aclk_rx_msgs.c
+1 -1
@@ -108,7 +108,7 @@ static inline int aclk_v2_payload_get_query(const char *payload, char **query_ur
108 }
109 start = payload + 4;
110
111 - if(!(end = strstr(payload, " HTTP/1.1\x0D\x0A"))) {
111 + if(!(end = strstr(payload, HTTP_1_1 HTTP_ENDL))) {
112 errno = 0;
113 netdata_log_error("Doesn't look like HTTP GET request.");
114 return 1;
aclk/aclk_util.c
+4
@@ -1,6 +1,9 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "aclk_util.h"
4 +
5 +#ifdef ENABLE_ACLK
6 +
7 #include "aclk_proxy.h"
8
9 #include "daemon/common.h"
@@ -437,6 +440,7 @@ void aclk_set_proxy(char **ohost, int *port, char **uname, char **pwd, enum mqtt
440
441 freez(proxy);
442 }
443 +#endif /* ENABLE_ACLK */
444
445 #if defined(OPENSSL_VERSION_NUMBER) && OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
446 static EVP_ENCODE_CTX *EVP_ENCODE_CTX_new(void)
aclk/aclk_util.h
+3
@@ -3,6 +3,8 @@
3 #define ACLK_UTIL_H
4
5 #include "libnetdata/libnetdata.h"
6 +
7 +#ifdef ENABLE_ACLK
8 #include "mqtt_wss_client.h"
9
10 #define CLOUD_EC_MALFORMED_NODE_ID 1
@@ -112,6 +114,7 @@ unsigned long int aclk_tbeb_delay(int reset, int base, unsigned long int min, un
114 #define aclk_tbeb_reset(x) aclk_tbeb_delay(1, 0, 0, 0)
115
116 void aclk_set_proxy(char **ohost, int *port, char **uname, char **pwd, enum mqtt_wss_proxy_type *type);
117 +#endif /* ENABLE_ACLK */
118
119 int base64_encode_helper(unsigned char *out, int *outl, const unsigned char *in, int in_len);
120
aclk/https_client.c
+62 -57
@@ -4,20 +4,12 @@
4
5 #include "https_client.h"
6
7 -#include "mqtt_websockets/c-rbuf/include/ringbuffer.h"
8 -
7 #include "aclk_util.h"
8
9 #include "daemon/global_statistics.h"
10
11 #define DEFAULT_CHUNKED_RESPONSE_BUFFER_SIZE (4096)
12
15 -enum http_parse_state {
16 - HTTP_PARSE_INITIAL = 0,
17 - HTTP_PARSE_HEADERS,
18 - HTTP_PARSE_CONTENT
19 -};
20 -
13 static const char *http_req_type_to_str(http_req_type_t req) {
14 switch (req) {
15 case HTTP_REQ_GET:
@@ -33,39 +25,33 @@ static const char *http_req_type_to_str(http_req_type_t req) {
25
26 #define TRANSFER_ENCODING_CHUNKED (-2)
27
36 -typedef struct {
37 - enum http_parse_state state;
38 - int content_length;
39 - int http_code;
40 -
41 - // for chunked data only
42 - char *chunked_response;
43 - size_t chunked_response_size;
44 - size_t chunked_response_written;
45 -
46 - enum chunked_content_state {
47 - CHUNKED_CONTENT_CHUNK_SIZE = 0,
48 - CHUNKED_CONTENT_CHUNK_DATA,
49 - CHUNKED_CONTENT_CHUNK_END_CRLF,
50 - CHUNKED_CONTENT_FINAL_CRLF
51 - } chunked_content_state;
52 -
53 - size_t chunk_size;
54 - size_t chunk_got;
55 -} http_parse_ctx;
56 -
28 #define HTTP_PARSE_CTX_INITIALIZER { .state = HTTP_PARSE_INITIAL, .content_length = -1, .http_code = 0 }
58 -static inline void http_parse_ctx_clear(http_parse_ctx *ctx) {
29 +void http_parse_ctx_create(http_parse_ctx *ctx)
30 +{
31 ctx->state = HTTP_PARSE_INITIAL;
32 ctx->content_length = -1;
33 ctx->http_code = 0;
34 + ctx->headers = c_rhash_new(0);
35 + ctx->flags = HTTP_PARSE_FLAGS_DEFAULT;
36 +}
37 +
38 +void http_parse_ctx_destroy(http_parse_ctx *ctx)
39 +{
40 + c_rhash_iter_t iter;
41 + const char *key;
42 +
43 + c_rhash_iter_t_initialize(&iter);
44 + while ( !c_rhash_iter_str_keys(ctx->headers, &iter, &key) ) {
45 + void *val;
46 + c_rhash_get_ptr_by_str(ctx->headers, key, &val);
47 + freez(val);
48 + }
49 +
50 + c_rhash_destroy(ctx->headers);
51 }
52
53 #define POLL_TO_MS 100
54
66 -#define NEED_MORE_DATA 0
67 -#define PARSE_SUCCESS 1
68 -#define PARSE_ERROR -1
55 #define HTTP_LINE_TERM "\x0D\x0A"
56 #define RESP_PROTO "HTTP/1.1 "
57 #define HTTP_KEYVAL_SEPARATOR ": "
@@ -76,7 +62,7 @@ static int process_http_hdr(http_parse_ctx *parse_ctx, const char *key, const ch
62 {
63 // currently we care only about specific headers
64 // we can skip the rest
79 - if (!strcmp("content-length", key)) {
65 + if (parse_ctx->content_length < 0 && !strcmp("content-length", key)) {
66 if (parse_ctx->content_length == TRANSFER_ENCODING_CHUNKED) {
67 netdata_log_error("Content-length and transfer-encoding: chunked headers are mutually exclusive");
68 return 1;
@@ -85,7 +71,7 @@ static int process_http_hdr(http_parse_ctx *parse_ctx, const char *key, const ch
71 netdata_log_error("Duplicate content-length header");
72 return 1;
73 }
88 - parse_ctx->content_length = atoi(val);
74 + parse_ctx->content_length = str2u(val);
75 if (parse_ctx->content_length < 0) {
76 netdata_log_error("Invalid content-length %d", parse_ctx->content_length);
77 return 1;
@@ -102,9 +88,20 @@ static int process_http_hdr(http_parse_ctx *parse_ctx, const char *key, const ch
88 }
89 return 0;
90 }
91 + char *val_cpy = strdupz(val);
92 + c_rhash_insert_str_ptr(parse_ctx->headers, key, val_cpy);
93 return 0;
94 }
95
96 +const char *get_http_header_by_name(http_parse_ctx *ctx, const char *name)
97 +{
98 + const char *ret;
99 + if (c_rhash_get_ptr_by_str(ctx->headers, name, (void**)&ret))
100 + return NULL;
101 +
102 + return ret;
103 +}
104 +
105 static int parse_http_hdr(rbuf_t buf, http_parse_ctx *parse_ctx)
106 {
107 int idx, idx_end;
@@ -169,8 +166,8 @@ static int process_chunked_content(rbuf_t buf, http_parse_ctx *parse_ctx)
166 case CHUNKED_CONTENT_CHUNK_SIZE:
167 if (!rbuf_find_bytes(buf, HTTP_LINE_TERM, strlen(HTTP_LINE_TERM), &idx)) {
168 if (rbuf_bytes_available(buf) >= rbuf_get_capacity(buf))
172 - return PARSE_ERROR;
173 - return NEED_MORE_DATA;
169 + return HTTP_PARSE_ERROR;
170 + return HTTP_PARSE_NEED_MORE_DATA;
171 }
172 if (idx == 0) {
173 parse_ctx->chunked_content_state = CHUNKED_CONTENT_FINAL_CRLF;
@@ -178,7 +175,7 @@ static int process_chunked_content(rbuf_t buf, http_parse_ctx *parse_ctx)
175 }
176 if (idx >= HTTP_HDR_BUFFER_SIZE) {
177 netdata_log_error("Chunk size is too long");
181 - return PARSE_ERROR;
178 + return HTTP_PARSE_ERROR;
179 }
180 char buf_size[HTTP_HDR_BUFFER_SIZE];
181 rbuf_pop(buf, buf_size, idx);
@@ -186,13 +183,13 @@ static int process_chunked_content(rbuf_t buf, http_parse_ctx *parse_ctx)
183 long chunk_size = strtol(buf_size, NULL, 16);
184 if (chunk_size < 0 || chunk_size == LONG_MAX) {
185 netdata_log_error("Chunk size out of range");
189 - return PARSE_ERROR;
186 + return HTTP_PARSE_ERROR;
187 }
188 parse_ctx->chunk_size = chunk_size;
189 if (parse_ctx->chunk_size == 0) {
190 if (errno == EINVAL) {
191 netdata_log_error("Invalid chunk size");
195 - return PARSE_ERROR;
192 + return HTTP_PARSE_ERROR;
193 }
194 parse_ctx->chunked_content_state = CHUNKED_CONTENT_CHUNK_END_CRLF;
195 continue;
@@ -204,7 +201,7 @@ static int process_chunked_content(rbuf_t buf, http_parse_ctx *parse_ctx)
201 // fallthrough
202 case CHUNKED_CONTENT_CHUNK_DATA:
203 if (!(bytes_to_copy = rbuf_bytes_available(buf)))
207 - return NEED_MORE_DATA;
204 + return HTTP_PARSE_NEED_MORE_DATA;
205 if (bytes_to_copy > parse_ctx->chunk_size - parse_ctx->chunk_got)
206 bytes_to_copy = parse_ctx->chunk_size - parse_ctx->chunk_got;
207 rbuf_pop(buf, parse_ctx->chunked_response + parse_ctx->chunked_response_written, bytes_to_copy);
@@ -217,19 +214,19 @@ static int process_chunked_content(rbuf_t buf, http_parse_ctx *parse_ctx)
214 case CHUNKED_CONTENT_FINAL_CRLF:
215 case CHUNKED_CONTENT_CHUNK_END_CRLF:
216 if (rbuf_bytes_available(buf) < strlen(HTTP_LINE_TERM))
220 - return NEED_MORE_DATA;
217 + return HTTP_PARSE_NEED_MORE_DATA;
218 char buf_crlf[strlen(HTTP_LINE_TERM)];
219 rbuf_pop(buf, buf_crlf, strlen(HTTP_LINE_TERM));
220 if (memcmp(buf_crlf, HTTP_LINE_TERM, strlen(HTTP_LINE_TERM))) {
221 netdata_log_error("CRLF expected");
225 - return PARSE_ERROR;
222 + return HTTP_PARSE_ERROR;
223 }
224 if (parse_ctx->chunked_content_state == CHUNKED_CONTENT_FINAL_CRLF) {
225 if (parse_ctx->chunked_response_size != parse_ctx->chunked_response_written)
226 netdata_log_error("Chunked response size mismatch");
227 chunked_response_buffer_grow_by(parse_ctx, 1);
228 parse_ctx->chunked_response[parse_ctx->chunked_response_written] = 0;
232 - return PARSE_SUCCESS;
229 + return HTTP_PARSE_SUCCESS;
230 }
231 if (parse_ctx->chunk_size == 0) {
232 parse_ctx->chunked_content_state = CHUNKED_CONTENT_FINAL_CRLF;
@@ -241,34 +238,34 @@ static int process_chunked_content(rbuf_t buf, http_parse_ctx *parse_ctx)
238 } while(1);
239 }
240
244 -static int parse_http_response(rbuf_t buf, http_parse_ctx *parse_ctx)
241 +http_parse_rc parse_http_response(rbuf_t buf, http_parse_ctx *parse_ctx)
242 {
243 int idx;
244 char rc[4];
245
246 do {
247 if (parse_ctx->state != HTTP_PARSE_CONTENT && !rbuf_find_bytes(buf, HTTP_LINE_TERM, strlen(HTTP_LINE_TERM), &idx))
251 - return NEED_MORE_DATA;
248 + return HTTP_PARSE_NEED_MORE_DATA;
249 switch (parse_ctx->state) {
250 case HTTP_PARSE_INITIAL:
251 if (rbuf_memcmp_n(buf, RESP_PROTO, strlen(RESP_PROTO))) {
252 netdata_log_error("Expected response to start with \"%s\"", RESP_PROTO);
256 - return PARSE_ERROR;
253 + return HTTP_PARSE_ERROR;
254 }
255 rbuf_bump_tail(buf, strlen(RESP_PROTO));
256 if (rbuf_pop(buf, rc, 4) != 4) {
257 netdata_log_error("Expected HTTP status code");
261 - return PARSE_ERROR;
258 + return HTTP_PARSE_ERROR;
259 }
260 if (rc[3] != ' ') {
261 netdata_log_error("Expected space after HTTP return code");
265 - return PARSE_ERROR;
262 + return HTTP_PARSE_ERROR;
263 }
264 rc[3] = 0;
265 parse_ctx->http_code = atoi(rc);
266 if (parse_ctx->http_code < 100 || parse_ctx->http_code >= 600) {
267 netdata_log_error("HTTP code not in range 100 to 599");
271 - return PARSE_ERROR;
268 + return HTTP_PARSE_ERROR;
269 }
270
271 rbuf_find_bytes(buf, HTTP_LINE_TERM, strlen(HTTP_LINE_TERM), &idx);
@@ -284,7 +281,7 @@ static int parse_http_response(rbuf_t buf, http_parse_ctx *parse_ctx)
281 break;
282 }
283 if (parse_http_hdr(buf, parse_ctx))
287 - return PARSE_ERROR;
284 + return HTTP_PARSE_ERROR;
285 rbuf_find_bytes(buf, HTTP_LINE_TERM, strlen(HTTP_LINE_TERM), &idx);
286 rbuf_bump_tail(buf, idx + strlen(HTTP_LINE_TERM));
287 break;
@@ -294,11 +291,14 @@ static int parse_http_response(rbuf_t buf, http_parse_ctx *parse_ctx)
291 return process_chunked_content(buf, parse_ctx);
292
293 if (parse_ctx->content_length < 0)
297 - return PARSE_SUCCESS;
294 + return HTTP_PARSE_SUCCESS;
295 +
296 + if (parse_ctx->flags & HTTP_PARSE_FLAG_DONT_WAIT_FOR_CONTENT)
297 + return HTTP_PARSE_SUCCESS;
298
299 if (rbuf_bytes_available(buf) >= (size_t)parse_ctx->content_length)
300 - return PARSE_SUCCESS;
301 - return NEED_MORE_DATA;
300 + return HTTP_PARSE_SUCCESS;
301 + return HTTP_PARSE_NEED_MORE_DATA;
302 }
303 } while(1);
304 }
@@ -486,7 +486,7 @@ static int read_parse_response(https_req_ctx_t *ctx) {
486 } while (ctx->poll_fd.events == 0 && rbuf_bytes_free(ctx->buf_rx) > 0);
487 } while (!(ret = parse_http_response(ctx->buf_rx, &ctx->parse_ctx)));
488
489 - if (ret != PARSE_SUCCESS) {
489 + if (ret != HTTP_PARSE_SUCCESS) {
490 netdata_log_error("Error parsing HTTP response");
491 return 1;
492 }
@@ -500,7 +500,7 @@ static int handle_http_request(https_req_ctx_t *ctx) {
500 BUFFER *hdr = buffer_create(TX_BUFFER_SIZE, &netdata_buffers_statistics.buffers_aclk);
501 int rc = 0;
502
503 - http_parse_ctx_clear(&ctx->parse_ctx);
503 + http_parse_ctx_create(&ctx->parse_ctx);
504
505 // Prepare data to send
506 switch (ctx->request->request_type) {
@@ -526,7 +526,7 @@ static int handle_http_request(https_req_ctx_t *ctx) {
526 buffer_strcat(hdr, ctx->request->url);
527 }
528
529 - buffer_strcat(hdr, " HTTP/1.1\x0D\x0A");
529 + buffer_strcat(hdr, HTTP_1_1 HTTP_ENDL);
530
531 //TODO Headers!
532 if (ctx->request->request_type != HTTP_REQ_CONNECT) {
@@ -661,12 +661,15 @@ int https_request(https_req_t *request, https_req_response_t *response) {
661 ctx->request = &req;
662 if (handle_http_request(ctx)) {
663 netdata_log_error("Failed to CONNECT with proxy");
664 + http_parse_ctx_destroy(&ctx->parse_ctx);
665 goto exit_sock;
666 }
667 if (ctx->parse_ctx.http_code != 200) {
668 netdata_log_error("Proxy didn't return 200 OK (got %d)", ctx->parse_ctx.http_code);
669 + http_parse_ctx_destroy(&ctx->parse_ctx);
670 goto exit_sock;
671 }
672 + http_parse_ctx_destroy(&ctx->parse_ctx);
673 netdata_log_info("Proxy accepted CONNECT upgrade");
674 }
675 ctx->request = request;
@@ -713,8 +716,10 @@ int https_request(https_req_t *request, https_req_response_t *response) {
716 // The actual request here
717 if (handle_http_request(ctx)) {
718 netdata_log_error("Couldn't process request");
719 + http_parse_ctx_destroy(&ctx->parse_ctx);
720 goto exit_SSL;
721 }
722 + http_parse_ctx_destroy(&ctx->parse_ctx);
723 response->http_code = ctx->parse_ctx.http_code;
724 if (ctx->parse_ctx.content_length == TRANSFER_ENCODING_CHUNKED) {
725 response->payload_size = ctx->parse_ctx.chunked_response_size;
aclk/https_client.h
+55
@@ -5,6 +5,9 @@
5
6 #include "libnetdata/libnetdata.h"
7
8 +#include "mqtt_websockets/c-rbuf/include/ringbuffer.h"
9 +#include "mqtt_websockets/c_rhash/include/c_rhash.h"
10 +
11 typedef enum http_req_type {
12 HTTP_REQ_GET = 0,
13 HTTP_REQ_POST,
@@ -77,4 +80,56 @@ void https_req_response_init(https_req_response_t *res);
80
81 int https_request(https_req_t *request, https_req_response_t *response);
82
83 +// we expose previously internal parser as this is usefull also from
84 +// other parts of the code
85 +enum http_parse_state {
86 + HTTP_PARSE_INITIAL = 0,
87 + HTTP_PARSE_HEADERS,
88 + HTTP_PARSE_CONTENT
89 +};
90 +
91 +typedef uint32_t parse_ctx_flags_t;
92 +
93 +#define HTTP_PARSE_FLAG_DONT_WAIT_FOR_CONTENT ((parse_ctx_flags_t)0x01)
94 +
95 +#define HTTP_PARSE_FLAGS_DEFAULT ((parse_ctx_flags_t)0)
96 +
97 +typedef struct {
98 + parse_ctx_flags_t flags;
99 +
100 + enum http_parse_state state;
101 + int content_length;
102 + int http_code;
103 +
104 + c_rhash headers;
105 +
106 + // for chunked data only
107 + char *chunked_response;
108 + size_t chunked_response_size;
109 + size_t chunked_response_written;
110 +
111 + enum chunked_content_state {
112 + CHUNKED_CONTENT_CHUNK_SIZE = 0,
113 + CHUNKED_CONTENT_CHUNK_DATA,
114 + CHUNKED_CONTENT_CHUNK_END_CRLF,
115 + CHUNKED_CONTENT_FINAL_CRLF
116 + } chunked_content_state;
117 +
118 + size_t chunk_size;
119 + size_t chunk_got;
120 +} http_parse_ctx;
121 +
122 +void http_parse_ctx_create(http_parse_ctx *ctx);
123 +void http_parse_ctx_destroy(http_parse_ctx *ctx);
124 +
125 +typedef enum {
126 + HTTP_PARSE_ERROR = -1,
127 + HTTP_PARSE_NEED_MORE_DATA = 0,
128 + HTTP_PARSE_SUCCESS = 1
129 +} http_parse_rc;
130 +
131 +http_parse_rc parse_http_response(rbuf_t buf, http_parse_ctx *parse_ctx);
132 +
133 +const char *get_http_header_by_name(http_parse_ctx *ctx, const char *name);
134 +
135 #endif /* NETDATA_HTTPS_CLIENT_H */
collectors/plugins.d/pluginsd_parser.c
+5
@@ -13,6 +13,11 @@ static ssize_t send_to_plugin(const char *txt, void *data) {
13 if(!txt || !*txt)
14 return 0;
15
16 +#ifdef ENABLE_H2O
17 + if(parser->h2o_ctx)
18 + return h2o_stream_write(parser->h2o_ctx, txt, strlen(txt));
19 +#endif
20 +
21 errno = 0;
22 spinlock_lock(&parser->writer.spinlock);
23 ssize_t bytes = -1;
collectors/plugins.d/pluginsd_parser.h
+3
@@ -104,6 +104,9 @@ typedef struct parser {
104 #ifdef ENABLE_HTTPS
105 NETDATA_SSL *ssl_output;
106 #endif
107 +#ifdef ENABLE_H2O
108 + void *h2o_ctx; // if set we use h2o_stream functions to send data
109 +#endif
110
111 PARSER_USER_OBJECT user; // User defined structure to hold extra state between calls
112
configure.ac
-3
@@ -936,9 +936,6 @@ if test "${enable_h2o}" != "no"; then
936 else
937 can_build_h2o="no"
938 fi
939 - if test "${with_zlib}" != "yes"; then
940 - can_build_h2o="no"
941 - fi
939 AC_MSG_RESULT([${can_build_h2o}])
940
941 if test "${can_build_h2o}" = "no" -a "${enable_h2o}" = "yes"; then
libnetdata/http/http_defs.h
+7
@@ -3,6 +3,13 @@
3 #ifndef NETDATA_HTTP_DEFS_H
4 #define NETDATA_HTTP_DEFS_H
5
6 +#define HTTP_1_1 " HTTP/1.1"
7 +#define HTTP_HDR_END "\r\n\r\n"
8 +#define HTTP_ENDL "\r\n"
9 +
10 +// HTTP_CODES 1XX
11 +#define HTTP_RESP_SWITCH_PROTO 101
12 +
13 // HTTP_CODES 2XX Success
14 #define HTTP_RESP_OK 200
15
libnetdata/socket/security.c
+7 -1
@@ -109,7 +109,7 @@ static void netdata_ssl_log_error_queue(const char *call, NETDATA_SSL *ssl, unsi
109 } while((err = ERR_get_error()));
110 }
111
112 -bool netdata_ssl_open(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd) {
112 +bool netdata_ssl_open_ext(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd, const unsigned char *alpn_protos, unsigned int alpn_protos_len) {
113 errno = 0;
114 ssl->ssl_errno = 0;
115
@@ -138,6 +138,8 @@ bool netdata_ssl_open(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd) {
138 ssl->state = NETDATA_SSL_STATE_FAILED;
139 return false;
140 }
141 + if (alpn_protos && alpn_protos_len > 0)
142 + SSL_set_alpn_protos(ssl->conn, alpn_protos, alpn_protos_len);
143 }
144
145 if(SSL_set_fd(ssl->conn, fd) != 1) {
@@ -153,6 +155,10 @@ bool netdata_ssl_open(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd) {
155 return true;
156 }
157
158 +bool netdata_ssl_open(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd) {
159 + return netdata_ssl_open_ext(ssl, ctx, fd, NULL, 0);
160 +}
161 +
162 void netdata_ssl_close(NETDATA_SSL *ssl) {
163 errno = 0;
164 ssl->ssl_errno = 0;
libnetdata/socket/security.h
+1
@@ -64,6 +64,7 @@ bool netdata_ssl_connect(NETDATA_SSL *ssl);
64 bool netdata_ssl_accept(NETDATA_SSL *ssl);
65
66 bool netdata_ssl_open(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd);
67 +bool netdata_ssl_open_ext(NETDATA_SSL *ssl, SSL_CTX *ctx, int fd, const unsigned char *alpn_protos, unsigned int alpn_protos_len);
68 void netdata_ssl_close(NETDATA_SSL *ssl);
69
70 ssize_t netdata_ssl_read(NETDATA_SSL *ssl, void *buf, size_t num);
streaming/README.md
+1
@@ -44,6 +44,7 @@ node**. This file is automatically generated by Netdata the first time it is sta
44 | `buffer size bytes` | `10485760` | The size of the buffer to use when sending metrics. The default `10485760` equals a buffer of 10MB, which is good for 60 seconds of data. Increase this if you expect latencies higher than that. The buffer is flushed on reconnect. |
45 | `reconnect delay seconds` | `5` | How long to wait until retrying to connect to the parent node. |
46 | `initial clock resync iterations` | `60` | Sync the clock of charts for how many seconds when starting. |
47 +| `parent using h2o` | `no` | Set to yes if you are connecting to parent trough it's h2o webserver/port. Currently there is no reason to set this to `yes` unless you are testing the new h2o based netdata webserver. When production ready this will be set to `yes` as default. |
48
49 ### `[API_KEY]` and `[MACHINE_GUID]` sections
50
streaming/common.h new
+9
@@ -0,0 +1,9 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef STREAMING_COMMON_H
4 +#define STREAMING_COMMON_H
5 +
6 +#define NETDATA_STREAM_URL "/stream"
7 +#define NETDATA_STREAM_PROTO_NAME "netdata_stream/2.0"
8 +
9 +#endif /* STREAMING_COMMON_H */
streaming/receiver.c
+28 -7
@@ -1,6 +1,7 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "rrdpush.h"
4 +#include "web/server/h2o/http_server.h"
5
6 extern struct config stream_config;
7
@@ -57,6 +58,11 @@ static inline int read_stream(struct receiver_state *r, char* buffer, size_t siz
58 return 0;
59 }
60
61 +#ifdef ENABLE_H2O
62 + if (is_h2o_rrdpush(r))
63 + return (int)h2o_stream_read(r->h2o_ctx, buffer, size);
64 +#endif
65 +
66 int tries = 100;
67 ssize_t bytes_read;
68
@@ -338,6 +344,10 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
344 parser = parser_init(&user, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl);
345 }
346
347 +#ifdef ENABLE_H2O
348 + parser->h2o_ctx = rpt->h2o_ctx;
349 +#endif
350 +
351 pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);
352
353 rrd_collector_started();
@@ -761,19 +771,30 @@ static void rrdpush_receive(struct receiver_state *rpt)
771 }
772
773 netdata_log_debug(D_STREAM, "Initial response to %s: %s", rpt->client_ip, initial_response);
764 - ssize_t bytes_sent = send_timeout(
774 +#ifdef ENABLE_H2O
775 + if (is_h2o_rrdpush(rpt)) {
776 + h2o_stream_write(rpt->h2o_ctx, initial_response, strlen(initial_response));
777 + } else {
778 +#endif
779 + ssize_t bytes_sent = send_timeout(
780 #ifdef ENABLE_HTTPS
766 - &rpt->ssl,
781 + &rpt->ssl,
782 #endif
768 - rpt->fd, initial_response, strlen(initial_response), 0, 60);
783 + rpt->fd, initial_response, strlen(initial_response), 0, 60);
784
770 - if(bytes_sent != (ssize_t)strlen(initial_response)) {
771 - internal_error(true, "Cannot send response, got %zd bytes, expecting %zu bytes", bytes_sent, strlen(initial_response));
772 - rrdpush_receive_log_status(rpt, "cannot reply back", "CANT REPLY DROPPING CONNECTION");
773 - goto cleanup;
785 + if(bytes_sent != (ssize_t)strlen(initial_response)) {
786 + internal_error(true, "Cannot send response, got %zd bytes, expecting %zu bytes", bytes_sent, strlen(initial_response));
787 + rrdpush_receive_log_status(rpt, "cannot reply back", "CANT REPLY DROPPING CONNECTION");
788 + goto cleanup;
789 + }
790 +#ifdef ENABLE_H2O
791 }
792 +#endif
793 }
794
795 +#ifdef ENABLE_H2O
796 + unless_h2o_rrdpush(rpt)
797 +#endif
798 {
799 // remove the non-blocking flag from the socket
800 if(sock_delnonblock(rpt->fd) < 0)
streaming/rrdpush.c
+5 -1
@@ -930,7 +930,7 @@ static void rrdpush_receiver_takeover_web_connection(struct web_client *w, struc
930 }
931
932 void *rrdpush_receiver_thread(void *ptr);
933 -int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_string) {
933 +int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_string, void *h2o_ctx) {
934
935 if(!service_running(ABILITY_STREAMING_CONNECTIONS))
936 return rrdpush_receiver_too_busy_now(w);
@@ -941,6 +941,10 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri
941
942 rpt->capabilities = STREAM_CAP_INVALID;
943
944 +#ifdef ENABLE_H2O
945 + rpt->h2o_ctx = h2o_ctx;
946 +#endif
947 +
948 __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_receivers, sizeof(*rpt), __ATOMIC_RELAXED);
949 __atomic_add_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(struct rrdhost_system_info), __ATOMIC_RELAXED);
950
streaming/rrdpush.h
+13 -1
@@ -137,6 +137,7 @@ typedef enum {
137 STREAM_HANDSHAKE_DISCONNECT_SOCKET_EOF = -24,
138 STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED = -25,
139 STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_TIMEOUT = -26,
140 + STREAM_HANDSHAKE_ERROR_HTTP_UPGRADE = -27,
141
142 } STREAM_HANDSHAKE;
143
@@ -247,6 +248,8 @@ struct sender_state {
248 usec_t last_flush_time_ut; // the last time the sender flushed the sending buffer in USEC
249 time_t last_buffer_recreate_s; // true when the sender buffer should be re-created
250 } atomic;
251 +
252 + int parent_using_h2o;
253 };
254
255 #define sender_lock(sender) spinlock_lock(&(sender)->spinlock)
@@ -378,8 +381,17 @@ struct receiver_state {
381 STREAM_NODE_INSTANCE *array;
382 } instances;
383 */
384 +
385 +#ifdef ENABLE_H2O
386 + void *h2o_ctx;
387 +#endif
388 };
389
390 +#ifdef ENABLE_H2O
391 +#define is_h2o_rrdpush(x) ((x)->h2o_ctx != NULL)
392 +#define unless_h2o_rrdpush(x) if(!is_h2o_rrdpush(x))
393 +#endif
394 +
395 struct rrdpush_destinations {
396 STRING *destination;
397 bool ssl;
@@ -436,7 +448,7 @@ void rrdpush_send_dyncfg(RRDHOST *host);
448 #define THREAD_TAG_STREAM_RECEIVER "RCVR" // "[host]" is appended
449 #define THREAD_TAG_STREAM_SENDER "SNDR" // "[host]" is appended
450
439 -int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_string);
451 +int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_string, void *h2o_ctx);
452 void rrdpush_sender_thread_stop(RRDHOST *host, STREAM_HANDSHAKE reason, bool wait);
453
454 void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, const RRDVAR_ACQUIRED *rva);
streaming/sender.c
+119 -2
@@ -1,6 +1,8 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "rrdpush.h"
4 +#include "common.h"
5 +#include "aclk/https_client.h"
6
7 #define WORKER_SENDER_JOB_CONNECT 0
8 #define WORKER_SENDER_JOB_PIPE_READ 1
@@ -534,6 +536,13 @@ static inline bool rrdpush_sender_validate_response(RRDHOST *host, struct sender
536 return false;
537 }
538
539 +unsigned char alpn_proto_list[] = {
540 + 18, 'n', 'e', 't', 'd', 'a', 't', 'a', '_', 's', 't', 'r', 'e', 'a', 'm', '/', '2', '.', '0',
541 + 8, 'h', 't', 't', 'p', '/', '1', '.', '1'
542 +};
543 +
544 +#define CONN_UPGRADE_VAL "upgrade"
545 +
546 static bool rrdpush_sender_connect_ssl(struct sender_state *s __maybe_unused) {
547 #ifdef ENABLE_HTTPS
548 RRDHOST *host = s->host;
@@ -544,7 +553,7 @@ static bool rrdpush_sender_connect_ssl(struct sender_state *s __maybe_unused) {
553 if(!ssl_required)
554 return true;
555
547 - if (netdata_ssl_open(&host->sender->ssl, netdata_ssl_streaming_sender_ctx, s->rrdpush_sender_socket)) {
556 + if (netdata_ssl_open_ext(&host->sender->ssl, netdata_ssl_streaming_sender_ctx, s->rrdpush_sender_socket, alpn_proto_list, sizeof(alpn_proto_list))) {
557 if(!netdata_ssl_connect(&host->sender->ssl)) {
558 // couldn't connect
559
@@ -579,6 +588,104 @@ static bool rrdpush_sender_connect_ssl(struct sender_state *s __maybe_unused) {
588 #endif
589 }
590
591 +static int rrdpush_http_upgrade_prelude(RRDHOST *host, struct sender_state *s) {
592 +
593 + char http[HTTP_HEADER_SIZE + 1];
594 + snprintfz(http, HTTP_HEADER_SIZE,
595 + "GET " NETDATA_STREAM_URL HTTP_1_1 HTTP_ENDL
596 + "Upgrade: " NETDATA_STREAM_PROTO_NAME HTTP_ENDL
597 + "Connection: Upgrade"
598 + HTTP_HDR_END);
599 +
600 + ssize_t bytes = send_timeout(
601 +#ifdef ENABLE_HTTPS
602 + &host->sender->ssl,
603 +#endif
604 + s->rrdpush_sender_socket,
605 + http,
606 + strlen(http),
607 + 0,
608 + 1000);
609 +
610 + bytes = recv_timeout(
611 +#ifdef ENABLE_HTTPS
612 + &host->sender->ssl,
613 +#endif
614 + s->rrdpush_sender_socket,
615 + http,
616 + HTTP_HEADER_SIZE,
617 + 0,
618 + 1000);
619 +
620 + if (bytes <= 0) {
621 + error_report("Error reading from remote");
622 + return 1;
623 + }
624 +
625 + rbuf_t buf = rbuf_create(bytes);
626 + rbuf_push(buf, http, bytes);
627 +
628 + http_parse_ctx ctx;
629 + http_parse_ctx_create(&ctx);
630 + ctx.flags |= HTTP_PARSE_FLAG_DONT_WAIT_FOR_CONTENT;
631 +
632 + int rc;
633 +// while((rc = parse_http_response(buf, &ctx)) == HTTP_PARSE_NEED_MORE_DATA);
634 + rc = parse_http_response(buf, &ctx);
635 +
636 + if (rc != HTTP_PARSE_SUCCESS) {
637 + error_report("Failed to parse HTTP response sent. (%d)", rc);
638 + goto err_cleanup;
639 + }
640 + if (ctx.http_code == HTTP_RESP_MOVED_PERM) {
641 + const char *hdr = get_http_header_by_name(&ctx, "location");
642 + if (hdr)
643 + error_report("HTTP response is %d Moved Permanently (location: \"%s\") instead of expected %d Switching Protocols.", ctx.http_code, hdr, HTTP_RESP_SWITCH_PROTO);
644 + else
645 + error_report("HTTP response is %d instead of expected %d Switching Protocols.", ctx.http_code, HTTP_RESP_SWITCH_PROTO);
646 + goto err_cleanup;
647 + }
648 + if (ctx.http_code == HTTP_RESP_NOT_FOUND) {
649 + error_report("HTTP response is %d instead of expected %d Switching Protocols. Parent version too old.", ctx.http_code, HTTP_RESP_SWITCH_PROTO);
650 + // TODO set some flag here that will signify parent is older version
651 + // and to try connection without rrdpush_http_upgrade_prelude next time
652 + goto err_cleanup;
653 + }
654 + if (ctx.http_code != HTTP_RESP_SWITCH_PROTO) {
655 + error_report("HTTP response is %d instead of expected %d Switching Protocols", ctx.http_code, HTTP_RESP_SWITCH_PROTO);
656 + goto err_cleanup;
657 + }
658 +
659 + const char *hdr = get_http_header_by_name(&ctx, "connection");
660 + if (!hdr) {
661 + error_report("Missing \"connection\" header in reply");
662 + goto err_cleanup;
663 + }
664 + if (strncmp(hdr, CONN_UPGRADE_VAL, strlen(CONN_UPGRADE_VAL))) {
665 + error_report("Expected \"connection: " CONN_UPGRADE_VAL "\"");
666 + goto err_cleanup;
667 + }
668 +
669 + hdr = get_http_header_by_name(&ctx, "upgrade");
670 + if (!hdr) {
671 + error_report("Missing \"upgrade\" header in reply");
672 + goto err_cleanup;
673 + }
674 + if (strncmp(hdr, NETDATA_STREAM_PROTO_NAME, strlen(NETDATA_STREAM_PROTO_NAME))) {
675 + error_report("Expected \"upgrade: " NETDATA_STREAM_PROTO_NAME "\"");
676 + goto err_cleanup;
677 + }
678 +
679 + netdata_log_debug(D_STREAM, "Stream sender upgrade to \"" NETDATA_STREAM_PROTO_NAME "\" successful");
680 + rbuf_free(buf);
681 + http_parse_ctx_destroy(&ctx);
682 + return 0;
683 +err_cleanup:
684 + rbuf_free(buf);
685 + http_parse_ctx_destroy(&ctx);
686 + return 1;
687 +}
688 +
689 static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_port, int timeout, struct sender_state *s) {
690
691 struct timeval tv = {
@@ -663,7 +770,7 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
770 "&NETDATA_SYSTEM_TOTAL_RAM=%s"
771 "&NETDATA_SYSTEM_TOTAL_DISK_SIZE=%s"
772 "&NETDATA_PROTOCOL_VERSION=%s"
666 - " HTTP/1.1\r\n"
773 + HTTP_1_1 HTTP_ENDL
774 "User-Agent: %s/%s\r\n"
775 "Accept: */*\r\n\r\n"
776 , host->rrdpush_send_api_key
@@ -718,6 +825,13 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
825 if(!rrdpush_sender_connect_ssl(s))
826 return false;
827
828 + if (s->parent_using_h2o && rrdpush_http_upgrade_prelude(host, s)) {
829 + rrdpush_sender_thread_close_socket(host);
830 + host->destination->reason = STREAM_HANDSHAKE_ERROR_HTTP_UPGRADE;
831 + host->destination->postpone_reconnection_until = now_realtime_sec() + 1 * 60;
832 + return false;
833 + }
834 +
835 ssize_t bytes, len = (ssize_t)strlen(http);
836
837 bytes = send_timeout(
@@ -1351,6 +1465,9 @@ void *rrdpush_sender_thread(void *ptr) {
1465 "initial clock resync iterations",
1466 remote_clock_resync_iterations); // TODO: REMOVE FOR SLEW / GAPFILLING
1467
1468 + s->parent_using_h2o = appconfig_get_boolean(
1469 + &stream_config, CONFIG_SECTION_STREAM, "parent using h2o", false);
1470 +
1471 // initialize rrdpush globals
1472 rrdhost_flag_clear(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
1473 rrdhost_flag_clear(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
web/server/h2o/connlist.c new
+156
@@ -0,0 +1,156 @@
1 +#include "connlist.h"
2 +
3 +conn_list_t conn_list = { NULL, NULL, 0, 0, PTHREAD_MUTEX_INITIALIZER };
4 +
5 +static h2o_stream_conn_t **conn_list_get_null_element_unsafe(conn_list_t *list)
6 +{
7 + struct conn_list_leaf *leaf = list->head;
8 + while (leaf != NULL) {
9 + for (int i = 0; i < CONN_LIST_MEMPOOL_SIZE; i++) {
10 + if (leaf->conn[i] == NULL)
11 + return &leaf->conn[i];
12 + }
13 + leaf = leaf->next;
14 + }
15 + return NULL;
16 +}
17 +
18 +void conn_list_insert(conn_list_t *list, h2o_stream_conn_t *conn)
19 +{
20 + pthread_mutex_lock(&list->lock);
21 +
22 + // in case the allocated capacity is not used up
23 + // we can reuse the null element
24 + if (list->capacity != list->size) {
25 + h2o_stream_conn_t **null_element = conn_list_get_null_element_unsafe(list);
26 + if (unlikely(null_element == NULL)) {
27 + pthread_mutex_unlock(&list->lock);
28 + error_report("conn_list_insert: capacity != size but no null element found");
29 + return;
30 + }
31 + *null_element = conn;
32 + list->size++;
33 + pthread_mutex_unlock(&list->lock);
34 + return;
35 + }
36 +
37 + // if not, we need to allocate a new leaf
38 + struct conn_list_leaf *old_tail = list->tail;
39 + list->tail = callocz(1, sizeof(struct conn_list_leaf));
40 + if (unlikely(old_tail == NULL))
41 + list->head = list->tail;
42 + else
43 + old_tail->next = list->tail;
44 +
45 + list->tail->conn[0] = conn;
46 + list->size++;
47 + list->capacity += CONN_LIST_MEMPOOL_SIZE;
48 +
49 + pthread_mutex_unlock(&list->lock);
50 +}
51 +
52 +typedef struct {
53 + conn_list_t *list;
54 + struct conn_list_leaf *leaf;
55 + int idx;
56 +} conn_list_iter_t;
57 +
58 +static inline void conn_list_iter_create_unsafe(conn_list_iter_t *iter, conn_list_t *list)
59 +{
60 + iter->list = list;
61 + iter->leaf = list->head;
62 + iter->idx = 0;
63 +}
64 +
65 +static inline int conn_list_iter_next_unsafe(conn_list_iter_t *iter, h2o_stream_conn_t **conn)
66 +{
67 + if (unlikely(iter->idx == iter->list->capacity))
68 + return 0;
69 +
70 + if (iter->idx && iter->idx % CONN_LIST_MEMPOOL_SIZE == 0) {
71 + iter->leaf = iter->leaf->next;
72 + }
73 +
74 + *conn = iter->leaf->conn[iter->idx++ % CONN_LIST_MEMPOOL_SIZE];
75 + return 1;
76 +}
77 +
78 +void conn_list_iter_all(conn_list_t *list, void (*cb)(h2o_stream_conn_t *conn))
79 +{
80 + pthread_mutex_lock(&list->lock);
81 + conn_list_iter_t iter;
82 + conn_list_iter_create_unsafe(&iter, list);
83 + h2o_stream_conn_t *conn;
84 + while (conn_list_iter_next_unsafe(&iter, &conn)) {
85 + if (conn == NULL)
86 + continue;
87 + cb(conn);
88 + }
89 + pthread_mutex_unlock(&list->lock);
90 +}
91 +
92 +static void conn_list_garbage_collect_unsafe(conn_list_t *list)
93 +{
94 + if (list->capacity - list->size > CONN_LIST_MEMPOOL_SIZE) {
95 + struct conn_list_leaf *new_tail = list->head;
96 + while (new_tail->next != list->tail)
97 + new_tail = new_tail->next;
98 +
99 + // check if the tail leaf is empty and move the data if not
100 + for (int i = 0; i < CONN_LIST_MEMPOOL_SIZE; i++) {
101 + if (list->tail->conn[i] != NULL) {
102 + h2o_stream_conn_t **null_element = conn_list_get_null_element_unsafe(list);
103 + if (unlikely(null_element == NULL)) {
104 + error_report("conn_list_garbage_collect_unsafe: list->capacity - list->size > CONN_LIST_MEMPOOL_SIZE but no null element found?");
105 + return;
106 + }
107 + *null_element = list->tail->conn[i];
108 + list->tail->conn[i] = NULL;
109 + }
110 + }
111 +
112 + freez(list->tail);
113 + new_tail->next = NULL;
114 + list->tail = new_tail;
115 + list->capacity -= CONN_LIST_MEMPOOL_SIZE;
116 + }
117 +}
118 +
119 +static inline int conn_list_iter_remove(conn_list_iter_t *iter, h2o_stream_conn_t *conn)
120 +{
121 + if (unlikely(iter->idx == iter->list->capacity))
122 + return -1;
123 +
124 + if (iter->idx && iter->idx % CONN_LIST_MEMPOOL_SIZE == 0) {
125 + iter->leaf = iter->leaf->next;
126 + }
127 +
128 + if(conn == iter->leaf->conn[iter->idx % CONN_LIST_MEMPOOL_SIZE]) {
129 + iter->leaf->conn[iter->idx % CONN_LIST_MEMPOOL_SIZE] = NULL;
130 +
131 + iter->idx++;
132 + return 1;
133 + }
134 +
135 + iter->idx++;
136 + return 0;
137 +}
138 +
139 +int conn_list_remove_conn(conn_list_t *list, h2o_stream_conn_t *conn)
140 +{
141 + pthread_mutex_lock(&list->lock);
142 + conn_list_iter_t iter;
143 + conn_list_iter_create_unsafe(&iter, list);
144 + int rc;
145 + while (!(rc = conn_list_iter_remove(&iter, conn)));
146 + if (rc == -1) {
147 + pthread_mutex_unlock(&list->lock);
148 + error_report("conn_list_remove_conn: conn not found");
149 + return 0;
150 + }
151 + list->size--;
152 + conn_list_garbage_collect_unsafe(list);
153 + pthread_mutex_unlock(&list->lock);
154 + return 1;
155 +}
156 +
web/server/h2o/connlist.h new
+30
@@ -0,0 +1,30 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef HTTPD_CONNLIST_H
4 +#define HTTPD_CONNLIST_H
5 +
6 +#include "streaming.h"
7 +
8 +// (-1) in following macro is to keep conn list + next pointer
9 +// be power of 2
10 +#define CONN_LIST_MEMPOOL_SIZE ((2^5)-1)
11 +struct conn_list_leaf {
12 + h2o_stream_conn_t *conn[CONN_LIST_MEMPOOL_SIZE];
13 + struct conn_list_leaf *next;
14 +};
15 +
16 +typedef struct {
17 + struct conn_list_leaf *head;
18 + struct conn_list_leaf *tail;
19 + int size;
20 + int capacity;
21 + pthread_mutex_t lock;
22 +} conn_list_t;
23 +
24 +extern conn_list_t conn_list;
25 +
26 +void conn_list_insert(conn_list_t *list, h2o_stream_conn_t *conn);
27 +void conn_list_iter_all(conn_list_t *list, void (*cb)(h2o_stream_conn_t *conn));
28 +int conn_list_remove_conn(conn_list_t *list, h2o_stream_conn_t *conn);
29 +
30 +#endif /* HTTPD_CONNLIST_H */
web/server/h2o/http_server.c
+49 -5
@@ -1,9 +1,12 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "daemon/common.h"
3 +#include "streaming/common.h"
4 #include "http_server.h"
5 +
6 #include "h2o.h"
7 +#include "h2o/http1.h"
8
9 +#include "streaming.h"
10 #include "h2o_utils.h"
11
12 static h2o_globalconf_t config;
@@ -188,6 +191,7 @@ static inline int _netdata_uberhandler(h2o_req_t *req, RRDHOST **host)
191 // individual response generators and thus remove the need to "emulate"
192 // the old webserver calling this function here and in ACLK
193 struct web_client w;
194 + memset(&w, 0, sizeof(w));
195 w.response.data = buffer_create(NBUF_INITIAL_SIZE_RESP, NULL);
196 w.response.header = buffer_create(NBUF_INITIAL_SIZE_RESP, NULL);
197 w.url_query_string_decoded = buffer_create(NBUF_INITIAL_SIZE_RESP, NULL);
@@ -241,17 +245,19 @@ static int netdata_uberhandler(h2o_handler_t *self, h2o_req_t *req)
245
246 int ret = _netdata_uberhandler(req, &host);
247
244 - char host_uuid_str[UUID_STR_LEN];
245 - uuid_unparse_lower(host->host_uuid, host_uuid_str);
246 -
248 if (!ret) {
249 + char host_uuid_str[UUID_STR_LEN];
250 +
251 + if (host != NULL)
252 + uuid_unparse_lower(host->host_uuid, host_uuid_str);
253 +
254 netdata_log_access("HTTPD OK method: " PRINTF_H2O_IOVEC_FMT
255 ", path: " PRINTF_H2O_IOVEC_FMT
256 ", as host: %s"
257 ", response: %d",
258 PRINTF_H2O_IOVEC(&req->method),
259 PRINTF_H2O_IOVEC(&req->input.path),
254 - host == localhost ? "localhost" : host_uuid_str,
260 + host == NULL ? "unknown" : (localhost ? "localhost" : host_uuid_str),
261 req->res.status);
262 } else {
263 netdata_log_access("HTTPD %d"
@@ -288,6 +294,33 @@ static int hdl_netdata_conf(h2o_handler_t *self, h2o_req_t *req)
294 return 0;
295 }
296
297 +static int hdl_stream(h2o_handler_t *self, h2o_req_t *req)
298 +{
299 + UNUSED(self);
300 + netdata_log_info("Streaming request trough h2o received");
301 + h2o_stream_conn_t *conn = mallocz(sizeof(*conn));
302 + h2o_stream_conn_t_init(conn);
303 +
304 + if (is_streaming_handshake(req)) {
305 + h2o_stream_conn_t_destroy(conn);
306 + freez(conn);
307 + return 1;
308 + }
309 +
310 + /* build response */
311 + req->res.status = HTTP_RESP_SWITCH_PROTO;
312 + req->res.reason = "Switching Protocols";
313 + h2o_add_header(&req->pool, &req->res.headers, H2O_TOKEN_UPGRADE, NULL, H2O_STRLIT(NETDATA_STREAM_PROTO_NAME));
314 +
315 +// TODO we should consider adding some nonce header here
316 +// h2o_add_header_by_str(&req->pool, &req->res.headers, H2O_STRLIT("whatever reply"), 0, NULL, accept_key,
317 +// strlen(accept_key));
318 +
319 + h2o_http1_upgrade(req, NULL, 0, stream_on_complete, conn);
320 +
321 + return 0;
322 +}
323 +
324 #define POLL_INTERVAL 100
325
326 void *h2o_main(void *ptr) {
@@ -308,6 +341,10 @@ void *h2o_main(void *ptr) {
341 h2o_handler_t *handler = h2o_create_handler(pathconf, sizeof(*handler));
342 handler->on_req = hdl_netdata_conf;
343
344 + pathconf = h2o_config_register_path(hostconf, NETDATA_STREAM_URL, 0);
345 + handler = h2o_create_handler(pathconf, sizeof(*handler));
346 + handler->on_req = hdl_stream;
347 +
348 pathconf = h2o_config_register_path(hostconf, "/", 0);
349 handler = h2o_create_handler(pathconf, sizeof(*handler));
350 handler->on_req = netdata_uberhandler;
@@ -328,12 +365,19 @@ void *h2o_main(void *ptr) {
365 return NULL;
366 }
367
368 + usec_t last_wpoll = now_monotonic_usec();
369 while (service_running(SERVICE_HTTPD)) {
370 int rc = h2o_evloop_run(ctx.loop, POLL_INTERVAL);
371 if (rc < 0 && errno != EINTR) {
372 netdata_log_error("h2o_evloop_run returned (%d) with errno other than EINTR. Aborting", rc);
373 break;
374 }
375 + usec_t now = now_monotonic_usec();
376 + if (now - last_wpoll > POLL_INTERVAL * USEC_PER_MS) {
377 + last_wpoll = now;
378 +
379 + h2o_stream_check_pending_write_reqs();
380 + }
381 }
382
383 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
web/server/h2o/http_server.h
+5
@@ -3,8 +3,13 @@
3 #ifndef HTTP_SERVER_H
4 #define HTTP_SERVER_H
5
6 +#include "libnetdata/libnetdata.h"
7 +
8 void *h2o_main(void * ptr);
9
10 +int h2o_stream_write(void *ctx, const char *data, size_t data_len);
11 +size_t h2o_stream_read(void *ctx, char *buf, size_t read_bytes);
12 +
13 int httpd_is_enabled();
14
15 #endif /* HTTP_SERVER_H */
web/server/h2o/streaming.c new
+381
@@ -0,0 +1,381 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "streaming.h"
4 +#include "connlist.h"
5 +#include "h2o_utils.h"
6 +#include "streaming/common.h"
7 +
8 +static int pending_write_reqs = 0;
9 +
10 +#define H2O2STREAM_BUF_SIZE (1024 * 1024)
11 +
12 +// h2o_stream_conn_t related functions
13 +void h2o_stream_conn_t_init(h2o_stream_conn_t *conn)
14 +{
15 + memset(conn, 0, sizeof(*conn));
16 + conn->rx = rbuf_create(H2O2STREAM_BUF_SIZE);
17 + conn->tx = rbuf_create(H2O2STREAM_BUF_SIZE);
18 +
19 + pthread_mutex_init(&conn->rx_buf_lock, NULL);
20 + pthread_mutex_init(&conn->tx_buf_lock, NULL);
21 + pthread_cond_init(&conn->rx_buf_cond, NULL);
22 + // no need to check for NULL as rbuf_create uses mallocz internally
23 +}
24 +
25 +void h2o_stream_conn_t_destroy(h2o_stream_conn_t *conn)
26 +{
27 + rbuf_free(conn->rx);
28 + rbuf_free(conn->tx);
29 +
30 + freez(conn->url);
31 + freez(conn->user_agent);
32 +
33 + pthread_mutex_destroy(&conn->rx_buf_lock);
34 + pthread_mutex_destroy(&conn->tx_buf_lock);
35 + pthread_cond_destroy(&conn->rx_buf_cond);
36 +}
37 +
38 +// streaming upgrade related functions
39 +int is_streaming_handshake(h2o_req_t *req)
40 +{
41 + /* method */
42 + if (!h2o_memis(req->input.method.base, req->input.method.len, H2O_STRLIT("GET")))
43 + return 1;
44 +
45 + if (!h2o_memis(req->path_normalized.base, req->path_normalized.len, H2O_STRLIT(NETDATA_STREAM_URL))) {
46 + return 1;
47 + }
48 +
49 + /* upgrade header */
50 + if (req->upgrade.base == NULL || !h2o_lcstris(req->upgrade.base, req->upgrade.len, H2O_STRLIT(NETDATA_STREAM_PROTO_NAME)))
51 + return 1;
52 +
53 + // TODO consider adding some key in form of random number
54 + // to prevent caching on route especially if TLS is not used
55 + // e.g. client sends random number
56 + // server replies with it xored
57 +
58 + return 0;
59 +}
60 +
61 +static void stream_on_close(h2o_stream_conn_t *conn);
62 +void stream_process(h2o_stream_conn_t *conn, int initial);
63 +
64 +void stream_on_complete(void *user_data, h2o_socket_t *sock, size_t reqsize)
65 +{
66 + h2o_stream_conn_t *conn = user_data;
67 +
68 + /* close the connection on error */
69 + if (sock == NULL) {
70 + stream_on_close(conn);
71 + return;
72 + }
73 +
74 + conn->sock = sock;
75 + sock->data = conn;
76 +
77 + conn_list_insert(&conn_list, conn);
78 +
79 + h2o_buffer_consume(&sock->input, reqsize);
80 + stream_process(conn, 1);
81 +}
82 +
83 +// handling of active streams
84 +static void stream_on_close(h2o_stream_conn_t *conn)
85 +{
86 + if (conn->sock != NULL)
87 + h2o_socket_close(conn->sock);
88 +
89 + conn_list_remove_conn(&conn_list, conn);
90 +
91 + pthread_mutex_lock(&conn->rx_buf_lock);
92 + conn->shutdown = 1;
93 + pthread_cond_broadcast(&conn->rx_buf_cond);
94 + pthread_mutex_unlock(&conn->rx_buf_lock);
95 +
96 + h2o_stream_conn_t_destroy(conn);
97 + freez(conn);
98 +}
99 +
100 +static void on_write_complete(h2o_socket_t *sock, const char *err)
101 +{
102 + h2o_stream_conn_t *conn = sock->data;
103 +
104 + if (err != NULL) {
105 + stream_on_close(conn);
106 + error_report("Streaming connection error \"%s\"", err);
107 + return;
108 + }
109 +
110 + pthread_mutex_lock(&conn->tx_buf_lock);
111 +
112 + rbuf_bump_tail(conn->tx, conn->tx_buf.len);
113 +
114 + conn->tx_buf.base = NULL;
115 + conn->tx_buf.len = 0;
116 +
117 + pthread_mutex_unlock(&conn->tx_buf_lock);
118 +
119 + stream_process(conn, 0);
120 +}
121 +
122 +static void stream_on_recv(h2o_socket_t *sock, const char *err)
123 +{
124 + h2o_stream_conn_t *conn = sock->data;
125 +
126 + if (err != NULL) {
127 + stream_on_close(conn);
128 + error_report("Streaming connection error \"%s\"", err);
129 + return;
130 + }
131 + stream_process(conn, 0);
132 +}
133 +
134 +#define PARSE_DONE 1
135 +#define PARSE_ERROR -1
136 +#define GIMME_MORE_OF_DEM_SWEET_BYTEZ 0
137 +
138 +#define STREAM_METHOD "STREAM "
139 +#define USER_AGENT "User-Agent: "
140 +
141 +#define NEED_MIN_BYTES(buf, bytes) \
142 +if (rbuf_bytes_available(buf) < bytes) \
143 + return GIMME_MORE_OF_DEM_SWEET_BYTEZ;
144 +
145 +// TODO check in streaming code this is probably defined somewhere already
146 +#define MAX_LEN_STREAM_HELLO (1024*2)
147 +
148 +static int process_STREAM_X_HTTP_1_1(http_stream_parse_state_t *parser_state, rbuf_t buf, char **url, char **user_agent)
149 +{
150 + int idx;
151 + switch(*parser_state) {
152 + case HTTP_STREAM:
153 + NEED_MIN_BYTES(buf, strlen(STREAM_METHOD));
154 + if (rbuf_memcmp_n(buf, H2O_STRLIT(STREAM_METHOD))) {
155 + error_report("Expected \"%s\"", STREAM_METHOD);
156 + return PARSE_ERROR;
157 + }
158 + rbuf_bump_tail(buf, strlen(STREAM_METHOD));
159 + *parser_state = HTTP_URL;
160 + /* FALLTHROUGH */
161 + case HTTP_URL:
162 + if (!rbuf_find_bytes(buf, " ", 1, &idx)) {
163 + if (rbuf_bytes_available(buf) >= MAX_LEN_STREAM_HELLO) {
164 + error_report("The initial \"STREAM [URL]" HTTP_1_1 "\" over max of %d", MAX_LEN_STREAM_HELLO);
165 + return PARSE_ERROR;
166 + }
167 + }
168 + *url = mallocz(idx + 1);
169 + rbuf_pop(buf, *url, idx);
170 + (*url)[idx] = 0;
171 +
172 + *parser_state = HTTP_PROTO;
173 + /* FALLTHROUGH */
174 + case HTTP_PROTO:
175 + NEED_MIN_BYTES(buf, strlen(HTTP_1_1));
176 + if (rbuf_memcmp_n(buf, H2O_STRLIT(HTTP_1_1))) {
177 + error_report("Expected \"%s\"", HTTP_1_1);
178 + return PARSE_ERROR;
179 + }
180 + rbuf_bump_tail(buf, strlen(HTTP_1_1));
181 + *parser_state = HTTP_USER_AGENT_KEY;
182 + /* FALLTHROUGH */
183 + case HTTP_USER_AGENT_KEY:
184 + // and OF COURSE EVERYTHING is passed in URL except
185 + // for user agent which we need and is passed as HTTP header
186 + // not worth writing a parser for this so we manually extract
187 + // just the single header we need and skip everything else
188 + if (!rbuf_find_bytes(buf, USER_AGENT, strlen(USER_AGENT), &idx)) {
189 + if (rbuf_bytes_available(buf) >= (size_t)(rbuf_get_capacity(buf) * 0.9)) {
190 + error_report("The initial \"STREAM [URL]" HTTP_1_1 "\" over max of %d", MAX_LEN_STREAM_HELLO);
191 + return PARSE_ERROR;
192 + }
193 + return GIMME_MORE_OF_DEM_SWEET_BYTEZ;
194 + }
195 + rbuf_bump_tail(buf, idx + strlen(USER_AGENT));
196 + *parser_state = HTTP_USER_AGENT_VALUE;
197 + /* FALLTHROUGH */
198 + case HTTP_USER_AGENT_VALUE:
199 + if (!rbuf_find_bytes(buf, "\r\n", 2, &idx)) {
200 + if (rbuf_bytes_available(buf) >= (size_t)(rbuf_get_capacity(buf) * 0.9)) {
201 + error_report("The initial \"STREAM [URL]" HTTP_1_1 "\" over max of %d", MAX_LEN_STREAM_HELLO);
202 + return PARSE_ERROR;
203 + }
204 + return GIMME_MORE_OF_DEM_SWEET_BYTEZ;
205 + }
206 +
207 + *user_agent = mallocz(idx + 1);
208 + rbuf_pop(buf, *user_agent, idx);
209 + (*user_agent)[idx] = 0;
210 +
211 + *parser_state = HTTP_HDR;
212 + /* FALLTHROUGH */
213 + case HTTP_HDR:
214 + if (!rbuf_find_bytes(buf, HTTP_HDR_END, strlen(HTTP_HDR_END), &idx)) {
215 + if (rbuf_bytes_available(buf) >= (size_t)(rbuf_get_capacity(buf) * 0.9)) {
216 + error_report("The initial \"STREAM [URL]" HTTP_1_1 "\" over max of %d", MAX_LEN_STREAM_HELLO);
217 + return PARSE_ERROR;
218 + }
219 + return GIMME_MORE_OF_DEM_SWEET_BYTEZ;
220 + }
221 + rbuf_bump_tail(buf, idx + strlen(HTTP_HDR_END));
222 +
223 + *parser_state = HTTP_DONE;
224 + return PARSE_DONE;
225 + case HTTP_DONE:
226 + error_report("Parsing is done. No need to call again.");
227 + return PARSE_DONE;
228 + default:
229 + error_report("Unknown parser state %d", (int)*parser_state);
230 + return PARSE_ERROR;
231 + }
232 +}
233 +
234 +#define SINGLE_WRITE_MAX (1024)
235 +
236 +void stream_process(h2o_stream_conn_t *conn, int initial)
237 +{
238 + int rc;
239 + struct web_client w;
240 +
241 + pthread_mutex_lock(&conn->tx_buf_lock);
242 + if (h2o_socket_is_writing(conn->sock) || rbuf_bytes_available(conn->tx)) {
243 + if (rbuf_bytes_available(conn->tx) && !conn->tx_buf.base) {
244 + conn->tx_buf.base = rbuf_get_linear_read_range(conn->tx, &conn->tx_buf.len);
245 + if (conn->tx_buf.base) {
246 + conn->tx_buf.len = MIN(conn->tx_buf.len, SINGLE_WRITE_MAX);
247 + h2o_socket_write(conn->sock, &conn->tx_buf, 1, on_write_complete);
248 + }
249 + }
250 + }
251 + pthread_mutex_unlock(&conn->tx_buf_lock);
252 +
253 + if (initial)
254 + h2o_socket_read_start(conn->sock, stream_on_recv);
255 +
256 + if (conn->sock->input->size) {
257 + size_t insert_max;
258 + pthread_mutex_lock(&conn->rx_buf_lock);
259 + char *insert_loc = rbuf_get_linear_insert_range(conn->rx, &insert_max);
260 + if (insert_loc == NULL) {
261 + pthread_cond_broadcast(&conn->rx_buf_cond);
262 + pthread_mutex_unlock(&conn->rx_buf_lock);
263 + return;
264 + }
265 + insert_max = MIN(insert_max, conn->sock->input->size);
266 + memcpy(insert_loc, conn->sock->input->bytes, insert_max);
267 + rbuf_bump_head(conn->rx, insert_max);
268 +
269 + h2o_buffer_consume(&conn->sock->input, insert_max);
270 +
271 + pthread_cond_broadcast(&conn->rx_buf_cond);
272 + pthread_mutex_unlock(&conn->rx_buf_lock);
273 + }
274 +
275 + switch (conn->state) {
276 + case STREAM_X_HTTP_1_1:
277 + // no conn->rx lock here as at this point we are still single threaded
278 + // until we call rrdpush_receiver_thread_spawn() later down
279 + rc = process_STREAM_X_HTTP_1_1(&conn->parse_state, conn->rx, &conn->url, &conn->user_agent);
280 + if (rc == PARSE_ERROR) {
281 + error_report("error parsing the STREAM hello");
282 + break;
283 + }
284 + if (rc != PARSE_DONE)
285 + break;
286 + conn->state = STREAM_X_HTTP_1_1_DONE;
287 + /* FALLTHROUGH */
288 + case STREAM_X_HTTP_1_1_DONE:
289 + memset(&w, 0, sizeof(w));
290 + w.response.data = buffer_create(1024, NULL);
291 +
292 + // get client ip from the conn->sock
293 + struct sockaddr client;
294 + socklen_t len = h2o_socket_getpeername(conn->sock, &client);
295 + char peername[NI_MAXHOST];
296 + size_t peername_len = h2o_socket_getnumerichost(&client, len, peername);
297 + memcpy(w.client_ip, peername, peername_len);
298 + w.client_ip[peername_len] = 0;
299 + w.user_agent = conn->user_agent;
300 +
301 + rc = rrdpush_receiver_thread_spawn(&w, conn->url, conn);
302 + if (rc != HTTP_RESP_OK) {
303 + error_report("HTTPD Failed to spawn the receiver thread %d", rc);
304 + conn->state = STREAM_CLOSE;
305 + stream_on_close(conn);
306 + } else {
307 + conn->state = STREAM_ACTIVE;
308 + }
309 + buffer_free(w.response.data);
310 + /* FALLTHROUGH */
311 + case STREAM_ACTIVE:
312 + break;
313 + default:
314 + error_report("Unknown conn->state");
315 + }
316 +}
317 +
318 +// read and write functions to be used by streaming parser
319 +int h2o_stream_write(void *ctx, const char *data, size_t data_len)
320 +{
321 + h2o_stream_conn_t *conn = (h2o_stream_conn_t *)ctx;
322 +
323 + pthread_mutex_lock(&conn->tx_buf_lock);
324 + size_t avail = rbuf_bytes_free(conn->tx);
325 + avail = MIN(avail, data_len);
326 + rbuf_push(conn->tx, data, avail);
327 + pthread_mutex_unlock(&conn->tx_buf_lock);
328 + __atomic_add_fetch(&pending_write_reqs, 1, __ATOMIC_SEQ_CST);
329 + return avail;
330 +}
331 +
332 +size_t h2o_stream_read(void *ctx, char *buf, size_t read_bytes)
333 +{
334 + int ret;
335 + h2o_stream_conn_t *conn = (h2o_stream_conn_t *)ctx;
336 +
337 + pthread_mutex_lock(&conn->rx_buf_lock);
338 + size_t avail = rbuf_bytes_available(conn->rx);
339 +
340 + if (!avail) {
341 + if (conn->shutdown) {
342 + pthread_mutex_unlock(&conn->rx_buf_lock);
343 + return -1;
344 + }
345 + pthread_cond_wait(&conn->rx_buf_cond, &conn->rx_buf_lock);
346 + if (conn->shutdown) {
347 + pthread_mutex_unlock(&conn->rx_buf_lock);
348 + return -1;
349 + }
350 + avail = rbuf_bytes_available(conn->rx);
351 + if (!avail) {
352 + pthread_mutex_unlock(&conn->rx_buf_lock);
353 + return 0;
354 + }
355 + }
356 +
357 + avail = MIN(avail, read_bytes);
358 +
359 + ret = rbuf_pop(conn->rx, buf, avail);
360 + pthread_mutex_unlock(&conn->rx_buf_lock);
361 +
362 + return ret;
363 +}
364 +
365 +// periodic check for pending write requests
366 +void check_tx_buf(h2o_stream_conn_t *conn)
367 +{
368 + pthread_mutex_lock(&conn->tx_buf_lock);
369 + if (rbuf_bytes_available(conn->tx)) {
370 + pthread_mutex_unlock(&conn->tx_buf_lock);
371 + stream_process(conn, 0);
372 + } else
373 + pthread_mutex_unlock(&conn->tx_buf_lock);
374 +}
375 +
376 +void h2o_stream_check_pending_write_reqs(void)
377 +{
378 + int _write_reqs = __atomic_exchange_n(&pending_write_reqs, 0, __ATOMIC_SEQ_CST);
379 + if (_write_reqs > 0)
380 + conn_list_iter_all(&conn_list, check_tx_buf);
381 +}
web/server/h2o/streaming.h new
+61
@@ -0,0 +1,61 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef HTTPD_STREAMING_H
4 +#define HTTPD_STREAMING_H
5 +
6 +#include "daemon/common.h"
7 +#include "mqtt_websockets/c-rbuf/include/ringbuffer.h"
8 +#include "h2o.h"
9 +
10 +typedef enum {
11 + STREAM_X_HTTP_1_1 = 0,
12 + STREAM_X_HTTP_1_1_DONE,
13 + STREAM_ACTIVE,
14 + STREAM_CLOSE
15 +} h2o_stream_state_t;
16 +
17 +typedef enum {
18 + HTTP_STREAM = 0,
19 + HTTP_URL,
20 + HTTP_PROTO,
21 + HTTP_USER_AGENT_KEY,
22 + HTTP_USER_AGENT_VALUE,
23 + HTTP_HDR,
24 + HTTP_DONE
25 +} http_stream_parse_state_t;
26 +
27 +typedef struct {
28 + h2o_socket_t *sock;
29 + h2o_stream_state_t state;
30 +
31 + rbuf_t rx;
32 + pthread_cond_t rx_buf_cond;
33 + pthread_mutex_t rx_buf_lock;
34 +
35 + rbuf_t tx;
36 + h2o_iovec_t tx_buf;
37 + pthread_mutex_t tx_buf_lock;
38 +
39 + http_stream_parse_state_t parse_state;
40 + char *url;
41 + char *user_agent;
42 +
43 + int shutdown;
44 +} h2o_stream_conn_t;
45 +
46 +// h2o_stream_conn_t related functions
47 +void h2o_stream_conn_t_init(h2o_stream_conn_t *conn);
48 +void h2o_stream_conn_t_destroy(h2o_stream_conn_t *conn);
49 +
50 +// streaming upgrade related functions
51 +int is_streaming_handshake(h2o_req_t *req);
52 +void stream_on_complete(void *user_data, h2o_socket_t *sock, size_t reqsize);
53 +
54 +// read and write functions to be used by streaming parser
55 +int h2o_stream_write(void *ctx, const char *data, size_t data_len);
56 +size_t h2o_stream_read(void *ctx, char *buf, size_t read_bytes);
57 +
58 +// call this periodically to check if there are any pending write requests
59 +void h2o_stream_check_pending_write_reqs(void);
60 +
61 +#endif /* HTTPD_STREAMING_H */
web/server/web_client.c
+1 -1
@@ -1743,7 +1743,7 @@ void web_client_process_request(struct web_client *w) {
1743 return;
1744 }
1745
1746 - w->response.code = rrdpush_receiver_thread_spawn(w, (char *)buffer_tostring(w->url_query_string_decoded));
1746 + w->response.code = rrdpush_receiver_thread_spawn(w, (char *)buffer_tostring(w->url_query_string_decoded), NULL);
1747 return;
1748
1749 case WEB_CLIENT_MODE_OPTIONS: