Skip to content

Commit

Permalink
Merge pull request #4337 from sysown/v2.x-TAP_ping
Browse files Browse the repository at this point in the history 
V2.x tap ping
  • Loading branch information
@renecannao
renecannao authored Sep 1, 2023
2 parents 94a5aa4 + 0ee88af commit 4d33507
Showing 1 changed file with 87 additions and 29 deletions.
116 changes: 87 additions & 29 deletions test/tap/tests/test_backend_conn_ping-t.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,10 @@ using srv_cfg = vector<pair<string,int>>;

int wait_timeout = 10;

#ifndef SESSIONS_FOR_CONNECTIONS_HANDLER
#define SESSIONS_FOR_CONNECTIONS_HANDLER 64
#endif

// if only 1 worker thread is running, wait_timeout should be bigger
// 1 worker thread : wait_timeout = 45
// 4 worker threads : wait_timeout = 10
Expand Down Expand Up @@ -89,6 +93,7 @@ int change_mysql_cfg(
return EXIT_FAILURE;
}

diag("Connecting to %s:%s with user %s", host.c_str(), port.c_str(), cl.mysql_username);
if (!mysql_real_connect(my_conn, host.c_str(), cl.mysql_username, cl.mysql_password, NULL, std::stol(port.c_str()), NULL, 0)) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(my_conn));
res = EXIT_FAILURE;
Expand All @@ -98,7 +103,9 @@ int change_mysql_cfg(
srv_cfg old_server_config {};

for (const pair<string,int>& config_var : new_srv_cfg) {
res = mysql_query(my_conn, string {"SELECT @@" + config_var.first}.c_str());
string query = "SELECT @@" + config_var.first;
diag("Line:%d : Running: %s", __LINE__, query.c_str());
res = mysql_query(my_conn, query.c_str());
if (res != EXIT_SUCCESS) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(my_conn));
res = EXIT_FAILURE;
Expand All @@ -118,13 +125,14 @@ int change_mysql_cfg(
res = EXIT_FAILURE;
break;
} else {
diag("Line:%d : Returned: %s = %s", __LINE__, config_var.first.c_str(), row[0]);
old_server_config.push_back({ config_var.first, std::stol(row[0]) });
}

mysql_free_result(my_res);

string query = string { "SET GLOBAL " + config_var.first + "=" + std::to_string(config_var.second) };
diag("Setting on %s:%s : %s", host.c_str(), port.c_str(), query.c_str());
query = string { "SET GLOBAL " + config_var.first + "=" + std::to_string(config_var.second) };
diag("Line:%d : Setting on %s:%s : %s", __LINE__ , host.c_str(), port.c_str(), query.c_str());
mysql_query(my_conn, query.c_str());
if (res != EXIT_SUCCESS) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(my_conn));
Expand Down Expand Up @@ -204,6 +212,7 @@ int check_backend_conns(
vector<MYSQL*> mysql_conns {};
int res = EXIT_SUCCESS;

diag("Line:%d : Creating %f connections on hg %d", __LINE__ , test_params.init_batch_size, hg);
for (uint32_t i = 0; i < test_params.init_batch_size; i++) {
int c_res = create_new_backend_conn(cl, hg, mysql_conns);
if (c_res != EXIT_SUCCESS) { return EXIT_FAILURE; }
Expand All @@ -213,6 +222,7 @@ int check_backend_conns(

// 1. Create server connections to monitor
for (uint32_t i = 0; i < test_params.its; i++) {
diag("Line:%d : Creating %f connections on hg %d , iteration %d", __LINE__ , test_params.batch_size, hg, i);
for (uint32_t j = 0; j < test_params.batch_size; j++) {
int c_res = create_new_backend_conn(cl, hg, mysql_conns);
if (c_res != EXIT_SUCCESS) { return EXIT_FAILURE; }
Expand Down Expand Up @@ -261,15 +271,17 @@ int check_backend_conns(
// Reset the previous value
act_mysql_conn_count = 0;

const string mysql_query {
const string mysql_query_string {
"SELECT count(*) FROM information_schema.processlist WHERE"
" COMMAND=\"Sleep\" and USER=\"" + string { cl.username } + "\" and DB=\"backend_conn_ping_test\""
" USER=\"" + string { cl.username } + "\""
//" USER=\"" + string { cl.username } + "\" and DB=\"backend_conn_ping_test\""
//" COMMAND=\"Sleep\" and USER=\"" + string { cl.username } + "\" and DB=\"backend_conn_ping_test\""
};

diag("Line:%d : Running: %s", __LINE__ , mysql_query_string.c_str());
for (MYSQL* mysql : svrs_conns) {
uint64_t tmp_mysql_conn_count = 0;

int q_res = get_query_result(mysql, mysql_query, tmp_mysql_conn_count);
int q_res = get_query_result(mysql, mysql_query_string, tmp_mysql_conn_count);
if (q_res != EXIT_SUCCESS) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, "get_query_result() failed");
break;
Expand Down Expand Up @@ -297,12 +309,22 @@ int check_backend_conns(
" AND schema='backend_conn_ping_test'"
" AND srv_port IN (" + srv_ports + ")"
};
diag("Line:%d : Running: %s", __LINE__ , proxy_query.c_str());
q_res = get_query_result(admin, proxy_query, act_proxy_free_conn_count);
if (q_res != EXIT_SUCCESS) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, "get_query_result() failed");
break;
}

{
string query = "SELECT hostgroup , srv_host , srv_port , status , ConnUsed , ConnFree , ConnOK , ConnERR FROM stats_mysql_connection_pool";
diag("Line:%d : Running: %s", __LINE__ , query.c_str());
MYSQL_QUERY(admin, query.c_str());
MYSQL_RES* result = mysql_store_result(admin);
while (MYSQL_ROW row = mysql_fetch_row(result)) {
diag("%s , %s , %s , %s , %s , %s , %s , %s", row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7]);
}
mysql_free_result(result);
}
const string proxy_used_query {
"SELECT ConnUsed from stats_mysql_connection_pool WHERE hostgroup=" + std::to_string(hg)
};
Expand All @@ -314,20 +336,30 @@ int check_backend_conns(
check_num, exp_conn_count, act_mysql_conn_count, act_proxy_free_conn_count, act_proxy_used_conn_count
);

diag("check_num = %u", check_num);
diag("exp_conn_count = %lu", exp_conn_count);
diag("act_mysql_conn_count = %lu", act_mysql_conn_count);
diag("act_proxy_free_conn_count = %lu", act_proxy_free_conn_count);
diag("act_proxy_used_conn_count = %lu", act_proxy_used_conn_count);

if (
act_mysql_conn_count >= exp_conn_count ||
(act_proxy_free_conn_count + act_proxy_used_conn_count) >= exp_conn_count
(act_proxy_free_conn_count + act_proxy_used_conn_count + SESSIONS_FOR_CONNECTIONS_HANDLER) >= exp_conn_count
) {
break;
}

sleep(intv);
if (intv) {
diag("Line:%d : Sleeping %d" , __LINE__ , intv);
sleep(intv);
}
}

ok(
q_res == EXIT_SUCCESS && act_mysql_conn_count >= exp_conn_count &&
((act_proxy_free_conn_count + act_proxy_used_conn_count) >= exp_conn_count) &&
act_mysql_conn_count == act_proxy_free_conn_count,
q_res == EXIT_SUCCESS && act_mysql_conn_count >= ((float) exp_conn_count * 0.95) // allow 5% margin of error
&&
((act_proxy_free_conn_count + act_proxy_used_conn_count + SESSIONS_FOR_CONNECTIONS_HANDLER) >= exp_conn_count)
//&& act_mysql_conn_count == act_proxy_free_conn_count // they can't be equal
,
"Created server connections should be properly maintained (pinged) by ProxySQL:"
" { ExpConns: %ld, ActMySQLConns: %ld, ActProxyConns: %ld }",
exp_conn_count, act_mysql_conn_count, act_proxy_free_conn_count
Expand All @@ -337,6 +369,7 @@ int check_backend_conns(
// 3. Check that no client side errors take place when exhausting backend connections
{
uint32_t broken_conns = 0;
diag("Line:%d : Running DO 1 on %lu connections", __LINE__, mysql_conns.size());
for (MYSQL* conn : mysql_conns) {
int rc = mysql_query(conn, string {"/* ;hostgroup=" + std::to_string(hg) + "*/ BEGIN"}.c_str());
if (rc != EXIT_SUCCESS) {
Expand Down Expand Up @@ -412,11 +445,11 @@ int main(int, char**) {
// Initialize connections
if (!proxy_mysql) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxy_mysql));
return EXIT_FAILURE;
return exit_status();
}
if (!mysql_real_connect(proxy_mysql, cl.host, cl.username, cl.password, NULL, cl.port, NULL, 0)) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxy_mysql));
return EXIT_FAILURE;
return exit_status();
}

// Create a new 'db' for connection filtering
Expand All @@ -427,16 +460,16 @@ int main(int, char**) {
MYSQL* proxy_admin = mysql_init(NULL);
if (!proxy_admin) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxy_admin));
return EXIT_FAILURE;
return exit_status();
}

if (!mysql_real_connect(proxy_admin, cl.host, cl.admin_username, cl.admin_password, NULL, cl.admin_port, NULL, 0)) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, mysql_error(proxy_admin));
return EXIT_FAILURE;
return exit_status();
}

