hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' of https://github.com/mariadb-corporation/MaxScale into develop

Browse Source
This commit is contained in:
VilhoRaatikka
2014-10-09 15:54:11 +03:00
parent 2bd93025a9 fe20df3596
commit 85b15cf5ce
21 changed files with 4691 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
query_classifier/test/CMakeLists.txt
View File

@ -1 +1,14 @@
add_subdirectory(canonical_tests)
if(${ERRMSG} MATCHES "ERRMSG-NOTFOUND")
message(FATAL_ERROR "The errmsg.sys file was not found, please define the path with -DERRMSG=<path>")
else()
if(${CMAKE_VERSION} VERSION_LESS 2.8)
execute_process(COMMAND cp ${ERRMSG} ${CMAKE_CURRENT_BINARY_DIR})
else()
file(COPY ${ERRMSG} DESTINATION ${CMAKE_CURRENT_BINARY_DIR})
endif()
endif()
add_subdirectory(canonical_tests)
add_executable(classify classify.c)
target_link_libraries(classify query_classifier fullcore)
add_test(TestQueryClassifier classify ${CMAKE_CURRENT_SOURCE_DIR}/input.sql ${CMAKE_CURRENT_SOURCE_DIR}/expected.sql)

175
query_classifier/test/classify.c Normal file
View File

@ -0,0 +1,175 @@
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string.h>
#include <query_classifier.h>
#include <buffer.h>
#include <mysql.h>
#include <unistd.h>
static char* server_options[] = {
"SkySQL Gateway",
"--no-defaults",
"--datadir=.",
"--language=.",
"--skip-innodb",
"--default-storage-engine=myisam",
NULL
};
const int num_elements = (sizeof(server_options) / sizeof(char *)) - 1;
static char* server_groups[] = {
"embedded",
"server",
"server",
NULL
};
int main(int argc, char** argv)
{
if(argc < 3){
fprintf(stderr,"Usage: classify <input> <expected output>");
return 1;
}
int rd = 0,buffsz = getpagesize(),strsz = 0,ex_val = 0;
char buffer[buffsz], *strbuff = (char*)calloc(buffsz,sizeof(char));
FILE *input,*expected;
if(mysql_library_init(num_elements, server_options, server_groups))
{
printf("Error: Cannot initialize Embedded Library.");
return 1;
}
input = fopen(argv[1],"rb");
expected = fopen(argv[2],"rb");
while((rd = fread(buffer,sizeof(char),buffsz,input))){
/**Fill the read buffer*/
if(strsz + rd >= buffsz){
char* tmp = (char*)calloc((buffsz*2),sizeof(char));
if(!tmp){
fprintf(stderr,"Error: Cannot allocate enough memory.");
return 1;
}
memcpy(tmp,strbuff,buffsz);
free(strbuff);
strbuff = tmp;
buffsz *= 2;
}
memcpy(strbuff+strsz,buffer,rd);
strsz += rd;
char *tok,*nlptr;
/**Remove newlines*/
while((nlptr = strpbrk(strbuff,"\n")) != NULL && (nlptr - strbuff) < strsz){
memmove(nlptr,nlptr+1,strsz - (nlptr + 1 - strbuff));
strsz -= 1;
}
/**Parse read buffer for full queries*/
while(strpbrk(strbuff,";") != NULL){
tok = strpbrk(strbuff,";");
unsigned int qlen = tok - strbuff + 1;
GWBUF* buff = gwbuf_alloc(qlen+6);
*((unsigned char*)(buff->start)) = qlen;
*((unsigned char*)(buff->start + 1)) = (qlen >> 8);
*((unsigned char*)(buff->start + 2)) = (qlen >> 16);
*((unsigned char*)(buff->start + 3)) = 0x00;
*((unsigned char*)(buff->start + 4)) = 0x03;
memcpy(buff->start+5, strbuff, qlen);
memmove(strbuff,tok + 1, strsz - qlen);
strsz -= qlen;
memset(strbuff + strsz,0,buffsz - strsz);
skygw_query_type_t type = query_classifier_get_type(buff);
char qtypestr[64];
char expbuff[256];
int expos = 0;
while((rd = fgetc(expected)) != '\n' && !feof(expected)){
expbuff[expos++] = rd;
}
expbuff[expos] = '\0';
if(type == QUERY_TYPE_UNKNOWN){
sprintf(qtypestr,"QUERY_TYPE_UNKNOWN");
}
if(type & QUERY_TYPE_LOCAL_READ){
sprintf(qtypestr,"QUERY_TYPE_LOCAL_READ");
}
if(type & QUERY_TYPE_READ){
sprintf(qtypestr,"QUERY_TYPE_READ");
}
if(type & QUERY_TYPE_WRITE){
sprintf(qtypestr,"QUERY_TYPE_WRITE");
}
if(type & QUERY_TYPE_MASTER_READ){
sprintf(qtypestr,"QUERY_TYPE_MASTER_READ");
}
if(type & QUERY_TYPE_SESSION_WRITE){
sprintf(qtypestr,"QUERY_TYPE_SESSION_WRITE");
}
if(type & QUERY_TYPE_USERVAR_READ){
sprintf(qtypestr,"QUERY_TYPE_USERVAR_READ");
}
if(type & QUERY_TYPE_SYSVAR_READ){
sprintf(qtypestr,"QUERY_TYPE_SYSVAR_READ");
}
if(type & QUERY_TYPE_GSYSVAR_READ){
sprintf(qtypestr,"QUERY_TYPE_GSYSVAR_READ");
}
if(type & QUERY_TYPE_GSYSVAR_WRITE){
sprintf(qtypestr,"QUERY_TYPE_GSYSVAR_WRITE");
}
if(type & QUERY_TYPE_BEGIN_TRX){
sprintf(qtypestr,"QUERY_TYPE_BEGIN_TRX");
}
if(type & QUERY_TYPE_ENABLE_AUTOCOMMIT){
sprintf(qtypestr,"QUERY_TYPE_ENABLE_AUTOCOMMIT");
}
if(type & QUERY_TYPE_DISABLE_AUTOCOMMIT){
sprintf(qtypestr,"QUERY_TYPE_DISABLE_AUTOCOMMIT");
}
if(type & QUERY_TYPE_ROLLBACK){
sprintf(qtypestr,"QUERY_TYPE_ROLLBACK");
}
if(type & QUERY_TYPE_COMMIT){
sprintf(qtypestr,"QUERY_TYPE_COMMIT");
}
if(type & QUERY_TYPE_PREPARE_NAMED_STMT){
sprintf(qtypestr,"QUERY_TYPE_PREPARE_NAMED_STMT");
}
if(type & QUERY_TYPE_PREPARE_STMT){
sprintf(qtypestr,"QUERY_TYPE_PREPARE_STMT");
}
if(type & QUERY_TYPE_EXEC_STMT){
sprintf(qtypestr,"QUERY_TYPE_EXEC_STMT");
}
if(type & QUERY_TYPE_CREATE_TMP_TABLE){
sprintf(qtypestr,"QUERY_TYPE_CREATE_TMP_TABLE");
}
if(type & QUERY_TYPE_READ_TMP_TABLE){
sprintf(qtypestr,"QUERY_TYPE_READ_TMP_TABLE");
}
if(strcmp(qtypestr,expbuff) != 0){
printf("Error in output: '%s' was expected but got '%s'",expbuff,qtypestr);
ex_val = 1;
}
gwbuf_free(buff);
}
}
fclose(input);
fclose(expected);
free(strbuff);
return ex_val;
}

13
query_classifier/test/expected.sql Normal file
View File

@ -0,0 +1,13 @@
QUERY_TYPE_READ
QUERY_TYPE_READ
QUERY_TYPE_WRITE
QUERY_TYPE_WRITE
QUERY_TYPE_CREATE_TMP_TABLE
QUERY_TYPE_GSYSVAR_WRITE
QUERY_TYPE_SYSVAR_READ
QUERY_TYPE_USERVAR_READ
QUERY_TYPE_COMMIT
QUERY_TYPE_DISABLE_AUTOCOMMIT
QUERY_TYPE_BEGIN_TRX
QUERY_TYPE_ROLLBACK
QUERY_TYPE_COMMIT

13
query_classifier/test/input.sql Normal file
View File

