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

Merge branch 'develop' into firewall

Browse Source 
Conflicts:
	server/modules/filter/test/CMakeLists.txt
	server/modules/filter/test/harness_common.c
This commit is contained in:
Markus Makela
2014-11-18 05:02:12 +02:00
parent 2e7419e942 7ef54219a6
commit 80b83f0ac0
17 changed files with 180 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
log_manager/log_manager.cc
View File

@ -57,6 +57,10 @@ static simple_mutex_t msg_mutex;
* actual library calls such as skygw_log_write.
*/
int lm_enabled_logfiles_bitmask = 0;
/**
* Thread-specific variable for session id which is used in trace logging.
*/
__thread size_t tls_sesid = 0;
/**
* BUFSIZ comes from the system. It equals with block size or
@ -720,17 +724,29 @@ static int logmanager_write_log(
{
/** Length of string that will be written, limited by bufsize */
int safe_str_len;
/** Length of session id */
int sesid_str_len;
/** 2 braces and 2 spaces */
if (id == LOGFILE_TRACE && tls_sesid > 0)
{
sesid_str_len = 2+2+get_decimal_len(tls_sesid);
}
else
{
sesid_str_len = 0;
}
timestamp_len = get_timestamp_len();
/** Findout how much can be safely written with current block size */
if (timestamp_len-1+str_len > lf->lf_buf_size)
if (timestamp_len-1+sesid_str_len-1+str_len > lf->lf_buf_size)
{
safe_str_len = lf->lf_buf_size;
}
else
{
safe_str_len = timestamp_len-1+str_len;
safe_str_len = timestamp_len-1+sesid_str_len-1+str_len;
}
/**
* Seek write position and register to block buffer.
@ -761,7 +777,7 @@ static int logmanager_write_log(
simple_mutex_unlock(&msg_mutex);
}
#endif
/** Book space for log string from buffer */
wp = blockbuf_get_writepos(&bb,
id,
safe_str_len,
@ -783,16 +799,28 @@ static int logmanager_write_log(
*/
timestamp_len = snprint_timestamp(wp, timestamp_len);
if (id == LOGFILE_TRACE)
{
/**
* Write session id
*/
snprintf(wp+timestamp_len, sesid_str_len, "[%lu] ", tls_sesid);
sesid_str_len -= 1; /*< remove terminating char */
}
/**
* Write next string to overwrite terminating null character
* of the timestamp string.
*/
if (use_valist) {
vsnprintf(wp+timestamp_len, safe_str_len-timestamp_len, str, valist);
vsnprintf(wp+timestamp_len+sesid_str_len,
safe_str_len-timestamp_len-sesid_str_len,
str,
valist);
} else {
snprintf(wp+timestamp_len, safe_str_len-timestamp_len, "%s", str);
snprintf(wp+timestamp_len+sesid_str_len,
safe_str_len-timestamp_len-sesid_str_len,
"%s",
str);
}
/** write to syslog */

6
log_manager/test/testorder.c
View File

@ -33,7 +33,8 @@ int main(int argc, char** argv)
char *message;
char** optstr;
long msg_index = 1;
struct timespec ts1;
ts1.tv_sec = 0;
memset(cwd,0,1024);
if( argc <4){
@ -95,7 +96,8 @@ int main(int argc, char** argv)
fprintf(stderr,"Error: log_manager returned %d",err);
break;
}
usleep(100);
ts1.tv_nsec = 100*1000000;
nanosleep(&ts1, NULL);
}
skygw_log_flush(LOGFILE_ERROR);

15
server/core/dcb.c
View File

@ -90,6 +90,21 @@ static int dcb_null_close(DCB *dcb);
static int dcb_null_auth(DCB *dcb, SERVER *server, SESSION *session, GWBUF *buf);
static int dcb_isvalid_nolock(DCB *dcb);
size_t dcb_get_session_id(
DCB* dcb)
{
size_t rval;
if (dcb != NULL && dcb->session != NULL)
{
rval = dcb->session->ses_id;
}
else
{
rval = 0;
}
return rval;
}
/**
* Return the pointer to the lsit of zombie DCB's
*

9
server/core/gateway.c
View File

@ -382,7 +382,12 @@ static bool file_write_header(
const char* header_buf3;
time_t* t = NULL;
struct tm* tm = NULL;
#if defined(LAPTOP_TEST)
struct timespec ts1;
ts1.tv_sec = 0;
ts1.tv_nsec = DISKWRITE_LATENCY*1000000;
#endif
if ((t = (time_t *)malloc(sizeof(time_t))) == NULL) {
goto return_succp;
}
@ -406,7 +411,7 @@ static bool file_write_header(
len2 = strlen(header_buf2);
len3 = strlen(header_buf3);
#if defined(LAPTOP_TEST)
usleep(DISKWRITE_LATENCY);
nanosleep(&ts1, NULL);
#else
wbytes1=fwrite((void*)header_buf1, len1, 1, outfile);
wbytes2=fwrite((void*)header_buf2, len2, 1, outfile);

9
server/core/poll.c
View File

@ -41,6 +41,8 @@
extern int lm_enabled_logfiles_bitmask;
extern __thread size_t tls_sesid;
/**
* @file poll.c - Abstraction of the epoll functionality
*
@ -677,6 +679,7 @@ uint32_t ev;
#else
atomic_add(&pollStats.n_write,
1);
LOGIF(LT, (tls_sesid = dcb_get_session_id(dcb)));
dcb->func.write_ready(dcb);
#endif
} else {
@ -712,6 +715,7 @@ uint32_t ev;
dcb->fd)));
atomic_add(
&pollStats.n_accept, 1);
LOGIF(LT, (tls_sesid = dcb_get_session_id(dcb)));
dcb->func.accept(dcb);
}
else
@ -724,6 +728,7 @@ uint32_t ev;
dcb,
dcb->fd)));
atomic_add(&pollStats.n_read, 1);
LOGIF(LT, (tls_sesid = dcb_get_session_id(dcb)));
dcb->func.read(dcb);
}
#if MUTEX_BLOCK
@ -759,6 +764,7 @@ uint32_t ev;
strerror(eno))));
}
atomic_add(&pollStats.n_error, 1);
LOGIF(LT, (tls_sesid = dcb_get_session_id(dcb)));
dcb->func.error(dcb);
}
@ -783,6 +789,7 @@ uint32_t ev;
{
dcb->flags |= DCBF_HUNG;
spinlock_release(&dcb->dcb_initlock);
LOGIF(LT, (tls_sesid = dcb_get_session_id(dcb)));
dcb->func.hangup(dcb);
}
else
@ -811,12 +818,14 @@ uint32_t ev;
{
dcb->flags |= DCBF_HUNG;
spinlock_release(&dcb->dcb_initlock);
LOGIF(LT, (tls_sesid = dcb_get_session_id(dcb)));
dcb->func.hangup(dcb);
}
else
spinlock_release(&dcb->dcb_initlock);
}
#endif
LOGIF(LT, tls_sesid = 0);
spinlock_acquire(&pollqlock);
if (dcb->evq.pending_events == 0)

29
server/core/session.c
View File

@ -43,7 +43,10 @@
#include <skygw_utils.h>
#include <log_manager.h>
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
/** Global session id; updated safely by holding session_spin */
static size_t session_id;
static SPINLOCK session_spin = SPINLOCK_INIT;
static SESSION *allSessions = NULL;
@ -216,10 +219,29 @@ session_alloc(SERVICE *service, DCB *client_dcb)
session->state = SESSION_STATE_ROUTER_READY;
spinlock_release(&session->ses_lock);
spinlock_acquire(&session_spin);
session->ses_id = ++session_id; /*< assign an id and increase */
session->next = allSessions;
allSessions = session;
spinlock_release(&session_spin);
if (session->client->user == NULL)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Started session [%lu] for %s service ",
session->ses_id,
service->name)));
}
else
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Started %s client session [%lu] for '%s' from %s",
service->name,
session->ses_id,
session->client->user,
session->client->remote)));
}
atomic_add(&service->stats.n_sessions, 1);
atomic_add(&service->stats.n_current, 1);
CHK_SESSION(session);
@ -352,6 +374,13 @@ bool session_free(
}
free(session->filters);
}
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Stopped %s client session [%lu]",
session->service->name,
session->ses_id)));
free(session);
succp = true;

