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

Use module parameters in topfilter and tpmfilter

Browse Source 
The topfilter and tpmfilter now use the module parameteres.

Also removed some unnecessary variables and included maxscale/cdefs.h in
tpmfilter.
This commit is contained in:
Markus Mäkelä
2017-01-09 09:59:14 +02:00
parent e6b7035766
commit d6d564c44d
2 changed files with 124 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
server/modules/filter/topfilter/topfilter.c
View File

@ -115,6 +115,14 @@ typedef struct
struct timeval disconnect;
} TOPN_SESSION;
static const MXS_ENUM_VALUE option_values[] =
{
{"ignorecase", REG_ICASE},
{"case", 0},
{"extended", REG_EXTENDED},
{NULL}
};
/**
* The module entry point routine. It is this routine that
* must populate the structure that is referred to as the
@ -153,6 +161,19 @@ MXS_MODULE* MXS_CREATE_MODULE()
NULL, /* Thread init. */
NULL, /* Thread finish. */
{
{"count", MXS_MODULE_PARAM_COUNT, "10"},
{"filebase", MXS_MODULE_PARAM_STRING, NULL, MXS_MODULE_OPT_REQUIRED},
{"match", MXS_MODULE_PARAM_STRING},
{"exclude", MXS_MODULE_PARAM_STRING},
{"source", MXS_MODULE_PARAM_STRING},
{"user", MXS_MODULE_PARAM_STRING},
{
"options",
MXS_MODULE_PARAM_ENUM,
"ignorecase",
MXS_MODULE_OPT_NONE,
option_values
},
{MXS_END_MODULE_PARAMS}
}
};
@ -176,81 +197,17 @@ createInstance(const char *name, char **options, CONFIG_PARAMETER *params)
if (my_instance)
{
my_instance->topN = 10;
my_instance->match = NULL;
my_instance->exclude = NULL;
my_instance->source = NULL;
my_instance->user = NULL;
my_instance->filebase = NULL;
my_instance->sessions = 0;
my_instance->topN = config_get_integer(params, "count");
my_instance->match = config_copy_string(params, "match");
my_instance->exclude = config_copy_string(params, "exclude");
my_instance->source = config_copy_string(params, "source");
my_instance->user = config_copy_string(params, "user");
my_instance->filebase = MXS_STRDUP_A(config_get_string(params, "filebase"));
int cflags = config_get_enum(params, "options", option_values);
bool error = false;
for (const CONFIG_PARAMETER *p = params; p; p = p->next)
{
if (!strcmp(p->name, "count"))
{
my_instance->topN = atoi(p->value);
}
else if (!strcmp(p->name, "filebase"))
{
my_instance->filebase = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "match"))
{
my_instance->match = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "exclude"))
{
my_instance->exclude = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "source"))
{
my_instance->source = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "user"))
{
my_instance->user = MXS_STRDUP_A(p->value);
}
else if (!filter_standard_parameter(p->name))
{
MXS_ERROR("topfilter: Unexpected parameter '%s'.", p->name);
error = true;
}
}
int cflags = REG_ICASE;
if (options)
{
for (int i = 0; options[i]; i++)
{
if (!strcasecmp(options[i], "ignorecase"))
{
cflags |= REG_ICASE;
}
else if (!strcasecmp(options[i], "case"))
{
cflags &= ~REG_ICASE;
}
else if (!strcasecmp(options[i], "extended"))
{
cflags |= REG_EXTENDED;
}
else
{
MXS_ERROR("topfilter: Unsupported option '%s'.",
options[i]);
error = true;
}
}
}
if (my_instance->filebase == NULL)
{
MXS_ERROR("topfilter: No 'filebase' parameter defined.");
error = true;
}
my_instance->sessions = 0;
if (my_instance->match &&
regcomp(&my_instance->re, my_instance->match, cflags))
{
@ -293,6 +250,7 @@ createInstance(const char *name, char **options, CONFIG_PARAMETER *params)
my_instance = NULL;
}
}
return (FILTER *) my_instance;
}

219
server/modules/filter/tpmfilter/tpmfilter.c
View File