if (compute_wait_timeout(proxy_admin) != EXIT_SUCCESS) {
return EXIT_FAILURE;
return exit_status();
}

double intv = 5;
Expand All @@ -462,10 +495,17 @@ int main(int, char**) {
fprintf(stderr, "File %s, line %d, Error: %s\n", __FILE__, __LINE__, err_msg);
}

return EXIT_FAILURE;
return exit_status();
}
diag("Setting mysql_servers config...");
{
string query = "UPDATE mysql_servers SET max_connections=2500";
diag("Running: %s", query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());
query = "LOAD MYSQL SERVERS TO RUNTIME";
diag("Running: %s", query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());
}
MYSQL_QUERY(proxy_admin, "UPDATE mysql_servers SET max_connections=2500");
MYSQL_QUERY(proxy_admin, "LOAD MYSQL SERVERS TO RUNTIME");

diag("Setting ProxySQL config...");
{
Expand All @@ -482,7 +522,9 @@ int main(int, char**) {
diag("%s", query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());
// Set a higher max_connection number for the servers
MYSQL_QUERY(proxy_admin, "LOAD MYSQL VARIABLES TO RUNTIME");
query = "LOAD MYSQL VARIABLES TO RUNTIME";
diag("%s", query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());
}
// Configure MySQL infra servers with: 'wait_timeout' and 'max_connections'
vector<pair<mysql_res_row, srv_cfg>> servers_old_configs {};
Expand All @@ -496,17 +538,18 @@ int main(int, char**) {

if (servers_rows.empty()) {
fprintf(stderr, "File %s, line %d, Error: Invalid result returned from 'mysql_servers'\n", __FILE__, __LINE__);
return EXIT_FAILURE;
return exit_status();
}

srv_cfg new_srv_cfg { { "wait_timeout", wait_timeout }, { "max_connections", 2500 } };

for (const mysql_res_row& srv_row : servers_rows) {
srv_cfg old_srv_cfg {};
diag("Line:%d : %s:%s", __LINE__ , srv_row[0].c_str(), srv_row[1].c_str());
int cfg_res = change_mysql_cfg(cl, srv_row[0], srv_row[1], new_srv_cfg, old_srv_cfg);

if (cfg_res != EXIT_SUCCESS) {
return EXIT_FAILURE;
return exit_status();
} else {
servers_old_configs.push_back({ srv_row, old_srv_cfg });
}
Expand All @@ -528,12 +571,23 @@ int main(int, char**) {
m_server_test.assign({ { "127.0.0.1", 13306 }, { "127.0.0.1", 13307 }, { "127.0.0.1", 13308 } });
}

for (const svr_addr& svr : s_server_test) {
diag("Line:%d : s_server_test: %s:%d", __LINE__, svr.first.c_str(), svr.second);
}
for (const svr_addr& svr : m_server_test) {
diag("Line:%d : m_server_test: %s:%d", __LINE__, svr.first.c_str(), svr.second);
}

diag("Performing 'check_backend_conns()' for servers: '%s'", nlohmann::json(s_server_test).dump().c_str());
int s_server_rc = check_backend_conns(cl, test_params, 0, s_server_test);
if (s_server_rc == EXIT_SUCCESS) {
diag("Cleaning up previous backend connections...");
MYSQL_QUERY(proxy_admin, "UPDATE mysql_servers SET max_connections=0");
MYSQL_QUERY(proxy_admin, "LOAD MYSQL SERVERS TO RUNTIME");
string query = "UPDATE mysql_servers SET max_connections=0";
diag("Line:%d : Running: %s", __LINE__ , query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());
query = "LOAD MYSQL SERVERS TO RUNTIME";
diag("Line:%d : Running: %s", __LINE__ , query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());

int w_res = wait_target_backend_conns(proxy_admin, 0, 10);
if (w_res != EXIT_SUCCESS) {
Expand All @@ -546,8 +600,12 @@ int main(int, char**) {
fprintf(stderr, "File %s, line %d, Error: \"%s\"\n", __FILE__, __LINE__, err_msg.c_str());
}

MYSQL_QUERY(proxy_admin, "UPDATE mysql_servers SET max_connections=2500");
MYSQL_QUERY(proxy_admin, "LOAD MYSQL SERVERS TO RUNTIME");
query = "UPDATE mysql_servers SET max_connections=2500";
diag("Line:%d : Running: %s", __LINE__ , query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());
query = "LOAD MYSQL SERVERS TO RUNTIME";
diag("Line:%d : Running: %s", __LINE__ , query.c_str());
MYSQL_QUERY(proxy_admin, query.c_str());

if (w_res == EXIT_SUCCESS) {
diag("Performing 'check_backend_conns()' for servers: '%s'", nlohmann::json(m_server_test).dump().c_str());
Expand Down

0 comments on commit 4d33507

Please sign in to comment.