#include // std::cout #include // std::sort #include // std::vector #include "re2/re2.h" #include "re2/regexp.h" #include "proxysql.h" #include "cpp.h" #include "MySQL_Logger.hpp" #include #include #include #include #include #include #include #include #include #include #include #include #include "SpookyV2.h" #include #include #define SELECT_VERSION_COMMENT "select @@version_comment limit 1" #define SELECT_VERSION_COMMENT_LEN 32 #define SELECT_DB_USER "select DATABASE(), USER() limit 1" #define SELECT_DB_USER_LEN 33 #define SELECT_CHARSET_VARIOUS "select @@character_set_client, @@character_set_connection, @@character_set_server, @@character_set_database limit 1" #define SELECT_CHARSET_VARIOUS_LEN 115 #define READ_ONLY_OFF "\x01\x00\x00\x01\x02\x23\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x0d\x56\x61\x72\x69\x61\x62\x6c\x65\x5f\x6e\x61\x6d\x65\x00\x0c\x21\x00\x0f\x00\x00\x00\xfd\x01\x00\x1f\x00\x00\x1b\x00\x00\x03\x03\x64\x65\x66\x00\x00\x00\x05\x56\x61\x6c\x75\x65\x00\x0c\x21\x00\x0f\x00\x00\x00\xfd\x01\x00\x1f\x00\x00\x05\x00\x00\x04\xfe\x00\x00\x02\x00\x0e\x00\x00\x05\x09\x72\x65\x61\x64\x5f\x6f\x6e\x6c\x79\x03\x4f\x46\x46\x05\x00\x00\x06\xfe\x00\x00\x02\x00" #define READ_ONLY_ON "\x01\x00\x00\x01\x02\x23\x00\x00\x02\x03\x64\x65\x66\x00\x00\x00\x0d\x56\x61\x72\x69\x61\x62\x6c\x65\x5f\x6e\x61\x6d\x65\x00\x0c\x21\x00\x0f\x00\x00\x00\xfd\x01\x00\x1f\x00\x00\x1b\x00\x00\x03\x03\x64\x65\x66\x00\x00\x00\x05\x56\x61\x6c\x75\x65\x00\x0c\x21\x00\x0f\x00\x00\x00\xfd\x01\x00\x1f\x00\x00\x05\x00\x00\x04\xfe\x00\x00\x02\x00\x0d\x00\x00\x05\x09\x72\x65\x61\x64\x5f\x6f\x6e\x6c\x79\x02\x4f\x4e\x05\x00\x00\x06\xfe\x00\x00\x02\x00" #ifdef __APPLE__ #ifndef MSG_NOSIGNAL #define MSG_NOSIGNAL 0 #endif // MSG_NOSIGNAL #endif // __APPLE__ #define SAFE_SQLITE3_STEP(_stmt) do {\ do {\ rc=sqlite3_step(_stmt);\ if (rc!=SQLITE_DONE) {\ assert(rc==SQLITE_LOCKED);\ usleep(100);\ }\ } while (rc!=SQLITE_DONE);\ } while (0) #define SAFE_SQLITE3_STEP2(_stmt) do {\ do {\ rc=sqlite3_step(_stmt);\ if (rc==SQLITE_LOCKED || rc==SQLITE_BUSY) {\ usleep(100);\ }\ } while (rc==SQLITE_LOCKED || rc==SQLITE_BUSY);\ } while (0) struct cpu_timer { cpu_timer() { begin = monotonic_time(); } ~cpu_timer() { unsigned long long end = monotonic_time(); #ifdef DEBUG std::cerr << double( end - begin ) / 1000000 << " secs.\n" ; #endif begin=end-begin; // make the compiler happy }; unsigned long long begin; }; static char *s_strdup(char *s) { char *ret=NULL; if (s) { ret=strdup(s); } return ret; } static int __SQLite3_Server_refresh_interval=1000; static bool testTimeoutSequence[] = {true, false, true, false, true, false, true, false}; static int testIndex = 7; static int testLag = 10; extern Query_Cache *GloQC; extern MySQL_Authentication *GloMyAuth; extern ProxySQL_Admin *GloAdmin; extern Query_Processor *GloQPro; extern MySQL_Threads_Handler *GloMTH; extern MySQL_Logger *GloMyLogger; extern MySQL_Monitor *GloMyMon; extern SQLite3_Server *GloSQLite3Server; #define PANIC(msg) { perror(msg); exit(EXIT_FAILURE); } static pthread_mutex_t sock_mutex = PTHREAD_MUTEX_INITIALIZER; static char * SQLite3_Server_variables_names[] = { (char *)"mysql_ifaces", (char *)"read_only", NULL }; static void * (*child_func[1]) (void *arg); typedef struct _main_args { int nfds; struct pollfd *fds; int *callback_func; volatile int *shutdown; } main_args; typedef struct _ifaces_desc_t { char **mysql_ifaces; } ifaces_desc_t; #define MAX_IFACES 128 #define MAX_SQLITE3SERVER_LISTENERS 128 class ifaces_desc { public: PtrArray *ifaces; ifaces_desc() { ifaces=new PtrArray(); } bool add(const char *iface) { for (unsigned int i=0; ilen; i++) { if (strcmp((const char *)ifaces->index(i),iface)==0) { return false; } } ifaces->add(strdup(iface)); return true; } ~ifaces_desc() { while(ifaces->len) { char *d=(char *)ifaces->remove_index_fast(0); free(d); } delete ifaces; } }; class sqlite3server_main_loop_listeners { private: int version; pthread_rwlock_t rwlock; char ** reset_ifaces(char **ifaces) { int i; if (ifaces) { for (i=0; iadd(token); i++; } free_tokenizer( &tok ); version++; wrunlock(); } bool update_ifaces(char *list, char ***_ifaces) { wrlock(); int i; char **ifaces=*_ifaces; tokenizer_t tok; tokenizer( &tok, list, ";", TOKENIZER_NO_EMPTIES ); const char* token; ifaces=reset_ifaces(ifaces); i=0; for ( token = tokenize( &tok ) ; token && i < MAX_IFACES ; token = tokenize( &tok ) ) { ifaces[i]=(char *)malloc(strlen(token)+1); strcpy(ifaces[i],token); i++; } free_tokenizer( &tok ); version++; wrunlock(); return true; } }; static sqlite3server_main_loop_listeners S_amll; void SQLite3_Server_session_handler(MySQL_Session *sess, void *_pa, PtrSize_t *pkt) { char *error=NULL; int cols; int affected_rows; bool run_query=true; SQLite3_result *resultset=NULL; char *strA=NULL; char *strB=NULL; int strAl, strBl; char *query=NULL; unsigned int query_length=pkt->size-sizeof(mysql_hdr); query=(char *)l_alloc(query_length); memcpy(query,(char *)pkt->ptr+sizeof(mysql_hdr)+1,query_length-1); query[query_length-1]=0; #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP) if (sess->client_myds->proxy_addr.addr == NULL) { struct sockaddr addr; socklen_t addr_len=sizeof(struct sockaddr); memset(&addr,0,addr_len); int rc; rc=getsockname(sess->client_myds->fd, &addr, &addr_len); if (rc==0) { char buf[512]; switch (addr.sa_family) { case AF_INET: { struct sockaddr_in *ipv4 = (struct sockaddr_in *)&addr; inet_ntop(addr.sa_family, &ipv4->sin_addr, buf, INET_ADDRSTRLEN); sess->client_myds->proxy_addr.addr = strdup(buf); } break; case AF_INET6: { struct sockaddr_in6 *ipv6 = (struct sockaddr_in6 *)&addr; inet_ntop(addr.sa_family, &ipv6->sin6_addr, buf, INET6_ADDRSTRLEN); sess->client_myds->proxy_addr.addr = strdup(buf); } break; default: sess->client_myds->proxy_addr.addr = strdup("unknown"); break; } } else { sess->client_myds->proxy_addr.addr = strdup("unknown"); } } #endif // TEST_AURORA || TEST_GALERA || TEST_GROUPREP char *query_no_space=(char *)l_alloc(query_length); memcpy(query_no_space,query,query_length); unsigned int query_no_space_length=remove_spaces(query_no_space); // fix bug #925 while (query_no_space[query_no_space_length-1]==';' || query_no_space[query_no_space_length-1]==' ') { query_no_space_length--; query_no_space[query_no_space_length]=0; } // fix bug #1047 if ( (!strncasecmp("BEGIN", query_no_space, strlen("BEGIN"))) || (!strncasecmp("START TRANSACTION", query_no_space, strlen("START TRANSACTION"))) || (!strncasecmp("COMMIT", query_no_space, strlen("COMMIT"))) || (!strncasecmp("ROLLBACK", query_no_space, strlen("ROLLBACK"))) || (!strncasecmp("SET character_set_results", query_no_space, strlen("SET character_set_results"))) || (!strncasecmp("SET SQL_AUTO_IS_NULL", query_no_space, strlen("SET SQL_AUTO_IS_NULL"))) || (!strncasecmp("SET NAMES", query_no_space, strlen("SET NAMES"))) || (!strncasecmp("SET AUTOCOMMIT", query_no_space, strlen("SET AUTOCOMMIT"))) ) { GloSQLite3Server->send_MySQL_OK(&sess->client_myds->myprot, NULL); run_query=false; goto __run_query; } if (query_no_space_length==SELECT_VERSION_COMMENT_LEN) { if (!strncasecmp(SELECT_VERSION_COMMENT, query_no_space, query_no_space_length)) { l_free(query_length,query); #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP) char *a = (char *)"SELECT '(ProxySQL Automated Test Server) - %s'"; query = (char *)malloc(strlen(a)+strlen(sess->client_myds->proxy_addr.addr)); sprintf(query,a,sess->client_myds->proxy_addr.addr); #else query=l_strdup("SELECT '(ProxySQL SQLite3 Server)'"); #endif // TEST_AURORA || TEST_GALERA || TEST_GROUPREP query_length=strlen(query)+1; goto __run_query; } } if (query_no_space_length==SELECT_DB_USER_LEN) { if (!strncasecmp(SELECT_DB_USER, query_no_space, query_no_space_length)) { l_free(query_length,query); char *query1=(char *)"SELECT \"admin\" AS 'DATABASE()', \"%s\" AS 'USER()'"; char *query2=(char *)malloc(strlen(query1)+strlen(sess->client_myds->myconn->userinfo->username)+10); sprintf(query2,query1,sess->client_myds->myconn->userinfo->username); query=l_strdup(query2); query_length=strlen(query2)+1; free(query2); goto __run_query; } } if (query_no_space_length==SELECT_CHARSET_VARIOUS_LEN) { if (!strncasecmp(SELECT_CHARSET_VARIOUS, query_no_space, query_no_space_length)) { l_free(query_length,query); char *query1=(char *)"select 'utf8' as '@@character_set_client', 'utf8' as '@@character_set_connection', 'utf8' as '@@character_set_server', 'utf8' as '@@character_set_database' limit 1"; query=l_strdup(query1); query_length=strlen(query1)+1; goto __run_query; } } if (!strncasecmp("SELECT @@version", query_no_space, strlen("SELECT @@version"))) { l_free(query_length,query); char *q=(char *)"SELECT '%s' AS '@@version'"; query_length=strlen(q)+20; query=(char *)l_alloc(query_length); sprintf(query,q,PROXYSQL_VERSION); goto __run_query; } if (!strncasecmp("SELECT version()", query_no_space, strlen("SELECT version()"))) { l_free(query_length,query); char *q=(char *)"SELECT '%s' AS 'version()'"; query_length=strlen(q)+20; query=(char *)l_alloc(query_length); sprintf(query,q,PROXYSQL_VERSION); goto __run_query; } if (strncasecmp("SHOW ", query_no_space, 5)) { goto __end_show_commands; // in the next block there are only SHOW commands } if (query_no_space_length==strlen("SHOW TABLES") && !strncasecmp("SHOW TABLES",query_no_space, query_no_space_length)) { l_free(query_length,query); query=l_strdup("SELECT name AS tables FROM sqlite_master WHERE type='table' AND name NOT IN ('sqlite_sequence') ORDER BY name"); query_length=strlen(query)+1; goto __run_query; } if ((query_no_space_length>17) && (!strncasecmp("SHOW TABLES FROM ", query_no_space, 17))) { strA=query_no_space+17; strAl=strlen(strA); strB=(char *)"SELECT name AS tables FROM %s.sqlite_master WHERE type='table' AND name NOT IN ('sqlite_sequence') ORDER BY name"; strBl=strlen(strB); int l=strBl+strAl-2; char *b=(char *)l_alloc(l+1); snprintf(b,l+1,strB,strA); b[l]=0; l_free(query_length,query); query=b; query_length=l+1; goto __run_query; } if ((query_no_space_length>17) && (!strncasecmp("SHOW TABLES LIKE ", query_no_space, 17))) { strA=query_no_space+17; strAl=strlen(strA); strB=(char *)"SELECT name AS tables FROM sqlite_master WHERE type='table' AND name LIKE '%s'"; strBl=strlen(strB); char *tn=NULL; // tablename tn=(char *)malloc(strlen(strA)); unsigned int i=0, j=0; while (i=3 && tbh[0]=='`' && tbh[strlen(tbh)-1]=='`') { // tablename is quoted char *tbh_tmp=(char *)malloc(strlen(tbh)-1); strncpy(tbh_tmp,tbh+1,strlen(tbh)-2); tbh_tmp[strlen(tbh)-2]=0; free(tbh); tbh=tbh_tmp; } int l=strBl+strlen(tbh)*3+strlen(dbh)-8; char *buff=(char *)l_alloc(l+1); snprintf(buff,l+1,strB,tbh,tbh,dbh,tbh); buff[l]=0; free(tbh); free(dbh); l_free(query_length,query); query=buff; query_length=l+1; goto __run_query; } if ( (query_no_space_length==strlen("SHOW DATABASES") && !strncasecmp("SHOW DATABASES",query_no_space, query_no_space_length)) || (query_no_space_length==strlen("SHOW SCHEMAS") && !strncasecmp("SHOW SCHEMAS",query_no_space, query_no_space_length)) ) { l_free(query_length,query); query=l_strdup("PRAGMA DATABASE_LIST"); query_length=strlen(query)+1; goto __run_query; } __end_show_commands: if (query_no_space_length==strlen("SELECT DATABASE()") && !strncasecmp("SELECT DATABASE()",query_no_space, query_no_space_length)) { l_free(query_length,query); query=l_strdup("SELECT \"main\" AS 'DATABASE()'"); query_length=strlen(query)+1; goto __run_query; } // see issue #1022 if (query_no_space_length==strlen("SELECT DATABASE() AS name") && !strncasecmp("SELECT DATABASE() AS name",query_no_space, query_no_space_length)) { l_free(query_length,query); query=l_strdup("SELECT \"main\" AS 'DATABASE()'"); query_length=strlen(query)+1; goto __run_query; } if (sess->session_type == PROXYSQL_SESSION_SQLITE) { // no admin if ( (strncasecmp("PRAGMA",query_no_space,6)==0) || (strncasecmp("ATTACH",query_no_space,6)==0) ) { proxy_error("[WARNING]: Commands executed from stats interface in Admin Module: \"%s\"\n", query_no_space); GloSQLite3Server->send_MySQL_ERR(&sess->client_myds->myprot, (char *)"Command not allowed"); run_query=false; } } __run_query: if (run_query) { #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP) if (strncasecmp("SELECT",query_no_space,6)==0) { #ifdef TEST_AURORA if (strstr(query_no_space,(char *)"REPLICA_HOST_STATUS")) { pthread_mutex_lock(&GloSQLite3Server->aurora_mutex); GloSQLite3Server->populate_aws_aurora_table(sess); } #endif // TEST_AURORA #ifdef TEST_GALERA if (strstr(query_no_space,(char *)"HOST_STATUS_GALERA")) { pthread_mutex_lock(&GloSQLite3Server->galera_mutex); GloSQLite3Server->populate_galera_table(sess); } #endif // TEST_GALERA #ifdef TEST_GROUPREP if (strstr(query_no_space,(char *)"GR_MEMBER_ROUTING_CANDIDATE_STATUS")) { pthread_mutex_lock(&GloSQLite3Server->grouprep_mutex); GloSQLite3Server->populate_grouprep_table(sess, testLag); if (testLag > 0) testLag--; } #endif // TEST_GROUPREP if (strstr(query_no_space,(char *)"Seconds_Behind_Master")) { free(query); char *a = (char *)"SELECT %d as Seconds_Behind_Master"; query = (char *)malloc(strlen(a)+4); sprintf(query,a,rand()%30+10); } } #endif // TEST_AURORA || TEST_GALERA || TEST_GROUPREP SQLite3_Session *sqlite_sess = (SQLite3_Session *)sess->thread->gen_args; sqlite_sess->sessdb->execute_statement(query, &error , &cols , &affected_rows , &resultset); #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP) if (strncasecmp("SELECT",query_no_space,6)==0) { #ifdef TEST_AURORA if (strstr(query_no_space,(char *)"REPLICA_HOST_STATUS")) { pthread_mutex_unlock(&GloSQLite3Server->aurora_mutex); if (rand() % 100 == 0) { // randomly add some latency on 1% of the traffic sleep(2); } } #endif // TEST_AURORA #ifdef TEST_GALERA if (strstr(query_no_space,(char *)"HOST_STATUS_GALERA")) { pthread_mutex_unlock(&GloSQLite3Server->galera_mutex); if (resultset->rows_count == 0) { PROXY_TRACE(); } if (rand() % 20 == 0) { // randomly add some latency on 5% of the traffic sleep(2); } } #endif // TEST_GALERA #ifdef TEST_GROUPREP if (strstr(query_no_space,(char *)"GR_MEMBER_ROUTING_CANDIDATE_STATUS")) { pthread_mutex_unlock(&GloSQLite3Server->grouprep_mutex); if (resultset->rows_count == 0) { PROXY_TRACE(); } if (strncmp("127.2.1.2", sess->client_myds->proxy_addr.addr,9) == 0) { if (testTimeoutSequence[testIndex--]) sleep(2); if (testIndex < 0) testIndex = 7; } else { if (rand() % 20 == 0) sleep(2); } } #endif // TEST_GROUPREP if (strstr(query_no_space,(char *)"Seconds_Behind_Master")) { if (rand() % 10 == 0) { // randomly add some latency on 10% of the traffic sleep(2); } } } #endif // TEST_AURORA || TEST_GALERA || TEST_GROUPREP sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot); delete resultset; } l_free(pkt->size-sizeof(mysql_hdr),query_no_space); // it is always freed here l_free(query_length,query); } SQLite3_Session::SQLite3_Session() { sessdb = new SQLite3DB(); sessdb->open((char *)"file:mem_sqlitedb?mode=memory&cache=shared", SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX); } SQLite3_Session::~SQLite3_Session() { delete sessdb; sessdb = NULL; } static void *child_mysql(void *arg) { int client = *(int *)arg; GloMTH->wrlock(); { char *s=GloMTH->get_variable((char *)"server_capabilities"); mysql_thread___server_capabilities=atoi(s); free(s); } GloMTH->wrunlock(); struct pollfd fds[1]; nfds_t nfds=1; int rc; pthread_mutex_unlock(&sock_mutex); MySQL_Thread *mysql_thr=new MySQL_Thread(); mysql_thr->curtime=monotonic_time(); SQLite3_Session *sqlite_sess = new SQLite3_Session(); mysql_thr->gen_args = (void *)sqlite_sess; GloQPro->init_thread(); mysql_thr->refresh_variables(); MySQL_Session *sess=mysql_thr->create_new_session_and_client_data_stream(client); sess->thread=mysql_thr; sess->session_type = PROXYSQL_SESSION_SQLITE; sess->handler_function=SQLite3_Server_session_handler; MySQL_Data_Stream *myds=sess->client_myds; fds[0].fd=client; fds[0].revents=0; fds[0].events=POLLIN|POLLOUT; free(arg); sess->client_myds->myprot.generate_pkt_initial_handshake(true,NULL,NULL, &sess->thread_session_id); while (__sync_fetch_and_add(&glovars.shutdown,0)==0) { if (myds->available_data_out()) { fds[0].events=POLLIN|POLLOUT; } else { fds[0].events=POLLIN; } fds[0].revents=0; rc=poll(fds,nfds,__sync_fetch_and_add(&__SQLite3_Server_refresh_interval,0)); if (rc == -1) { if (errno == EINTR) { continue; } else { goto __exit_child_mysql; } } myds->revents=fds[0].revents; myds->read_from_net(); if (myds->net_failure) goto __exit_child_mysql; myds->read_pkts(); sess->to_process=1; int rc=sess->handler(); if (rc==-1) goto __exit_child_mysql; } __exit_child_mysql: delete sqlite_sess; delete mysql_thr; return NULL; } static void * sqlite3server_main_loop(void *arg) { int i; int version=0; struct sockaddr_in addr; struct pollfd *fds=((struct _main_args *)arg)->fds; int nfds=((struct _main_args *)arg)->nfds; int *callback_func=((struct _main_args *)arg)->callback_func; volatile int *shutdown=((struct _main_args *)arg)->shutdown; char *socket_names[MAX_SQLITE3SERVER_LISTENERS]; for (i=0;ischeduler_run_once(); unsigned long long poll_wait=500000; if (next_run < curtime + 500000) { poll_wait=next_run-curtime; } if (poll_wait > 500000) { poll_wait=500000; } poll_wait=poll_wait/1000; // conversion to millisecond int rc; rc=poll(fds,nfds,poll_wait); if ((rc == -1 && errno == EINTR) || rc==0) { // poll() timeout, try again goto __end_while_pool; } for (i=1;ipipefd[0]; fds[nfds].events=POLLIN; fds[nfds].revents=0; nfds++; unsigned int j; i=0; j=0; for (j=0; jifaces->len; j++) { char *add=NULL; char *port=NULL; char *sn=(char *)S_amll.ifaces_mysql->ifaces->index(j); char *h = NULL; if (*sn == '[') { char *p = strchr(sn, ']'); if (p == NULL) proxy_error("Invalid IPv6 address: %s\n", sn); h = ++sn; // remove first '[' *p = '\0'; sn = p++; // remove last ']' add = h; port = ++p; // remove ':' } else { c_split_2(sn, ":" , &add, &port); } int s = ( atoi(port) ? listen_on_port(add, atoi(port), 128) : listen_on_unix(add, 128)); if (s>0) { fds[nfds].fd=s; fds[nfds].events=POLLIN; fds[nfds].revents=0; callback_func[nfds]=0; socket_names[nfds]=strdup(sn); nfds++; } if (add) free(add); if (port) free(port); } S_amll.wrunlock(); } } //if (__sync_add_and_fetch(shutdown,0)==0) __sync_add_and_fetch(shutdown,1); for (i=0; iopen((char *)"file:mem_sqlitedb?mode=memory&cache=shared", SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_FULLMUTEX); variables.read_only=false; #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP) string s = ""; #ifdef TEST_AURORA init_aurora_ifaces_string(s); #endif // TEST_AURORA #ifdef TEST_GALERA init_galera_ifaces_string(s); #endif // TEST_GALERA #ifdef TEST_GROUPREP init_grouprep_ifaces_string(s); #endif // TEST_GROUPREP variables.mysql_ifaces=strdup(s.c_str()); #else variables.mysql_ifaces=strdup("127.0.0.1:6030"); #endif // TEST_AURORA || TEST_GALERA || TEST_GROUPREP }; #ifdef TEST_GALERA void SQLite3_Server::populate_galera_table(MySQL_Session *sess) { // this function needs to be called with lock on mutex galera_mutex already acquired sessdb->execute("BEGIN TRANSACTION"); char *error=NULL; int cols=0; int affected_rows=0; SQLite3_result *resultset=NULL; //sqlite3 *mydb3=sessdb->get_db(); string myip = string(sess->client_myds->proxy_addr.addr); string clu_id_s = myip.substr(6,1); unsigned int cluster_id = atoi(clu_id_s.c_str()); cluster_id--; int hg_id = 2270+(cluster_id*10)+1; char buf[1024]; sprintf(buf, (char *)"SELECT * FROM HOST_STATUS_GALERA WHERE hostgroup_id = %d LIMIT 1", hg_id); sessdb->execute_statement(buf, &error , &cols , &affected_rows , &resultset); if (resultset->rows_count==0) { //sessdb->execute("DELETE FROM HOST_STATUS_GALERA"); sqlite3_stmt *statement=NULL; int rc; char *query=(char *)"INSERT INTO HOST_STATUS_GALERA VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)"; //rc=sqlite3_prepare_v2(mydb3, query, -1, &statement, 0); rc = sessdb->prepare_v2(query, &statement); ASSERT_SQLITE_OK(rc, sessdb); for (unsigned int i=0; iexecute("COMMIT"); } #endif // TEST_GALERA #ifdef TEST_AURORA void SQLite3_Server::populate_aws_aurora_table(MySQL_Session *sess) { // this function needs to be called with lock on mutex aurora_mutex already acquired sessdb->execute("DELETE FROM REPLICA_HOST_STATUS"); sqlite3_stmt *statement=NULL; //sqlite3 *mydb3=sessdb->get_db(); int rc; char *query=(char *)"INSERT INTO REPLICA_HOST_STATUS VALUES (?1, ?2, ?3, ?4, ?5)"; //rc=sqlite3_prepare_v2(mydb3, query, -1, &statement, 0); rc = sessdb->prepare_v2(query, &statement); ASSERT_SQLITE_OK(rc, sessdb); time_t __timer; char lut[30]; struct tm __tm_info; time(&__timer); localtime_r(&__timer, &__tm_info); strftime(lut, 25, "%Y-%m-%d %H:%M:%S", &__tm_info); string myip = string(sess->client_myds->proxy_addr.addr); string clu_id_s = myip.substr(6,1); unsigned int cluster_id = atoi(clu_id_s.c_str()); cluster_id--; //if (rand() % 200 == 0) { if (rand() % 20000 == 0) { // simulate a failover cur_aurora_writer[cluster_id] = rand() % num_aurora_servers[cluster_id]; proxy_info("Simulating a failover for AWS Aurora cluster %d , HGs (%d:%d)\n", cluster_id, 1270 + cluster_id*2+1 , 1270 + cluster_id*2+2); } if (rand() % 1000 == 0) { if (num_aurora_servers[cluster_id] < max_num_aurora_servers) { num_aurora_servers[cluster_id]++; proxy_info("Simulating the add of a new server for AWS Aurora Cluster %d , HGs (%d:%d). Now adding server num %d\n", cluster_id, 1270 + cluster_id*2+1 , 1270 + cluster_id*2+2, num_aurora_servers[cluster_id]); } } if (rand() % 1000 == 0) { if (num_aurora_servers[cluster_id] > 1) { if (cur_aurora_writer[cluster_id] != (num_aurora_servers[cluster_id] - 1) ) { num_aurora_servers[cluster_id]--; proxy_info("Simulating the deletion of a server from AWS Aurora Cluster %d , HGs (%d:%d). Removing server num %d\n", cluster_id, 1270 + cluster_id*2+1 , 1270 + cluster_id*2+2, num_aurora_servers[cluster_id]+1); } } } for (unsigned int i=0; iexecute("DELETE FROM GR_MEMBER_ROUTING_CANDIDATE_STATUS"); string myip = string(sess->client_myds->proxy_addr.addr); string server_id = myip.substr(8,1); if (server_id == "1") sessdb->execute("INSERT INTO GR_MEMBER_ROUTING_CANDIDATE_STATUS (viable_candidate, read_only, transactions_behind) values ('YES', 'NO', 0)"); else { std::stringstream ss; ss << "INSERT INTO GR_MEMBER_ROUTING_CANDIDATE_STATUS (viable_candidate, read_only, transactions_behind) values ('YES', 'YES', " << txs_behind << ")"; sessdb->execute(ss.str().c_str()); } } #endif // TEST_GALERA #if defined(TEST_AURORA) || defined(TEST_GALERA) || defined(TEST_GROUPREP) void SQLite3_Server::insert_into_tables_defs(std::vector *tables_defs, const char *table_name, const char *table_def) { table_def_t *td = new table_def_t; td->table_name=strdup(table_name); td->table_def=strdup(table_def); tables_defs->push_back(td); }; void SQLite3_Server::check_and_build_standard_tables(SQLite3DB *db, std::vector *tables_defs) { table_def_t *td; db->execute("PRAGMA foreign_keys = OFF"); for (std::vector::iterator it=tables_defs->begin(); it!=tables_defs->end(); ++it) { td=*it; db->check_and_build_table(td->table_name, td->table_def); } db->execute("PRAGMA foreign_keys = ON"); }; void SQLite3_Server::drop_tables_defs(std::vector *tables_defs) { table_def_t *td; while (!tables_defs->empty()) { td=tables_defs->back(); free(td->table_name); td->table_name=NULL; free(td->table_def); td->table_def=NULL; tables_defs->pop_back(); delete td; } }; #endif // TEST_AURORA || TEST_GALERA || defined(TEST_GROUPREP) void SQLite3_Server::wrlock() { pthread_rwlock_wrlock(&rwlock); }; void SQLite3_Server::wrunlock() { pthread_rwlock_unlock(&rwlock); }; void SQLite3_Server::print_version() { fprintf(stderr,"Standard ProxySQL SQLite3 Server rev. %s -- %s -- %s\n", PROXYSQL_SQLITE3_SERVER_VERSION, __FILE__, __TIMESTAMP__); }; bool SQLite3_Server::init() { cpu_timer cpt; #ifdef TEST_AURORA tables_defs_aurora = new std::vector; insert_into_tables_defs(tables_defs_aurora, (const char *)"REPLICA_HOST_STATUS", (const char *)"CREATE TABLE REPLICA_HOST_STATUS (SERVER_ID VARCHAR NOT NULL, SESSION_ID VARCHAR NOT NULL, CPU REAL NOT NULL, LAST_UPDATE_TIMESTAMP VARCHAR NOT NULL, REPLICA_LAG_IN_MILLISECONDS REAL NOT NULL)"); check_and_build_standard_tables(sessdb, tables_defs_aurora); GloAdmin->enable_aurora_testing(); #endif // TEST_AURORA #ifdef TEST_GALERA tables_defs_galera = new std::vector; insert_into_tables_defs(tables_defs_galera, (const char *)"HOST_STATUS_GALERA", (const char *)"CREATE TABLE HOST_STATUS_GALERA (hostgroup_id INT NOT NULL , hostname VARCHAR NOT NULL , port INT NOT NULL , wsrep_local_state VARCHAR , read_only VARCHAR , wsrep_local_recv_queue VARCHAR , wsrep_desync VARCHAR , wsrep_reject_queries VARCHAR , wsrep_sst_donor_rejects_queries VARCHAR , wsrep_cluster_status VARCHAR , PRIMARY KEY (hostgroup_id, hostname, port))"); check_and_build_standard_tables(sessdb, tables_defs_galera); GloAdmin->enable_galera_testing(); #endif // TEST_GALERA #ifdef TEST_GROUPREP tables_defs_grouprep = new std::vector; insert_into_tables_defs(tables_defs_grouprep, (const char *)"GR_MEMBER_ROUTING_CANDIDATE_STATUS", (const char*)"CREATE TABLE GR_MEMBER_ROUTING_CANDIDATE_STATUS (viable_candidate varchar not null, read_only varchar not null, transactions_behind int not null)"); check_and_build_standard_tables(sessdb, tables_defs_grouprep); GloAdmin->enable_grouprep_testing(); #endif // TEST_GALERA child_func[0]=child_mysql; main_shutdown=0; main_poll_nfds=0; main_poll_fds=NULL; main_callback_func=NULL; main_callback_func=(int *)malloc(sizeof(int)*MAX_SQLITE3SERVER_LISTENERS); main_poll_fds=(struct pollfd *)malloc(sizeof(struct pollfd)*MAX_SQLITE3SERVER_LISTENERS); main_poll_nfds=0; S_amll.update_ifaces(variables.mysql_ifaces, &S_amll.ifaces_mysql); pthread_t SQLite3_Server_thr; struct _main_args *arg=(struct _main_args *)malloc(sizeof(struct _main_args)); arg->nfds=main_poll_nfds; arg->fds=main_poll_fds; arg->shutdown=&main_shutdown; arg->callback_func=main_callback_func; if (pthread_create(&SQLite3_Server_thr, NULL, sqlite3server_main_loop, (void *)arg) !=0 ) { perror("Thread creation"); exit(EXIT_FAILURE); } #ifdef DEBUG std::cerr << "SQLite3 Server initialized in "; #endif return true; }; char **SQLite3_Server::get_variables_list() { size_t l=sizeof(SQLite3_Server_variables_names)/sizeof(char *); unsigned int i; char **ret=(char **)malloc(sizeof(char *)*l); for (i=0;iget_myds(); myds->DSS=STATE_QUERY_SENT_DS; myprot->generate_pkt_OK(true,NULL,NULL,1,rows,0,2,0,msg); myds->DSS=STATE_SLEEP; } void SQLite3_Server::send_MySQL_ERR(MySQL_Protocol *myprot, char *msg) { assert(myprot); MySQL_Data_Stream *myds=myprot->get_myds(); myds->DSS=STATE_QUERY_SENT_DS; myprot->generate_pkt_ERR(true,NULL,NULL,1,1045,(char *)"28000",msg); myds->DSS=STATE_SLEEP; }