@cryptotaxi247 / netdata-1 / commits / 1bb3047ee

Reduce connection timeout and fallback to IPV4 for ACLK connections (#18568)

Stelios Fragkakis committed Sep 18, 2024 at 17:31 UTC 1bb3047ee60a46cf210b16c563b15be4fafc1045

9 files changed +62 -28

src/aclk/aclk.c

+5 -3

@@ -566,6 +566,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)

566 url_t mqtt_url;

567 #endif

568

569 + bool fallback_ipv4 = false;

570 while (service_running(SERVICE_ACLK)) {

571 aclk_cloud_base_url = cloud_config_url_get();

572 if (aclk_cloud_base_url == NULL) {

@@ -612,7 +613,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)

613 }

614 aclk_env = callocz(1, sizeof(aclk_env_t));

615

615 - ret = aclk_get_env(aclk_env, base_url.host, base_url.port);

616 + ret = aclk_get_env(aclk_env, base_url.host, base_url.port, &fallback_ipv4);

617 url_t_destroy(&base_url);

618 if(ret) switch(ret) {

619 case 1:

@@ -680,7 +681,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)

681 continue;

682 }

683

683 - ret = aclk_get_mqtt_otp(aclk_private_key, (char **)&mqtt_conn_params.clientid, (char **)&mqtt_conn_params.username, (char **)&mqtt_conn_params.password, &auth_url);

684 + ret = aclk_get_mqtt_otp(aclk_private_key, (char **)&mqtt_conn_params.clientid, (char **)&mqtt_conn_params.username, (char **)&mqtt_conn_params.password, &auth_url, &fallback_ipv4);

685 url_t_destroy(&auth_url);

686 if (ret) {

687 aclk_status = ACLK_STATUS_INVALID_OTP;

@@ -725,7 +726,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)

726 ret = mqtt_wss_connect(client, base_url.host, base_url.port, &mqtt_conn_params, ACLK_SSL_FLAGS, &proxy_conf);

727 url_t_destroy(&base_url);

728 #else

728 - ret = mqtt_wss_connect(client, mqtt_url.host, mqtt_url.port, &mqtt_conn_params, ACLK_SSL_FLAGS, &proxy_conf);

729 + ret = mqtt_wss_connect(client, mqtt_url.host, mqtt_url.port, &mqtt_conn_params, ACLK_SSL_FLAGS, &proxy_conf, &fallback_ipv4);

730 url_t_destroy(&mqtt_url);

731

732 freez((char*)mqtt_conn_params.clientid);

@@ -744,6 +745,7 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)

745 aclk_status = ACLK_STATUS_CONNECTED;

746 nd_log(NDLS_ACCESS, NDLP_INFO, "ACLK CONNECTED");

747 mqtt_connected_actions(client);

748 + fallback_ipv4 = false;

749 return 0;

750 }

751

src/aclk/aclk_otp.c

+12 -12

@@ -4,7 +4,7 @@

4 #include "aclk_util.h"

5 #include "aclk.h"