11
server/include/dcb.h
View File

@ -307,16 +307,11 @@ int dcb_remove_callback(DCB *, DCB_REASON, int (*)(struct dcb *, DCB_REASON, vo
void *);
int dcb_isvalid(DCB *); /* Check the DCB is in the linked list */
bool dcb_set_state(
DCB* dcb,
dcb_state_t new_state,
dcb_state_t* old_state);
void dcb_call_foreach (DCB_REASON reason);
bool dcb_set_state(DCB* dcb, dcb_state_t new_state, dcb_state_t* old_state);
void dcb_call_foreach (DCB_REASON reason);
size_t dcb_get_session_id(DCB* dcb);;
void dcb_call_foreach (
DCB_REASON reason);
/**
* DCB flags values
*/

1
server/include/session.h
View File

@ -110,6 +110,7 @@ typedef struct session {
#endif
SPINLOCK ses_lock;
session_state_t state; /**< Current descriptor state */
size_t ses_id; /**< unique session identifier */
struct dcb *client; /**< The client connection */
void *data; /**< The session data */
void *router_session;/**< The router instance data */

10
server/modules/filter/test/CMakeLists.txt
View File

@ -8,9 +8,9 @@ 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 ${EMBEDDED_LIB})
target_link_libraries(harness fullcore ${EMBEDDED_LIB})
file(COPY ${ERRMSG} DESTINATION ${CMAKE_CURRENT_BINARY_DIR})
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/fwtest.cnf.in ${CMAKE_CURRENT_BINARY_DIR}/fwtest.cnf @ONLY)
target_link_libraries(harness_ui fullcore log_manager utils)
target_link_libraries(harness fullcore)
execute_process(COMMAND ${CMAKE_COMMAND} -E copy ${ERRMSG} ${CMAKE_CURRENT_BINARY_DIR})
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 -e ${CMAKE_CURRENT_SOURCE_DIR}/hint_testing.expected")
add_test(TestFwfilter /bin/sh -c "MAXSCALE_HOME=\"${CMAKE_BINARY_DIR}\" ${CMAKE_CURRENT_BINARY_DIR}/harness -i ${CMAKE_CURRENT_SOURCE_DIR}/fwtest.input -o ${CMAKE_CURRENT_BINARY_DIR}/fwtest.output -c ${CMAKE_CURRENT_SOURCE_DIR}/fwtest.cnf -t 1 -s 1 -e ${CMAKE_CURRENT_SOURCE_DIR}/fwtest.expected")
add_test(TestRegexfilter /bin/sh -c "MAXSCALE_HOME=\"${CMAKE_BINARY_DIR}\" ${CMAKE_CURRENT_BINARY_DIR}/harness -i ${CMAKE_CURRENT_SOURCE_DIR}/regextest.input -o ${CMAKE_CURRENT_BINARY_DIR}/regextest.output -c ${CMAKE_CURRENT_SOURCE_DIR}/regextest.cnf -t 1 -s 1 -e ${CMAKE_CURRENT_SOURCE_DIR}/regextest.expected")

