| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #define WEB_SERVER_INTERNALS 1 |
| 4 | #include "static-threaded.h" |
| 5 | |
| 6 | int web_client_timeout = DEFAULT_DISCONNECT_IDLE_WEB_CLIENTS_AFTER_SECONDS; |
| 7 | int web_client_first_request_timeout = DEFAULT_TIMEOUT_TO_RECEIVE_FIRST_WEB_REQUEST; |
| 8 | long web_client_streaming_rate_t = 0L; |
| 9 | |
| 10 | #define WORKER_JOB_ADD_CONNECTION 0 |
| 11 | #define WORKER_JOB_DEL_COLLECTION 1 |
| 12 | #define WORKER_JOB_ADD_FILE 2 |
| 13 | #define WORKER_JOB_DEL_FILE 3 |
| 14 | #define WORKER_JOB_READ_FILE 4 |
| 15 | #define WORKER_JOB_WRITE_FILE 5 |
| 16 | #define WORKER_JOB_RCV_DATA 6 |
| 17 | #define WORKER_JOB_SND_DATA 7 |
| 18 | #define WORKER_JOB_PROCESS 8 |
| 19 | |
| 20 | #if (WORKER_UTILIZATION_MAX_JOB_TYPES < 9) |
| 21 | #error Please increase WORKER_UTILIZATION_MAX_JOB_TYPES to at least 8 |
| 22 | #endif |
| 23 | |
| 24 | /* |
| 25 | * -------------------------------------------------------------------------------------------------------------------- |
| 26 | * Build web_client state from the pollinfo that describes an accepted connection. |
| 27 | */ |
| 28 | static struct web_client *web_client_create_on_fd(POLLINFO *pi) { |
| 29 | struct web_client *w; |
| 30 | |
| 31 | pulse_web_client_connected(); |
| 32 | w = web_client_get_from_cache(); |
| 33 | w->fd = pi->fd; |
| 34 | |
| 35 | strncpyz(w->user_auth.client_ip, pi->client_ip, sizeof(w->user_auth.client_ip) - 1); |
| 36 | strncpyz(w->client_port, pi->client_port, sizeof(w->client_port) - 1); |
| 37 | strncpyz(w->client_host, pi->client_host, sizeof(w->client_host) - 1); |
| 38 | |
| 39 | if(unlikely(!*w->user_auth.client_ip)) strcpy(w->user_auth.client_ip, "-"); |
| 40 | if(unlikely(!*w->client_port)) strcpy(w->client_port, "-"); |
| 41 | w->port_acl = pi->port_acl; |
| 42 | |
| 43 | int flag = 1; |
| 44 | if(unlikely( |
| 45 | web_client_check_conn_tcp(w) && setsockopt(w->fd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)) != 0)) |
| 46 | netdata_log_debug(D_WEB_CLIENT, "%llu: failed to enable TCP_NODELAY on socket fd %d.", w->id, w->fd); |
| 47 | |
| 48 | flag = 1; |
| 49 | if(unlikely(setsockopt(w->fd, SOL_SOCKET, SO_KEEPALIVE, (char *) &flag, sizeof(int)) != 0)) |
| 50 | netdata_log_debug(D_WEB_CLIENT, "%llu: failed to enable SO_KEEPALIVE on socket fd %d.", w->id, w->fd); |
| 51 | |
| 52 | web_client_update_acl_matches(w); |
| 53 | web_client_enable_wait_receive(w); |
| 54 | |
| 55 | web_server_log_connection(w, "CONNECTED"); |
| 56 | |
| 57 | return(w); |
| 58 | } |
| 59 | |
| 60 | // -------------------------------------------------------------------------------------- |
| 61 | // the main socket listener - STATIC-THREADED |
| 62 | |
| 63 | struct web_server_static_threaded_worker { |
| 64 | ND_THREAD *thread; |
| 65 | |
| 66 | int id; |
| 67 | bool initializing; |
| 68 | SPINLOCK spinlock; |
| 69 | |
| 70 | size_t max_sockets; |
| 71 | |
| 72 | volatile size_t connected; |
| 73 | volatile size_t disconnected; |
| 74 | volatile size_t receptions; |
| 75 | volatile size_t sends; |
| 76 | volatile size_t max_concurrent; |
| 77 | }; |
| 78 | |
| 79 | static long long static_threaded_workers_count = 1; |
| 80 | |
| 81 | static struct web_server_static_threaded_worker *static_workers_private_data = NULL; |
| 82 | static __thread struct web_server_static_threaded_worker *worker_private = NULL; |
| 83 | |
| 84 | // ---------------------------------------------------------------------------- |
| 85 | |
| 86 | static inline int web_server_check_client_status(struct web_client *w) { |
| 87 | if(unlikely(web_client_check_dead(w) || (!web_client_has_wait_receive(w) && !web_client_has_wait_send(w)))) |
| 88 | return -1; |
| 89 | |
| 90 | return 0; |
| 91 | } |
| 92 | |
| 93 | // ---------------------------------------------------------------------------- |
| 94 | // web server clients |
| 95 | |
| 96 | static void *web_server_add_callback(POLLINFO *pi, nd_poll_event_t *events, void *data __maybe_unused) { |
| 97 | worker_is_busy(WORKER_JOB_ADD_CONNECTION); |
| 98 | worker_private->connected++; |
| 99 | |
| 100 | size_t concurrent = worker_private->connected - worker_private->disconnected; |
| 101 | if(unlikely(concurrent > worker_private->max_concurrent)) |
| 102 | worker_private->max_concurrent = concurrent; |
| 103 | |
| 104 | *events = ND_POLL_READ; |
| 105 | |
| 106 | netdata_log_debug(D_WEB_CLIENT_ACCESS, "LISTENER on %d: new connection.", pi->fd); |
| 107 | struct web_client *w = web_client_create_on_fd(pi); |
| 108 | |
| 109 | if (!strncmp(pi->client_port, "UNIX", 4)) { |
| 110 | web_client_set_conn_unix(w); |
| 111 | } else { |
| 112 | web_client_set_conn_tcp(w); |
| 113 | } |
| 114 | |
| 115 | if ((web_client_check_conn_tcp(w)) && (netdata_ssl_web_server_ctx)) { |
| 116 | sock_setnonblock(w->fd, false); |
| 117 | |
| 118 | //Read the first 7 bytes from the message, but the message |
| 119 | //is not removed from the queue, because we are using MSG_PEEK |
| 120 | char test[8]; |
| 121 | if ( recv(w->fd,test, 7, MSG_PEEK) == 7 ) { |
| 122 | test[7] = '\0'; |
| 123 | } |
| 124 | else { |
| 125 | // we couldn't read 7 bytes |
| 126 | sock_setnonblock(w->fd, true); |
| 127 | goto cleanup; |
| 128 | } |
| 129 | |
| 130 | if(test[0] > 0x17) { |
| 131 | // no SSL |
| 132 | netdata_ssl_close(&w->ssl); // free any previous SSL data |
| 133 | } |
| 134 | else { |
| 135 | // SSL |
| 136 | if(!netdata_ssl_open(&w->ssl, netdata_ssl_web_server_ctx, w->fd) || !netdata_ssl_accept(&w->ssl)) |
| 137 | WEB_CLIENT_IS_DEAD(w); |
| 138 | } |
| 139 | |
| 140 | sock_setnonblock(w->fd, true); |
| 141 | } |
| 142 | |
| 143 | netdata_log_debug(D_WEB_CLIENT, "%llu: ADDED CLIENT FD %d", w->id, pi->fd); |
| 144 | |
| 145 | cleanup: |
| 146 | worker_is_idle(); |
| 147 | return w; |
| 148 | } |
| 149 | |
| 150 | // TCP client disconnected |
| 151 | static void web_server_del_callback(POLLINFO *pi) { |
| 152 | worker_is_busy(WORKER_JOB_DEL_COLLECTION); |
| 153 | |
| 154 | worker_private->disconnected++; |
| 155 | |
| 156 | struct web_client *w = (struct web_client *)pi->data; |
| 157 | |
| 158 | if(web_client_flag_check(w, WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET)) |
| 159 | pi->flags |= POLLINFO_FLAG_DONT_CLOSE; |
| 160 | |
| 161 | netdata_log_debug(D_WEB_CLIENT, "%llu: CLOSING CLIENT FD %d", w->id, pi->fd); |
| 162 | web_server_log_connection(w, "DISCONNECTED"); |
| 163 | web_client_request_done(w); |
| 164 | web_client_release_to_cache(w); |
| 165 | pulse_web_client_disconnected(); |
| 166 | |
| 167 | worker_is_idle(); |
| 168 | } |
| 169 | |
| 170 | static __thread POLLINFO *current_thread_pollinfo = NULL; |
| 171 | |
| 172 | void web_server_remove_current_socket_from_poll(void) { |
| 173 | if(!current_thread_pollinfo) return; |
| 174 | poll_process_remove_from_poll(current_thread_pollinfo); |
| 175 | } |
| 176 | |
| 177 | static int web_server_rcv_callback(POLLINFO *pi, nd_poll_event_t *events) { |
| 178 | int ret = -1; |
| 179 | worker_is_busy(WORKER_JOB_RCV_DATA); |
| 180 | |
| 181 | worker_private->receptions++; |
| 182 | |
| 183 | struct web_client *w = (struct web_client *)pi->data; |
| 184 | int fd = pi->fd; |
| 185 | |
| 186 | ssize_t bytes; |
| 187 | bytes = web_client_receive(w); |
| 188 | |
| 189 | if (likely(bytes > 0)) { |
| 190 | pulse_web_server_received_bytes(bytes); |
| 191 | |
| 192 | netdata_log_debug(D_WEB_CLIENT, "%llu: processing received data on fd %d.", w->id, fd); |
| 193 | worker_is_idle(); |
| 194 | worker_is_busy(WORKER_JOB_PROCESS); |
| 195 | current_thread_pollinfo = pi; |
| 196 | web_client_process_request_from_web_server(w); |
| 197 | current_thread_pollinfo = NULL; |
| 198 | |
| 199 | // Request processing may block for long-running functions. |
| 200 | // Refresh receive timestamp so idle timeout uses the actual return time. |
| 201 | pi->last_received_t = now_boottime_sec(); |
| 202 | |
| 203 | // The first-request timeout protects request ingress only. |
| 204 | // Once we no longer wait to receive request bytes, the first request is complete. |
| 205 | if(unlikely(!(pi->flags & POLLINFO_FLAG_FIRST_REQUEST_RECEIVED) && !web_client_has_wait_receive(w))) |
| 206 | pi->flags |= POLLINFO_FLAG_FIRST_REQUEST_RECEIVED; |
| 207 | |
| 208 | if (unlikely(w->mode == HTTP_REQUEST_MODE_STREAM)) { |
| 209 | ssize_t rc = web_client_send(w); |
| 210 | if(rc > 0) |
| 211 | pulse_web_server_sent_bytes(rc); |
| 212 | } |
| 213 | else if(unlikely(w->fd == fd && web_client_has_wait_receive(w))) |
| 214 | *events |= ND_POLL_READ; |
| 215 | |
| 216 | if(unlikely(w->fd == fd && web_client_has_wait_send(w))) |
| 217 | *events |= ND_POLL_WRITE; |
| 218 | |
| 219 | } else if(unlikely(bytes < 0)) { |
| 220 | ret = -1; |
| 221 | goto cleanup; |
| 222 | } else if (unlikely(bytes == 0)) { |
| 223 | if(unlikely(w->fd == fd && web_client_has_ssl_wait_receive(w))) |
| 224 | *events |= ND_POLL_READ; |
| 225 | |
| 226 | if(unlikely(w->fd == fd && web_client_has_ssl_wait_send(w))) |
| 227 | *events |= ND_POLL_WRITE; |
| 228 | } |
| 229 | |
| 230 | ret = web_server_check_client_status(w); |
| 231 | |
| 232 | cleanup: |
| 233 | worker_is_idle(); |
| 234 | return ret; |
| 235 | } |
| 236 | |
| 237 | static int web_server_snd_callback(POLLINFO *pi, nd_poll_event_t *events) { |
| 238 | int retval = -1; |
| 239 | worker_is_busy(WORKER_JOB_SND_DATA); |
| 240 | |
| 241 | worker_private->sends++; |
| 242 | |
| 243 | struct web_client *w = (struct web_client *)pi->data; |
| 244 | int fd = pi->fd; |
| 245 | |
| 246 | netdata_log_debug(D_WEB_CLIENT, "%llu: sending data on fd %d.", w->id, fd); |
| 247 | |
| 248 | current_thread_pollinfo = pi; |
| 249 | ssize_t ret = web_client_send(w); |
| 250 | current_thread_pollinfo = NULL; |
| 251 | |
| 252 | if(unlikely(ret < 0)) { |
| 253 | retval = -1; |
| 254 | goto cleanup; |
| 255 | } |
| 256 | |
| 257 | pulse_web_server_sent_bytes(ret); |
| 258 | |
| 259 | if(unlikely(w->fd == fd && web_client_has_wait_receive(w))) |
| 260 | *events |= ND_POLL_READ; |
| 261 | |
| 262 | if(unlikely(w->fd == fd && web_client_has_wait_send(w))) |
| 263 | *events |= ND_POLL_WRITE; |
| 264 | |
| 265 | retval = web_server_check_client_status(w); |
| 266 | |
| 267 | cleanup: |
| 268 | worker_is_idle(); |
| 269 | return retval; |
| 270 | } |
| 271 | |
| 272 | // ---------------------------------------------------------------------------- |
| 273 | // web server worker thread |
| 274 | |
| 275 | static void socket_listen_main_static_threaded_worker_cleanup(void *pptr) { |
| 276 | worker_private = CLEANUP_FUNCTION_GET_PTR(pptr); |
| 277 | if(!worker_private) return; |
| 278 | |
| 279 | netdata_log_info("stopped after %zu connects, %zu disconnects (max concurrent %zu), %zu receptions and %zu sends", |
| 280 | worker_private->connected, |
| 281 | worker_private->disconnected, |
| 282 | worker_private->max_concurrent, |
| 283 | worker_private->receptions, |
| 284 | worker_private->sends |
| 285 | ); |
| 286 | |
| 287 | worker_unregister(); |
| 288 | } |
| 289 | |
| 290 | static bool web_server_should_stop(void) { |
| 291 | return !service_running(SERVICE_WEB_SERVER); |
| 292 | } |
| 293 | |
| 294 | void socket_listen_main_static_threaded_worker(void *ptr) { |
| 295 | worker_private = ptr; |
| 296 | spinlock_lock(&worker_private->spinlock); |
| 297 | worker_private->initializing = false; |
| 298 | spinlock_unlock(&worker_private->spinlock); |
| 299 | worker_register("WEB"); |
| 300 | worker_register_job_name(WORKER_JOB_ADD_CONNECTION, "connect"); |
| 301 | worker_register_job_name(WORKER_JOB_DEL_COLLECTION, "disconnect"); |
| 302 | worker_register_job_name(WORKER_JOB_ADD_FILE, "file start"); |
| 303 | worker_register_job_name(WORKER_JOB_DEL_FILE, "file end"); |
| 304 | worker_register_job_name(WORKER_JOB_READ_FILE, "file read"); |
| 305 | worker_register_job_name(WORKER_JOB_WRITE_FILE, "file write"); |
| 306 | worker_register_job_name(WORKER_JOB_RCV_DATA, "receive"); |
| 307 | worker_register_job_name(WORKER_JOB_SND_DATA, "send"); |
| 308 | worker_register_job_name(WORKER_JOB_PROCESS, "process"); |
| 309 | |
| 310 | CLEANUP_FUNCTION_REGISTER(socket_listen_main_static_threaded_worker_cleanup) cleanup_ptr = worker_private; |
| 311 | poll_events(&api_sockets |
| 312 | , web_server_add_callback |
| 313 | , web_server_del_callback |
| 314 | , web_server_rcv_callback |
| 315 | , web_server_snd_callback |
| 316 | , NULL |
| 317 | , web_server_should_stop |
| 318 | , web_allow_connections_from |
| 319 | , web_allow_connections_dns |
| 320 | , NULL |
| 321 | , web_client_first_request_timeout |
| 322 | , web_client_timeout |
| 323 | , nd_profile.update_every * 1000 // timer_milliseconds |
| 324 | , ptr // timer_data |
| 325 | , worker_private->max_sockets |
| 326 | ); |
| 327 | } |
| 328 | |
| 329 | |
| 330 | // ---------------------------------------------------------------------------- |
| 331 | // web server main thread - also becomes a worker |
| 332 | |
| 333 | static void socket_listen_main_static_threaded_cleanup(void *pptr) { |
| 334 | struct netdata_static_thread *static_thread = CLEANUP_FUNCTION_GET_PTR(pptr); |
| 335 | if(!static_thread) return; |
| 336 | |
| 337 | static_thread->enabled = NETDATA_MAIN_THREAD_EXITING; |
| 338 | |
| 339 | netdata_log_info("closing all web server sockets..."); |
| 340 | listen_sockets_close(&api_sockets); |
| 341 | |
| 342 | netdata_log_info("all static web threads stopped."); |
| 343 | |
| 344 | // Lets join all threads |
| 345 | for (int i = 1; i < static_threaded_workers_count; i++) { |
| 346 | bool initializing; |
| 347 | do { |
| 348 | spinlock_lock(&static_workers_private_data[i].spinlock); |
| 349 | initializing = static_workers_private_data[i].initializing; |
| 350 | spinlock_unlock(&static_workers_private_data[i].spinlock); |
| 351 | if (unlikely(initializing)) |
| 352 | sleep_usec(1000); |
| 353 | } while(initializing); |
| 354 | (void) nd_thread_join(static_workers_private_data[i].thread); |
| 355 | } |
| 356 | |
| 357 | static_thread->enabled = NETDATA_MAIN_THREAD_EXITED; |
| 358 | } |
| 359 | |
| 360 | void socket_listen_main_static_threaded(void *ptr) { |
| 361 | CLEANUP_FUNCTION_REGISTER(socket_listen_main_static_threaded_cleanup) cleanup_ptr = ptr; |
| 362 | web_server_mode = WEB_SERVER_MODE_STATIC_THREADED; |
| 363 | |
| 364 | if(!api_sockets.opened) |
| 365 | fatal("LISTENER: no listen sockets available."); |
| 366 | |
| 367 | netdata_ssl_validate_certificate = !inicfg_get_boolean(&netdata_config, CONFIG_SECTION_WEB, "ssl skip certificate verification", !netdata_ssl_validate_certificate); |
| 368 | |
| 369 | if(!netdata_ssl_validate_certificate_sender) |
| 370 | netdata_log_info("SSL: web server will skip SSL certificates verification."); |
| 371 | |
| 372 | netdata_ssl_initialize_ctx(NETDATA_SSL_WEB_SERVER_CTX); |
| 373 | |
| 374 | static_threaded_workers_count = netdata_conf_web_query_threads(); |
| 375 | |
| 376 | size_t max_sockets = (size_t)inicfg_get_number(&netdata_config, CONFIG_SECTION_WEB, "web server max sockets", |
| 377 | (long long int)(rlimit_nofile.rlim_cur / 4)); |
| 378 | |
| 379 | static_workers_private_data = callocz((size_t)static_threaded_workers_count, |
| 380 | sizeof(struct web_server_static_threaded_worker)); |
| 381 | |
| 382 | int i; |
| 383 | spinlock_init(&static_workers_private_data[0].spinlock); |
| 384 | static_workers_private_data[0].initializing = true; |
| 385 | for (i = 1; i < static_threaded_workers_count; i++) { |
| 386 | static_workers_private_data[i].id = i; |
| 387 | static_workers_private_data[i].max_sockets = max_sockets / static_threaded_workers_count; |
| 388 | |
| 389 | char tag[50 + 1]; |
| 390 | snprintfz(tag, sizeof(tag) - 1, "WEB[%d]", i+1); |
| 391 | |
| 392 | spinlock_init(&static_workers_private_data[i].spinlock); |
| 393 | static_workers_private_data[i].initializing = true; |
| 394 | static_workers_private_data[i].thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT, |
| 395 | socket_listen_main_static_threaded_worker, |
| 396 | (void *)&static_workers_private_data[i]); |
| 397 | } |
| 398 | |
| 399 | // and the main one |
| 400 | static_workers_private_data[0].max_sockets = max_sockets / static_threaded_workers_count; |
| 401 | socket_listen_main_static_threaded_worker((void *)&static_workers_private_data[0]); |
| 402 | } |