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

Merge branch 'develop' into tee_fixes

Browse Source
This commit is contained in:
Markus Makela
2015-03-06 18:27:41 +02:00
parent 3b76ed43c4 8c3409cf32
commit 2aaa367c37
49 changed files with 4515 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/core/CMakeLists.txt
View File

@ -7,7 +7,7 @@ add_executable(maxscale atomic.c buffer.c spinlock.c gateway.c
gw_utils.c utils.c dcb.c load_utils.c session.c service.c server.c
poll.c config.c users.c hashtable.c dbusers.c thread.c gwbitmask.c
monitor.c adminusers.c secrets.c filter.c modutil.c hint.c
housekeeper.c memlog.c)
housekeeper.c memlog.c resultset.c)
target_link_libraries(maxscale ${EMBEDDED_LIB} log_manager utils ssl aio pthread crypt dl crypto inih z rt m stdc++)
install(TARGETS maxscale DESTINATION bin)

58
server/core/dcb.c
View File

@ -803,7 +803,8 @@ int dcb_read(
if (r <= 0 &&
l_errno != EAGAIN &&
l_errno != EWOULDBLOCK)
l_errno != EWOULDBLOCK &&
l_errno != 0)
{
n = -1;
goto return_n;
@ -1586,8 +1587,10 @@ va_list args;
int
dcb_isclient(DCB *dcb)
{
if(dcb->session) {
if (dcb->session->client) {
if (dcb->state != DCB_STATE_LISTENING && dcb->session)
{
if (dcb->session->client)
{
return (dcb->session && dcb == dcb->session->client);
}
}
@ -2172,3 +2175,52 @@ dcb_null_auth(DCB *dcb, SERVER *server, SESSION *session, GWBUF *buf)
{
return 0;
}
/**
* Return DCB counts optionally filtered by usage
*
* @param usage The usage of the DCB
* @return A count of DCBs in the desired state
*/
int
dcb_count_by_usage(DCB_USAGE usage)
{
int rval = 0;
DCB *ptr;
spinlock_acquire(&dcbspin);
ptr = allDCBs;
while (ptr)
{
switch (usage)
{
case DCB_USAGE_CLIENT:
if (dcb_isclient(ptr))
rval++;
break;
case DCB_USAGE_LISTENER:
if (ptr->state == DCB_STATE_LISTENING)
rval++;
break;
case DCB_USAGE_BACKEND:
if (dcb_isclient(ptr) == 0
&& ptr->dcb_role == DCB_ROLE_REQUEST_HANDLER)
rval++;
break;
case DCB_USAGE_INTERNAL:
if (ptr->dcb_role == DCB_ROLE_REQUEST_HANDLER)
rval++;
break;
case DCB_USAGE_ZOMBIE:
if (DCB_ISZOMBIE(ptr))
rval++;
break;
case DCB_USAGE_ALL:
rval++;
break;
}
ptr = ptr->next;
}
spinlock_release(&dcbspin);
return rval;
}

30
server/core/gateway.c
View File

@ -79,6 +79,8 @@
# define _GNU_SOURCE
#endif
time_t MaxScaleStarted;
extern char *program_invocation_name;
extern char *program_invocation_short_name;
@ -148,6 +150,7 @@ static struct option long_options[] = {
{"config", required_argument, 0, 'f'},
{"nodaemon", no_argument, 0, 'd'},
{"log", required_argument, 0, 'l'},
{"syslog", required_argument, 0, 's'},
{"version", no_argument, 0, 'v'},
{"help", no_argument, 0, '?'},
{0, 0, 0, 0}
@ -993,6 +996,8 @@ static void usage(void)
" (default: $MAXSCALE_HOME/etc/MaxScale.cnf)\n"
" -l|--log=... log to file or shared memory\n"
" -lfile or -lshm - defaults to shared memory\n"
" -s|--syslog= log messages to syslog"
" true or false - defaults to true"
" -v|--version print version info and exit\n"
" -?|--help show this help\n"
, progname);
@ -1055,6 +1060,7 @@ int main(int argc, char **argv)
void* log_flush_thr = NULL;
int option_index;
int logtofile = 0; /* Use shared memory or file */
int syslog_enabled = 1; /** Log to syslog */
ssize_t log_flush_timeout_ms = 0;
sigset_t sigset;
sigset_t sigpipe_mask;
@ -1094,7 +1100,7 @@ int main(int argc, char **argv)
goto return_main;
}
}
while ((opt = getopt_long(argc, argv, "dc:f:l:v?",
while ((opt = getopt_long(argc, argv, "dc:f:l:vs:?",
long_options, &option_index)) != -1)
{
bool succp = true;
@ -1199,7 +1205,17 @@ int main(int argc, char **argv)
succp = false;
}
break;
case 's':
if(strstr(optarg,"="))
{
strtok(optarg,"= ");
syslog_enabled = config_truth_value(strtok(NULL,"= "));
}
else
{
syslog_enabled = config_truth_value(optarg);
}
break;
case '?':
usage();
rc = EXIT_SUCCESS;
@ -1561,6 +1577,8 @@ int main(int argc, char **argv)
argv[0] = "MaxScale";
argv[1] = "-j";
argv[2] = buf;
logmanager_enable_syslog(syslog_enabled);
if (logtofile)
{
@ -1797,6 +1815,8 @@ int main(int argc, char **argv)
LOGIF(LM, (skygw_log_write(LOGFILE_MESSAGE,
"MaxScale started with %d server threads.",
config_threadcount())));
MaxScaleStarted = time(0);
/*<
* Serve clients.
*/
@ -1951,3 +1971,9 @@ static int write_pid_file(char *home_dir) {
/* success */
return 0;
}
int
MaxScaleUptime()
{
return time(0) - MaxScaleStarted;
}

11
server/core/housekeeper.c
View File

@ -116,9 +116,20 @@ HKTASK *task, *ptr;
ptr = ptr->next;
}
if (ptr)
{
if (strcmp(ptr->name, name) == 0)
{
spinlock_release(&tasklock);
free(task->name);
free(task);
return 0;
}
ptr->next = task;
}
else
{
tasks = task;
}
spinlock_release(&tasklock);
return task->nextdue;

77
server/core/load_utils.c
View File

@ -408,3 +408,80 @@ MODULES *ptr = registered;
}
dcb_printf(dcb, "----------------+-------------+---------+-------+-------------------------\n\n");
}
/**
* Provide a row to the result set that defines the set of modules
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
moduleRowCallback(RESULTSET *set, void *data)
{
int *rowno = (int *)data;
int i = 0;;
char *stat, buf[20];
RESULT_ROW *row;
MODULES *ptr;
ptr = registered;
while (i < *rowno && ptr)
{
i++;
ptr = ptr->next;
}
if (ptr == NULL)
{
free(data);
return NULL;
}
(*rowno)++;
row = resultset_make_row(set);
resultset_row_set(row, 0, ptr->module);
resultset_row_set(row, 1, ptr->type);
resultset_row_set(row, 2, ptr->version);
sprintf(buf, "%d.%d.%d", ptr->info->api_version.major,
ptr->info->api_version.minor,
ptr->info->api_version.patch);
resultset_row_set(row, 3, buf);
resultset_row_set(row, 4, ptr->info->status == MODULE_IN_DEVELOPMENT
? "In Development"
: (ptr->info->status == MODULE_ALPHA_RELEASE
? "Alpha"
: (ptr->info->status == MODULE_BETA_RELEASE
? "Beta"
: (ptr->info->status == MODULE_GA
? "GA"
: (ptr->info->status == MODULE_EXPERIMENTAL
? "Experimental" : "Unknown")))));
return row;
}
/**
* Return a resultset that has the current set of modules in it
*
* @return A Result set
*/
RESULTSET *
moduleGetList()
{
RESULTSET *set;
int *data;
if ((data = (int *)malloc(sizeof(int))) == NULL)
return NULL;
*data = 0;
if ((set = resultset_create(moduleRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Module Name", 18, COL_TYPE_VARCHAR);
resultset_add_column(set, "Module Type", 12, COL_TYPE_VARCHAR);
resultset_add_column(set, "Version", 10, COL_TYPE_VARCHAR);
resultset_add_column(set, "API Version", 8, COL_TYPE_VARCHAR);
resultset_add_column(set, "Status", 15, COL_TYPE_VARCHAR);
return set;
}

19
server/core/modutil.c
View File

@ -63,6 +63,23 @@ unsigned char *ptr;
return ptr[4] == 0x03; // COM_QUERY
}
/**
* Check if a GWBUF structure is a MySQL COM_STMT_PREPARE packet
*
* @param buf Buffer to check
* @return True if GWBUF is a COM_STMT_PREPARE packet
*/
int
modutil_is_SQL_prepare(GWBUF *buf)
{
unsigned char *ptr;
if (GWBUF_LENGTH(buf) < 5)
return 0;
ptr = GWBUF_DATA(buf);
return ptr[4] == 0x16 ; // COM_STMT_PREPARE
}
/**
* Extract the SQL portion of a COM_QUERY packet
*
@ -243,7 +260,7 @@ modutil_get_SQL(GWBUF *buf)
unsigned int len, length;
char *ptr, *dptr, *rval = NULL;
if (!modutil_is_SQL(buf))
if (!modutil_is_SQL(buf) && !modutil_is_SQL_prepare(buf))
return rval;
ptr = GWBUF_DATA(buf);
length = *ptr++;

63
server/core/monitor.c
View File

@ -365,3 +365,66 @@ monitorSetNetworkTimeout(MONITOR *mon, int type, int value) {
mon->module->setNetworkTimeout(mon->handle, type, value);
}
}
/**
* Provide a row to the result set that defines the set of monitors
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
monitorRowCallback(RESULTSET *set, void *data)
{
int *rowno = (int *)data;
int i = 0;;
char buf[20];
RESULT_ROW *row;
MONITOR *ptr;
spinlock_acquire(&monLock);
ptr = allMonitors;
while (i < *rowno && ptr)
{
i++;
ptr = ptr->next;
}
if (ptr == NULL)
{
spinlock_release(&monLock);
free(data);
return NULL;
}
(*rowno)++;
row = resultset_make_row(set);
resultset_row_set(row, 0, ptr->name);
resultset_row_set(row, 1, ptr->state & MONITOR_STATE_RUNNING
? "Running" : "Stopped");
spinlock_release(&monLock);
return row;
}
/**
* Return a resultset that has the current set of monitors in it
*
* @return A Result set
*/
RESULTSET *
monitorGetList()
{
RESULTSET *set;
int *data;
if ((data = (int *)malloc(sizeof(int))) == NULL)
return NULL;
*data = 0;
if ((set = resultset_create(monitorRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Monitor", 20, COL_TYPE_VARCHAR);
resultset_add_column(set, "Status", 10, COL_TYPE_VARCHAR);
return set;
}

107
server/core/poll.c
View File

@ -30,7 +30,9 @@
#include <gw.h>
#include <config.h>
#include <housekeeper.h>
#include <config.h>
#include <mysql.h>
#include <resultset.h>
#define PROFILE_POLL 0
@ -151,8 +153,8 @@ static struct {
int n_hup; /*< Number of hangup events */
int n_accept; /*< Number of accept events */
int n_polls; /*< Number of poll cycles */
int n_pollev; /*< Number of polls returnign events */
int n_nbpollev; /*< Number of polls returnign events */
int n_pollev; /*< Number of polls returning events */
int n_nbpollev; /*< Number of polls returning events */
int n_nothreads; /*< Number of times no threads are polling */
int n_fds[MAXNFDS]; /*< Number of wakeups with particular
n_fds value */
@ -1524,3 +1526,104 @@ int i;
dcb_printf(pdcb, " > %2d00ms | %-10d | %-10d\n", N_QUEUE_TIMES,
queueStats.qtimes[N_QUEUE_TIMES], queueStats.exectimes[N_QUEUE_TIMES]);
}
/**
* Return a poll statistic from the polling subsystem
*
* @param stat The required statistic
* @return The value of that statistic
*/
int
poll_get_stat(POLL_STAT stat)
{
switch (stat)
{
case POLL_STAT_READ:
return pollStats.n_read;
case POLL_STAT_WRITE:
return pollStats.n_write;
case POLL_STAT_ERROR:
return pollStats.n_error;
case POLL_STAT_HANGUP:
return pollStats.n_hup;
case POLL_STAT_ACCEPT:
return pollStats.n_accept;
case POLL_STAT_EVQ_LEN:
return pollStats.evq_length;
case POLL_STAT_EVQ_PENDING:
return pollStats.evq_pending;
case POLL_STAT_EVQ_MAX:
return pollStats.evq_max;
case POLL_STAT_MAX_QTIME:
return (int)queueStats.maxqtime;
case POLL_STAT_MAX_EXECTIME:
return (int)queueStats.maxexectime;
}
return 0;
}
/**
* Provide a row to the result set that defines the event queue statistics
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
eventTimesRowCallback(RESULTSET *set, void *data)
{
int *rowno = (int *)data;
char buf[40];
RESULT_ROW *row;
if (*rowno >= N_QUEUE_TIMES)
{
free(data);
return NULL;
}
row = resultset_make_row(set);
if (*rowno == 0)
resultset_row_set(row, 0, "< 100ms");
else if (*rowno == N_QUEUE_TIMES - 1)
{
sprintf(buf, "> %2d00ms", N_QUEUE_TIMES);
resultset_row_set(row, 0, buf);
}
else
{
sprintf(buf, "%2d00 - %2d00ms", *rowno, (*rowno) + 1);
resultset_row_set(row, 0, buf);
}
sprintf(buf, "%d", queueStats.qtimes[*rowno]);
resultset_row_set(row, 1, buf);
sprintf(buf, "%d", queueStats.exectimes[*rowno]);
resultset_row_set(row, 2, buf);
(*rowno)++;
return row;
}
/**
* Return a resultset that has the current set of services in it
*
* @return A Result set
*/
RESULTSET *
eventTimesGetList()
{
RESULTSET *set;
int *data;
if ((data = (int *)malloc(sizeof(int))) == NULL)
return NULL;
*data = 0;
if ((set = resultset_create(eventTimesRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Duration", 20, COL_TYPE_VARCHAR);
resultset_add_column(set, "No. Events Queued", 12, COL_TYPE_VARCHAR);
resultset_add_column(set, "No. Events Executed", 12, COL_TYPE_VARCHAR);
return set;
}

467
server/core/resultset.c Normal file
View File

@ -0,0 +1,467 @@
/*
* This file is distributed as part of the MariaDB Corporation MaxScale. 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 MariaDB Corporation Ab 2013-2014
*/
/**
* @file resultset.c - Implementation of a generic result set mechanism
*
* @verbatim
* Revision History
*
* Date Who Description
* 17/02/15 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <string.h>
#include <ctype.h>
#include <resultset.h>
#include <buffer.h>
#include <dcb.h>
static int mysql_send_fieldcount(DCB *, int);
static int mysql_send_columndef(DCB *, char *, int, int, uint8_t);
static int mysql_send_eof(DCB *, int);
static int mysql_send_row(DCB *, RESULT_ROW *, int);
/**
* Create a generic result set
*
* @param func Function to call for each row
* @param data Data to pass to the row retrieval function
* @return An empty resultset or NULL on error
*/
RESULTSET *
resultset_create(RESULT_ROW_CB func, void *data)
{
RESULTSET *rval;
if ((rval = (RESULTSET *)malloc(sizeof(RESULTSET))) != NULL)
{
rval->n_cols = 0;
rval->column = NULL;
rval->userdata = data;
rval->fetchrow = func;
}
return rval;
}
/**
* Free a previously allocated resultset
*
* @param resultset The result set to free
*/
void
resultset_free(RESULTSET *resultset)
{
RESULT_COLUMN *col;
if (resultset)
return;
col = resultset->column;
while (col)
{
RESULT_COLUMN *next;
next = col->next;
resultset_column_free(col);
col = next;
}
free(resultset);
}
/**
* Add a new column to a result set. Columns are added to the right
* of the result set, i.e. the existing order is maintained.
*
* @param set The result set
* @param name The column name
* @param len The column length
* @param type The column type
* @return The numebr of columns added to the result set
*/
int
resultset_add_column(RESULTSET *set, char *name, int len, RESULT_COL_TYPE type)
{
RESULT_COLUMN *newcol, *ptr;
if ((newcol = (RESULT_COLUMN *)malloc(sizeof(RESULT_COLUMN))) == NULL)
return 0;
if ((newcol->name = strdup(name)) == NULL)
{
free(newcol);
return 0;
}
newcol->type = type;
newcol->len = len;
newcol->next = NULL;
if (set->column == NULL)
set->column = newcol;
else
{
ptr = set->column;
while (ptr->next)
ptr = ptr->next;
ptr->next = newcol;
}
set->n_cols++;
return 1;
}
/**
* Free a result set column
*
* @param col Column to free
*/
void
resultset_column_free(RESULT_COLUMN *col)
{
free(col->name);
free(col);
}
/**
* Create a blank row, a row with all values NULL, for a result
* set.
*
* @param set The result set the row will be part of
* @return The NULL result set row
*/
RESULT_ROW *
resultset_make_row(RESULTSET *set)
{
RESULT_ROW *row;
int i;
if ((row = (RESULT_ROW *)malloc(sizeof(RESULT_ROW))) == NULL)
return NULL;
row->n_cols = set->n_cols;
if ((row->cols = (char **)malloc(row->n_cols * sizeof(char *))) == NULL)
{
free(row);
return NULL;
}
for (i = 0; i < set->n_cols; i++)
row->cols[i] = NULL;
return row;
}
/**
* Free a result set row. If a column in the row has a non-null values
* then the data is assumed to be a malloc'd pointer and will be free'd.
* If any value is not a malloc'd pointer it should be removed before
* making this call.
*
* @param row The row to free
*/
void
resultset_free_row(RESULT_ROW *row)
{
int i;
for (i = 0; i < row->n_cols; i++)
if (row->cols[i])
free(row->cols[i]);
free(row->cols);
free(row);
}
/**
* Add a value in a particular column of the row . The value is
* a NULL terminated string and will be copied into malloc'd
* storage by this routine.
*
* @param row The row ro add the column into
* @param col The column number (0 to n_cols - 1)
* @param value The column value, may be NULL
* @return The number of columns inserted
*/
int
resultset_row_set(RESULT_ROW *row, int col, char *value)
{
if (col < 0 || col >= row->n_cols)
return 0;
if (value)
{
if ((row->cols[col] = strdup(value)) == NULL)
return 0;
return 1;
}
else if (row->cols[col])
free(row->cols[col]);
row->cols[col] = NULL;
return 1;
}
/**
* Stream a result set using the MySQL protocol for encodign the result
* set. Each row is retrieved by calling the function passed in the
* argument list.
*
* @param set The result set to stream
* @param dcb The connection to stream the result set to
*/
void
resultset_stream_mysql(RESULTSET *set, DCB *dcb)
{
RESULT_COLUMN *col;
RESULT_ROW *row;
uint8_t seqno = 2;
mysql_send_fieldcount(dcb, set->n_cols);
col = set->column;
while (col)
{
mysql_send_columndef(dcb, col->name, col->type, col->len, seqno++);
col = col->next;
}
mysql_send_eof(dcb, seqno++);
while ((row = (*set->fetchrow)(set, set->userdata)) != NULL)
{
mysql_send_row(dcb, row, seqno++);
resultset_free_row(row);
}
mysql_send_eof(dcb, seqno);
}
/**
* Send the field count packet in a response packet sequence.
*
* @param dcb DCB of connection to send result set to
* @param count Number of columns in the result set
* @return Non-zero on success
*/
static int
mysql_send_fieldcount(DCB *dcb, int count)
{
GWBUF *pkt;
uint8_t *ptr;
if ((pkt = gwbuf_alloc(5)) == NULL)
return 0;
ptr = GWBUF_DATA(pkt);
*ptr++ = 0x01; // Payload length
*ptr++ = 0x00;
*ptr++ = 0x00;
*ptr++ = 0x01; // Sequence number in response
*ptr++ = count; // Length of result string
return dcb->func.write(dcb, pkt);
}
/**
* Send the column definition packet in a response packet sequence.
*
* @param dcb The DCB of the connection
* @param name Name of the column
* @param type Column type
* @param len Column length
* @param seqno Packet sequence number
* @return Non-zero on success
*/
static int
mysql_send_columndef(DCB *dcb, char *name, int type, int len, uint8_t seqno)
{
GWBUF *pkt;
uint8_t *ptr;
int plen;
if ((pkt = gwbuf_alloc(26 + strlen(name))) == NULL)
return 0;
ptr = GWBUF_DATA(pkt);
plen = 22 + strlen(name);
*ptr++ = plen & 0xff;
*ptr++ = (plen >> 8) & 0xff;
*ptr++ = (plen >> 16)& 0xff;
*ptr++ = seqno; // Sequence number in response
*ptr++ = 3; // Catalog is always def
*ptr++ = 'd';
*ptr++ = 'e';
*ptr++ = 'f';
*ptr++ = 0; // Schema name length
*ptr++ = 0; // virtual table name length
*ptr++ = 0; // Table name length
*ptr++ = strlen(name); // Column name length;
while (*name)
*ptr++ = *name++; // Copy the column name
*ptr++ = 0; // Orginal column name
*ptr++ = 0x0c; // Length of next fields always 12
*ptr++ = 0x3f; // Character set
*ptr++ = 0;
*ptr++ = len & 0xff; // Length of column
*ptr++ = (len >> 8) & 0xff;
*ptr++ = (len >> 16) & 0xff;
*ptr++ = (len >> 24) & 0xff;
*ptr++ = type;
*ptr++ = 0x81; // Two bytes of flags
if (type == 0xfd)
*ptr++ = 0x1f;
else
*ptr++ = 0x00;
*ptr++= 0;
*ptr++= 0;
*ptr++= 0;
return dcb->func.write(dcb, pkt);
}
/**
* Send an EOF packet in a response packet sequence.
*
* @param dcb The client connection
* @param seqno The sequence number of the EOF packet
* @return Non-zero on success
*/
static int
mysql_send_eof(DCB *dcb, int seqno)
{
GWBUF *pkt;
uint8_t *ptr;
if ((pkt = gwbuf_alloc(9)) == NULL)
return 0;
ptr = GWBUF_DATA(pkt);
*ptr++ = 0x05;
*ptr++ = 0x00;
*ptr++ = 0x00;
*ptr++ = seqno; // Sequence number in response
*ptr++ = 0xfe; // Length of result string
*ptr++ = 0x00; // No Errors
*ptr++ = 0x00;
*ptr++ = 0x02; // Autocommit enabled
*ptr++ = 0x00;
return dcb->func.write(dcb, pkt);
}
/**
* Send a row packet in a response packet sequence.
*
* @param dcb The client connection
* @param row The row to send
* @param seqno The sequence number of the EOF packet
* @return Non-zero on success
*/
static int
mysql_send_row(DCB *dcb, RESULT_ROW *row, int seqno)
{
GWBUF *pkt;
int i, len = 4;
uint8_t *ptr;
for (i = 0; i < row->n_cols; i++)
{
if (row->cols[i])
len += strlen(row->cols[i]);
len++;
}
if ((pkt = gwbuf_alloc(len)) == NULL)
return 0;
ptr = GWBUF_DATA(pkt);
len -= 4;
*ptr++ = len & 0xff;
*ptr++ = (len >> 8) & 0xff;
*ptr++ = (len >> 16) & 0xff;
*ptr++ = seqno;
for (i = 0; i < row->n_cols; i++)
{
if (row->cols[i])
{
len = strlen(row->cols[i]);
*ptr++ = len;
strncpy((char *)ptr, row->cols[i], len);
ptr += len;
}
else
{
*ptr++ = 0; // NULL column
}
}
return dcb->func.write(dcb, pkt);
}
/**
* Return true if the string only contains numerics
*
* @param value String to test
* @return Non-zero if the string is made of of numeric values
*/
static int
value_is_numeric(char *value)
{
while (*value)
{
if (!isdigit(*value))
return 0;
value++;
}
return 1;
}
/**
* Stream a result set encoding it as a JSON object
* Each row is retrieved by calling the function passed in the
* argument list.
*
* @param set The result set to stream
* @param dcb The connection to stream the result set to
*/
void
resultset_stream_json(RESULTSET *set, DCB *dcb)
{
RESULT_COLUMN *col;
RESULT_ROW *row;
int rowno = 0;
dcb_printf(dcb, "[ ");
while ((row = (*set->fetchrow)(set, set->userdata)) != NULL)
{
int i = 0;
if (rowno++ > 0)
dcb_printf(dcb, ",\n");
dcb_printf(dcb, "{ ");
col = set->column;
while (col)
{
dcb_printf(dcb, "\"%s\" : ", col->name);
if (row->cols[i] && value_is_numeric(row->cols[i]))
dcb_printf(dcb, "%s", row->cols[i]);
else if (row->cols[i])
dcb_printf(dcb, "\"%s\"", row->cols[i]);
else
dcb_printf(dcb, "NULL");
i++;
col = col->next;
if (col)
dcb_printf(dcb, ", ");
}
resultset_free_row(row);
dcb_printf(dcb, "}");
}
dcb_printf(dcb, "]\n");
}

72
server/core/server.c
View File

@ -659,3 +659,75 @@ SERVER_PARAM *param = server->parameters;
}
return NULL;
}
/**
* Provide a row to the result set that defines the set of servers
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
serverRowCallback(RESULTSET *set, void *data)
{
int *rowno = (int *)data;
int i = 0;;
char *stat, buf[20];
RESULT_ROW *row;
SERVER *ptr;
spinlock_acquire(&server_spin);
ptr = allServers;
while (i < *rowno && ptr)
{
i++;
ptr = ptr->next;
}
if (ptr == NULL)
{
spinlock_release(&server_spin);
free(data);
return NULL;
}
(*rowno)++;
row = resultset_make_row(set);
resultset_row_set(row, 0, ptr->unique_name);
resultset_row_set(row, 1, ptr->name);
sprintf(buf, "%d", ptr->port);
resultset_row_set(row, 2, buf);
sprintf(buf, "%d", ptr->stats.n_current);
resultset_row_set(row, 3, buf);
stat = server_status(ptr);
resultset_row_set(row, 4, stat);
free(stat);
spinlock_release(&server_spin);
return row;
}
/**
* Return a resultset that has the current set of servers in it
*
* @return A Result set
*/
RESULTSET *
serverGetList()
{
RESULTSET *set;
int *data;
if ((data = (int *)malloc(sizeof(int))) == NULL)
return NULL;
*data = 0;
if ((set = resultset_create(serverRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Server", 20, COL_TYPE_VARCHAR);
resultset_add_column(set, "Address", 15, COL_TYPE_VARCHAR);
resultset_add_column(set, "Port", 5, COL_TYPE_VARCHAR);
resultset_add_column(set, "Connections", 8, COL_TYPE_VARCHAR);
resultset_add_column(set, "Status", 20, COL_TYPE_VARCHAR);
return set;
}

179
server/core/service.c
View File

@ -34,6 +34,7 @@
* 09/09/14 Massimiliano Pinto Added service option for localhost authentication
* 13/10/14 Massimiliano Pinto Added hashtable for resources (i.e database names for MySQL services)
* 06/02/15 Mark Riddoch Added caching of authentication data
* 18/02/15 Mark Riddoch Added result set management
*
* @endverbatim
*/
@ -58,6 +59,8 @@
#include <sys/stat.h>
#include <sys/types.h>
#include <housekeeper.h>
#include <resultset.h>
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
@ -1535,3 +1538,179 @@ void service_shutdown()
}
spinlock_release(&service_spin);
}
/**
* Return the count of all sessions active for all services
*
* @return Count of all active sessions
*/
int
serviceSessionCountAll()
{
SERVICE *ptr;
int rval = 0;
spinlock_acquire(&service_spin);
ptr = allServices;
while (ptr)
{
rval += ptr->stats.n_current;
ptr = ptr->next;
}
spinlock_release(&service_spin);
return rval;
}
/**
* Provide a row to the result set that defines the set of service
* listeners
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
serviceListenerRowCallback(RESULTSET *set, void *data)
{
int *rowno = (int *)data;
int i = 0;;
char buf[20];
RESULT_ROW *row;
SERVICE *ptr;
SERV_PROTOCOL *lptr = NULL;
spinlock_acquire(&service_spin);
ptr = allServices;
if (ptr)
lptr = ptr->ports;
while (i < *rowno && ptr)
{
lptr = ptr->ports;
while (i < *rowno && lptr)
{
if ((lptr = lptr->next) != NULL)
i++;
}
if (i < *rowno)
{
ptr = ptr->next;
if (ptr && (lptr = ptr->ports) != NULL)
i++;
}
}
if (lptr == NULL)
{
spinlock_release(&service_spin);
free(data);
return NULL;
}
(*rowno)++;
row = resultset_make_row(set);
resultset_row_set(row, 0, ptr->name);
resultset_row_set(row, 1, lptr->protocol);
resultset_row_set(row, 2, (lptr && lptr->address) ? lptr->address : "*");
sprintf(buf, "%d", lptr->port);
resultset_row_set(row, 3, buf);
resultset_row_set(row, 4,
(!lptr->listener || !lptr->listener->session ||
lptr->listener->session->state == SESSION_STATE_LISTENER_STOPPED) ?
"Stopped" : "Running");
spinlock_release(&service_spin);
return row;
}
/**
* Return a resultset that has the current set of services in it
*
* @return A Result set
*/
RESULTSET *
serviceGetListenerList()
{
RESULTSET *set;
int *data;
if ((data = (int *)malloc(sizeof(int))) == NULL)
return NULL;
*data = 0;
if ((set = resultset_create(serviceListenerRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Service Name", 25, COL_TYPE_VARCHAR);
resultset_add_column(set, "Protocol Module", 20, COL_TYPE_VARCHAR);
resultset_add_column(set, "Address", 15, COL_TYPE_VARCHAR);
resultset_add_column(set, "Port", 5, COL_TYPE_VARCHAR);
resultset_add_column(set, "State", 8, COL_TYPE_VARCHAR);
return set;
}
/**
* Provide a row to the result set that defines the set of services
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
serviceRowCallback(RESULTSET *set, void *data)
{
int *rowno = (int *)data;
int i = 0;;
char buf[20];
RESULT_ROW *row;
SERVICE *ptr;
spinlock_acquire(&service_spin);
ptr = allServices;
while (i < *rowno && ptr)
{
i++;
ptr = ptr->next;
}
if (ptr == NULL)
{
spinlock_release(&service_spin);
free(data);
return NULL;
}
(*rowno)++;
row = resultset_make_row(set);
resultset_row_set(row, 0, ptr->name);
resultset_row_set(row, 1, ptr->routerModule);
sprintf(buf, "%d", ptr->stats.n_current);
resultset_row_set(row, 2, buf);
sprintf(buf, "%d", ptr->stats.n_sessions);
resultset_row_set(row, 3, buf);
spinlock_release(&service_spin);
return row;
}
/**
* Return a resultset that has the current set of services in it
*
* @return A Result set
*/
RESULTSET *
serviceGetList()
{
RESULTSET *set;
int *data;
if ((data = (int *)malloc(sizeof(int))) == NULL)
return NULL;
*data = 0;
if ((set = resultset_create(serviceRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Service Name", 25, COL_TYPE_VARCHAR);
resultset_add_column(set, "Router Module", 20, COL_TYPE_VARCHAR);
resultset_add_column(set, "No. Sessions", 10, COL_TYPE_VARCHAR);
resultset_add_column(set, "Total Sessions", 10, COL_TYPE_VARCHAR);
return set;
}

100
server/core/session.c
View File

@ -941,4 +941,102 @@ void session_close_timeouts(void* data)
spinlock_release(&session_spin);
}
}
}
/**
* Callback structure for the session list extraction
*/
typedef struct {
int index;
SESSIONLISTFILTER filter;
} SESSIONFILTER;
/**
* Provide a row to the result set that defines the set of sessions
*
* @param set The result set
* @param data The index of the row to send
* @return The next row or NULL
*/
static RESULT_ROW *
sessionRowCallback(RESULTSET *set, void *data)
{
SESSIONFILTER *cbdata = (SESSIONFILTER *)data;
int i = 0;
char buf[20];
RESULT_ROW *row;
SESSION *ptr;
spinlock_acquire(&session_spin);
ptr = allSessions;
/* Skip to the first non-listener if not showing listeners */
while (ptr && cbdata->filter == SESSION_LIST_CONNECTION &&
ptr->state == SESSION_STATE_LISTENER)
{
ptr = ptr->next;
}
while (i < cbdata->index && ptr)
{
if (cbdata->filter == SESSION_LIST_CONNECTION &&
ptr->state != SESSION_STATE_LISTENER)
{
i++;
}
else if (cbdata->filter == SESSION_LIST_ALL)
{
i++;
}
ptr = ptr->next;
}
/* Skip to the next non-listener if not showing listeners */
while (ptr && cbdata->filter == SESSION_LIST_CONNECTION &&
ptr->state == SESSION_STATE_LISTENER)
{
ptr = ptr->next;
}
if (ptr == NULL)
{
spinlock_release(&session_spin);
free(data);
return NULL;
}
cbdata->index++;
row = resultset_make_row(set);
sprintf(buf, "%p", ptr);
resultset_row_set(row, 0, buf);
resultset_row_set(row, 1, ((ptr->client && ptr->client->remote)
? ptr->client->remote : ""));
resultset_row_set(row, 2, (ptr->service && ptr->service->name
? ptr->service->name : ""));
resultset_row_set(row, 3, session_state(ptr->state));
spinlock_release(&session_spin);
return row;
}
/**
* Return a resultset that has the current set of sessions in it
*
* @return A Result set
*/
RESULTSET *
sessionGetList(SESSIONLISTFILTER filter)
{
RESULTSET *set;
SESSIONFILTER *data;
if ((data = (SESSIONFILTER *)malloc(sizeof(SESSIONFILTER))) == NULL)
return NULL;
data->index = 0;
data->filter = filter;
if ((set = resultset_create(sessionRowCallback, data)) == NULL)
{
free(data);
return NULL;
}
resultset_add_column(set, "Session", 16, COL_TYPE_VARCHAR);
resultset_add_column(set, "Client", 15, COL_TYPE_VARCHAR);
resultset_add_column(set, "Service", 15, COL_TYPE_VARCHAR);
resultset_add_column(set, "State", 15, COL_TYPE_VARCHAR);
return set;
}

28
server/core/utils.c
View File

@ -42,6 +42,7 @@
#include <poll.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <secrets.h>
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
@ -235,3 +236,30 @@ int gw_getsockerrno(
return_eno:
return eno;
}
/**
* Create a HEX(SHA1(SHA1(password)))
*
* @param password The password to encrypt
* @return The new allocated encrypted password, that the caller must free
*
*/
char *create_hex_sha1_sha1_passwd(char *passwd) {
uint8_t hash1[SHA_DIGEST_LENGTH]="";
uint8_t hash2[SHA_DIGEST_LENGTH]="";
char *hexpasswd=NULL;
if ((hexpasswd = (char *)calloc(SHA_DIGEST_LENGTH * 2 + 1, 1)) == NULL)
return NULL;
/* hash1 is SHA1(real_password) */
gw_sha1_str((uint8_t *)passwd, strlen(passwd), hash1);
/* hash2 is the SHA1(input data), where input_data = SHA1(real_password) */
gw_sha1_str(hash1, SHA_DIGEST_LENGTH, hash2);
/* dbpass is the HEX form of SHA1(SHA1(real_password)) */
gw_bin2hex(hexpasswd, hash2, SHA_DIGEST_LENGTH);
return hexpasswd;
}