@ -0,0 +1,13 @@
select sleep(2);
select * from tst where lname like '%e%' order by fname;
insert into tst values ("Jane","Doe"),("Daisy","Duck"),("Marie","Curie");
update tst set fname="Farmer", lname="McDonald" where lname="%Doe" and fname="John";
create temporary table tmp as select * from t1;
/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
select @@server_id;
select @OLD_SQL_NOTES;
SET autocommit=1;
SET autocommit=0;
BEGIN;
ROLLBACK;
COMMIT;

7
server/core/test/testserver.c
View File

@ -48,8 +48,11 @@ char *status;
ss_dfprintf(stderr,
"testserver : creating server called MyServer");
server = server_alloc("MyServer", "HTTPD", 9876);
ss_info_dassert(NULL != service, "New server with valid protocol and port must not be null");
ss_info_dassert(0 != service_isvalid(service), "Service must be valid after creation");
//ss_info_dassert(NULL != service, "New server with valid protocol and port must not be null");
//ss_info_dassert(0 != service_isvalid(service), "Service must be valid after creation");
ss_dfprintf(stderr, "\t..done\nTest Parameter for Server.");
ss_info_dassert(NULL == serverGetParameter(server, "name"), "Parameter should be null when not set");
serverAddParameter(server, "name", "value");

2
server/core/test/testusers.c
View File

@ -48,7 +48,7 @@ int result, count;
ss_dfprintf(stderr,
"testusers : Initialise the user table.");
users = users_alloc();
ss_info_dassert(NULL != servers, "Allocating user table should not return NULL.")
ss_info_dassert(NULL != users, "Allocating user table should not return NULL.")
ss_dfprintf(stderr, "\t..done\nAdd a user");
count = users_add(users, "username", "authorisation");
ss_info_dassert(1 == count, "Should add one user");

6
server/modules/filter/CMakeLists.txt
View File

@ -24,4 +24,8 @@ add_library(topfilter SHARED topfilter.c)
target_link_libraries(topfilter log_manager utils)
install(TARGETS topfilter DESTINATION modules)
add_subdirectory(hint)
add_subdirectory(hint)
if(BUILD_TESTS)
add_subdirectory(test)
endif()

13
server/modules/filter/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,13 @@
aux_source_directory(${CMAKE_SOURCE_DIR}/server/core CORE_ALL)
foreach(VAR ${CORE_ALL})
if(NOT( (${VAR} MATCHES "max[a-z_]*.c") OR (${VAR} MATCHES "gateway.c")))
list(APPEND CORE ${VAR})
endif()
endforeach()
include_directories(${CMAKE_CURRENT_SOURCE_DIR})
add_executable(harness_ui harness_ui.c harness_common.c)
add_executable(harness harness_util.c harness_common.c ${CORE})
target_link_libraries(harness_ui fullcore log_manager utils)
target_link_libraries(harness fullcore)
add_test(TestHintfilter /bin/sh -c "MAXSCALE_HOME=\"${CMAKE_BINARY_DIR}\" ${CMAKE_CURRENT_BINARY_DIR}/harness -i ${CMAKE_CURRENT_SOURCE_DIR}/hint_testing.input -o ${CMAKE_CURRENT_BINARY_DIR}/hint_testing.output -c ${CMAKE_CURRENT_SOURCE_DIR}/hint_testing.cnf -t 1 -s 1 && diff ${CMAKE_CURRENT_SOURCE_DIR}/hint_testing.expected ${CMAKE_CURRENT_BINARY_DIR}/hint_testing.output;exit $?")

80
server/modules/filter/test/Makefile Executable file
View File