@ -40,6 +40,8 @@
* @endverbatim
*/
#include <maxscale/cdefs.h>
#include <ctype.h>
#include <stdio.h>
#include <fcntl.h>
@ -60,18 +62,15 @@
#include <maxscale/atomic.h>
#include <maxscale/query_classifier.h>
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
/* The maximum size for query statements in a transaction (64MB) */
static size_t sql_size_limit = 64 * 1024 * 1024;
static size_t buf_size = 10;
static size_t sql_size_limit = 64 * 1024 *
1024; /* The maximum size for query statements in a transaction (64MB) */
static const int default_sql_size = 4 * 1024;
static const char* default_query_delimiter = "@@@";
static const char* default_log_delimiter = ":::";
static const char* default_file_name = "tpm.log";
static const char* default_named_pipe = "/tmp/tpmfilter";
#define DEFAULT_QUERY_DELIMITER "@@@"
#define DEFAULT_LOG_DELIMITER ":::"
#define DEFAULT_FILE_NAME "tpm.log"
#define DEFAULT_NAMED_PIPE "/tmp/tpmfilter"
/*
* The filter entry points
@ -172,6 +171,12 @@ MXS_MODULE* MXS_CREATE_MODULE()
NULL, /* Thread init. */
NULL, /* Thread finish. */
{
{"named_pipe", MXS_MODULE_PARAM_STRING, DEFAULT_NAMED_PIPE},
{"filename", MXS_MODULE_PARAM_STRING, DEFAULT_FILE_NAME},
{"delimiter", MXS_MODULE_PARAM_STRING, DEFAULT_LOG_DELIMITER},
{"query_delimiter", MXS_MODULE_PARAM_STRING, DEFAULT_QUERY_DELIMITER},
{"source", MXS_MODULE_PARAM_STRING},
{"user", MXS_MODULE_PARAM_STRING},
{MXS_END_MODULE_PARAMS}
}
};
@ -188,131 +193,95 @@ MXS_MODULE* MXS_CREATE_MODULE()
*
* @return The instance data for this new instance
*/
static FILTER *
static FILTER *
createInstance(const char *name, char **options, CONFIG_PARAMETER *params)
{
int i, ret;
TPM_INSTANCE *my_instance;
TPM_INSTANCE *my_instance = MXS_CALLOC(1, sizeof(TPM_INSTANCE));
if ((my_instance = MXS_CALLOC(1, sizeof(TPM_INSTANCE))) != NULL)
if (my_instance)
{
my_instance->source = NULL;
my_instance->user = NULL;
my_instance->log_enabled = false;
/* set default log filename */
my_instance->filename = MXS_STRDUP_A(default_file_name);
/* set default delimiter */
my_instance->delimiter = MXS_STRDUP_A(default_log_delimiter);
/* set default query delimiter */
my_instance->query_delimiter = MXS_STRDUP_A(default_query_delimiter);
my_instance->query_delimiter_size = 3;
/* set default named pipe */
my_instance->named_pipe = MXS_STRDUP_A(default_named_pipe);
for (const CONFIG_PARAMETER *p = params; p; p = p->next)
{
if (!strcmp(p->name, "filename"))
{
MXS_FREE(my_instance->filename);
my_instance->filename = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "source"))
{
my_instance->source = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "user"))
{
my_instance->user = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "delimiter"))
{
MXS_FREE(my_instance->delimiter);
my_instance->delimiter = MXS_STRDUP_A(p->value);
}
else if (!strcmp(p->name, "query_delimiter"))
{
MXS_FREE(my_instance->query_delimiter);
my_instance->query_delimiter = MXS_STRDUP_A(p->value);
my_instance->query_delimiter_size = strlen(my_instance->query_delimiter);
}
else if (!strcmp(p->name, "named_pipe"))
{
if (p->value == NULL)
{
MXS_ERROR("You need to specify 'named_pipe' for tpmfilter.");
MXS_FREE(my_instance);
return NULL;
}
else
{
my_instance->named_pipe = MXS_STRDUP_A(p->value);
// check if the file exists first.
if (access(my_instance->named_pipe, F_OK) == 0)
{
// if exists, check if it is a named pipe.
struct stat st;
ret = stat(my_instance->named_pipe, &st);
// check whether the file is named pipe.
if (ret == -1 && errno != ENOENT)
{
MXS_ERROR("stat() failed on named pipe: %s", strerror(errno));
MXS_FREE(my_instance);
return NULL;
}
if (ret == 0 && S_ISFIFO(st.st_mode))
{
// if it is a named pipe, we delete it and recreate it.
unlink(my_instance->named_pipe);
}
else
{
MXS_ERROR("The file '%s' already exists and it is not a named pipe.", my_instance->named_pipe);
MXS_FREE(my_instance);
return NULL;
}
}
// now create the named pipe.
ret = mkfifo(my_instance->named_pipe, 0660);
if (ret == -1)
{
MXS_ERROR("mkfifo() failed on named pipe: %s", strerror(errno));
MXS_FREE(my_instance);
return NULL;
}
}
}
}
if (my_instance->named_pipe == NULL)
{
MXS_ERROR("You need to specify 'named_pipe' for tpmfilter.");
MXS_FREE(my_instance);
return NULL;
}
my_instance->sessions = 0;
my_instance->log_enabled = false;
my_instance->filename = MXS_STRDUP_A(config_get_string(params, "filename"));
my_instance->delimiter = MXS_STRDUP_A(config_get_string(params, "delimiter"));
my_instance->query_delimiter = MXS_STRDUP_A(config_get_string(params, "query_delimiter"));
my_instance->query_delimiter_size = strlen(my_instance->query_delimiter);
my_instance->named_pipe = MXS_STRDUP_A(config_get_string(params, "named_pipe"));
my_instance->source = config_copy_string(params, "source");
my_instance->user = config_copy_string(params, "user");
bool error = false;
// check if the file exists first.
if (access(my_instance->named_pipe, F_OK) == 0)
{
// if exists, check if it is a named pipe.
struct stat st;
int ret = stat(my_instance->named_pipe, &st);
// check whether the file is named pipe.
if (ret == -1 && errno != ENOENT)
{
MXS_ERROR("stat() failed on named pipe: %s", strerror(errno));
error = true;
}
else if (ret == 0 && S_ISFIFO(st.st_mode))
{
// if it is a named pipe, we delete it and recreate it.
unlink(my_instance->named_pipe);
}
else
{
MXS_ERROR("The file '%s' already exists and it is not "
"a named pipe.", my_instance->named_pipe);
error = true;
}
}
// now create the named pipe.
if (mkfifo(my_instance->named_pipe, 0660) == -1)
{
MXS_ERROR("mkfifo() failed on named pipe: %s", strerror(errno));
error = true;
}
my_instance->fp = fopen(my_instance->filename, "w");
if (my_instance->fp == NULL)
{
MXS_ERROR("Opening output file '%s' for tpmfilter failed due to %d, %s", my_instance->filename, errno,
strerror(errno));
return NULL;
MXS_ERROR("Opening output file '%s' for tpmfilter failed due to %d, %s",
my_instance->filename, errno, strerror(errno));
error = true;
}
/*
* Launch a thread that checks the named pipe.
*/
THREAD thread;
if (!error && thread_start(&thread, checkNamedPipe, (void*)my_instance) == NULL)
{
MXS_ERROR("Couldn't create a thread to check the named pipe: %s", strerror(errno));
error = true;
}
if (error)
{
MXS_FREE(my_instance->delimiter);
MXS_FREE(my_instance->filename);
MXS_FREE(my_instance->named_pipe);
MXS_FREE(my_instance->query_delimiter);
MXS_FREE(my_instance->source);
MXS_FREE(my_instance->user);
if (my_instance->fp)
{
fclose(my_instance->fp);
}
MXS_FREE(my_instance);
}
}
/*
* Launch a thread that checks the named pipe.
*/
THREAD thread;
if (thread_start(&thread, checkNamedPipe, (void*) my_instance) == NULL)
{
MXS_ERROR("Couldn't create a thread to check the named pipe: %s", strerror(errno));
MXS_FREE(my_instance);
return NULL;
}
return (FILTER *)my_instance;
return(FILTER *)my_instance;
}
/**
@ -339,7 +308,6 @@ newSession(FILTER *instance, SESSION *session)
my_session->max_sql_size = default_sql_size; // default max query size of 4k.
my_session->sql = (char*)MXS_CALLOC(my_session->max_sql_size, sizeof(char));
memset(my_session->sql, 0x00, my_session->max_sql_size);
my_session->buf = (char*)MXS_CALLOC(buf_size, sizeof(char));
my_session->sql_index = 0;
my_session->n_statements = 0;
my_session->total.tv_sec = 0;
@ -411,7 +379,6 @@ freeSession(FILTER *instance, void *session)
MXS_FREE(my_session->clientHost);
MXS_FREE(my_session->userName);
MXS_FREE(my_session->sql);
MXS_FREE(my_session->buf);
MXS_FREE(session);
return;
}