test/cetus
1
0
Fork 0
mirror of https://gitee.com/wangbin579/cetus.git synced 2024-12-02 20:08:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity
5cf749ef0a
cetus/plugins/admin/admin-plugin.c

2556 lines
90 KiB
C
Raw Blame History

/* $%BEGINLICENSE%$
Copyright (c) 2007, 2012, Oracle and/or its affiliates. All rights reserved.
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License as
published by the Free Software Foundation; version 2 of the
License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
02110-1301 USA
$%ENDLICENSE%$ */
#define _GNU_SOURCE
#include <ctype.h>
#include <errno.h>
#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
#include <malloc.h>
#include "cetus-users.h"
#include "cetus-util.h"
#include "cetus-variable.h"
#include "character-set.h"
#include "chassis-event.h"
#include "chassis-options.h"
#include "cetus-monitor.h"
#include "glib-ext.h"
#include "network-mysqld-packet.h"
#include "network-mysqld-proto.h"
#include "network-mysqld.h"
#include "server-session.h"
#include "sys-pedantic.h"
#ifndef PLUGIN_VERSION
#ifdef CHASSIS_BUILD_TAG
#define PLUGIN_VERSION CHASSIS_BUILD_TAG
#else
#define PLUGIN_VERSION PACKAGE_VERSION
#endif
#endif
struct chassis_plugin_config {
gchar *address; /**< listening address of the admin interface */
gchar *admin_username; /**< login username */
gchar *admin_password; /**< login password */
gchar *allow_ip; /**< allow ip addr list */
GHashTable *allow_ip_table;
gchar *deny_ip; /**< deny ip addr list */
GHashTable *deny_ip_table;
gboolean has_shard_plugin; /**< another plugin name is shard or proxy, TRUE is shard, FALSE is proxy */
network_mysqld_con *listen_con;
};
static struct event* g_sampling_timer = NULL;
/*
* tokenize input, alloc and return nth token
* n -> [0,..)
*/
static char *str_nth_token(const char *input, int n)
{
char *t = NULL;
char **tokens = g_strsplit(input, " ", -1);
if (g_strv_length(tokens) > n) {
t = g_strdup(tokens[n]);
}
g_strfreev(tokens);
return t;
}
/* get config->has_shard_plugin */
static gboolean has_shard_plugin(GPtrArray *modules)
{
int i;
for (i = 0; i < modules->len; i++) {
chassis_plugin *plugin = modules->pdata[i];
if (strcmp(plugin->name, "shard") == 0) {
return TRUE;
}
}
return FALSE;
}
NETWORK_MYSQLD_PLUGIN_PROTO(server_con_init) {
network_mysqld_auth_challenge *challenge;
GString *packet;
challenge = network_mysqld_auth_challenge_new();
challenge->server_version_str = g_strdup_printf("%s admin", PACKAGE_STRING);
challenge->server_version = 50099;
challenge->charset = charset_get_number("latin1");
challenge->capabilities = CETUS_DEFAULT_FLAGS;
challenge->server_status = SERVER_STATUS_AUTOCOMMIT;
challenge->thread_id = 1;
/* generate a random challenge */
network_mysqld_auth_challenge_set_challenge(challenge);
packet = g_string_new(NULL);
network_mysqld_proto_append_auth_challenge(packet, challenge);
con->client->challenge = challenge;
network_mysqld_queue_append(con->client, con->client->send_queue, S(packet));
g_string_free(packet, TRUE);
con->state = ST_SEND_HANDSHAKE;
g_assert(con->plugin_con_state == NULL);
return NETWORK_SOCKET_SUCCESS;
}
NETWORK_MYSQLD_PLUGIN_PROTO(server_read_auth) {
network_packet packet;
network_socket *recv_sock, *send_sock;
network_mysqld_auth_response *auth;
GString *excepted_response;
GString *hashed_pwd;
recv_sock = con->client;
send_sock = con->client;
packet.data = g_queue_peek_head(recv_sock->recv_queue->chunks);
packet.offset = 0;
/* decode the packet */
network_mysqld_proto_skip_network_header(&packet);
auth = network_mysqld_auth_response_new(con->client->challenge->capabilities);
if (network_mysqld_proto_get_auth_response(&packet, auth)) {
network_mysqld_auth_response_free(auth);
return NETWORK_SOCKET_ERROR;
}
if (!(auth->client_capabilities & CLIENT_PROTOCOL_41)) {
/* should use packet-id 0 */
network_mysqld_queue_append(con->client, con->client->send_queue,
C("\xff\xd7\x07" "4.0 protocol is not supported"));
network_mysqld_auth_response_free(auth);
return NETWORK_SOCKET_ERROR;
}
con->client->response = auth;
/* Check client addr in admin-allow-ip and admin-deny-ip */
gboolean check_ip;
char *ip_err_msg = NULL;
if (con->config->allow_ip_table || con->config->deny_ip_table) {
char *client_addr = con->client->src->name->str;
char **client_addr_arr = g_strsplit(client_addr, ":", -1);
char *client_ip = client_addr_arr[0];
if (g_hash_table_size(con->config->allow_ip_table) != 0 &&
(g_hash_table_lookup(con->config->allow_ip_table, client_ip) || g_hash_table_lookup(con->config->allow_ip_table, "*"))) {
check_ip = FALSE;
} else if (g_hash_table_size(con->config->deny_ip_table) != 0 &&
(g_hash_table_lookup(con->config->deny_ip_table, client_ip) || g_hash_table_lookup(con->config->deny_ip_table, "*"))) {
check_ip = TRUE;
ip_err_msg = g_strdup_printf("Access denied for user '%s'@'%s'",
con->config->admin_username, client_ip);
} else {
check_ip = FALSE;
}
g_strfreev(client_addr_arr);
} else {
check_ip = FALSE;
}
if (check_ip) {
network_mysqld_con_send_error_full(send_sock, L(ip_err_msg), 1045, "28000");
g_free(ip_err_msg);
con->state = ST_SEND_ERROR;
} else {
/* check if the password matches */
excepted_response = g_string_new(NULL);
hashed_pwd = g_string_new(NULL);
if (!strleq(S(con->client->response->username),
con->config->admin_username,
strlen(con->config->admin_username)))
{
network_mysqld_con_send_error_full(send_sock,
C("unknown user"), 1045, "28000");
/* close the connection after we have sent this packet */
con->state = ST_SEND_ERROR;
} else if (network_mysqld_proto_password_hash(hashed_pwd,
con->config->admin_password,
strlen(con->config->admin_password)))
{
} else if (network_mysqld_proto_password_scramble(excepted_response,
S(recv_sock->challenge->auth_plugin_data),
S(hashed_pwd))) {
network_mysqld_con_send_error_full(send_sock,
C("scrambling failed"), 1045, "28000");
/* close the connection after we have sent this packet */
con->state = ST_SEND_ERROR;
} else if (!g_string_equal(excepted_response, auth->auth_plugin_data)) {
network_mysqld_con_send_error_full(send_sock,
C("password doesn't match"), 1045, "28000");
/* close the connection after we have sent this packet */
con->state = ST_SEND_ERROR;
} else {
network_mysqld_con_send_ok(send_sock);
con->state = ST_SEND_AUTH_RESULT;
}
g_string_free(hashed_pwd, TRUE);
g_string_free(excepted_response, TRUE);
}
g_string_free(g_queue_pop_tail(recv_sock->recv_queue->chunks), TRUE);
if (recv_sock->recv_queue->chunks->length > 0) {
g_warning("%s: client-recv-queue-len = %d",
G_STRLOC, recv_sock->recv_queue->chunks->length);
}
return NETWORK_SOCKET_SUCCESS;
}
static const char *get_conn_xa_state_name(network_mysqld_con_dist_tran_state_t state) {
switch (state) {
case NEXT_ST_XA_START: return "XS";
case NEXT_ST_XA_QUERY: return "XQ";
case NEXT_ST_XA_END: return "XE";
case NEXT_ST_XA_PREPARE: return "XP";
case NEXT_ST_XA_COMMIT: return "XC";
case NEXT_ST_XA_ROLLBACK: return "XR";
case NEXT_ST_XA_CANDIDATE_OVER: return "XCO";
case NEXT_ST_XA_OVER: return "XO";
default:
break;
}
return "NX";
}
static char *states[] = {
"unknown",
"up",
"down",
"maintaining",
"deleted",
};
static char *types[] = {
"unknown",
"rw",
"ro",
};
static int admin_send_backends_info(network_mysqld_con *admin_con, const char *sql)
{
chassis *chas = admin_con->srv;
chassis_private *priv = chas->priv;
chassis_plugin_config *config = admin_con->config;
GPtrArray *fields = g_ptr_array_new_with_free_func(
(GDestroyNotify)network_mysqld_proto_fielddef_free);
MYSQL_FIELD *field;
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("backend_ndx");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("address");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("state");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("type");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("slave delay");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("uuid");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("idle_conns");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("used_conns");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("total_conns");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
if (config->has_shard_plugin) {
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("group");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
}
GPtrArray *rows = g_ptr_array_new_with_free_func(
(GDestroyNotify)network_mysqld_mysql_field_row_free);
network_backends_t *bs = priv->backends;
int len = bs->backends->len;
int i;
char buffer[32];
for (i = 0; i < len; i++) {
network_backend_t *backend = bs->backends->pdata[i];
GPtrArray *row = g_ptr_array_new_with_free_func(g_free);
snprintf(buffer, sizeof(buffer), "%d", i+1);
g_ptr_array_add(row, g_strdup(buffer));
g_ptr_array_add(row, g_strdup(backend->addr->name->str));
g_ptr_array_add(row, g_strdup(states[(int) (backend->state)]));
g_ptr_array_add(row, g_strdup(types[(int) (backend->type)]));
snprintf(buffer, sizeof(buffer), "%d", backend->slave_delay_msec);
g_ptr_array_add(row, backend->type == BACKEND_TYPE_RO ? g_strdup(buffer) : NULL);
g_ptr_array_add(row, backend->uuid->len ? g_strdup(backend->uuid->str) : NULL);
snprintf(buffer, sizeof(buffer), "%d", backend->pool->cur_idle_connections);
g_ptr_array_add(row, g_strdup(buffer));
snprintf(buffer, sizeof(buffer), "%d", backend->connected_clients);
g_ptr_array_add(row, g_strdup(buffer));
snprintf(buffer, sizeof(buffer), "%d",
backend->pool->cur_idle_connections + backend->connected_clients);
g_ptr_array_add(row, g_strdup(buffer));
g_ptr_array_add(row, backend->server_group->len ?
g_strdup(backend->server_group->str) : NULL);
g_ptr_array_add(rows, row);
}
network_mysqld_con_send_resultset(admin_con->client, fields, rows);
/* Free data */
g_ptr_array_free(rows, TRUE);
g_ptr_array_free(fields, TRUE);
return PROXY_SEND_RESULT;
}
static void g_table_free_all(gpointer q) {
GHashTable *table = q;
g_hash_table_destroy(table);
}
typedef struct used_conns {
int num;
} used_conns_t;
static int admin_send_backend_detail_info(network_mysqld_con *admin_con, const char *sql)
{
chassis *chas = admin_con->srv;
chassis_private *priv = chas->priv;
int i, j, len;
char buffer[32];
GPtrArray *fields;
GPtrArray *rows;
GPtrArray *row;
MYSQL_FIELD *field;
GHashTable *back_user_conn_hash_table = g_hash_table_new_full(g_str_hash,
g_str_equal, g_free, g_table_free_all);
network_backends_t *bs = priv->backends;
len = bs->backends->len;
for (i = 0; i < len; i++) {
network_backend_t *backend = bs->backends->pdata[i];
GHashTable *table = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, g_free);
g_hash_table_insert(back_user_conn_hash_table, g_strdup(backend->addr->name->str), table);
}
len = priv->cons->len;
for (i = 0; i < len; i++) {
network_mysqld_con *con = priv->cons->pdata[i];
if (!con->client || !con->client->response) {
continue;
}
#ifndef SIMPLE_PARSER
if (con->servers == NULL) {
continue;
}
for (j = 0; j < con->servers->len; j++) {
server_session_t *pmd = g_ptr_array_index(con->servers, j);
GHashTable *table = g_hash_table_lookup(back_user_conn_hash_table,
pmd->backend->addr->name->str);
if (table == NULL) {
g_warning("%s: table is null for backend:%s", G_STRLOC,
pmd->backend->addr->name->str);
continue;
}
used_conns_t *total_used = g_hash_table_lookup(table,
con->client->response->username->str);
if (total_used == NULL) {
total_used = g_new0(used_conns_t, 1);
g_hash_table_insert(table,
g_strdup(con->client->response->username->str), total_used);
}
total_used->num++;
}
#else
if (con->servers != NULL) {
for (j = 0; j < con->servers->len; j++) {
network_socket *sock = g_ptr_array_index(con->servers, j);
GHashTable *table = g_hash_table_lookup(back_user_conn_hash_table,
sock->dst->name->str);
if (table == NULL) {
g_warning("%s: table is null for backend:%s", G_STRLOC,
sock->dst->name->str);
continue;
}
used_conns_t *total_used = g_hash_table_lookup(table,
con->client->response->username->str);
if (total_used == NULL) {
total_used = g_new0(used_conns_t, 1);
g_hash_table_insert(table,
g_strdup(con->client->response->username->str), total_used);
}
total_used->num++;
}
} else {
if (con->server == NULL) {
continue;
}
GHashTable *table = g_hash_table_lookup(back_user_conn_hash_table,
con->server->dst->name->str);
if (table == NULL) {
g_warning("%s: table is null for backend:%s", G_STRLOC,
con->server->dst->name->str);
continue;
}
used_conns_t *total_used = g_hash_table_lookup(table,
con->client->response->username->str);
if (total_used == NULL) {
total_used = g_new0(used_conns_t, 1);
g_hash_table_insert(table,
g_strdup(con->client->response->username->str), total_used);
}
total_used->num++;
}
#endif
}
fields = g_ptr_array_new_with_free_func((void *) network_mysqld_proto_fielddef_free);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("backend_ndx");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("username");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("idle_conns");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("used_conns");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("total_conns");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
rows = g_ptr_array_new_with_free_func((void *) network_mysqld_mysql_field_row_free);
len = bs->backends->len;
for (i = 0; i < len; i++) {
network_backend_t *backend = bs->backends->pdata[i];
GHashTable *table = g_hash_table_lookup(back_user_conn_hash_table,
backend->addr->name->str);
if (table == NULL) {
g_warning("%s: table is null for backend:%s", G_STRLOC,
backend->addr->name->str);
continue;
}
GHashTable *users = backend->pool->users;
if (users != NULL) {
GHashTableIter iter;
GString *key;
GQueue *queue;
g_hash_table_iter_init(&iter, users);
/* count all users' pooled connections */
while (g_hash_table_iter_next(&iter, (void **)&key, (void **)&queue)) {
row = g_ptr_array_new_with_free_func(g_free);
snprintf(buffer, sizeof(buffer),"%d", i+1);
g_ptr_array_add(row, g_strdup(buffer));
g_ptr_array_add(row, g_strdup(key->str));
snprintf(buffer, sizeof(buffer), "%d", queue->length);
g_ptr_array_add(row, g_strdup(buffer));
used_conns_t *total_used = g_hash_table_lookup(table, key->str);
if (total_used) {
snprintf(buffer, sizeof(buffer), "%d", total_used->num);
} else {
snprintf(buffer, sizeof(buffer), "%d", 0);
}
g_ptr_array_add(row, g_strdup(buffer));
if (total_used) {
snprintf(buffer, sizeof(buffer), "%d", queue->length + total_used->num);
} else {
snprintf(buffer, sizeof(buffer), "%d", queue->length);
}
g_ptr_array_add(row, g_strdup(buffer));
g_ptr_array_add(rows, row);
}
}
}
network_mysqld_con_send_resultset(admin_con->client, fields, rows);
/* Free data */
g_ptr_array_free(rows, TRUE);
g_ptr_array_free(fields, TRUE);
g_hash_table_destroy(back_user_conn_hash_table);
return PROXY_SEND_RESULT;
}
static int admin_show_connectionlist(network_mysqld_con *admin_con, const char *sql)
{
char *arg = str_nth_token(sql, 2);
int number = 65536;
if (arg) {
number = atoi(arg);
g_free(arg);
}
chassis *chas = admin_con->srv;
chassis_private *priv = chas->priv;
chassis_plugin_config *config = admin_con->config;
int i, len;
char buffer[32];
GPtrArray *fields;
GPtrArray *rows;
GPtrArray *row;
MYSQL_FIELD *field;
fields = g_ptr_array_new_with_free_func((void *) network_mysqld_proto_fielddef_free);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("User");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Host");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("db");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Command");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Time");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Trans");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("PS");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("State");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
if (config->has_shard_plugin) {
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Xa");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Xid");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
}
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Server");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Info");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
rows = g_ptr_array_new_with_free_func((void *) network_mysqld_mysql_field_row_free);
struct timeval now;
gettimeofday(&(now), NULL);
len = priv->cons->len;
int count = 0;
for (i = 0; i < len; i++) {
network_mysqld_con *con = priv->cons->pdata[i];
if (!con->client) {
continue;
}
if (count >= number) {
break;
}
count++;
row = g_ptr_array_new_with_free_func(g_free);
if (con->client->response != NULL) {
g_ptr_array_add(row, g_strdup(con->client->response->username->str));
} else {
g_ptr_array_add(row, NULL);
}
g_ptr_array_add(row, g_strdup(con->client->src->name->str));
g_ptr_array_add(row, g_strdup(con->client->default_db->str));
if (con->state <= ST_READ_QUERY) {
g_ptr_array_add(row, g_strdup("Sleep"));
g_ptr_array_add(row, g_strdup("0"));
} else {
g_ptr_array_add(row, g_strdup("Query"));
int diff = now.tv_sec - con->req_recv_time.tv_sec;
if (diff > 7200) {
g_critical("%s:too slow connection(%s) processing for con:%p",
G_STRLOC, con->client->src->name->str, con);
}
diff = diff * 1000;
diff += (now.tv_usec - con->req_recv_time.tv_usec) / 1000;
snprintf(buffer, sizeof(buffer), "%d", diff);
g_ptr_array_add(row, g_strdup(buffer));
}
g_ptr_array_add(row, g_strdup(con->is_in_transaction ? "Y" : "N"));
g_ptr_array_add(row, g_strdup(con->is_prepared ? "Y" : "N"));
g_ptr_array_add(row, g_strdup(network_mysqld_con_st_name(con->state)));
if (config->has_shard_plugin) {
g_ptr_array_add(row, g_strdup(get_conn_xa_state_name(con->dist_tran_state)));
if (con->dist_tran) {
snprintf(buffer, sizeof(buffer), "%lld", con->xa_id);
g_ptr_array_add(row, g_strdup(buffer));
} else {
g_ptr_array_add(row, NULL);
}
}
if (con->servers != NULL) {
int j;
GString *servers = g_string_new(NULL);
for (j = 0; j < con->servers->len; j++) {
server_session_t *pmd = g_ptr_array_index(con->servers, j);
if (pmd && pmd->server) {
if (pmd->server->src) {
g_string_append_len(servers, S(pmd->server->src->name));
char *delim = "->";
g_string_append_len(servers, delim, strlen(delim));
}
g_string_append_len(servers, S(pmd->server->dst->name));
g_string_append_c(servers, ' ');
}
}
g_ptr_array_add(row, g_strdup(servers->str));
g_string_free(servers, TRUE);
} else {
if (con->server != NULL) {
GString *server = g_string_new(NULL);
if (con->server->src) {
g_string_append_len(server, S(con->server->src->name));
char *delim = "->";
g_string_append_len(server, delim, strlen(delim));
}
g_string_append_len(server, S(con->server->dst->name));
g_ptr_array_add(row, g_strdup(server->str));
g_string_free(server, TRUE);
} else {
g_ptr_array_add(row, NULL);
}
}
if (con->orig_sql->len) {
if (con->state == ST_READ_QUERY) {
g_ptr_array_add(row, NULL);
} else {
g_ptr_array_add(row, g_strdup(con->orig_sql->str));
}
} else {
g_ptr_array_add(row, NULL);
}
g_ptr_array_add(rows, row);
}
network_mysqld_con_send_resultset(admin_con->client, fields, rows);
g_ptr_array_free(rows, TRUE);
g_ptr_array_free(fields, TRUE);
return PROXY_SEND_RESULT;
}
static GList *network_mysqld_admin_plugin_allow_ip_get(chassis_plugin_config *config) {
if (config && config->allow_ip_table) {
return g_hash_table_get_keys(config->allow_ip_table);
}
return NULL;
}
static GList *network_mysqld_admin_plugin_deny_ip_get(chassis_plugin_config *config) {
if (config && config->deny_ip_table) {
return g_hash_table_get_keys(config->deny_ip_table);
}
return NULL;
}
static gboolean
network_mysqld_admin_plugin_allow_ip_add(chassis_plugin_config *config, char *addr)
{
if (!config || !addr) return FALSE;
if (!config->allow_ip_table) {
config->allow_ip_table = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
}
gboolean success = FALSE;
if (!g_hash_table_lookup(config->allow_ip_table, addr)) {
g_hash_table_insert(config->allow_ip_table, g_strdup(addr), (void *) TRUE);
success = TRUE;
}
return success;
}
static gboolean
network_mysqld_admin_plugin_deny_ip_add(chassis_plugin_config *config, char *addr)
{
if (!config || !addr) return FALSE;
if (!config->deny_ip_table) {
config->deny_ip_table = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
}
gboolean success = FALSE;
if (!g_hash_table_lookup(config->deny_ip_table, addr)) {
g_hash_table_insert(config->deny_ip_table, g_strdup(addr), (void *) TRUE);
success = TRUE;
}
return success;
}
static gboolean
network_mysqld_admin_plugin_allow_ip_del(chassis_plugin_config *config, char *addr)
{
if (!config || !addr || !config->allow_ip_table) return FALSE;
return g_hash_table_remove(config->allow_ip_table, addr);
}
static gboolean
network_mysqld_admin_plugin_deny_ip_del(chassis_plugin_config *config, char *addr)
{
if (!config || !addr || !config->deny_ip_table) return FALSE;
return g_hash_table_remove(config->deny_ip_table, addr);
}
static int admin_show_allow_ip(network_mysqld_con *con, const char *sql) {
char *module_name = str_nth_token(sql, 2);
if (!module_name) {
return PROXY_NO_DECISION;
}
if (strcmp(module_name, "admin") != 0
&& strcmp(module_name, "proxy") != 0
&& strcmp(module_name, "shard") != 0) {
g_free(module_name);
return PROXY_NO_DECISION;
}
chassis *chas = con->srv;
GPtrArray *fields = g_ptr_array_new_with_free_func((void *) network_mysqld_proto_fielddef_free);
MYSQL_FIELD *field;
GPtrArray *rows = g_ptr_array_new_with_free_func((void *) network_mysqld_mysql_field_row_free);
GPtrArray *row;
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Plugin");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Address");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
int i;
for (i = 0; i < chas->modules->len; i++) {
chassis_plugin *plugin = chas->modules->pdata[i];
if (strcmp(plugin->name, module_name) == 0) {
GList *allow_ip_list = plugin->allow_ip_get(plugin->config);
if (allow_ip_list) {
GList *cur_p = allow_ip_list;
while (cur_p) {
row = g_ptr_array_new_with_free_func(g_free);
g_ptr_array_add(row, g_strdup(module_name));
g_ptr_array_add(row, g_strdup((char *) cur_p->data));
g_ptr_array_add(rows, row);
cur_p = cur_p->next;
}
}
break;
}
}
network_mysqld_con_send_resultset(con->client, fields, rows);
/* Free data */
g_ptr_array_free(rows, TRUE);
g_ptr_array_free(fields, TRUE);
g_free(module_name);
return PROXY_SEND_RESULT;
}
static int admin_show_deny_ip(network_mysqld_con *con, const char *sql) {
char *module_name = str_nth_token(sql, 2);
if (!module_name) {
return PROXY_NO_DECISION;
}
if (strcmp(module_name, "admin") != 0
&& strcmp(module_name, "proxy") != 0
&& strcmp(module_name, "shard") != 0) {
g_free(module_name);
return PROXY_NO_DECISION;
}
chassis *chas = con->srv;
GPtrArray *fields = g_ptr_array_new_with_free_func((void *) network_mysqld_proto_fielddef_free);
MYSQL_FIELD *field;
GPtrArray *rows = g_ptr_array_new_with_free_func((void *) network_mysqld_mysql_field_row_free);
GPtrArray *row;
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Plugin");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("Address");
field->type = MYSQL_TYPE_STRING;
g_ptr_array_add(fields, field);
int i;
for (i = 0; i < chas->modules->len; i++) {
chassis_plugin *plugin = chas->modules->pdata[i];
if (strcmp(plugin->name, module_name) == 0) {
GList *deny_ip_list = plugin->deny_ip_get(plugin->config);
if (deny_ip_list) {
GList *cur_p = deny_ip_list;
while (cur_p) {
row = g_ptr_array_new_with_free_func(g_free);
g_ptr_array_add(row, g_strdup(module_name));
g_ptr_array_add(row, g_strdup((char *) cur_p->data));
g_ptr_array_add(rows, row);
cur_p = cur_p->next;
}
}
break;
}
}
network_mysqld_con_send_resultset(con->client, fields, rows);
/* Free data */
g_ptr_array_free(rows, TRUE);
g_ptr_array_free(fields, TRUE);
g_free(module_name);
return PROXY_SEND_RESULT;
}
static int admin_add_allow_ip(network_mysqld_con *con, const char *sql)
{
char *module_name = str_nth_token(sql, 2);
if (!module_name)
return PROXY_NO_DECISION;
char *addr = str_nth_token(sql, 3);
if (!addr)
return PROXY_NO_DECISION;
if (strcmp(module_name, "admin") != 0
&& strcmp(module_name, "proxy") != 0
&& strcmp(module_name, "shard") != 0) {
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
chassis *chas = con->srv;
int i;
gboolean success = FALSE;
for (i = 0; i < chas->modules->len; i++) {
chassis_plugin *plugin = chas->modules->pdata[i];
if (strcmp(plugin->name, module_name) == 0) {
success = plugin->allow_ip_add(plugin->config, addr);
break;
}
}
network_mysqld_con_send_ok_full(con->client, success, 0, SERVER_STATUS_AUTOCOMMIT, 0);
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
static int admin_add_deny_ip(network_mysqld_con *con, const char *sql)
{
char *module_name = str_nth_token(sql, 2);
if (!module_name)
return PROXY_NO_DECISION;
char *addr = str_nth_token(sql, 3);
if (!addr)
return PROXY_NO_DECISION;
if (strcmp(module_name, "admin") != 0
&& strcmp(module_name, "proxy") != 0
&& strcmp(module_name, "shard") != 0) {
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
chassis *chas = con->srv;
int i;
gboolean success = FALSE;
for (i = 0; i < chas->modules->len; i++) {
chassis_plugin *plugin = chas->modules->pdata[i];
if (strcmp(plugin->name, module_name) == 0) {
success = plugin->deny_ip_add(plugin->config, addr);
break;
}
}
network_mysqld_con_send_ok_full(con->client, success, 0, SERVER_STATUS_AUTOCOMMIT, 0);
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
static int admin_delete_allow_ip(network_mysqld_con *con, const char *sql)
{
char *module_name = str_nth_token(sql, 2);
if (!module_name)
return PROXY_NO_DECISION;
char *addr = str_nth_token(sql, 3);
if (!addr)
return PROXY_NO_DECISION;
if (strcmp(module_name, "admin") != 0
&& strcmp(module_name, "proxy") != 0
&& strcmp(module_name, "shard") != 0) {
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
chassis *chas = con->srv;
int i;
gboolean success = FALSE;
for (i = 0; i < chas->modules->len; i++) {
chassis_plugin *plugin = chas->modules->pdata[i];
if (strcmp(plugin->name, module_name) == 0) {
success = plugin->allow_ip_del(plugin->config, addr);
break;
}
}
network_mysqld_con_send_ok_full(con->client, success, 0, SERVER_STATUS_AUTOCOMMIT, 0);
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
static int admin_delete_deny_ip(network_mysqld_con *con, const char *sql)
{
char *module_name = str_nth_token(sql, 2);
if (!module_name)
return PROXY_NO_DECISION;
char *addr = str_nth_token(sql, 3);
if (!addr)
return PROXY_NO_DECISION;
if (strcmp(module_name, "admin") != 0
&& strcmp(module_name, "proxy") != 0
&& strcmp(module_name, "shard") != 0) {
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
chassis *chas = con->srv;
int i;
gboolean success = FALSE;
for (i = 0; i < chas->modules->len; i++) {
chassis_plugin *plugin = chas->modules->pdata[i];
if (strcmp(plugin->name, module_name) == 0) {
success = plugin->deny_ip_del(plugin->config, addr);
break;
}
}
network_mysqld_con_send_ok_full(con->client, success, 0, SERVER_STATUS_AUTOCOMMIT, 0);
g_free(module_name);
g_free(addr);
return PROXY_SEND_RESULT;
}
#define MAKE_FIELD_DEF_1_COL(fields, col_name) \
do {\
MYSQL_FIELD *field = network_mysqld_proto_fielddef_new();\
field->name = g_strdup((col_name)); \
field->type = FIELD_TYPE_VAR_STRING;\
g_ptr_array_add(fields, field); \
}while(0)
#define MAKE_FIELD_DEF_2_COL(fields, col1_name, col2_name) \
do {\
MYSQL_FIELD *field = network_mysqld_proto_fielddef_new();\
field->name = g_strdup((col1_name)); \
field->type = FIELD_TYPE_VAR_STRING;\
g_ptr_array_add(fields, field); \
field = network_mysqld_proto_fielddef_new(); \
field->name = g_strdup((col2_name)); \
field->type = FIELD_TYPE_VAR_STRING;\
g_ptr_array_add(fields, field); \
}while(0)
#define APPEND_ROW_1_COL(rows, row_data) \
do {\
GPtrArray *row = g_ptr_array_new();\
g_ptr_array_add(row, (row_data)); \
g_ptr_array_add(rows, row);\
}while(0)
#define APPEND_ROW_2_COL(rows, col1, col2) \
do {\
GPtrArray *row = g_ptr_array_new();\
g_ptr_array_add(row, (col1)); \
g_ptr_array_add(row, (col2)); \
g_ptr_array_add(rows, row);\
}while(0)
static void strip_extra_spaces(char *str) {
int i,x;
for(i=x=0; str[i]; ++i)
if (!isspace(str[i]) || (i > 0 && !isspace(str[i-1])))
str[x++] = str[i];
str[x] = '\0';
}
static void str_replace(char *p, const char *x, char y)
{
int i,j;
for (i=j=0; p[i]; ++i) {
if (p[i] == x[0] && p[i+1] == x[1]) {
++i;
p[j++] = y;
} else {
p[j++] = p[i];
}
}
p[j] = '\0';
}
static void normalize_equal_sign(char *p)
{
str_replace(p, "= ", '=');
str_replace(p, " =", '=');
}
/*
* tolower, but leave "quoted strings" unmodified
*/
static void lower_identifiers(char *str)
{
gboolean in_string = FALSE;
int i;
for (i = 0; str[i]; ++i) {
if (in_string)
continue;
if (str[i] == '\'' || str[i] == '"')
in_string = !in_string;
if (isalpha(str[i]))
str[i] = tolower(str[i]);
}
}
/* only match % wildcard, case insensitive */
static gboolean sql_pattern_like(const char *pattern, const char *string)
{
if (!pattern || pattern[0] == '\0')
return TRUE;
char *glob = g_strdup(pattern);
int i;
for (i = 0; glob[i]; ++i) {
if (glob[i] == '%') glob[i] = '*';
glob[i] = tolower(glob[i]);
}
char *lower_str = g_ascii_strdown(string, -1);
gboolean rc = g_pattern_match_simple(glob, lower_str);
g_free(glob);
g_free(lower_str);
return rc;
}
/* returned list must be freed */
static GList *admin_get_all_options(chassis *chas)
{
GList* options = g_list_copy(chas->options->options); /* shallow copy */
return options;
}
static int admin_show_variables(network_mysqld_con *con, const char *sql)
{
char **tokens = g_strsplit(sql, " ", -1);
int token_count = g_strv_length(tokens);
char *pattern = NULL;
if (token_count == 2) {
pattern = "%";
} else if (token_count == 4) {
pattern = tokens[3];
cetus_string_dequote(pattern);
} else {
network_mysqld_con_send_error(con->client, C("error syntax"));
g_strfreev(tokens);
return PROXY_SEND_RESULT;
}
GList *options = admin_get_all_options(con->srv);
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "Variable_name", "Value");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
GList *freelist = NULL;
GList *l = NULL;
for (l = options; l; l = l->next) {
chassis_option_t *opt = l->data;
/* just support these for now */
if (opt->arg != OPTION_ARG_INT && opt->arg != OPTION_ARG_INT64
&&opt->arg != OPTION_ARG_DOUBLE && opt->arg != OPTION_ARG_STRING
&&opt->arg != OPTION_ARG_NONE)
continue;
if (sql_pattern_like(pattern, opt->long_name)) {
char *value = chassis_option_get_value_str(opt);
freelist = g_list_append(freelist, value);
APPEND_ROW_2_COL(rows, (char *)opt->long_name, value);
}
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
g_list_free_full(freelist, g_free);
g_list_free(options);
g_strfreev(tokens);
return PROXY_SEND_RESULT;
}
static int admin_show_status(network_mysqld_con *con, const char *sql)
{
char **tokens = g_strsplit(sql, " ", -1);
int token_count = g_strv_length(tokens);
char *pattern = NULL;
if (token_count == 2) {
pattern = "%";
} else if (token_count == 4) {
pattern = tokens[3];
cetus_string_dequote(pattern);
} else {
network_mysqld_con_send_error(con->client, C("error syntax"));
g_strfreev(tokens);
return PROXY_SEND_RESULT;
}
cetus_variable_t *variables = con->srv->priv->stats_variables;
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "Variable_name", "Value");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
GList *freelist = NULL;
int i = 0;
for (i = 0; variables[i].name; ++i) {
if (sql_pattern_like(pattern, variables[i].name)) {
char *value = cetus_variable_get_value_str(&variables[i]);
freelist = g_list_append(freelist, value);
APPEND_ROW_2_COL(rows, variables[i].name, value);
}
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
g_list_free_full(freelist, g_free);
g_strfreev(tokens);
return PROXY_SEND_RESULT;
}
static int admin_set_reduce_conns(network_mysqld_con *con, const char *sql)
{
char *mode = str_nth_token(sql, 2);
if (mode) {
if (strcasecmp(mode, "true") == 0) {
con->srv->is_reduce_conns = 1;
} else if (strcasecmp(mode, "false") == 0) {
con->srv->is_reduce_conns = 0;
}
g_free(mode);
}
network_mysqld_con_send_ok(con->client);
return PROXY_SEND_RESULT;
}
static int admin_reduce_memory(network_mysqld_con *con, const char *sql)
{
struct mallinfo m;
m = mallinfo();
g_message("%s:Total allocated space (bytes): %d", G_STRLOC, m.uordblks);
g_message("%s:Total free space (bytes): %d", G_STRLOC, m.fordblks);
g_message("%s:Top-most, releasable space (bytes): %d", G_STRLOC, m.keepcost);
if (m.fordblks > m.uordblks) {
malloc_trim(0);
m = mallinfo();
g_message("%s:After trim, total allocated space (bytes): %d", G_STRLOC, m.uordblks);
g_message("%s:After trim, total free space (bytes): %d", G_STRLOC, m.fordblks);
g_message("%s:After trim, top-most, releasable space (bytes): %d", G_STRLOC, m.keepcost);
}
network_mysqld_con_send_ok_full(con->client, 1, 0, SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_set_maintain(network_mysqld_con *con, const char *sql)
{
gboolean error = FALSE;
char *mode = str_nth_token(sql, 2);
if (mode) {
if (strcasecmp(mode, "true") == 0) {
con->srv->maintain_close_mode = 1;
} else if (strcasecmp(mode, "false") == 0) {
con->srv->maintain_close_mode = 0;
} else {
error = TRUE;
}
g_free(mode);
} else {
error = TRUE;
}
if (error)
return PROXY_NO_DECISION;
network_mysqld_con_send_ok(con->client);
return PROXY_SEND_RESULT;
}
static int admin_reload_shard(network_mysqld_con *con, const char *sql)
{
network_mysqld_con_send_ok_full(con->client, 1, 0, SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_send_version(network_mysqld_con *con, const char *sql)
{
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_1_COL(fields, "cetus version");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
APPEND_ROW_1_COL(rows, PLUGIN_VERSION);
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
return PROXY_SEND_RESULT;
}
static int admin_send_connection_stat(network_mysqld_con *con, const char *sql)
{
int backend_ndx = -1;
char user[128] = {0};
sscanf(sql,
"select conn_num from backends where backend_ndx=%d and user=%*['\"]%64[^'\"]%*['\"]",
&backend_ndx, user);
if (backend_ndx == -1 || user[0] == '\0')
return PROXY_NO_DECISION;
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_1_COL(fields, "connection_num");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
char *numstr = NULL;
chassis_private *g = con->srv->priv;
backend_ndx -= 1; /* index in sql start from 1, not 0 */
if (backend_ndx >= 0 && backend_ndx < network_backends_count(g->backends)) {
network_backend_t *backend = network_backends_get(g->backends, backend_ndx);
GString *user_name = g_string_new(user);
/* TODO: if robbed, conns is not for user_name */
GQueue *conns = network_connection_pool_get_conns(backend->pool, user_name, NULL);
if (conns) {
numstr = g_strdup_printf("%d", conns->length);
}
g_string_free(user_name, TRUE);
}
if (numstr) {
APPEND_ROW_1_COL(rows, numstr);
} else {
APPEND_ROW_1_COL(rows, "0");
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
if (numstr)
g_free(numstr);
return PROXY_SEND_RESULT;
}
static void bytes_to_hex_str(char *pin, int len, char *pout)
{
const char *hex = "0123456789ABCDEF";
int i = 0;
for(; i < len; ++i) {
*pout++ = hex[(*pin>>4)&0xF];
*pout++ = hex[(*pin++)&0xF];
}
*pout = 0;
}
static int admin_send_user_password(network_mysqld_con *con, const char *sql)
{
char from_table[128] = {0};
char user[128] = {0};
char *where_start = strcasestr(sql, "where");
if (where_start) {
sscanf(sql, "select * from %s where user=%*['\"]%64[^'\"]%*['\"]", from_table, user);
if (from_table[0] == '\0' || user[0] == '\0')
return PROXY_NO_DECISION;
} else {
sscanf(sql, "select * from %s", from_table);
if (from_table[0] == '\0')
return PROXY_NO_DECISION;
}
chassis_private *g = con->srv->priv;
enum cetus_pwd_type pwd_type;
if (strcmp(from_table, "user_pwd") == 0) {
pwd_type = CETUS_SERVER_PWD;
} else if (strcmp(from_table, "app_user_pwd") == 0) {
pwd_type = CETUS_CLIENT_PWD;
} else {
g_critical("error target db");
return PROXY_NO_DECISION;
}
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "user", "password(sha1)");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
if (user[0]) { /* one user */
GString *hashpwd = g_string_new(0);
cetus_users_get_hashed_pwd(g->users, user, pwd_type, hashpwd);
char *pwdhex = NULL;
if (hashpwd->len > 0) {
pwdhex = g_malloc0(hashpwd->len *2 + 10);
bytes_to_hex_str(hashpwd->str, hashpwd->len, pwdhex);
APPEND_ROW_2_COL(rows, user, pwdhex);
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
g_string_free(hashpwd, TRUE);
if (pwdhex)
g_free(pwdhex);
} else { /* all users */
GList *strings_to_free = NULL;
GHashTableIter iter;
char *username = NULL;
GString *hashpwd = g_string_new(0);
char *hack = NULL; /* don't use value directly */
g_hash_table_iter_init(&iter, g->users->records);
while (g_hash_table_iter_next(&iter, (gpointer *)&username, (gpointer *)&hack)) {
cetus_users_get_hashed_pwd(g->users, username, pwd_type, hashpwd);
char *pwdhex = NULL;
if (hashpwd->len > 0) {
pwdhex = g_malloc0(hashpwd->len *2 + 10);
bytes_to_hex_str(hashpwd->str, hashpwd->len, pwdhex);
strings_to_free = g_list_append(strings_to_free, pwdhex);
}
APPEND_ROW_2_COL(rows, username, pwdhex);
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
g_string_free(hashpwd, TRUE);
if (strings_to_free)
g_list_free_full(strings_to_free, g_free);
}
return PROXY_SEND_RESULT;
}
/* update or insert */
static int admin_update_user_password(network_mysqld_con *con, const char *sql)
{
char from_table[128] = {0};
char user[128] = {0};
char new_pwd[128] = {0};
sscanf(sql, "update %64s set password=%*['\"]%64[^'\"]%*['\"] where user=%*['\"]%64[^'\"]%*['\"]",
from_table, new_pwd, user);
if (new_pwd[0] == '\0' || from_table[0] == '\0' || user[0] == '\0')
return PROXY_NO_DECISION;
chassis_private *g = con->srv->priv;
enum cetus_pwd_type pwd_type = CETUS_UNKNOWN_PWD;
if (strcmp(from_table, "user_pwd") == 0) {
pwd_type = CETUS_SERVER_PWD;
} else if (strcmp(from_table, "app_user_pwd") == 0) {
pwd_type = CETUS_CLIENT_PWD;
}
gboolean affected = cetus_users_update_record(g->users, user, new_pwd, pwd_type);
if (affected)
cetus_users_write_json(g->users);
network_mysqld_con_send_ok_full(con->client, affected?1:0, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_delete_user_password(network_mysqld_con *con, const char *sql)
{
char from_table[128] = {0};
char user[128] = {0};
sscanf(sql, "delete from %64s where user=%*['\"]%64[^'\"]%*['\"]", from_table, user);
if (from_table[0] == '\0' || user[0] == '\0')
return PROXY_NO_DECISION;
chassis_private *g = con->srv->priv;
gboolean affected = cetus_users_delete_record(g->users, user);
if (affected)
cetus_users_write_json(g->users);
network_mysqld_con_send_ok_full(con->client, affected?1:0, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static backend_type_t backend_type(const char *str)
{
backend_type_t type = BACKEND_TYPE_UNKNOWN;
if (strcasecmp(str, "ro") == 0)
type = BACKEND_TYPE_RO;
else if (strcasecmp(str, "rw") == 0)
type = BACKEND_TYPE_RW;
return type;
}
static backend_state_t backend_state(const char *str)
{
backend_state_t state = BACKEND_STATE_UNKNOWN;
if (strcasecmp(str, "up") == 0)
state = BACKEND_STATE_UP;
else if (strcasecmp(str, "down") == 0)
state = BACKEND_STATE_DOWN;
else if (strcasecmp(str, "maintaining") == 0)
state = BACKEND_STATE_MAINTAINING;
return state;
}
static int admin_insert_backend(network_mysqld_con *con, const char *sql)
{
/* TODO: to which group */
char address[128] = {0};
char type_str[64] = {0};
char state_str[64] = {0};
sscanf(sql, "insert into backends values (%*['\"]%64[0-9:.@a-zA-Z]%*['\"], %*['\"]%32[rowRWO]%*['\"], %*['\"]%32[a-zA-Z]%*['\"])",
address, type_str, state_str);
if (address[0] == '\0' || type_str[0] == '\0' || state_str[0] == '\0')
return PROXY_NO_DECISION;
chassis_private *g = con->srv->priv;
int affected = network_backends_add(g->backends, address,
backend_type(type_str),
backend_state(state_str),
con->srv);
network_mysqld_con_send_ok_full(con->client, affected == 0?1:0, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_add_backend(network_mysqld_con *con, const char *sql)
{
char address[128] = {0};
backend_type_t type = BACKEND_TYPE_UNKNOWN;
if (strncasecmp(sql, "add master", 10) == 0) {
sscanf(sql, "add master %*['\"]%64[0-9:.@a-zA-Z]%*['\"]", address);
type = BACKEND_TYPE_RW;
} else if (strncasecmp(sql, "add slave", 9) == 0) {
sscanf(sql, "add slave %*['\"]%64[0-9:.@a-zA-Z]%*['\"]", address);
type = BACKEND_TYPE_RO;
}
if (address[0] == '\0' || type == BACKEND_TYPE_UNKNOWN)
return PROXY_NO_DECISION;
chassis_private *g = con->srv->priv;
int affected = network_backends_add(g->backends, address, type, BACKEND_STATE_UNKNOWN, con->srv);
network_mysqld_con_send_ok_full(con->client, affected == 0?1:0, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_update_backend(network_mysqld_con *con, const char *sql)
{
char *set_start = strcasestr(sql, "set");
char *where_start = strcasestr(sql, "where");
char *set_cols_str = NULL;
if (where_start) {
set_cols_str = g_strndup(set_start+4, where_start-set_start-4);
} else {
set_cols_str = g_strdup(set_start+4);
}
char **set_cols_array = g_strsplit(set_cols_str, ",", -1);
int i = 0;
char type_str[64] = {0};
char state_str[64] = {0};
char *col;
for (; (col = set_cols_array[i]) != NULL; ++i) {
g_strstrip(col);
if (strncasecmp(col, "type", 4) == 0) {
sscanf(col, "type=%*['\"]%32[a-zA-Z]%*['\"]", type_str);
} else if (strncasecmp(col, "state", 5) == 0) {
sscanf(col, "state=%*['\"]%32[a-zA-Z]%*['\"]", state_str);
}
}
g_free(set_cols_str);
g_strfreev(set_cols_array);
if (type_str[0] == '\0' && state_str[0] == '\0')
return PROXY_NO_DECISION;
chassis_private *g = con->srv->priv;
int affected_rows = 0;
if (where_start) {
int backend_ndx = -1;
sscanf(where_start, "where backend_ndx=%d", &backend_ndx);
backend_ndx -= 1; /* index in sql start from 1, not 0 */
if (backend_ndx < 0) {
char address[128] = {0};
sscanf(where_start, "where address=%*['\"]%64[0-9:.]%*['\"]", address);
backend_ndx = network_backends_find_address(g->backends, address);
}
network_backend_t *bk = network_backends_get(g->backends, backend_ndx);
if (!bk) {
network_mysqld_con_send_error(con->client, C("no such backend"));
return PROXY_SEND_RESULT;
}
int type = type_str[0] != '\0' ? backend_type(type_str) : bk->type;
int state = state_str[0] != '\0' ? backend_state(state_str) : bk->state;
network_backends_modify(g->backends, backend_ndx, type, state);
affected_rows = 1;
} else {
for (i = 0; i < network_backends_count(g->backends); ++i) {
network_backend_t *bk = network_backends_get(g->backends, i);
if (bk) {
int type = type_str[0] != '\0' ? backend_type(type_str) : bk->type;
int state = state_str[0] != '\0' ? backend_state(state_str) : bk->state;
network_backends_modify(g->backends, i, type, state);
affected_rows += 1;
}
}
}
network_mysqld_con_send_ok_full(con->client, affected_rows, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_delete_backend(network_mysqld_con *con, const char *sql)
{
char *where_start = strcasestr(sql, "where");
chassis_private *g = con->srv->priv;
int backend_ndx = -1;
if (where_start) {/* delete from backends where xx=xx */
sscanf(where_start, "where backend_ndx=%d", &backend_ndx);
backend_ndx -= 1; /* index in sql start from 1, not 0 */
if (backend_ndx < 0) {
char address[128] = {0};
sscanf(where_start, "where address=%*['\"]%64[0-9:.]%*['\"]", address);
backend_ndx = network_backends_find_address(g->backends, address);
}
} else {/* remove backend xxx */
sscanf(sql, "remove backend %d", &backend_ndx);
backend_ndx -= 1; /* index in sql start from 1, not 0 */
}
if (backend_ndx >= 0 && backend_ndx < network_backends_count(g->backends)) {
network_backends_remove(g->backends, backend_ndx);/* TODO: just change state? */
network_mysqld_con_send_ok_full(con->client, 1, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
} else {
network_mysqld_con_send_ok_full(con->client, 0, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
}
static int admin_supported_stats(network_mysqld_con *con)
{
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_1_COL(fields, "name");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
APPEND_ROW_1_COL(rows, "client_query");
APPEND_ROW_1_COL(rows, "proxyed_query");
APPEND_ROW_1_COL(rows, "reset");
APPEND_ROW_1_COL(rows, "query_time_table");
APPEND_ROW_1_COL(rows, "server_query_details");
APPEND_ROW_1_COL(rows, "query_wait_table");
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
return PROXY_SEND_RESULT;
}
static int admin_get_stats(network_mysqld_con *con, const char *sql)
{
const char *p = sql + 9;
if (*p == '\0') { /* just "stats get", no argument */
return admin_supported_stats(con);
} else {
++p; /* stats get [xxx], point to xxx */
}
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "name", "value");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
chassis *chas = con->srv;
query_stats_t *stats = &(chas->query_stats);
char buf1[32] = {0};
char buf2[32] = {0};
int i;
if (strcasecmp(p, "client_query") == 0) {
snprintf(buf1, 32, "%lu", stats->client_query.ro);
snprintf(buf2, 32, "%lu", stats->client_query.rw);
APPEND_ROW_2_COL(rows, "client_query.ro", buf1);
APPEND_ROW_2_COL(rows, "client_query.rw", buf2);
} else if (strcasecmp(p, "proxyed_query") == 0) {
snprintf(buf1, 32, "%lu", stats->proxyed_query.ro);
snprintf(buf2, 32, "%lu", stats->proxyed_query.rw);
APPEND_ROW_2_COL(rows, "proxyed_query.ro", buf1);
APPEND_ROW_2_COL(rows, "proxyed_query.rw", buf2);
} else if (strcasecmp(p, "query_time_table") == 0) {
for (i = 0; i < MAX_QUERY_TIME && stats->query_time_table[i]; ++i) {
GPtrArray *row = g_ptr_array_new_with_free_func(g_free);
g_ptr_array_add(row, g_strdup_printf("query_time_table.%d", i+1));
g_ptr_array_add(row, g_strdup_printf("%lu", stats->query_time_table[i]));
g_ptr_array_add(rows, row);
}
} else if (strcasecmp(p, "query_wait_table") == 0) {
for (i = 0; i < MAX_QUERY_TIME && stats->query_wait_table[i]; ++i) {
GPtrArray *row = g_ptr_array_new_with_free_func(g_free);
g_ptr_array_add(row, g_strdup_printf("query_wait_table.%d", i+1));
g_ptr_array_add(row, g_strdup_printf("%lu", stats->query_wait_table[i]));
g_ptr_array_add(rows, row);
}
} else if (strcasecmp(p, "server_query_details") == 0) {
for (i = 0; i < MAX_SERVER_NUM && i < network_backends_count(chas->priv->backends); ++i) {
GPtrArray *row = g_ptr_array_new_with_free_func(g_free);
g_ptr_array_add(row, g_strdup_printf("server_query_details.%d.ro", i+1));
g_ptr_array_add(row, g_strdup_printf("%lu", stats->server_query_details[i].ro));
g_ptr_array_add(rows, row);
row = g_ptr_array_new_with_free_func(g_free);
g_ptr_array_add(row, g_strdup_printf("server_query_details.%d.rw", i+1));
g_ptr_array_add(row, g_strdup_printf("%lu", stats->server_query_details[i].rw));
g_ptr_array_add(rows, row);
}
} else if (strcasecmp(p, "reset") == 0) {
APPEND_ROW_2_COL(rows, "reset", "0");
} else {
APPEND_ROW_2_COL(rows, (char *)p, (char *)p);
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
return PROXY_SEND_RESULT;
}
static int admin_supported_config(network_mysqld_con *con)
{
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_1_COL(fields, "name");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
APPEND_ROW_1_COL(rows, "common");
APPEND_ROW_1_COL(rows, "pool");
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
return PROXY_SEND_RESULT;
}
static int admin_get_config(network_mysqld_con *con, const char *sql)
{
const char *p = sql + 10;
if (*p == '\0') { /* just "config get", no argument */
return admin_supported_config(con);
} else {
++p; /* config get [xxx], point to xxx */
}
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "name", "value");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
chassis *chas = con->srv;
char buf1[32] = {0}, buf2[32] = {0};
char buf3[32] = {0}, buf4[32] = {0};
if (strcasecmp(p, "common") == 0) {
snprintf(buf1, 32, "%d", chas->check_slave_delay);
snprintf(buf2, 32, "%f", chas->slave_delay_down_threshold_sec);
snprintf(buf3, 32, "%f", chas->slave_delay_recover_threshold_sec);
snprintf(buf4, 32, "%d", chas->long_query_time);
APPEND_ROW_2_COL(rows, "common.check_slave_delay", buf1);
APPEND_ROW_2_COL(rows, "common.slave_delay_down_threshold_sec", buf2);
APPEND_ROW_2_COL(rows, "common.slave_delay_recover_threshold_sec", buf3);
APPEND_ROW_2_COL(rows, "common.long_query_time", buf4);
} else if (strcasecmp(p, "pool") == 0) {
snprintf(buf1, 32, "%d", chas->mid_idle_connections);
snprintf(buf2, 32, "%d", chas->max_idle_connections);
snprintf(buf3, 32, "%d", chas->max_resp_len);
snprintf(buf4, 32, "%d", chas->master_preferred);
APPEND_ROW_2_COL(rows, "pool.default_pool_size", buf1);
APPEND_ROW_2_COL(rows, "pool.max_pool_size", buf2);
APPEND_ROW_2_COL(rows, "pool.max_resp_len", buf3);
APPEND_ROW_2_COL(rows, "pool.master_preferred", buf4);
} else {
APPEND_ROW_2_COL(rows, (char *)p, (char *)p);
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
return PROXY_SEND_RESULT;
}
static int admin_set_config(network_mysqld_con *con, const char *sql)
{
char key[128] = {0};
int val = -1;
sscanf(sql, "config set %64[a-zA-Z0-9_.]=%d", key, &val);
if (key[0] == '\0' || val == -1) {
network_mysqld_con_send_ok_full(con->client, 0, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
chassis *chas = con->srv;
int affected_rows = 1;
if (strcasecmp(key, "common.check_slave_delay") == 0) {
chas->check_slave_delay = val;
} else if (strcasecmp(key, "common.slave_delay_down_threshold_sec") == 0) {
chas->slave_delay_down_threshold_sec = val;
} else if (strcasecmp(key, "common.slave_delay_recover_threshold_sec") == 0) {
chas->slave_delay_recover_threshold_sec = val;
} else if (strcasecmp(key, "common.long_query_time") == 0) {
chas->long_query_time = val;
} else if (strcasecmp(key, "pool.default_pool_size") == 0) {
chas->mid_idle_connections = val;
} else if (strcasecmp(key, "pool.max_pool_size") == 0) {
chas->max_idle_connections = val;
} else if (strcasecmp(key, "pool.max_resp_len") == 0) {
chas->max_resp_len = val;
} else if (strcasecmp(key, "pool.master_preferred") == 0) {
chas->master_preferred = val;
} else {
affected_rows = 0;
}
network_mysqld_con_send_ok_full(con->client, affected_rows, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_reset_stats(network_mysqld_con *con, const char *sql)
{
query_stats_t *stats = &con->srv->query_stats;
memset(stats, 0, sizeof(*stats));
network_mysqld_con_send_ok_full(con->client, 1, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static int admin_save_settings(network_mysqld_con *con, const char *sql)
{
/* TODO:*/
network_mysqld_con_send_ok_full(con->client, 1, 0,
SERVER_STATUS_AUTOCOMMIT, 0);
return PROXY_SEND_RESULT;
}
static void calc_qps_average(char *buf, int len);
static void calc_tps_average(char *buf, int len);
static void get_module_names(chassis *chas, GString *plugin_names)
{
int i;
for (i = 0; i < chas->modules->len; ++i) {
chassis_plugin *p = g_ptr_array_index(chas->modules, i);
g_string_append(plugin_names, p->name);
g_string_append_c(plugin_names, ' ');
}
}
static int admin_send_status(network_mysqld_con *con, const char *sql)
{
chassis_private *g = con->srv->priv;
chassis_plugin_config *config = con->config;
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "Status", "Value");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
APPEND_ROW_2_COL(rows, "Cetus version", PLUGIN_VERSION);
char start_time[32];
chassis_epoch_to_string(&con->srv->startup_time, C(start_time));
APPEND_ROW_2_COL(rows, "Startup time", start_time);
GString *plugin_names = g_string_new(0);
get_module_names(con->srv, plugin_names);
APPEND_ROW_2_COL(rows, "Loaded modules", plugin_names->str);
const int bsize = 32;
static char buf1[32], buf2[32], buf3[32];
snprintf(buf1, bsize, "%d", network_backends_idle_conns(g->backends));
APPEND_ROW_2_COL(rows, "Idle backend connections", buf1);
snprintf(buf2, bsize, "%d", network_backends_used_conns(g->backends));
APPEND_ROW_2_COL(rows, "Used backend connections", buf2);
snprintf(buf3, bsize, "%d", g->cons->len);
APPEND_ROW_2_COL(rows, "Client connections", buf3);
query_stats_t *stats = &(con->srv->query_stats);
char qcount[32];
snprintf(qcount, 32, "%ld", stats->client_query.ro+stats->client_query.rw);
APPEND_ROW_2_COL(rows, "Query count", qcount);
if (config->has_shard_plugin) {
char xacount[32];
snprintf(xacount, 32, "%ld", stats->xa_count);
APPEND_ROW_2_COL(rows, "XA count", xacount);
}
char qps[64];
calc_qps_average(C(qps));
APPEND_ROW_2_COL(rows, "QPS (1min, 5min, 15min)", qps);
char tps[64];
calc_tps_average(C(tps));
APPEND_ROW_2_COL(rows, "TPS (1min, 5min, 15min)", tps);
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
g_string_free(plugin_names, TRUE);
return PROXY_SEND_RESULT;
}
static int admin_send_group_info(network_mysqld_con *con, const char *sql)
{
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MYSQL_FIELD *field = network_mysqld_proto_fielddef_new();
field->name = g_strdup("group");
field->type = FIELD_TYPE_VAR_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup(("master"));
field->type = FIELD_TYPE_VAR_STRING;
g_ptr_array_add(fields, field);
field = network_mysqld_proto_fielddef_new();
field->name = g_strdup(("slaves"));
field->type = FIELD_TYPE_VAR_STRING;
g_ptr_array_add(fields, field);
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
GList *free_list = NULL;
network_backends_t *bs = con->srv->priv->backends;
int i;
for (i = 0; i < bs->groups->len; ++i) {
network_group_t *gp = g_ptr_array_index(bs->groups, i);
GPtrArray *row = g_ptr_array_new();
g_ptr_array_add(row, gp->name->str);
if (gp->master) {
g_ptr_array_add(row, gp->master->addr->name->str);
} else {
g_ptr_array_add(row, "");
}
GString *slaves = g_string_new(0);
network_group_get_slave_names(gp, slaves);
g_ptr_array_add(row, slaves->str);
free_list = g_list_append(free_list, slaves);
g_ptr_array_add(rows, row);
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
g_list_free_full(free_list, g_string_true_free);
return PROXY_SEND_RESULT;
}
static int admin_help(network_mysqld_con *con, const char *sql);
typedef int(*sql_handler_func)(network_mysqld_con *, const char *);
struct sql_handler_entry_t {
const char *prefix;
sql_handler_func func;
const char *pattern;
const char *desc;
};
static struct sql_handler_entry_t sql_handler_shard_map[] = {
{"select conn_details from backend", admin_send_backend_detail_info,
"select conn_details from backend", "display the idle conns"},
{"select * from backends", admin_send_backends_info,
"select * from backends", "list the backends and their state"},
{"select * from groups", admin_send_group_info,
"select * from groups"},
{"show connectionlist", admin_show_connectionlist,
"show connectionlist [<num>]", "show <num> connections"},
{"show allow_ip ", admin_show_allow_ip,
"show allow_ip <module>", "show allow_ip rules of module, currently admin|proxy|shard"},
{"show deny_ip ", admin_show_deny_ip,
"show deny_ip <module>", "show deny_ip rules of module, currently admin|proxy|shard"},
{"add allow_ip ", admin_add_allow_ip,
"add allow_ip <module> <address>", "add address to white list of module"},
{"add deny_ip ", admin_add_deny_ip,
"add deny_ip <module> <address>", "add address to white list of module"},
{"delete allow_ip ", admin_delete_allow_ip,
"delete allow_ip <module> <address>", "delete address from white list of module"},
{"delete deny_ip ", admin_delete_deny_ip,
"delete deny_ip <module> <address>", "delete address from white list of module"},
{"set reduce_conns ", admin_set_reduce_conns,
"set reduce_conns (true|false)", "reduce idle connections if set to true"},
{"reduce memory", admin_reduce_memory,
"reduce memory", "reduce memory occupied by system"},
{"set maintain ", admin_set_maintain,
"set maintain (true|false)", "close all client connections if set to true"},
{"reload shard", admin_reload_shard,
"reload shard", "reload sharding config from remote db"},
{"show status", admin_show_status,
"show status [like '%pattern%']", "show select/update/insert/delete statistics"},
{"show variables", admin_show_variables,
"show variables [like '%pattern%']"},
{"select version", admin_send_version,
"select version", "cetus version"},
{"select conn_num from backends where", admin_send_connection_stat,
"select conn_num from backends where backend_ndx=<index> and user='<name>')"},
{"select * from user_pwd", admin_send_user_password,
"select * from user_pwd [where user='<name>']"},
{"select * from app_user_pwd", admin_send_user_password,
"select * from app_user_pwd [where user='<name>']"},
{"update user_pwd set password", admin_update_user_password,
"update user_pwd set password='xx' where user='<name>'"},
{"update app_user_pwd set password", admin_update_user_password,
"update app_user_pwd set password='xx' where user='<name>'"},
{"delete from user_pwd where", admin_delete_user_password,
"delete from user_pwd where user='<name>'"},
{"delete from app_user_pwd where", admin_delete_user_password,
"delete from app_user_pwd where user='<name>'"},
{"insert into backends values", admin_insert_backend,
"insert into backends values ('<ip:port@group>', '(ro|rw)', '<state>')",
"add mysql instance to backends list"},
{"update backends set", admin_update_backend,
"update backends set (type|state)=x where (backend_ndx=<index>|address=<'ip:port'>)",
"update mysql instance type or state"},
{"delete from backends", admin_delete_backend,
"delete from backends where (backend_ndx=<index>|address=<'ip:port'>)"},
{"remove backend ", admin_delete_backend, /* TODO: unify */
"remove backend where (backend_ndx=<index>|address=<'ip:port'>)"},
{"add master", admin_add_backend, "add master <'ip:port@group'>"},
{"add slave", admin_add_backend, "add slave <'ip:port@group'>"},
{"stats get", admin_get_stats, "stats get [<item>]", "show query statistics"},
{"config get", admin_get_config, "config get [<item>]", "show config"},
{"config set ", admin_set_config, "config set <key>=<value>"},
{"stats reset", admin_reset_stats, "stats reset", "reset query statistics"},
{"save settings ", admin_save_settings, "save settings", "not implemented"},
{"select * from help", admin_help, "select * from help", "show this help"},
{"select help", admin_help, "select help", "show this help"},
{"cetus", admin_send_status, "cetus", "Show overall status of Cetus"},
{NULL, NULL, NULL, NULL}
};
static struct sql_handler_entry_t sql_handler_rw_map[] = {
{"select conn_details from backend", admin_send_backend_detail_info,
"select conn_details from backend", "display the idle conns"},
{"select * from backends", admin_send_backends_info,
"select * from backends", "list the backends and their state"},
{"show connectionlist", admin_show_connectionlist,
"show connectionlist [<num>]", "show <num> connections"},
{"show allow_ip ", admin_show_allow_ip,
"show allow_ip <module>", "show allow_ip rules of module, currently admin|proxy|shard"},
{"show deny_ip ", admin_show_deny_ip,
"show deny_ip <module>", "show deny_ip rules of module, currently admin|proxy|shard"},
{"add allow_ip ", admin_add_allow_ip,
"add allow_ip <module> <address>", "add address to white list of module"},
{"add deny_ip ", admin_add_deny_ip,
"add deny_ip <module> <address>", "add address to white list of module"},
{"delete allow_ip ", admin_delete_allow_ip,
"delete allow_ip <module> <address>", "delete address from white list of module"},
{"delete deny_ip ", admin_delete_deny_ip,
"delete deny_ip <module> <address>", "delete address from white list of module"},
{"set reduce_conns ", admin_set_reduce_conns,
"set reduce_conns (true|false)", "reduce idle connections if set to true"},
{"reduce memory", admin_reduce_memory,
"reduce memory", "reduce memory occupied by system"},
{"set maintain ", admin_set_maintain,
"set maintain (true|false)", "close all client connections if set to true"},
{"show status", admin_show_status,
"show status [like '%pattern%']", "show select/update/insert/delete statistics"},
{"show variables", admin_show_variables,
"show variables [like '%pattern%']"},
{"select version", admin_send_version,
"select version", "cetus version"},
{"select conn_num from backends where", admin_send_connection_stat,
"select conn_num from backends where backend_ndx=<index> and user='<name>')"},
{"select * from user_pwd", admin_send_user_password,
"select * from user_pwd [where user='<name>']"},
{"select * from app_user_pwd", admin_send_user_password,
"select * from app_user_pwd [where user='<name>']"},
{"update user_pwd set password", admin_update_user_password,
"update user_pwd set password='xx' where user='<name>'"},
{"update app_user_pwd set password", admin_update_user_password,
"update app_user_pwd set password='xx' where user='<name>'"},
{"delete from user_pwd where", admin_delete_user_password,
"delete from user_pwd where user='<name>'"},
{"delete from app_user_pwd where", admin_delete_user_password,
"delete from app_user_pwd where user='<name>'"},
{"insert into backends values", admin_insert_backend,
"insert into backends values ('<ip:port>', '(ro|rw)', '<state>')",
"add mysql instance to backends list"},
{"update backends set", admin_update_backend,
"update backends set (type|state)=x where (backend_ndx=<index>|address=<'ip:port'>)",
"update mysql instance type or state"},
{"delete from backends", admin_delete_backend,
"delete from backends where (backend_ndx=<index>|address=<'ip:port'>)"},
{"remove backend ", admin_delete_backend, /* TODO: unify */
"remove backend where (backend_ndx=<index>|address=<'ip:port'>)"},
{"add master", admin_add_backend, "add master <'ip:port'>"},
{"add slave", admin_add_backend, "add slave <'ip:port'>"},
{"stats get", admin_get_stats, "stats get [<item>]", "show query statistics"},
{"config get", admin_get_config, "config get [<item>]", "show config"},
{"config set ", admin_set_config, "config set <key>=<value>"},
{"stats reset", admin_reset_stats, "stats reset", "reset query statistics"},
{"save settings ", admin_save_settings, "save settings", "not implemented"},
{"select * from help", admin_help, "select * from help", "show this help"},
{"select help", admin_help, "select help", "show this help"},
{"cetus", admin_send_status, "cetus", "Show overall status of Cetus"},
{NULL, NULL, NULL, NULL}
};
static int admin_help(network_mysqld_con *con, const char *sql)
{
GPtrArray *fields = network_mysqld_proto_fielddefs_new();
MAKE_FIELD_DEF_2_COL(fields, "Command", "Description");
GPtrArray *rows = g_ptr_array_new_with_free_func(
(void *)network_mysqld_mysql_field_row_free);
chassis_plugin_config *config = con->config;
struct sql_handler_entry_t *sql_handler_map =
config->has_shard_plugin ? sql_handler_shard_map : sql_handler_rw_map;
int i;
for (i = 0; sql_handler_map[i].prefix; ++i) {
struct sql_handler_entry_t *e = &(sql_handler_map[i]);
char *pattern = e->pattern ? (char *)e->pattern : "";
char *desc = e->desc ? (char *)e->desc : "";
APPEND_ROW_2_COL(rows, pattern, desc);
}
network_mysqld_con_send_resultset(con->client, fields, rows);
network_mysqld_proto_fielddefs_free(fields);
g_ptr_array_free(rows, TRUE);
return PROXY_SEND_RESULT;
}
static network_mysqld_stmt_ret
admin_process_query(network_mysqld_con *con)
{
char command = -1;
network_socket *recv_sock = con->client;
GList *chunk = recv_sock->recv_queue->chunks->head;
GString *packet = chunk->data;
if (packet->len < NET_HEADER_SIZE) {
/* packet too short */
return PROXY_SEND_QUERY;
}
command = packet->str[NET_HEADER_SIZE + 0];
if (COM_QUERY == command) {
/* we need some more data after the COM_QUERY */
if (packet->len < NET_HEADER_SIZE + 2) return PROXY_SEND_QUERY;
/* LOAD DATA INFILE is nasty */
if (packet->len - NET_HEADER_SIZE - 1 >= sizeof("LOAD ") - 1 &&
!g_ascii_strncasecmp(packet->str + NET_HEADER_SIZE + 1, C("LOAD ")))
{
return PROXY_SEND_QUERY;
}
}
g_string_assign_len(con->orig_sql, packet->str + (NET_HEADER_SIZE + 1),
packet->len - (NET_HEADER_SIZE + 1));
g_strstrip(con->orig_sql->str); /* strip leading and trailing spaces */
strip_extra_spaces(con->orig_sql->str); /* replace multiple spaces with one */
normalize_equal_sign(con->orig_sql->str); /* remove spaces on the side of = */
lower_identifiers(con->orig_sql->str);
con->orig_sql->len = strlen(con->orig_sql->str);
const char *sql = con->orig_sql->str;
chassis_plugin_config *config = con->config;
struct sql_handler_entry_t *sql_handler_map =
config->has_shard_plugin ? sql_handler_shard_map : sql_handler_rw_map;
int i;
for (i = 0; sql_handler_map[i].prefix; ++i) {
if (strcasestr(sql, sql_handler_map[i].prefix)) {
return sql_handler_map[i].func(con, sql);
}
}
network_mysqld_con_send_error(con->client, C("request error, \"select * from help\" for usage"));
return PROXY_SEND_RESULT;
}
/**
* gets called after a query has been read
*/
NETWORK_MYSQLD_PLUGIN_PROTO(server_read_query) {
network_socket *recv_sock;
network_mysqld_stmt_ret ret;
gettimeofday(&(con->req_recv_time), NULL);
recv_sock = con->client;
if (recv_sock->recv_queue->chunks->length != 1) {
g_message("%s: client-recv-queue-len = %d", G_STRLOC,
recv_sock->recv_queue->chunks->length);
}
ret = admin_process_query(con);
switch (ret) {
case PROXY_NO_DECISION:
network_mysqld_con_send_error(con->client,
C("request error, \"select * from help\" for usage"));
con->state = ST_SEND_QUERY_RESULT;
break;
case PROXY_SEND_RESULT:
con->state = ST_SEND_QUERY_RESULT;
break;
default:
network_mysqld_con_send_error(con->client,
C("need a resultset + proxy.PROXY_SEND_RESULT, got something else"));
con->state = ST_SEND_ERROR;
break;
}
g_string_free(g_queue_pop_tail(recv_sock->recv_queue->chunks), TRUE);
return NETWORK_SOCKET_SUCCESS;
}
/**
* cleanup the admin specific data on the current connection
*
* @return NETWORK_SOCKET_SUCCESS
*/
NETWORK_MYSQLD_PLUGIN_PROTO(admin_disconnect_client) {
/* private state is not used in admin-plugin */
con->plugin_con_state = NULL;
return NETWORK_SOCKET_SUCCESS;
}
static int network_mysqld_server_connection_init(network_mysqld_con *con) {
con->plugins.con_init = server_con_init;
con->plugins.con_read_auth = server_read_auth;
con->plugins.con_read_query = server_read_query;
con->plugins.con_cleanup = admin_disconnect_client;
return 0;
}
static chassis_plugin_config *network_mysqld_admin_plugin_new(void) {
chassis_plugin_config *config;
config = g_new0(chassis_plugin_config, 1);
return config;
}
static void network_mysqld_admin_plugin_free(chassis *chas, chassis_plugin_config *config) {
if (config->listen_con) {
/* the socket will be freed by network_mysqld_free() */
}
if (config->address) {
chassis_config_unregister_service(chas->config_manager, config->address);
g_free(config->address);
}
if (g_sampling_timer) {
evtimer_del(g_sampling_timer);
g_free(g_sampling_timer);
g_sampling_timer = NULL;
}
if (config->admin_username) g_free(config->admin_username);
if (config->admin_password) g_free(config->admin_password);
if (config->allow_ip) g_free(config->allow_ip);
if (config->allow_ip_table) g_hash_table_destroy(config->allow_ip_table);
if (config->deny_ip) g_free(config->deny_ip);
if (config->deny_ip_table) g_hash_table_destroy(config->deny_ip_table);
g_free(config);
}
/**
* add the proxy specific options to the cmdline interface
*/
static GList *
network_mysqld_admin_plugin_get_options(chassis_plugin_config *config)
{
chassis_options_t opts = {0};
chassis_options_add(&opts, "admin-address",
0, 0, OPTION_ARG_STRING, &(config->address),
"listening address:port of the admin-server (default: :4041)",
"<host:port>");
chassis_options_add(&opts, "admin-username",
0, 0, OPTION_ARG_STRING, &(config->admin_username),
"username to allow to log in", "<string>");
chassis_options_add(&opts, "admin-password",
0, 0, OPTION_ARG_STRING, &(config->admin_password),
"password to allow to log in", "<string>");
chassis_options_add(&opts, "admin-allow-ip",
0, 0, OPTION_ARG_STRING, &(config->allow_ip),
"ip address allowed to connect to admin", "<string>");
chassis_options_add(&opts, "admin-deny-ip",
0, 0, OPTION_ARG_STRING, &(config->deny_ip),
"ip address denyed to connect to admin", "<string>");
return opts.options;
}
/* ring buffer from: https://github.com/AndersKaloer/Ring-Buffer */
#define RING_BUFFER_SIZE 128 /* must be power of 2, !! index [0, 126] !!*/
#define RING_BUFFER_MASK (RING_BUFFER_SIZE-1)
typedef struct ring_buffer_t {
int head;
int tail;
guint64 buffer[RING_BUFFER_SIZE];
} ring_buffer_t;
static void ring_buffer_add(ring_buffer_t *buffer, guint64 data) {
if (((buffer->head - buffer->tail) & RING_BUFFER_MASK) == RING_BUFFER_MASK)
buffer->tail = ((buffer->tail + 1) & RING_BUFFER_MASK);
buffer->buffer[buffer->head] = data;
buffer->head = ((buffer->head + 1) & RING_BUFFER_MASK);
}
static guint64 ring_buffer_get(ring_buffer_t *buffer, int index) {
if (index >= ((buffer->head - buffer->tail) & RING_BUFFER_MASK))
return 0;
int data_index = ((buffer->tail + index) & RING_BUFFER_MASK);
return buffer->buffer[data_index];
}
static ring_buffer_t g_sql_count = {126, 0};
static ring_buffer_t g_trx_count = {126, 0};
static void calc_qps_average(char *buf, int len)
{
const int MOST_RECENT = 126;
guint64 c_now = ring_buffer_get(&g_sql_count, MOST_RECENT);
guint64 c_1min = ring_buffer_get(&g_sql_count, MOST_RECENT - 6);
guint64 c_5min = ring_buffer_get(&g_sql_count, MOST_RECENT - 6*5);
guint64 c_15min = ring_buffer_get(&g_sql_count, MOST_RECENT - 6*15);
snprintf(buf, len, "%.2f, %.2f, %.2f",
(c_now-c_1min)/60.0, (c_now-c_5min)/300.0, (c_now-c_15min)/900.0);
}
static void calc_tps_average(char *buf, int len)
{
const int MOST_RECENT = 126;
guint64 c_now = ring_buffer_get(&g_trx_count, MOST_RECENT);
guint64 c_1min = ring_buffer_get(&g_trx_count, MOST_RECENT - 6);
guint64 c_5min = ring_buffer_get(&g_trx_count, MOST_RECENT - 6*5);
guint64 c_15min = ring_buffer_get(&g_trx_count, MOST_RECENT - 6*15);
snprintf(buf, len, "%.2f, %.2f, %.2f",
(c_now-c_1min)/60.0, (c_now-c_5min)/300.0, (c_now-c_15min)/900.0);
}
struct _timer_func_arg_t {
chassis *chas;
struct event *ev;
};
/* sample interval is 10-sec, 127 samples takes about 21-min */
static void sql_stats_sampling_func(int fd, short what, void *arg)
{
chassis *chas = arg;
query_stats_t *stats = &(chas->query_stats);
ring_buffer_add(&g_sql_count, stats->client_query.ro + stats->client_query.rw);
ring_buffer_add(&g_trx_count, stats->xa_count);
static struct timeval ten_sec = {10, 0};
/* EV_PERSIST not work for libevent1.4, re-activate timer each time */
chassis_event_add_with_timeout(chas, g_sampling_timer, &ten_sec);
}
/**
* init the plugin with the parsed config
*/
static int
network_mysqld_admin_plugin_apply_config(chassis *chas,
chassis_plugin_config *config)
{
network_mysqld_con *con;
network_socket *listen_sock;
if (!config->address) {
config->address = g_strdup(":4041");
} else {
chas->proxy_address = config->address;
g_message("set proxy address for chassis:%s", config->address);
}
if (!config->admin_username) {
g_critical("%s: --admin-username needs to be set",
G_STRLOC);
return -1;
}
if (!config->admin_password) {
g_critical("%s: --admin-password needs to be set",
G_STRLOC);
return -1;
}
if (!g_strcmp0(config->admin_password, "")) {
g_critical("%s: --admin-password cannot be empty",
G_STRLOC);
return -1;
}
g_message("%s:admin-server listening on port", G_STRLOC);
GHashTable *allow_ip_table = NULL;
if (config->allow_ip) {
allow_ip_table = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
char **ip_arr = g_strsplit(config->allow_ip, ",", -1);
int i;
for (i = 0; ip_arr[i]; i++) {
g_hash_table_insert(allow_ip_table, g_strdup(ip_arr[i]), (void *) TRUE);
}
g_strfreev(ip_arr);
}
config->allow_ip_table = allow_ip_table;
GHashTable *deny_ip_table = NULL;
if (config->deny_ip) {
deny_ip_table = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
char **ip_arr = g_strsplit(config->deny_ip, ",", -1);
int i;
for (i = 0; ip_arr[i]; i++) {
g_hash_table_insert(deny_ip_table, g_strdup(ip_arr[i]), (void *) TRUE);
}
g_strfreev(ip_arr);
}
config->deny_ip_table = deny_ip_table;
g_message("%s:admin-server listening on port", G_STRLOC);
/**
* create a connection handle for the listen socket
*/
con = network_mysqld_con_new();
network_mysqld_add_connection(chas, con, TRUE);
con->config = config;
config->listen_con = con;
listen_sock = network_socket_new();
con->server = listen_sock;
/*
* set the plugin hooks as we want to apply them to the new
* connections too later
*/
network_mysqld_server_connection_init(con);
g_message("%s:admin-server listening on port:%s", G_STRLOC, config->address);
/* FIXME: network_socket_set_address() */
if (0 != network_address_set_address(listen_sock->dst,
config->address))
{
return -1;
}
g_message("%s:admin-server listening on port", G_STRLOC);
/* FIXME: network_socket_bind() */
if (0 != network_socket_bind(listen_sock)) {
return -1;
}
g_message("admin-server listening on port %s", config->address);
/* set config->has_shard_plugin */
config->has_shard_plugin = has_shard_plugin(chas->modules);
/**
* call network_mysqld_con_accept() with this connection when we are done
*/
event_set(&(listen_sock->event), listen_sock->fd,
EV_READ|EV_PERSIST, network_mysqld_con_accept, con);
chassis_event_add(chas, &(listen_sock->event));
chassis_config_register_service(chas->config_manager, config->address, "admin");
/* EV_PERSIST not work for libevent 1.4 */
g_sampling_timer = g_new0(struct event, 1);
evtimer_set(g_sampling_timer, sql_stats_sampling_func, chas);
struct timeval ten_sec = {10, 0};
chassis_event_add_with_timeout(chas, g_sampling_timer, &ten_sec);
return 0;
}
G_MODULE_EXPORT int plugin_init(chassis_plugin *p) {
p->magic = CHASSIS_PLUGIN_MAGIC;
p->name = g_strdup("admin");
p->version = g_strdup(PLUGIN_VERSION);
p->init = network_mysqld_admin_plugin_new;
p->get_options = network_mysqld_admin_plugin_get_options;
p->apply_config = network_mysqld_admin_plugin_apply_config;
p->destroy = network_mysqld_admin_plugin_free;
/* For allow_ip configs */
p->allow_ip_get = network_mysqld_admin_plugin_allow_ip_get;
p->allow_ip_add = network_mysqld_admin_plugin_allow_ip_add;
p->allow_ip_del = network_mysqld_admin_plugin_allow_ip_del;
/* For deny_ip configs */
p->deny_ip_get = network_mysqld_admin_plugin_deny_ip_get;
p->deny_ip_add = network_mysqld_admin_plugin_deny_ip_add;
p->deny_ip_del = network_mysqld_admin_plugin_deny_ip_del;
return 0;
}
Reference in New Issue View Git Blame Copy Permalink