33
server/modules/filter/test/harness_common.c
View File

@ -11,6 +11,12 @@ int harness_init(int argc, char** argv, HARNESS_INSTANCE** inst){
int i = 0;
MYSQL_session* mysqlsess;
DCB* dcb;
char cwd[1024];
char tmp[2048];
char** optstr;
if(!(argc == 2 && strcmp(argv[1],"-h") == 0)){
skygw_logmanager_init(0,NULL);
}
@ -32,8 +38,8 @@ int harness_init(int argc, char** argv, HARNESS_INSTANCE** inst){
instance.last_ind = -1;
instance.sess_ind = -1;
instance.session = calloc(1,sizeof(SESSION));
MYSQL_session* mysqlsess = calloc(1,sizeof(MYSQL_session));
DCB* dcb = calloc(1,sizeof(DCB));
dcb = calloc(1,sizeof(DCB));
mysqlsess = calloc(1,sizeof(MYSQL_session));
sprintf(mysqlsess->user,"dummyuser");
sprintf(mysqlsess->db,"dummydb");
@ -43,6 +49,17 @@ int harness_init(int argc, char** argv, HARNESS_INSTANCE** inst){
instance.session->client = (void*)dcb;
instance.session->data = (void*)mysqlsess;
getcwd(cwd,sizeof(cwd));
sprintf(tmp,"%s",cwd);
optstr = (char**)malloc(sizeof(char*)*4);
optstr[0] = strdup("log_manager");
optstr[1] = strdup("-j");
optstr[2] = strdup(tmp);
optstr[3] = NULL;
skygw_logmanager_init( 3, optstr);
free(optstr);
process_opts(argc,argv);
if(!(instance.thrpool = malloc(instance.thrcount * sizeof(pthread_t)))){
@ -849,6 +866,8 @@ void route_buffers()
while(instance.buff_ind < instance.buffer_count){
pthread_mutex_unlock(&instance.work_mtx);
while(instance.last_ind < instance.session_count){
struct timespec ts1;
ts1.tv_sec = 0;
tprg = ((bprg + (float)instance.last_ind)/fin);
if(!instance.verbose){
@ -857,7 +876,8 @@ void route_buffers()
trig += step;
}
}
usleep(100);
ts1.tv_nsec = 100*1000000;
nanosleep(&ts1, NULL);
}
pthread_mutex_lock(&instance.work_mtx);
instance.buff_ind++;
@ -892,7 +912,11 @@ void work_buffer(void* thr_num)
index < instance.session_count &&
instance.buff_ind < instance.buffer_count)
{
struct timespec ts1;
ts1.tv_sec = 0;
if(instance.head->instance->routeQuery(instance.head->filter,
instance.head->session[index],
instance.buffer[instance.buff_ind]) == 0){
if(instance.outfile > 0){
@ -906,7 +930,8 @@ void work_buffer(void* thr_num)
fake_ok);
}
atomic_add(&instance.last_ind,1);
usleep(1000*instance.rt_delay);
ts1.tv_nsec = 1000*instance.rt_delay*1000000;
nanosleep(&ts1, NULL);
}
}

162
server/modules/filter/test/hint_testing.cnf
View File

@ -1,165 +1,3 @@
#
# 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

5
server/modules/filter/test/regextest.cnf Normal file
View File

@ -0,0 +1,5 @@
[Regex]
type=filter
module=regexfilter
match=wrong
replace=right

3
server/modules/filter/test/regextest.expected Executable file
View File

@ -0,0 +1,3 @@
select * from mysql.right;
select right from test.table;
select * from test.table where name='right';

3
server/modules/filter/test/regextest.input Normal file
View File

@ -0,0 +1,3 @@
select * from mysql.wrong;
select wrong from test.table;
select * from test.table where name='wrong';

9
server/modules/protocol/mysql_client.c
View File

@ -1111,7 +1111,7 @@ int gw_MySQLAccept(DCB *listener)
int sendbuf = GW_BACKEND_SO_SNDBUF;
socklen_t optlen = sizeof(sendbuf);
int eno = 0;
int syseno = 0;
int syseno = 0;
int i = 0;
CHK_DCB(listener);
@ -1151,6 +1151,8 @@ int gw_MySQLAccept(DCB *listener)
}
else if (eno == ENFILE || eno == EMFILE)
{
struct timespec ts1;
ts1.tv_sec = 0;
/**
* Exceeded system's (ENFILE) or processes
* (EMFILE) max. number of files limit.
@ -1173,8 +1175,9 @@ int gw_MySQLAccept(DCB *listener)
strerror(eno))));
}
i++;
usleep(100*i*i);
ts1.tv_nsec = 100*i*i*1000000;
nanosleep(&ts1, NULL);
if (i<10) {
goto retry_accept;
}

2
server/modules/protocol/mysql_common.c
View File

@ -1847,7 +1847,7 @@ void protocol_add_srv_command(
MySQLProtocol* p,
mysql_server_cmd_t cmd)
{
#if defined(SS_DEBUG)
#if defined(EXTRA_SS_DEBUG)
server_command_t* c;
#endif
spinlock_acquire(&p->protocol_lock);

32
utils/skygw_utils.cc
View File

@ -1224,12 +1224,16 @@ void acquire_lock(
int* l)
{
register int misscount = 0;
struct timespec ts1;
ts1.tv_sec = 0;
while (atomic_add(l, 1) != 0) {
atomic_add(l, -1);
misscount += 1;
if (misscount > 10) {
usleep(rand()%misscount);
if (misscount > 10)
{
ts1.tv_nsec = (rand()%misscount)*1000000;
nanosleep(&ts1, NULL);
}
}
}
@ -1636,7 +1640,12 @@ static bool file_write_header(
const char* header_buf4;
time_t* t;
struct tm* tm;
#if defined(LAPTOP_TEST)
struct timespec ts1;
ts1.tv_sec = 0;
ts1.tv_nsec = DISKWRITE_LATENCY*1000000;
#endif
t = (time_t *)malloc(sizeof(time_t));
tm = (struct tm *)malloc(sizeof(struct tm));
*t = time(NULL);
@ -1662,7 +1671,7 @@ static bool file_write_header(
len3 = strlen(header_buf3);
len4 = strlen(header_buf4);
#if defined(LAPTOP_TEST)
usleep(DISKWRITE_LATENCY);
nanosleep(&ts1, NULL);
#else
wbytes1=fwrite((void*)header_buf1, len1, 1, file->sf_file);
wbytes2=fwrite((void*)header_buf2, len2, 1, file->sf_file);
@ -1709,7 +1718,12 @@ static bool file_write_footer(
const char* header_buf1;
char* header_buf3 = NULL;
const char* header_buf4;
#if defined(LAPTOP_TEST)
struct timespec ts1;
ts1.tv_sec = 0;
ts1.tv_nsec = DISKWRITE_LATENCY*1000000;
#endif
CHK_FILE(file);
if (shutdown)
@ -1734,7 +1748,7 @@ static bool file_write_footer(
len1 = strlen(header_buf1);
len4 = strlen(header_buf4);
#if defined(LAPTOP_TEST)
usleep(DISKWRITE_LATENCY);
nanosleep(&ts1, NULL);
#else
wbytes3=fwrite((void*)header_buf3, tslen, 1, file->sf_file);
wbytes1=fwrite((void*)header_buf1, len1, 1, file->sf_file);
@ -1784,11 +1798,15 @@ int skygw_file_write(
size_t nwritten;
int fd;
static int writecount;
#else
struct timespec ts1;
ts1.tv_sec = 0;
ts1.tv_nsec = DISKWRITE_LATENCY*1000000;
#endif
CHK_FILE(file);
#if defined(LAPTOP_TEST)
usleep(DISKWRITE_LATENCY);
nanosleep(&ts1, NULL);
#else
nwritten = fwrite(data, nbytes, 1, file->sf_file);