test/cetus
1
0
Fork 0
mirror of https://gitee.com/wangbin579/cetus.git synced 2024-11-29 18:37:36 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge commit '62858e829801a1c0a9ae8a899f0159d76b44baa7'

Browse Source
This commit is contained in:
lazio579 2019-02-26 09:32:22 +08:00
commit 874d151313
6 changed files with 198 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
plugins/admin/admin-commands.c
View File

@ -30,6 +30,7 @@
static gint save_setting(chassis *srv, gint *effected_rows);
static void send_result(network_socket *client, gint ret, gint affected);
static void admin_update_remote_backend(network_mysqld_con* con, chassis_config_t *conf);
static const char *get_conn_xa_state_name(network_mysqld_con_dist_tran_state_t state) {
switch (state) {
@ -1242,8 +1243,24 @@ void admin_insert_backend(network_mysqld_con* con, char *addr, char *type, char
case BACKEND_OPERATE_SUCCESS:
{
if(con->srv->config_manager->type == CHASSIS_CONF_MYSQL) {
network_mysqld_con_send_ok_full(con->client, 1, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
chassis_config_t* conf = con->srv->config_manager;
if(backend_type(type) == BACKEND_TYPE_RO) {
conf->key = g_strdup("proxy-read-only-backend-addresses");
conf->value = admin_get_value_by_key(con, conf->key);
conf->reserve1 = NULL;
conf->reserve2 = NULL;
admin_update_remote_backend(con, conf);
return;
} else if (backend_type(type) == BACKEND_TYPE_RW) {
conf->key = g_strdup("proxy-backend-addresses");
conf->value = admin_get_value_by_key(con, conf->key);
conf->reserve1 = NULL;
conf->reserve2 = NULL;
admin_update_remote_backend(con, conf);
return;
} else {
network_mysqld_con_send_error(con->client, C("insert backend succeed, but save to remote failed."));
}
} else {
gint effected_rows = 0;
ret = save_setting(con->srv, &effected_rows);
@ -1331,8 +1348,13 @@ void admin_update_backend(network_mysqld_con* con, GList* equations,
}
int affected = (network_backends_modify(g->backends, backend_ndx, type, state, NO_PREVIOUS_STATE)==0)?1:0;
if(con->srv->config_manager->type == CHASSIS_CONF_MYSQL) {
network_mysqld_con_send_ok_full(con->client, affected,
0, SERVER_STATUS_AUTOCOMMIT, 0);
chassis_config_t* conf = con->srv->config_manager;
conf->key = g_strdup("proxy-read-only-backend-addresses");
conf->value = admin_get_value_by_key(con, conf->key);
conf->reserve1 = g_strdup("proxy-backend-addresses");
conf->reserve2 = admin_get_value_by_key(con, conf->reserve1);
admin_update_remote_backend(con, conf);
return;
} else {
gint ret = CHANGE_SAVE_ERROR;
gint effected_rows = 0;
@ -1363,8 +1385,13 @@ void admin_delete_backend(network_mysqld_con* con, char *key, char *val)
if (backend_ndx >= 0 && backend_ndx < network_backends_count(g->backends)) {
network_backends_remove(g->backends, backend_ndx);/*TODO: just change state? */
if(con->srv->config_manager->type == CHASSIS_CONF_MYSQL) {
network_mysqld_con_send_ok_full(con->client, 1, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
chassis_config_t* conf = con->srv->config_manager;
conf->key = g_strdup("proxy-read-only-backend-addresses");
conf->value = admin_get_value_by_key(con, conf->key);
conf->reserve1 = g_strdup("proxy-backend-addresses");
conf->reserve2 = admin_get_value_by_key(con, conf->reserve1);
admin_update_remote_backend(con, conf);
return;
} else {
gint effected_rows = 0;
gint ret = save_setting(con->srv, &effected_rows);
@ -1592,6 +1619,80 @@ static void admin_set_remote_config(network_mysqld_con* con, chassis_config_t *c
con->is_admin_waiting_resp = 1;
}
void admin_update_remote_backend_callback(int fd, short what, void *arg)
{
g_message("%s call admin_update_remote_backend_callback", G_STRLOC);
network_mysqld_con* con = arg;
chassis *chas = con->srv;
chassis_config_t* conf = con->srv->config_manager;
if (conf->options_update_flag) {
conf->retries++;
if (conf->retries == 0) {
network_mysqld_con_send_error(con->client, C("update remote backend timeout"));
con->is_admin_waiting_resp = 0;
send_admin_resp(con->srv, con);
g_message("%s call admin_update_remote_backend_callback timeout, last timeout:%d", G_STRLOC, conf->ms_timeout);
goto fun_end;
}
conf->ms_timeout = conf->ms_timeout << 1;
int sec = conf->ms_timeout / 1000;
int usec = (conf->ms_timeout - 1000 * sec) * 1000;
struct timeval check_interval = {sec, usec};
chassis_event_add_with_timeout(chas, &chas->remote_config_event, &check_interval);
g_message("%s call admin_update_remote_backend_callback, timeout:%d", G_STRLOC, conf->ms_timeout);
return;
}
if (!conf->options_success_flag) {
network_mysqld_con_send_error(con->client,C("backend is set locally but cannot replace remote"));
con->is_admin_waiting_resp = 0;
send_admin_resp(con->srv, con);
g_message("%s call send_admin_resp over", G_STRLOC);
goto fun_end;
}
network_mysqld_con_send_ok_full(con->client, 1, 0, SERVER_STATUS_AUTOCOMMIT, 0);
con->is_admin_waiting_resp = 0;
send_admin_resp(con->srv, con);
g_message("%s call send_admin_resp over", G_STRLOC);
fun_end:
if (conf->key) {
g_free(conf->key);
conf->key = NULL;
}
if (conf->value) {
g_free(conf->value);
conf->value = NULL;
}
if (conf->reserve1) {
g_free(conf->reserve1);
conf->reserve1 = NULL;
}
if (conf->reserve2) {
g_free(conf->reserve2);
conf->reserve2 = NULL;
}
}
static void admin_update_remote_backend(network_mysqld_con* con, chassis_config_t *conf)
{
g_message("%s call admin_update_remote_backend", G_STRLOC);
chassis* chas = con->srv;
conf->options_update_flag = 1;
chas->asynchronous_type = ASYNCHRONOUS_UPDATE_BACKENDS;
evtimer_set(&chas->remote_config_event, admin_update_remote_backend_callback, con);
struct timeval check_interval = {0, 2000};
conf->ms_timeout = 2;
conf->retries = 1;
chassis_event_add_with_timeout(chas, &chas->remote_config_event, &check_interval);
con->is_admin_waiting_resp = 1;
}
void admin_set_config(network_mysqld_con* con, char* key, char* value)
{
if (con->is_processed_by_subordinate) {
@ -2678,3 +2779,21 @@ void admin_select_version_comment(network_mysqld_con* con) {
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
}
char* admin_get_value_by_key(network_mysqld_con* con, const char *key) {
GList *options = admin_get_all_options(con->srv);
GList *l = NULL;
for (l = options; l; l = l->next) {
chassis_option_t *opt = l->data;
if (g_strcasecmp(key, opt->long_name) == 0) {
struct external_param param = {0};
param.chas = con->srv;
param.opt_type = opt->opt_property;
char *value = opt->show_hook != NULL? opt->show_hook(&param) : NULL;
return value;
}
}
g_list_free(options);
return NULL;
}

1
plugins/admin/admin-commands.h
View File

@ -66,4 +66,5 @@ void admin_sql_log_status(network_mysqld_con* con);
void admin_kill_query(network_mysqld_con* con, guint32);
void admin_comment_handle(network_mysqld_con* con);
void admin_select_version_comment(network_mysqld_con* con);
char* admin_get_value_by_key(network_mysqld_con* con, const char *key);
#endif // ADMIN_COMMANDS_H

4
src/cetus-process-cycle.c
View File

@ -180,6 +180,10 @@ gpointer retrieve_remote_config_mainloop(gpointer user_data) {
g_message("%s: ASYNCHRONOUS_SET_CONFIG visited", G_STRLOC);
chassis_config_set_remote_options(conf, conf->key, conf->value);
break;
case ASYNCHRONOUS_UPDATE_BACKENDS:
g_message("%s: ASYNCHRONOUS_UPDATE_BACKENDS visited", G_STRLOC);
chassis_config_set_remote_backends(conf, conf->key, conf->value, conf->reserve1, conf->reserve2);
break;
default:
break;
}

63
src/chassis-config.c
View File

@ -369,6 +369,69 @@ recycle_mysql_resources:
return status;
}
gboolean
chassis_config_set_remote_backends(chassis_config_t *conf, gchar* key1, gchar* value1, gchar* key2, gchar* value2)
{
if(conf->type != CHASSIS_CONF_MYSQL){
return TRUE;
}
MYSQL *conn = chassis_config_get_mysql_connection(conf);
if (!conn) {
g_warning("%s:Cannot connect to mysql server.", G_STRLOC);
conf->options_update_flag = 0;
conf->options_success_flag = 0;
return FALSE;
}
gboolean status = FALSE, status1 = FALSE, status2 = FALSE;
gchar sql1[1024] = { 0 }, real_value1[1024] = { 0 };
gchar sql2[1024] = { 0 }, real_value2[1024] = { 0 };
if (key1) {
if (value1) {
mysql_real_escape_string(conn, real_value1, value1, strlen(value1));
snprintf(sql1, sizeof(sql1),
"INSERT INTO %s.`settings`(option_key,option_value) VALUES ('%s', '%s') ON DUPLICATE KEY UPDATE option_value = '%s'", conf->schema, key1, real_value1, real_value1);
} else {
snprintf(sql1, sizeof(sql1),
"DELETE FROM %s.`settings` where option_key = '%s'", conf->schema, key1);
}
if (mysql_query(conn, sql1)) {
g_warning("sql failed: %s | error: %s", sql1, mysql_error(conn));
} else {
status1 = TRUE;
}
}
if (status1 && key2) {
if (value2) {
mysql_real_escape_string(conn, real_value2, value2, strlen(value2));
snprintf(sql2, sizeof(sql2),
"INSERT INTO %s.`settings`(option_key,option_value) VALUES ('%s', '%s') ON DUPLICATE KEY UPDATE option_value = '%s'", conf->schema, key2, real_value2, real_value2);
} else {
snprintf(sql2, sizeof(sql2),
"DELETE FROM %s.`settings` where option_key = '%s'", conf->schema, key2);
}
if (mysql_query(conn, sql2)) {
g_warning("sql failed: %s | error: %s", sql1, mysql_error(conn));
} else {
status2 = TRUE;
}
}
status = status1 && (key2 == NULL? TRUE:status2);
recycle_mysql_resources:
if (status) {
conf->options_update_flag = 0;
conf->options_success_flag = 1;
} else {
conf->options_success_flag = 0;
conf->options_update_flag = 0;
}
mysql_close(conf->mysql_conn);
conf->mysql_conn = NULL;
return status;
}
gint chassis_config_reload_options(chassis_config_t *conf)
{
switch (conf->type) {

4
src/chassis-config.h
View File

@ -64,6 +64,8 @@ struct chassis_config_t {
void *user_data;
void *key;
void *value;
void *reserve1;
void *reserve2;
};
/**
@ -84,6 +86,8 @@ gboolean chassis_config_load_options_mysql(chassis_config_t *conf);
gboolean chassis_config_set_remote_options(chassis_config_t *conf, gchar* key, gchar* value);
gboolean chassis_config_set_remote_backends(chassis_config_t *conf, gchar* key1, gchar* value1, gchar* key2, gchar* value2);
GHashTable *chassis_config_get_options(chassis_config_t *);
gboolean chassis_config_parse_options(chassis_config_t *, GList *entries);

1
src/chassis-mainloop.h
View File

@ -79,6 +79,7 @@ enum asynchronous_admin_type {
ASYNCHRONOUS_UPDATE_OR_DELETE_USER_PASSWORD,
ASYNCHRONOUS_CONFIG_REMOTE_SHARD,
ASYNCHRONOUS_SET_CONFIG,
ASYNCHRONOUS_UPDATE_BACKENDS,
};
typedef struct rw_op_t {