7 -static int aclk_https_request(https_req_t *request, https_req_response_t *response) {

7 +static int aclk_https_request(https_req_t *request, https_req_response_t *response, bool *fallback_ipv4) {

8 int rc;

9 // wrapper for ACLK only which loads ACLK specific proxy settings

10 // then only calls https_request

@@ -18,7 +18,7 @@ static int aclk_https_request(https_req_t *request, https_req_response_t *respon

18 request->proxy_password = proxy_conf.password;

19 }

21 - rc = https_request(request, response);

21 + rc = https_request(request, response, fallback_ipv4);

22 freez((char*)proxy_conf.host);

23 freez((char*)proxy_conf.username);

24 freez((char*)proxy_conf.password);

@@ -303,7 +303,7 @@ inline static int base64_decode_helper(unsigned char *out, int *outl, const unsi

303 }

304

305 #define OTP_URL_PREFIX "/api/v1/auth/node/"

306 -int aclk_get_otp_challenge(url_t *target, const char *agent_id, unsigned char **challenge, int *challenge_bytes)

306 +int aclk_get_otp_challenge(url_t *target, const char *agent_id, unsigned char **challenge, int *challenge_bytes, bool *fallback_ipv4)

307 {

308 int rc = 1;

309 https_req_t req = HTTPS_REQ_T_INITIALIZER;

@@ -316,7 +316,7 @@ int aclk_get_otp_challenge(url_t *target, const char *agent_id, unsigned char **

316 buffer_sprintf(url, "%s/node/%s/challenge", target->path, agent_id);

317 req.url = (char *)buffer_tostring(url);

318

319 - if (aclk_https_request(&req, &resp)) {

319 + if (aclk_https_request(&req, &resp, fallback_ipv4)) {

320 netdata_log_error("ACLK_OTP Challenge failed");

321 buffer_free(url);

322 return 1;

@@ -373,7 +373,7 @@ cleanup_resp:

373 return rc;

374 }

375

376 -int aclk_send_otp_response(const char *agent_id, const unsigned char *response, int response_bytes, url_t *target, struct auth_data *mqtt_auth)

376 +int aclk_send_otp_response(const char *agent_id, const unsigned char *response, int response_bytes, url_t *target, struct auth_data *mqtt_auth, bool *fallback_ipv4)

377 {

378 int len;

379 int rc = 1;

@@ -399,7 +399,7 @@ int aclk_send_otp_response(const char *agent_id, const unsigned char *response,

399 req.payload = (char *)buffer_tostring(resp_json);

400 req.payload_size = strlen(req.payload);

401

402 - if (aclk_https_request(&req, &resp)) {

402 + if (aclk_https_request(&req, &resp, fallback_ipv4)) {

403 netdata_log_error("ACLK_OTP Password error trying to post result to password");

404 goto cleanup_buffers;

405 }

@@ -475,9 +475,9 @@ static int private_decrypt(RSA *p_key, unsigned char * enc_data, int data_len, u

475 }

476

477 #if OPENSSL_VERSION_NUMBER >= OPENSSL_VERSION_300

478 -int aclk_get_mqtt_otp(EVP_PKEY *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target)

478 +int aclk_get_mqtt_otp(EVP_PKEY *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target, bool *fallback_ipv4)

479 #else

480 -int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target)

480 +int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target, bool *fallback_ipv4)

481 #endif

482 {

483 unsigned char *challenge = NULL;

@@ -490,7 +490,7 @@ int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_p

490 }

491

492 // Get Challenge

493 - if (aclk_get_otp_challenge(target, claim_id.str, &challenge, &challenge_bytes)) {

493 + if (aclk_get_otp_challenge(target, claim_id.str, &challenge, &challenge_bytes, fallback_ipv4)) {

494 netdata_log_error("Error getting challenge");

495 return 1;

496 }

@@ -508,7 +508,7 @@ int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_p

508

509 // Encode and Send Challenge

510 struct auth_data data = { .client_id = NULL, .passwd = NULL, .username = NULL };

511 - if (aclk_send_otp_response(claim_id.str, response_plaintext, response_plaintext_bytes, target, &data)) {

511 + if (aclk_send_otp_response(claim_id.str, response_plaintext, response_plaintext_bytes, target, &data, fallback_ipv4)) {

512 netdata_log_error("Error getting response");

513 freez(response_plaintext);

514 return 1;

@@ -814,7 +814,7 @@ exit:

814 return 1;

815 }

816

817 -int aclk_get_env(aclk_env_t *env, const char* aclk_hostname, int aclk_port) {

817 +int aclk_get_env(aclk_env_t *env, const char* aclk_hostname, int aclk_port, bool *fallback_ipv4) {

818 BUFFER *buf = buffer_create(1024, &netdata_buffers_statistics.buffers_aclk);

819

820 https_req_t req = HTTPS_REQ_T_INITIALIZER;

@@ -834,7 +834,7 @@ int aclk_get_env(aclk_env_t *env, const char* aclk_hostname, int aclk_port) {

834 req.host = (char*)aclk_hostname;

835 req.port = aclk_port;

836 req.url = buf->buffer;

837 - if (aclk_https_request(&req, &resp)) {

837 + if (aclk_https_request(&req, &resp, fallback_ipv4)) {

838 netdata_log_error("Error trying to contact env endpoint");

839 https_req_response_free(&resp);

840 buffer_free(buf);

src/aclk/aclk_otp.h

+3 -3

@@ -9,10 +9,10 @@

9 #include "aclk_util.h"

11 #if OPENSSL_VERSION_NUMBER >= OPENSSL_VERSION_300

12 -int aclk_get_mqtt_otp(EVP_PKEY *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target);

12 +int aclk_get_mqtt_otp(EVP_PKEY *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target, bool *fallback_ipv4);

13 #else

14 -int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target);

14 +int aclk_get_mqtt_otp(RSA *p_key, char **mqtt_id, char **mqtt_usr, char **mqtt_pass, url_t *target, bool *fallback_ipv4);

15 #endif

16 -int aclk_get_env(aclk_env_t *env, const char *aclk_hostname, int aclk_port);

16 +int aclk_get_env(aclk_env_t *env, const char *aclk_hostname, int aclk_port, bool *fallback_ipv4);

18 #endif /* ACLK_OTP_H */

src/aclk/https_client.c

+4 -3

@@ -621,13 +621,14 @@ static int cert_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)

621 return preverify_ok;

622 }

623

624 -int https_request(https_req_t *request, https_req_response_t *response) {

624 +int https_request(https_req_t *request, https_req_response_t *response, bool *fallback_ipv4)

625 +{

626 int rc = 1, ret;

627 char connect_port_str[PORT_STR_MAX_BYTES];

628

629 const char *connect_host = request->proxy_host ? request->proxy_host : request->host;

630 int connect_port = request->proxy_host ? request->proxy_port : request->port;

630 - struct timeval timeout = { .tv_sec = request->timeout_s, .tv_usec = 0 };

631 + struct timeval timeout = { .tv_sec = 10, .tv_usec = 0 };

632

633 https_req_ctx_t *ctx = callocz(1, sizeof(https_req_ctx_t));

634 ctx->req_start_time = now_realtime_sec();

@@ -640,7 +641,7 @@ int https_request(https_req_t *request, https_req_response_t *response) {

641

642 snprintfz(connect_port_str, PORT_STR_MAX_BYTES, "%d", connect_port);

643

643 - ctx->sock = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, connect_host, 0, connect_port_str, &timeout);

644 + ctx->sock = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, connect_host, 0, connect_port_str, &timeout, fallback_ipv4);

645 if (ctx->sock < 0) {

646 netdata_log_error("Error connecting TCP socket to \"%s\"", connect_host);

647 goto exit_buf_rx;

src/aclk/https_client.h

+1 -1

@@ -75,7 +75,7 @@ void https_req_response_free(https_req_response_t *res);

75 .proxy_port = 8080 \

76 }

78 -int https_request(https_req_t *request, https_req_response_t *response);

78 +int https_request(https_req_t *request, https_req_response_t *response, bool *fallback_ipv4);

80 // we expose previously internal parser as this is usefull also from

81 // other parts of the code

src/aclk/mqtt_websockets/mqtt_wss_client.c

+5 -2

@@ -510,7 +510,8 @@ int mqtt_wss_connect(

510 int port,

511 struct mqtt_connect_params *mqtt_params,

512 int ssl_flags,

513 - struct mqtt_wss_proxy *proxy)

513 + struct mqtt_wss_proxy *proxy,

514 + bool *fallback_ipv4)

515 {

516 if (!mqtt_params) {

517 mws_error(client->log, "mqtt_params can't be null!");

@@ -566,7 +567,9 @@ int mqtt_wss_connect(

567

568 char port_str[16];

569 snprintf(port_str, sizeof(port_str) -1, "%d", client->port);

569 - int fd = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, client->host, 0, port_str, NULL);

570 +

571 + struct timeval timeout = { .tv_sec = 10, .tv_usec = 0 };

572 + int fd = connect_to_this_ip46(IPPROTO_TCP, SOCK_STREAM, client->host, 0, port_str, &timeout, fallback_ipv4);

573 if (fd < 0) {

574 mws_error(client->log, "Could not connect to remote endpoint \"%s\", port %d.\n", client->host, port);

575 return -3;

src/aclk/mqtt_websockets/mqtt_wss_client.h

+8 -1

@@ -65,7 +65,14 @@ struct mqtt_wss_proxy;

65 * @param mqtt_params pointer to mqtt_connect_params structure which contains MQTT credentials and settings

66 * @param ssl_flags parameters for OpenSSL, 0=MQTT_WSS_SSL_CERT_CHECK_FULL

67 */

68 -int mqtt_wss_connect(mqtt_wss_client client, char *host, int port, struct mqtt_connect_params *mqtt_params, int ssl_flags, struct mqtt_wss_proxy *proxy);

68 +int mqtt_wss_connect(

69 + mqtt_wss_client client,

70 + char *host,

71 + int port,

72 + struct mqtt_connect_params *mqtt_params,

73 + int ssl_flags,

74 + struct mqtt_wss_proxy *proxy,

75 + bool *fallback_ipv4);

76 int mqtt_wss_service(mqtt_wss_client client, int timeout_ms);

77 void mqtt_wss_disconnect(mqtt_wss_client client, int timeout_ms);

src/libnetdata/socket/socket.c

+16 -2

@@ -857,7 +857,15 @@ static inline int connect_to_unix(const char *path, struct timeval *timeout) {

857 // service the service name or port to connect to

858 // timeout the timeout for establishing a connection

859

860 -int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t scope_id, const char *service, struct timeval *timeout) {

860 +int connect_to_this_ip46(

861 + int protocol,

862 + int socktype,

863 + const char *host,

864 + uint32_t scope_id,

865 + const char *service,

866 + struct timeval *timeout,

867 + bool *fallback_ipv4)

868 +{

869 struct addrinfo hints;

870 struct addrinfo *ai_head = NULL, *ai = NULL;

871

@@ -890,6 +898,9 @@ int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t

898 for (ai = ai_head; ai != NULL && fd == -1; ai = ai->ai_next) {

899 if(nd_thread_signaled_to_cancel()) break;

900

901 + if (fallback_ipv4 && *fallback_ipv4 && ai->ai_family == PF_INET6)

902 + continue;

903 +

904 if (ai->ai_family == PF_INET6) {

905 struct sockaddr_in6 *pSadrIn6 = (struct sockaddr_in6 *) ai->ai_addr;

906 if(pSadrIn6->sin6_scope_id == 0) {

@@ -969,6 +980,9 @@ int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t

980

981 close(fd);

982 fd = -1;

983 +

984 + if (fallback_ipv4 && ai->ai_family == PF_INET6)

985 + *fallback_ipv4 = true;

986 break;

987

988 default:

@@ -1090,7 +1104,7 @@ int connect_to_this(const char *definition, int default_port, struct timeval *ti

1104 service = default_service;

1105

1106

1093 - return connect_to_this_ip46(protocol, socktype, host, scope_id, service, timeout);

1107 + return connect_to_this_ip46(protocol, socktype, host, scope_id, service, timeout,NULL);

1108 }

1109

1110 void foreach_entry_in_connection_string(const char *destination, bool (*callback)(char *entry, void *data), void *data) {

src/libnetdata/socket/socket.h

+8 -1

@@ -33,7 +33,14 @@ int listen_sockets_setup(LISTEN_SOCKETS *sockets);

33 void listen_sockets_close(LISTEN_SOCKETS *sockets);

35 void foreach_entry_in_connection_string(const char *destination, bool (*callback)(char *entry, void *data), void *data);

36 -int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t scope_id, const char *service, struct timeval *timeout);

36 +int connect_to_this_ip46(

37 + int protocol,

38 + int socktype,

39 + const char *host,

40 + uint32_t scope_id,

41 + const char *service,

42 + struct timeval *timeout,

43 + bool *fallback_ipv4);

44 int connect_to_this(const char *definition, int default_port, struct timeval *timeout);

45 int connect_to_one_of(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter, char *connected_to, size_t connected_to_size);

46 int connect_to_one_of_urls(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter, char *connected_to, size_t connected_to_size);