@ -0,0 +1,80 @@
# This file is distributed as part of MaxScale form SkySQL. It 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.
#
# 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 Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright SkySQL Ab 2014
include ../../../../build_gateway.inc
LOGPATH := $(ROOT_PATH)/log_manager
UTILSPATH := $(ROOT_PATH)/utils
QCLASSPATH := $(ROOT_PATH)/query_classifier
COREPATH := $(ROOT_PATH)/server/core
CC=cc
CFLAGS=-c -fPIC -I/usr/include -I../../include -I../../../include \
-I$(LOGPATH) -I$(UTILSPATH) -I./ -I$(ROOT_PATH)/server/inih -I$(QCLASSPATH) \
$(MYSQL_HEADERS) -Wall -g
include ../../../../makefile.inc
LDFLAGS=-rdynamic -L$(LOGPATH) -L$(UTILSPATH) -L$(EMBEDDED_LIB) \
-Wl,-rpath,$(DEST)/lib \
-Wl,-rpath,$(LOGPATH) -Wl,-rpath,$(UTILSPATH) \
-Wl,-rpath,$(EMBEDDED_LIB)
SRCS=harness_util.c harness_common.c
OBJ=$(SRCS:.c=.o)
COREOBJ=$(COREPATH)/load_utils.o $(COREPATH)/dcb.o $(COREPATH)/utils.o \
$(COREPATH)/gw_utils.o $(COREPATH)/buffer.o $(COREPATH)/poll.o \
$(COREPATH)/spinlock.o $(COREPATH)/gwbitmask.o $(COREPATH)/session.o \
$(COREPATH)/atomic.o $(COREPATH)/hashtable.o $(COREPATH)/filter.o $(COREPATH)/modutil.o $(ROOT_PATH)/server/inih/ini.o \
$(COREPATH)/hint.o $(COREPATH)/config.o $(COREPATH)/service.o $(COREPATH)/server.o $(COREPATH)/monitor.o $(COREPATH)/housekeeper.o $(COREPATH)/adminusers.o $(COREPATH)/dbusers.o $(COREPATH)/thread.o $(COREPATH)/users.o $(COREPATH)/secrets.o
LIBS= $(UTILSPATH)/skygw_utils.o -lssl -pthread -llog_manager -lmysqld -ldl -lcrypto -lcrypt -lm
MODULES := $(wildcard ../*.so)
all: build
build:$(OBJ)
$(CC) $(OBJ) $(COREOBJ) $(LDFLAGS) $(LIBS) -o harness
$(MAKE) -C ../
cp ../*.so ./
%.o: %.c
$(CC) $(CFLAGS) $< -o $@
clean:
rm -f *.o
rm -f *.so
rm -f harness
cleantests:clean
rm *.output
buildtests:build
testall:
$(MAKE) cleantests
$(MAKE) buildtests
$(MAKE) runtests
runtests:
@echo ""
@echo "-------------------------------"
@echo "$(shell date)"
@echo "Test Filter harness"
@echo "-------------------------------"
@echo "Testing hints... "
@./hint_tests.sh
@echo ""
documentation:
doxygen doxygen.conf

20
server/modules/filter/test/README Normal file
View File

@ -0,0 +1,20 @@
Filter Test Harness
For a more detailed description of the filter harness, either generate the documentation or read the harness.h file.
Running the program without arguments enters the interactive mode. Type 'help' for a list of all commands.
The default values for threads and sessions are stored in the 'harness.cnf' file
Mandatory parameters for the command line mode are -c and -i.
Parameters for the command line:
-h Display this information
-c Path to the MaxScale configuration file to parse for filters
-i Name of the input file for buffers
-o Name of the output file for results
-q Suppress printing to stdout
-t Number of threads
-s Number of sessions
-d Routing delay

2303
server/modules/filter/test/doxygen.conf Normal file
View File

File diff suppressed because it is too large Load Diff

2
server/modules/filter/test/harness.cnf Normal file
View File

@ -0,0 +1,2 @@
threads=2
sessions=4

362
server/modules/filter/test/harness.h Normal file
View File

@ -0,0 +1,362 @@
#ifndef _FILTER_HARNESS_H
#define _FILTER_HARNESS_H
/*
* This file is distributed as part of the SkySQL Gateway. It 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.
*
* 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 Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright SkySQL Ab 2013
*/
/**
* @mainpage
* Test harness for independent testing of filters
*
* A test harness that feeds a GWBUF to a chain of filters and prints the results
* either into a file or to the standard output.
*
* The contents of the GWBUF and the filter parameters are either manually set through
* the command line or read from a file.
* @verbatim
* Options for the configuration file 'harness.cnf'':
*
* threads Number of threads to use when routing buffers
* sessions Number of sessions
*
* Options for the command line:
*
* -c Path to the MaxScale configuration file to parse for filters
* -i Name of the input file for buffers
* -o Name of the output file for results
* -q Suppress printing to stdout
* -s Number of sessions
* -t Number of threads
* -d Routing delay, in milliseconds
*
*
* Revision History
*
* Date Who Description
* 01/07/14 Markus Makela Initial implementation
*
* @endverbatim
*/
#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <filter.h>
#include <buffer.h>
#include <modules.h>
#include <modutil.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <atomic.h>
#include <ini.h>
#include <hint.h>
#include <modutil.h>
/**
* A single name-value pair and a link to the next item in the
* configuration.
*/
typedef struct CONFIG_ITEM_T
{
char* name;
char* value;
struct CONFIG_ITEM_T* next;
}CONFIG_ITEM;
/**
*A simplified version of a MaxScale configuration context used to load filters
* and their options.
*/
typedef struct CONFIG_T
{
char* section;
CONFIG_ITEM* item;
struct CONFIG_T* next;
}CONFIG;
/**
*A structure that holds all the necessary information to emulate a working
* filter environment.
*/
struct FILTERCHAIN_T
{
FILTER* filter; /**An instance of a particular filter*/
FILTER_OBJECT* instance; /**Dynamically loaded module*/
SESSION** session; /**A list of sessions*/
DOWNSTREAM** down; /** A list of next filters downstreams*/
UPSTREAM** up; /** A list of next filters upstreams*/
char* name; /**Module name*/
struct FILTERCHAIN_T* next;
};
typedef struct FILTERCHAIN_T FILTERCHAIN;
/**
* A container for all the filters, query buffers and user specified parameters
*/
typedef struct
{
int running;
int verbose; /**Whether to print to stdout*/
int infile; /**A file where the queries are loaded from*/
int error;
char* mod_dir; /**Module directory absolute path*/
char* infile_name;
int outfile; /**A file where the output of the filters is logged*/
char* outfile_name;
FILTERCHAIN* head; /**The head of the filter chain*/
FILTERCHAIN* tail; /**The tail of the filter chain*/
GWBUF** buffer; /**Buffers that are fed to the filter chain*/
int buffer_count;
int session_count;
DOWNSTREAM dummyrouter; /**Dummy downstream router for data extraction*/
UPSTREAM dummyclient; /**Dummy downstream router for data extraction*/
CONFIG* conf; /**Configurations loaded from a file*/
pthread_mutex_t work_mtx; /**Mutex for buffer routing*/
int buff_ind; /**Index of first unrouted buffer*/
int sess_ind;/**Index of first unused session*/
int last_ind; /**Index of last used session*/
pthread_t* thrpool;
int thrcount; /**Number of active threads*/
int rt_delay; /**Delay each thread waits after routing a query, in milliseconds*/
}HARNESS_INSTANCE;
static HARNESS_INSTANCE instance;
/**
*A list of available actions.
*/
typedef enum
{
UNDEFINED,
RUNFILTERS,
LOAD_FILTER,
DELETE_FILTER,
LOAD_CONFIG,
SET_INFILE,
SET_OUTFILE,
THR_COUNT,
SESS_COUNT,
OK,
QUIT
} operation_t;
typedef enum
{
PACKET_OK,
PACKET_ERROR,
PACKET_RESULT_SET
} packet_t;
typedef packet_t PACKET;
/**
* Initialize the static instance.
*/
int harness_init(int argc,char** argv);
/**
* Frees all the query buffers
*/
void free_buffers();
/**
* Frees all the loaded filters
*/
void free_filters();
/**
* Converts the passed string into an operation
*
* @param tk The string to parse
* @return The operation to perform or UNDEFINED, if parsing failed
*/
operation_t user_input(char* tk);
/**
*Prints a list of available commands.
*/
void print_help();
/**
* Prints the current status of loaded filters and queries, number of threads
* and sessions and possible output files.
*/
void print_status();
/**
*Opens a file for reading and/or writing with adequate permissions.
*
* @param str Path to file
* @param write Non-zero for write permissions, zero for read only.
* @return The assigned file descriptor or -1 in case an error occurred
*/
int open_file(char* str, unsigned int write);
/**
* Reads filter parameters from the command line as name-value pairs.
*
*@param paramc The number of parameters read is assigned to this variable
*@return The newly allocated list of parameters with the last one being NULL
*/
FILTER_PARAMETER** read_params(int* paramc);
/**
* Dummy endpoint for the queries of the filter chain
*
* Prints and logs the contents of the GWBUF after it has passed through all the filters.
* The packet is handled as a COM_QUERY packet and the packet header is not printed.
*/
int routeQuery(void* instance, void* session, GWBUF* queue);
/**
* Dummy endpoint for the replies of the filter chain
*
* Prints and logs the contents of the GWBUF after it has passed through all the filters.
* The packet is handled as a OK packet with no message and the packet header is not printed.
*/
int clientReply(void* ins, void* session, GWBUF* queue);
/**
*Manual input if a query string
*
* Reads a single query from the standard input and inserts it into a GWBUF.
*/
void manual_query();
/**
*Loads the file pointed by @{instance.infile}
* @return Zero if successful, non-zero if an error occurred
*/
int load_query();
/**
* Handler for the INI file parser that builds a linked list
* of all the sections and their name-value pairs.
* @param user Current configuration.
* @param section Name of the section.
* @param name Name of the item.
* @param value Value of the item.
* @return Non-zero on success, zero in case parsing is finished.
* @see load_config()
*/
int handler(void* user, const char* section, const char* name,const char* value);
/**
* Removes all non-filter modules from the configuration
*
* @param conf A pointer to a configuration struct
* @return The stripped version of the configuration
* @see load_config()
*/
CONFIG* process_config(CONFIG* conf);
/**
* Loads the filter module and link it to the filter chain
*
* The downstream is set to point to the current head of the filter chain
*
* @param str Name of the filter module
* @return Pointer to the newly initialized FILTER_CHAIN element or NULL in case module loading failed
* @see load_filter()
*/
FILTERCHAIN* load_filter_module(char* str);
/**
* Loads a new instance of a filter and starts a new session.
* This function assumes that the filter module is already loaded.
* Passing NULL as the CONFIG parameter causes the parameters to be
* read from the command line one at a time.
*
* @param fc The FILTERCHAIN where the new instance and session are created
* @param cnf A configuration read from a file
* @return 1 on success, 0 in case an error occurred
* @see load_filter_module()
*/
int load_filter(FILTERCHAIN* fc, CONFIG* cnf);
/**
* Reads a MaxScale configuration (or any INI file using MaxScale notation) file and loads only the filter modules in it.
*
* @param fname Configuration file name
* @return Non-zero on success, zero in case an error occurred.
*/
int load_config(char* fname);
/**
* Initializes the indexes used while routing buffers and prints the progress
* of the routing process.
*/
void route_buffers();
/**
* Worker function for threads.
* Routes a query buffer if there are unrouted buffers left.
*
* @param thr_num ID number of the thread
*/
void work_buffer(void* thr_num);
/**
* Generates a fake packet used to emulate a response from the backend.
*
* Current implementation only works with PACKET_OK and the packet has no message.
* The caller is responsible for freeing the allocated memory by calling gwbuf_free().
* @param pkt The packet type
* @return The newly generated packet or NULL if an error occurred
*/
GWBUF* gen_packet(PACKET pkt);
/**
* Process the command line parameters and the harness configuration file.
*
* Reads the contents of the 'harness.cnf' file and command line parameters
* and parses them. Options are interpreted accoding to the following table.
* If no command line arguments are given, interactive mode is used.
*
* By default if no input file is given or no configuration file or specific
* filters are given, but other options are, the program exits with 0.
*
* Options for the configuration file 'harness.cnf'':
*
* threads Number of threads to use when routing buffers
* sessions Number of sessions
*
* Options for the command line:
*
* -h Display this information
* -c Path to the MaxScale configuration file to parse for filters
* -i Name of the input file for buffers
* -o Name of the output file for results
* -q Suppress printing to stdout
* -t Number of threads
* -s Number of sessions
* -d Routing delay
*
* @param argc Number of arguments
* @param argv List of argument strings
* @return 1 if successful, 0 if no input file, configuration file or specific
* filters are given, but other options are, or if an error occurs.
*/
int process_opts(int argc, char** argv);
#endif

997
server/modules/filter/test/harness_common.c Normal file
View File

@ -0,0 +1,997 @@
#include <harness.h>
int harness_init(int argc, char** argv){
int i = 0;
if(!(argc == 2 && strcmp(argv[1],"-h") == 0)){
skygw_logmanager_init(0,NULL);
}
if(!(instance.head = calloc(1,sizeof(FILTERCHAIN))))
{
printf("Error: Out of memory\n");
skygw_log_write(LOGFILE_ERROR,"Error: Out of memory\n");
return 1;
}
instance.running = 1;
instance.infile = -1;
instance.outfile = -1;
instance.buff_ind = -1;
instance.last_ind = -1;
instance.sess_ind = -1;
process_opts(argc,argv);
if(!(instance.thrpool = malloc(instance.thrcount * sizeof(pthread_t)))){
printf("Error: Out of memory\n");
skygw_log_write(LOGFILE_ERROR,"Error: Out of memory\n");
return 1;
}
/**Initialize worker threads*/
pthread_mutex_lock(&instance.work_mtx);
size_t thr_num = 1;
for(i = 0;i<instance.thrcount;i++){
pthread_create(&instance.thrpool[i],NULL,(void*)work_buffer,(void*)thr_num++);
}
return 0;
}
void free_filters()
{
int i;
if(instance.head){
while(instance.head->next){
FILTERCHAIN* tmph = instance.head;
instance.head = instance.head->next;
if(tmph->instance){
for(i = 0;i<instance.session_count;i++){
if(tmph->filter && tmph->session[i]){
tmph->instance->freeSession(tmph->filter,tmph->session[i]);
}
}
}
free(tmph->filter);
free(tmph->session);
free(tmph->down);
free(tmph->name);
free(tmph);
}
}
}
void free_buffers()
{
if(instance.buffer){
int i;
for(i = 0;i<instance.buffer_count;i++){
gwbuf_free(instance.buffer[i]);
}
free(instance.buffer);
instance.buffer = NULL;
instance.buffer_count = 0;
}
if(instance.infile >= 0){
close(instance.infile);
free(instance.infile_name);
instance.infile = -1;
}
}
int open_file(char* str, unsigned int write)
{
int mode;
if(write){
mode = O_RDWR|O_CREAT;
}else{
mode = O_RDONLY;
}
return open(str,mode,S_IRWXU|S_IRGRP|S_IXGRP|S_IXOTH);
}
FILTER_PARAMETER** read_params(int* paramc)
{
char buffer[256];
char* token;
char* names[64];
char* values[64];
int pc = 0, do_read = 1, val_len = 0;
int i;
memset(names,0,64);
memset(values,0,64);
printf("Enter filter parametes as <name>=<value>, enter \"done\" to stop.\n");
while(do_read){
memset(buffer,0,256);
printf(">");
fgets(buffer,255,stdin);
if(strcmp("done\n",buffer) == 0){
do_read = 0;
}else{
token = strtok(buffer,"=\n");
if(token!=NULL){
val_len = strcspn(token," \n\0");
if((names[pc] = calloc((val_len + 1),sizeof(char))) != NULL){
memcpy(names[pc],token,val_len);
}
}
token = strtok(NULL,"=\n");
if(token!=NULL){
val_len = strcspn(token," \n\0");
if((values[pc] = calloc((val_len + 1),sizeof(char))) != NULL){
memcpy(values[pc],token,val_len);
}
pc++;
}
}
if(pc >= 64){
do_read = 0;
}
}
FILTER_PARAMETER** params;
if((params = malloc(sizeof(FILTER_PARAMETER*)*(pc+1)))!=NULL){
for(i = 0;i<pc;i++){
params[i] = malloc(sizeof(FILTER_PARAMETER));
if(params[i]){
params[i]->name = strdup(names[i]);
params[i]->value = strdup(values[i]);
}
free(names[i]);
free(values[i]);
}
}
params[pc] = NULL;
*paramc = pc;
return params;
}
int routeQuery(void* ins, void* session, GWBUF* queue)
{
unsigned int buffsz = 0;
char *qstr;
buffsz = (char*)queue->end - ((char*)queue->start + 5);
if(queue->hint){
buffsz += 40;
if(queue->hint->data){
buffsz += strnlen(queue->hint->data,1024);
}
if(queue->hint->value){
buffsz += strnlen(queue->hint->value,1024);
}
}
qstr = calloc(buffsz + 1,sizeof(char));
if(qstr){
memcpy(qstr,queue->start + 5,buffsz);
if(queue->hint){
char *ptr = qstr + strlen(qstr);
switch(queue->hint->type){
case HINT_ROUTE_TO_MASTER:
sprintf(ptr,"|HINT_ROUTE_TO_MASTER");
break;
case HINT_ROUTE_TO_SLAVE:
sprintf(ptr,"|HINT_ROUTE_TO_SLAVE");
break;
case HINT_ROUTE_TO_NAMED_SERVER:
sprintf(ptr,"|HINT_ROUTE_TO_NAMED_SERVER");
break;
case HINT_ROUTE_TO_UPTODATE_SERVER:
sprintf(ptr,"|HINT_ROUTE_TO_UPTODATE_SERVER");
break;
case HINT_ROUTE_TO_ALL:
sprintf(ptr,"|HINT_ROUTE_TO_ALL");
break;
case HINT_PARAMETER:
sprintf(ptr,"|HINT_PARAMETER");
break;
default:
sprintf(ptr,"|HINT_UNDEFINED");
break;
}
ptr = qstr + strlen(qstr);
if(queue->hint->data){
sprintf(ptr,"|%s",(char*)queue->hint->data);
ptr = qstr + strlen(qstr);
}
if(queue->hint->value){
sprintf(ptr,"|%s",(char*)queue->hint->value);
ptr = qstr + strlen(qstr);
}
}
}else{
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
return 0;
}
if(instance.verbose){
printf("Query endpoint: %s\n", qstr);
}
if(instance.outfile>=0){
write(instance.outfile,qstr,strlen(qstr));
write(instance.outfile,"\n",1);
}
free(qstr);
return 1;
}
int clientReply(void* ins, void* session, GWBUF* queue)
{
if(instance.verbose){
pthread_mutex_lock(&instance.work_mtx);
unsigned char* ptr = (unsigned char*)queue->start;
unsigned int i,pktsize = 4 + ptr[0] + (ptr[1] << 8) + (ptr[2] << 16);
printf("Reply endpoint: ");
for(i = 0;i<pktsize;i++){
printf("%.2x ",*ptr++);
}
printf("\n");
pthread_mutex_unlock(&instance.work_mtx);
}
if(instance.outfile>=0){
int qlen = queue->end - queue->start;
write(instance.outfile,"Reply: ",strlen("Reply: "));
write(instance.outfile,queue->start,qlen);
write(instance.outfile,"\n",1);
}
return 1;
}
int load_query()
{
char** query_list;
char* buff;
char rc;
int i, qcount = 0, qbuff_sz = 10, buff_sz = 2048;
int offset = 0;
unsigned int qlen = 0;
if((buff = calloc(buff_sz,sizeof(char))) == NULL ||
(query_list = calloc(qbuff_sz,sizeof(char*))) == NULL){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
return 1;
}
while(read(instance.infile,&rc,1)){
if(rc != '\n' && rc != '\0'){
if(offset >= buff_sz){
char* tmp = malloc(sizeof(char)*2*buff_sz);
if(tmp){
memcpy(tmp,buff,buff_sz);
free(buff);
buff = tmp;
buff_sz *= 2;
}else{
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
free(buff);
return 1;
}
}
buff[offset++] = rc;
}else{
if(qcount >= qbuff_sz){
char** tmpcl = malloc(sizeof(char*) * (qcount * 2 + 1));
if(!tmpcl){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
return 1;
}
for(i = 0;i < qbuff_sz;i++){
tmpcl[i] = query_list[i];
}
free(query_list);
query_list = tmpcl;
qbuff_sz = qcount * 2 + 1;
}
query_list[qcount] = malloc(sizeof(char)*(offset + 1));
memcpy(query_list[qcount],buff,offset);
query_list[qcount][offset] = '\0';
offset = 0;
qcount++;
}
}
GWBUF** tmpbff = malloc(sizeof(GWBUF*)*(qcount + 1));
if(tmpbff){
for(i = 0;i<qcount;i++){
tmpbff[i] = gwbuf_alloc(strnlen(query_list[i],buff_sz) + 6);
gwbuf_set_type(tmpbff[i],GWBUF_TYPE_MYSQL);
memcpy(tmpbff[i]->sbuf->data + 5,query_list[i],strnlen(query_list[i],buff_sz));
qlen = strnlen(query_list[i],buff_sz);
tmpbff[i]->sbuf->data[0] = qlen;
tmpbff[i]->sbuf->data[1] = (qlen << 8);
tmpbff[i]->sbuf->data[2] = (qlen << 16);
tmpbff[i]->sbuf->data[3] = 0x00;
tmpbff[i]->sbuf->data[4] = 0x03;
}
tmpbff[qcount] = NULL;
instance.buffer = tmpbff;
}else{
printf("Error: cannot allocate enough memory for buffers.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory for buffers.\n");
free_buffers();
return 1;
}
if(qcount < 1){
return 1;
}
instance.buffer_count = qcount;
return 0;
}
int handler(void* user, const char* section, const char* name,
const char* value)
{
CONFIG* conf = instance.conf;
if(conf == NULL){/**No sections handled*/
if((conf = malloc(sizeof(CONFIG))) &&
(conf->item = malloc(sizeof(CONFIG_ITEM)))){
conf->section = strdup(section);
conf->item->name = strdup(name);
conf->item->value = strdup(value);
conf->item->next = NULL;
conf->next = NULL;
}
}else{
CONFIG* iter = instance.conf;
/**Finds the matching section*/
while(iter){
if(strcmp(iter->section,section) == 0){
CONFIG_ITEM* item = malloc(sizeof(CONFIG_ITEM));
if(item){
item->name = strdup(name);
item->value = strdup(value);
item->next = iter->item;
iter->item = item;
break;
}
}else{
iter = iter->next;
}
}
/**Section not found, creating a new one*/
if(iter == NULL){
CONFIG* nxt = malloc(sizeof(CONFIG));
if(nxt && (nxt->item = malloc(sizeof(CONFIG_ITEM)))){
nxt->section = strdup(section);
nxt->item->name = strdup(name);
nxt->item->value = strdup(value);
nxt->item->next = NULL;
nxt->next = conf;
conf = nxt;
}
}
}
instance.conf = conf;
return 1;
}
CONFIG* process_config(CONFIG* conf)
{
CONFIG* tmp;
CONFIG* tail = conf;
CONFIG* head = NULL;
CONFIG_ITEM* item;
while(tail){
item = tail->item;
while(item){
if(strcmp("type",item->name) == 0 &&
strcmp("filter",item->value) == 0){
tmp = tail->next;
tail->next = head;
head = tail;
tail = tmp;
break;
}else{
item = item->next;
}
}
if(item == NULL){
tail = tail->next;
}
}
return head;
}
int load_config( char* fname)
{
CONFIG* iter;
CONFIG_ITEM* item;
int config_ok = 1;
free_filters();
if(ini_parse(fname,handler,instance.conf) < 0){
printf("Error parsing configuration file!\n");
skygw_log_write(LOGFILE_ERROR,"Error parsing configuration file!\n");
config_ok = 0;
goto cleanup;
}
if(instance.verbose){
printf("Configuration loaded from %s\n\n",fname);
}
if(instance.conf == NULL){
printf("Nothing valid was read from the file.\n");
skygw_log_write(LOGFILE_MESSAGE,"Nothing valid was read from the file.\n");
config_ok = 0;
goto cleanup;
}
instance.conf = process_config(instance.conf);
if(instance.conf){
if(instance.verbose){
printf("Modules Loaded:\n");
}
iter = instance.conf;
}else{
printf("No filters found in the configuration file.\n");
skygw_log_write(LOGFILE_MESSAGE,"No filters found in the configuration file.\n");
config_ok = 0;
goto cleanup;
}
while(iter){
item = iter->item;
while(item){
if(!strcmp("module",item->name)){
if(instance.mod_dir){
char* modstr = malloc(sizeof(char)*(strlen(instance.mod_dir) + strlen(item->value) + 1));
strcpy(modstr,instance.mod_dir);
strcat(modstr,"/");
strcat(modstr,item->value);
instance.head = load_filter_module(modstr);
free(modstr);
}else{
instance.head = load_filter_module(item->value);
}
if(!instance.head || !load_filter(instance.head,instance.conf)){
printf("Error creating filter instance!\nModule: %s\n",item->value);
skygw_log_write(LOGFILE_ERROR,"Error creating filter instance!\nModule: %s\n",item->value);
config_ok = 0;
goto cleanup;
}else{
if(instance.verbose){
printf("\t%s\n",iter->section);
}
}
}
item = item->next;
}
iter = iter->next;
}
while(instance.conf){
item = instance.conf->item;
while(item){
item = instance.conf->item;
instance.conf->item = instance.conf->item->next;
free(item->name);
free(item->value);
free(item);
item = instance.conf->item;
}
instance.conf = instance.conf->next;
}
cleanup:
while(instance.conf){
iter = instance.conf;
instance.conf = instance.conf->next;
item = iter->item;
while(item){
free(item->name);
free(item->value);
free(item);
iter->item = iter->item->next;
item = iter->item;
}
free(iter);
}
instance.conf = NULL;
return config_ok;
}
int load_filter(FILTERCHAIN* fc, CONFIG* cnf)
{
FILTER_PARAMETER** fparams;
int i, paramc = -1;
if(cnf == NULL){
fparams = read_params(&paramc);
}else{
CONFIG* iter = cnf;
CONFIG_ITEM* item;
while(iter){
paramc = -1;
item = iter->item;
while(item){
/**Matching configuration found*/
if(!strcmp(item->name,"module") && !strcmp(item->value,fc->name)){
paramc = 0;
item = iter->item;
while(item){
if(strcmp(item->name,"module") && strcmp(item->name,"type")){
paramc++;
}
item = item->next;
}
item = iter->item;
fparams = calloc((paramc + 1),sizeof(FILTER_PARAMETER*));
if(fparams){
int i = 0;
while(item){
if(strcmp(item->name,"module") != 0 &&
strcmp(item->name,"type") != 0){
fparams[i] = malloc(sizeof(FILTER_PARAMETER));
if(fparams[i]){
fparams[i]->name = strdup(item->name);
fparams[i]->value = strdup(item->value);
i++;
}
}
item = item->next;
}
}
}
if(paramc > -1){
break;
}else{
item = item->next;
}
}
if(paramc > -1){
break;
}else{
iter = iter->next;
}
}
}
int sess_err = 0;
if(fc && fc->instance){
fc->filter = (FILTER*)fc->instance->createInstance(NULL,fparams);
for(i = 0;i<instance.session_count;i++){
if((fc->session[i] = fc->instance->newSession(fc->filter, fc->session[i])) &&
(fc->down[i] = calloc(1,sizeof(DOWNSTREAM))) &&
(fc->up[i] = calloc(1,sizeof(UPSTREAM)))){
fc->up[i]->session = NULL;
fc->up[i]->instance = NULL;
fc->up[i]->clientReply = (void*)clientReply;
if(fc->instance->setUpstream && fc->instance->clientReply){
fc->instance->setUpstream(fc->filter, fc->session[i], fc->up[i]);
}else{
skygw_log_write(LOGFILE_MESSAGE,
"Warning: The filter %s does not support client relies.\n",fc->name);
}
if(fc->next && fc->next->next){
fc->down[i]->routeQuery = (void*)fc->next->instance->routeQuery;
fc->down[i]->session = fc->next->session[i];
fc->down[i]->instance = fc->next->filter;
fc->instance->setDownstream(fc->filter, fc->session[i], fc->down[i]);
fc->next->up[i]->clientReply = (void*)fc->instance->clientReply;
fc->next->up[i]->session = fc->session[i];
fc->next->up[i]->instance = fc->filter;
if(fc->instance->setUpstream && fc->instance->clientReply){
fc->next->instance->setUpstream(fc->next->filter,fc->next->session[i],fc->next->up[i]);
}
}else{ /**The dummy router is the next one*/
fc->down[i]->routeQuery = (void*)routeQuery;
fc->down[i]->session = NULL;
fc->down[i]->instance = NULL;
fc->instance->setDownstream(fc->filter, fc->session[i], fc->down[i]);
}
}
if(!fc->session[i] || !fc->down[i] || !fc->up[i]){
sess_err = 1;
break;
}
}
if(sess_err){
for(i = 0;i<instance.session_count;i++){
if(fc->filter && fc->session[i]){
fc->instance->freeSession(fc->filter, fc->session[i]);
}
free(fc->down[i]);
}
free(fc->session);
free(fc->down);
free(fc->name);
free(fc);
}
}
if(cnf){
int x;
for(x = 0;x<paramc;x++){
free(fparams[x]->name);
free(fparams[x]->value);
}
free(fparams);
}
return sess_err ? 0 : 1;
}
FILTERCHAIN* load_filter_module(char* str)
{
FILTERCHAIN* flt_ptr = NULL;
if((flt_ptr = calloc(1,sizeof(FILTERCHAIN))) != NULL &&
(flt_ptr->session = calloc(instance.session_count,sizeof(SESSION*))) != NULL &&
(flt_ptr->down = calloc(instance.session_count,sizeof(DOWNSTREAM*))) != NULL &&
(flt_ptr->up = calloc(instance.session_count,sizeof(UPSTREAM*))) != NULL){
flt_ptr->next = instance.head;
}
if((flt_ptr->instance = (FILTER_OBJECT*)load_module(str, MODULE_FILTER)) == NULL)
{
printf("Error: Module loading failed: %s\n",str);
skygw_log_write(LOGFILE_ERROR,"Error: Module loading failed: %s\n",str);
free(flt_ptr->down);
free(flt_ptr);
return NULL;
}
flt_ptr->name = strdup(str);
return flt_ptr;
}
void route_buffers()
{
if(instance.buffer_count > 0){
float tprg = 0.f, bprg = 0.f, trig = 0.f,
fin = instance.buffer_count*instance.session_count,
step = (fin/50.f)/fin;
FILTERCHAIN* fc = instance.head;
while(fc->next->next){
fc = fc->next;
}
instance.tail = fc;
instance.buff_ind = 0;
instance.sess_ind = 0;
instance.last_ind = 0;
printf("Routing queries...\n");
if(!instance.verbose){
printf("%s","|0%");
float f;
for(f = 0.f;f<1.f - step*7;f += step){
printf(" ");
}
printf("%s\n","100%|");
write(1,"|",1);
}
while(instance.buff_ind < instance.buffer_count){
pthread_mutex_unlock(&instance.work_mtx);
while(instance.last_ind < instance.session_count){
tprg = ((bprg + (float)instance.last_ind)/fin);
if(!instance.verbose){
if(tprg >= trig){
write(1,"-",1);
trig += step;
}
}
usleep(100);
}
pthread_mutex_lock(&instance.work_mtx);
instance.buff_ind++;
bprg += instance.last_ind;
instance.sess_ind = 0;
instance.last_ind = 0;
}
if(!instance.verbose){
write(1,"|\n",2);
}
printf("Queries routed.\n");
}
}
void work_buffer(void* thr_num)
{
unsigned int index = instance.session_count;
GWBUF* fake_ok = gen_packet(PACKET_OK);
while(instance.running){
pthread_mutex_lock(&instance.work_mtx);
pthread_mutex_unlock(&instance.work_mtx);
index = atomic_add(&instance.sess_ind,1);
if(instance.running &&
index < instance.session_count &&
instance.buff_ind < instance.buffer_count)
{
instance.head->instance->routeQuery(instance.head->filter,
instance.head->session[index],
instance.buffer[instance.buff_ind]);
if(instance.tail->instance->clientReply){
instance.tail->instance->clientReply(instance.tail->filter,
instance.tail->session[index],
fake_ok);
}
atomic_add(&instance.last_ind,1);
usleep(1000*instance.rt_delay);
}
}
gwbuf_free(fake_ok);
}
GWBUF* gen_packet(PACKET pkt)
{
unsigned int psize = 0;
GWBUF* buff = NULL;
unsigned char* ptr;
switch(pkt){
case PACKET_OK:
psize = 11;
break;
default:
break;
}
if(psize > 0){
buff = gwbuf_alloc(psize);
ptr = (unsigned char*)buff->start;
switch(pkt){
case PACKET_OK:
ptr[0] = 7; /**Packet size*/
ptr[1] = 0;
ptr[2] = 0;
ptr[3] = 1; /**sequence_id*/
ptr[4] = 0; /**OK header*/
ptr[5] = 0; /**affected_rows*/
ptr[6] = 0; /**last_insert_id*/
ptr[7] = 0; /**status_flags*/
ptr[8] = 0;
ptr[9] = 0; /**warnings*/
ptr[10] = 0;
break;
default:
break;
}
}
return buff;
}
int process_opts(int argc, char** argv)
{
int fd = open_file("harness.cnf",1), buffsize = 1024;
int rd,fsize;
char *buff = calloc(buffsize,sizeof(char)), *tok = NULL;
/**Parse 'harness.cnf' file*/
fsize = lseek(fd,0,SEEK_END);
lseek(fd,0,SEEK_SET);
instance.thrcount = 1;
instance.session_count = 1;
read(fd,buff,fsize);
tok = strtok(buff,"=");
while(tok){
if(!strcmp(tok,"threads")){
tok = strtok(NULL,"\n\0");
instance.thrcount = strtol(tok,0,0);
}else if(!strcmp(tok,"sessions")){
tok = strtok(NULL,"\n\0");
instance.session_count = strtol(tok,0,0);
}
tok = strtok(NULL,"=");
}
free(buff);
instance.verbose = 1;
if(argc < 2){
return 1;
}
char* conf_name = NULL;
while((rd = getopt(argc,argv,"m:c:i:o:s:t:d:qh")) > 0){
switch(rd){
case 'o':
instance.outfile = open_file(optarg,1);
printf("Output is written to: %s\n",optarg);
break;
case 'i':
instance.infile = open_file(optarg,0);
printf("Input is read from: %s\n",optarg);
break;
case 'c':
conf_name = strdup(optarg);
break;
case 'q':
instance.verbose = 0;
break;
case 's':
instance.session_count = atoi(optarg);
printf("Sessions: %i ",instance.session_count);
break;
case 't':
instance.thrcount = atoi(optarg);
printf("Threads: %i ",instance.thrcount);
break;
case 'd':
instance.rt_delay = atoi(optarg);
printf("Routing delay: %i ",instance.rt_delay);
break;
case 'h':
printf(
"\nOptions for the configuration file 'harness.cnf'':\n\n"
"\tthreads\tNumber of threads to use when routing buffers\n"
"\tsessions\tNumber of sessions\n\n"
"Options for the command line:\n\n"
"\t-h\tDisplay this information\n"
"\t-c\tPath to the MaxScale configuration file to parse for filters\n"
"\t-i\tName of the input file for buffers\n"
"\t-o\tName of the output file for results\n"
"\t-q\tSuppress printing to stdout\n"
"\t-s\tNumber of sessions\n"
"\t-t\tNumber of threads\n"
"\t-d\tRouting delay\n");
break;
case 'm':
instance.mod_dir = strdup(optarg);
printf("Module directory: %s",optarg);
break;
default:
break;
}
}
printf("\n");
if(conf_name && load_config(conf_name)){
load_query();
}else{
instance.running = 0;
}
return 0;
}

403
server/modules/filter/test/harness_ui.c Executable file
View File

@ -0,0 +1,403 @@
#include <harness.h>
int main(int argc, char** argv){
int i;
char buffer[256];
char* tk;
FILTERCHAIN* tmp_chn;
FILTERCHAIN* del_chn;
if(harness_init(argc,argv)){
printf("Error: Initialization failed.\n");
skygw_log_write(LOGFILE_ERROR,"Error: Initialization failed.\n");
skygw_logmanager_done();
skygw_logmanager_exit();
return 1;
}
if(instance.verbose){
printf("\n\n\tFilter Test Harness\n\n");
}
while(instance.running){
printf("Harness> ");
memset(buffer,0,256);
fgets(buffer,256,stdin);
tk = strtok(buffer," \n");
switch(user_input(tk))
{
case RUNFILTERS:
if(instance.head->next == NULL){
printf("No filters loaded.\n");
break;
}
if(instance.buffer == NULL){
if(instance.infile<0){
manual_query();
}else{
load_query();
}
}
route_buffers();
break;
case LOAD_FILTER:
tk = strtok(NULL," \n");
tmp_chn = load_filter_module(tk);
if(!tmp_chn || !load_filter(tmp_chn,instance.conf)){
printf("Error creating filter instance.\n");
skygw_log_write(LOGFILE_ERROR,"Error: Error creating filter instance.\n");
}else{
instance.head = tmp_chn;
}
break;
case DELETE_FILTER:
tk = strtok(NULL," \n\0");
tmp_chn = instance.head;
del_chn = instance.head;
if(tk){
if(strcmp(instance.head->name,tk) == 0){
instance.head = instance.head->next;
}else{
while(del_chn->next){
if(strcmp(del_chn->name,tk) == 0){
tmp_chn->next = del_chn->next;
break;
}else{
tmp_chn = del_chn;
del_chn = del_chn->next;
}
}
}
if(del_chn && del_chn->next){
printf("Deleted %s.\n",del_chn->name);
if(del_chn->instance){
del_chn->instance->freeSession(del_chn->filter,del_chn->session);
}
free(del_chn->filter);
free(del_chn->down);
free(del_chn->name);
free(del_chn);
}else{
printf("No matching filter found.\n");
}
}
break;
case LOAD_CONFIG:
tk = strtok(NULL," \n\0");
if(!load_config(tk)){
free_filters();
}
break;
case SET_INFILE:
tk = strtok(NULL," \n\0");
if(instance.infile >= 0){
close(instance.infile);
free(instance.infile_name);
}
if(tk!= NULL){
free_buffers();
instance.infile = open_file(tk,0);
if(instance.infile >= 0){
load_query();
instance.infile_name = strdup(tk);
if(instance.verbose){
printf("Loaded %d queries from file '%s'\n",instance.buffer_count,instance.infile_name);
}
}
}else{
instance.infile = -1;
printf("Queries are read from: command line\n");
}
break;
case SET_OUTFILE:
tk = strtok(NULL," \n\0");
if(instance.outfile >= 0){
close(instance.outfile);
free(instance.outfile_name);
}
if(tk!= NULL){
instance.outfile = open_file(tk,1);
if(instance.outfile >= 0){
instance.outfile_name = strdup(tk);
printf("Output is logged to: %s\n",tk);
}
}else{
instance.outfile = -1;
printf("Output logging disabled.\n");
}
break;
case SESS_COUNT:
tk = strtok(NULL," \n\0");
free_buffers();
free_filters();
instance.session_count = atoi(tk);
printf("Sessions set to: %d\n", instance.session_count);
break;
case THR_COUNT:
instance.running = 0;
pthread_mutex_unlock(&instance.work_mtx);
for(i = 0;i<instance.thrcount;i++){
pthread_join(instance.thrpool[i],NULL);
}
pthread_mutex_lock(&instance.work_mtx);
instance.running = 1;
tk = strtok(NULL," \n\0");
instance.thrcount = atoi(tk);
void* t_thr_pool;
if(!(t_thr_pool = realloc(instance.thrpool,instance.thrcount * sizeof(pthread_t)))){
printf("Error: Out of memory\n");
skygw_log_write(LOGFILE_ERROR,"Error: Out of memory\n");
instance.running = 0;
break;
}
instance.thrpool = t_thr_pool;
int thr_num = 1;
for(i = 0;i<instance.thrcount;i++){
pthread_create(&instance.thrpool[i],
NULL,
(void*)work_buffer,
(void*)thr_num++);
}
printf("Threads set to: %d\n", instance.thrcount);
break;
case QUIT:
instance.running = 0;
pthread_mutex_unlock(&instance.work_mtx);
for(i = 0;i<instance.thrcount;i++){
pthread_join(instance.thrpool[i],NULL);
}
break;
case UNDEFINED:
printf("Command not found, enter \"help\" for a list of commands\n");
break;
default:
break;
}
}
if(instance.infile >= 0){
close(instance.infile);
}
if(instance.outfile >= 0){
close(instance.outfile);
}
free_buffers();
free_filters();
skygw_logmanager_done();
skygw_logmanager_exit();
free(instance.head);
return 0;
}
operation_t user_input(char* tk)
{
if(tk){
char cmpbuff[256];
int tklen = strcspn(tk," \n\0");
memset(cmpbuff,0,256);
if(tklen > 0 && tklen < 256){
strncpy(cmpbuff,tk,tklen);
strcat(cmpbuff,"\0");
if(strcmp(tk,"run")==0 || strcmp(tk,"r")==0){
return RUNFILTERS;
}else if(strcmp(cmpbuff,"add")==0){
return LOAD_FILTER;
}else if(strcmp(cmpbuff,"delete")==0){
return DELETE_FILTER;
}else if(strcmp(cmpbuff,"clear")==0){
tk = strtok(NULL," \n\0");
if(tk && !strcmp(tk,"queries")){
free_buffers();
printf("Queries cleared.\n");
}else if(tk && !strcmp(tk,"filters")){
printf("Filters cleared.\n");
free_filters();
}else{
printf("All cleared.\n");
free_buffers();
free_filters();
}
return OK;
}else if(strcmp(cmpbuff,"config")==0){
return LOAD_CONFIG;
}else if(strcmp(cmpbuff,"in")==0){
return SET_INFILE;
}else if(strcmp(cmpbuff,"out")==0){
return SET_OUTFILE;
}else if(strcmp(cmpbuff,"exit")==0 || strcmp(cmpbuff,"quit")==0 || strcmp(cmpbuff,"q")==0){
return QUIT;
}else if(strcmp(cmpbuff,"help")==0){
print_help();
return OK;
}else if(strcmp(cmpbuff,"status")==0){
print_status();
return OK;
}else if(strcmp(cmpbuff,"quiet")==0){
instance.verbose = 0;
return OK;
}else if(strcmp(cmpbuff,"verbose")==0){
instance.verbose = 1;
return OK;
}else if(strcmp(cmpbuff,"sessions")==0){
return SESS_COUNT;
}else if(strcmp(cmpbuff,"threads")==0){
return THR_COUNT;
}
}
}
return UNDEFINED;
}
void print_help()
{
printf("\nFilter Test Harness\n\n"
"List of commands:\n %-32s%s\n %-32s%s\n %-32s%s\n %-32s%s\n %-32s%s\n "
"%-32s%s\n %-32s%s\n %-32s%s\n %-32s%s\n %-32s%s\n %-32s%s\n %-32s%s\n "
"%-32s%s\n %-32s%s\n"
,"help","Prints this help message."
,"run","Feeds the contents of the buffer to the filter chain."
,"add <filter name>","Loads a filter and appeds it to the end of the chain."
,"delete <filter name>","Deletes a filter."
,"status","Lists all loaded filters and queries"
,"clear","Clears the filter chain."
,"config <file name>","Loads filter configurations from a file."
,"in <file name>","Source file for the SQL statements."
,"out <file name>","Destination file for the SQL statements. Defaults to stdout if no parameters were passed."
,"threads <number>","Sets the amount of threads to use"
,"sessions <number>","How many sessions to create for each filter. This clears all loaded filters."
,"quiet","Print only error messages."
,"verbose","Print everything."
,"exit","Exit the program"
);
}
void manual_query()
{
char query[1024];
unsigned int qlen;
GWBUF** tmpbuf;
free_buffers();
printf("Enter query: ");
fgets(query,1024,stdin);
qlen = strnlen(query, 1024);
if((tmpbuf = malloc(sizeof(GWBUF*)))== NULL){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
return;
}
instance.buffer = tmpbuf;
instance.buffer_count = 1;
instance.buffer[0] = gwbuf_alloc(qlen + 5);
gwbuf_set_type(instance.buffer[0],GWBUF_TYPE_MYSQL);
memcpy(instance.buffer[0]->sbuf->data + 5,query,qlen);
instance.buffer[0]->sbuf->data[0] = (qlen);
instance.buffer[0]->sbuf->data[1] = (qlen << 8);
instance.buffer[0]->sbuf->data[2] = (qlen << 16);
instance.buffer[0]->sbuf->data[3] = 0x00;
instance.buffer[0]->sbuf->data[4] = 0x03;
}
void print_status()
{
if(instance.head->filter){
printf("Filters currently loaded:\n\n");
FILTERCHAIN* hd = instance.head;
int i = 1;
while(hd->filter){
printf("%d: %s\n", i++, hd->name);
hd = hd->next;
}
}else{
printf("No filters loaded.\n");
}
printf("\n");
if(instance.buffer_count > 0){
printf("%d queries loaded.\n",instance.buffer_count);
}else{
printf("No queries loaded.\n");
}
printf("Using %d threads and %d sessions.\n",instance.thrcount,instance.session_count);
if(instance.infile_name){
printf("Input is read from %s.\n",instance.infile_name);
}
if(instance.outfile_name){
printf("Output is written to %s.\n",instance.outfile_name);
}
}

14
server/modules/filter/test/harness_util.c Normal file
View File

@ -0,0 +1,14 @@
#include <harness.h>
int main(int argc,char** argv)
{
if(harness_init(argc,argv) || instance.error){
printf("Error: Initialization failed.\n");
skygw_log_write(LOGFILE_ERROR,"Error: Initialization failed.\n");
skygw_logmanager_done();
skygw_logmanager_exit();
return 1;
}
route_buffers();
return 0;
}

165
server/modules/filter/test/hint_testing.cnf Executable file
View File

@ -0,0 +1,165 @@
#
# Example MaxScale.cnf configuration file
#
#
#
# Number of server threads
# Valid options are:
# threads=<number of threads>
[maxscale]
threads=1
# Define a monitor that can be used to determine the state and role of
# the servers.
#
# Valid options are:
#
# module=<name of module to load>
# servers=<server name>,<server name>,...
# user =<user name - must have slave replication and
# slave client privileges>
# passwd=<password of the above user, plain text currently>
# monitor_interval=<sampling interval in milliseconds,
# default value is 10000>
[MySQL Monitor]
type=monitor
module=mysqlmon
servers=server1,server2,server3,server4
user=maxuser
passwd=maxpwd
# A series of service definition
#
# Valid options are:
#
# router=<name of router module>
# servers=<server name>,<server name>,...
# user=<User to fetch password inforamtion with>
# passwd=<Password of the user, plain text currently>
# enable_root_user=<0 or 1, default is 0>
# version_string=<specific string for server handshake,
# default is the MariaDB embedded library version>
#
# Valid router modules currently are:
# readwritesplit, readconnroute and debugcli
[RW Split Router]
type=service
router=readwritesplit
servers=server1,server2,server3,server4
max_slave_connections=90%
user=maxuser
passwd=maxpwd
#filters=MQ
[RW Split Hint Router]
type=service
router=readwritesplit
servers=server1,server2,server3,server4
max_slave_connections=90%
user=maxuser
passwd=maxpwd
filters=Hint
[Read Connection Router]
type=service
router=readconnroute
router_options=master
servers=server1
user=maxuser
passwd=maxpwd
[HTTPD Router]
type=service
router=testroute
servers=server1,server2,server3
[Debug Interface]
type=service
router=debugcli
[Hint]
type=filter
module=hintfilter
#[MQ]
#type=filter
#module=mqfilter
#exchange=x1
#key=k1
#queue=q1
#port=5673
# Listener definitions for the services
#
# Valid options are:
#
# service=<name of service defined elsewhere>
# protocol=<name of protocol module with which to listen>
# port=<Listening port>
# address=<Address to bind to>
# socket=<Listening socket>
[RW Split Listener]
type=listener
service=RW Split Router
protocol=MySQLClient
port=4006
[RW Split Hint Listener]
type=listener
service=RW Split Hint Router
protocol=MySQLClient
port=4009
[Read Connection Listener]
type=listener
service=Read Connection Router
protocol=MySQLClient
port=4008
#socket=/tmp/readconn.sock
[Debug Listener]
type=listener
service=Debug Interface
protocol=telnetd
port=4442
#address=127.0.0.1
[HTTPD Listener]
type=listener
service=HTTPD Router
protocol=HTTPD
port=6444
# Definition of the servers
[server1]
type=server
address=127.0.0.1
port=3000
protocol=MySQLBackend
[server2]
type=server
address=127.0.0.1
port=3001
protocol=MySQLBackend
[server3]
type=server
address=127.0.0.1
port=3002
protocol=MySQLBackend
[server4]
type=server
address=127.0.0.1
port=3003
protocol=MySQLBackend

48
server/modules/filter/test/hint_testing.expected Executable file
View File

@ -0,0 +1,48 @@
select @@server_id; -- maxscale begin route to master|HINT_ROUTE_TO_MASTER
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; -- maxscale route to server server3|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; -- maxscale end
select @@server_id; -- maxscale named1 prepare route to master
select @@server_id; -- maxscale named1 begin|HINT_ROUTE_TO_MASTER
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; -- maxscale route to server server3|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; -- maxscale end
select @@server_id; -- maxscale shorthand1 begin route to server server2|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id;|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id; -- maxscale route to server server3|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id; -- maxscale end
select @@server_id; # maxscale begin route to master|HINT_ROUTE_TO_MASTER
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; # maxscale route to server server3|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; # maxscale end
select @@server_id; # maxscale named2 prepare route to master
select @@server_id; # maxscale named2 begin|HINT_ROUTE_TO_MASTER
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; # maxscale route to server server3|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id; # maxscale end
select @@server_id; # maxscale shorthand2 begin route to server server2|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id;|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id; # maxscale route to server server3|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id; # maxscale end
select @@server_id/* maxscale begin route to master */;|HINT_ROUTE_TO_MASTER
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id/* maxscale route to server server3 */;|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id/* maxscale end */;
select @@server_id/* maxscale named3 prepare route to master */;
select @@server_id/* maxscale named3 begin */;|HINT_ROUTE_TO_MASTER
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id/* maxscale route to server server3 */;|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_MASTER
select @@server_id/* maxscale end */;
select @@server_id/* maxscale shorthand3 begin route to server server2 */; |HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id;|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id/* maxscale route to server server3 */;|HINT_ROUTE_TO_NAMED_SERVER|server3
select @@server_id;|HINT_ROUTE_TO_NAMED_SERVER|server2
select @@server_id/* maxscale end */;

48
server/modules/filter/test/hint_testing.input Normal file
View File

@ -0,0 +1,48 @@
select @@server_id; -- maxscale begin route to master
select @@server_id;
select @@server_id; -- maxscale route to server server3
select @@server_id;
select @@server_id; -- maxscale end
select @@server_id; -- maxscale named1 prepare route to master
select @@server_id; -- maxscale named1 begin
select @@server_id;
select @@server_id; -- maxscale route to server server3
select @@server_id;
select @@server_id; -- maxscale end
select @@server_id; -- maxscale shorthand1 begin route to server server2
select @@server_id;
select @@server_id; -- maxscale route to server server3
select @@server_id;
select @@server_id; -- maxscale end
select @@server_id; # maxscale begin route to master
select @@server_id;
select @@server_id; # maxscale route to server server3
select @@server_id;
select @@server_id; # maxscale end
select @@server_id; # maxscale named2 prepare route to master
select @@server_id; # maxscale named2 begin
select @@server_id;
select @@server_id; # maxscale route to server server3
select @@server_id;
select @@server_id; # maxscale end
select @@server_id; # maxscale shorthand2 begin route to server server2
select @@server_id;
select @@server_id; # maxscale route to server server3
select @@server_id;
select @@server_id; # maxscale end
select @@server_id/* maxscale begin route to master */;
select @@server_id;
select @@server_id/* maxscale route to server server3 */;
select @@server_id;
select @@server_id/* maxscale end */;
select @@server_id/* maxscale named3 prepare route to master */;
select @@server_id/* maxscale named3 begin */;
select @@server_id;
select @@server_id/* maxscale route to server server3 */;
select @@server_id;
select @@server_id/* maxscale end */;
select @@server_id/* maxscale shorthand3 begin route to server server2 */;
select @@server_id;
select @@server_id/* maxscale route to server server3 */;
select @@server_id;
select @@server_id/* maxscale end */;

9
server/modules/filter/test/hint_tests.sh Executable file
View File

@ -0,0 +1,9 @@
#! /bin/bash
./harness -q -i hint_testing.input -c hint_testing.cnf -o hint_testing.output -t 1 -s 1 -q &>/dev/null
diff hint_testing.expected hint_testing.output &>/dev/null
if [[ "$?" == "0" ]]
then
echo "PASSED"
else
echo "FAILED"
fi

1
server/modules/filter/test/querysmall Normal file
View File

@ -0,0 +1 @@
SELECT * FROM test_table;