gfwleak/tango-maat
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加性能测试用例，处理无法获取redis time的异常。

Browse Source
This commit is contained in:
zhengchao
2019-07-28 19:13:04 +06:00
parent 718bbc59dd
commit 3473cf6dd4
8 changed files with 278 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/CMakeLists.txt
View File

@@ -1,7 +1,7 @@
cmake_minimum_required(VERSION 3.5)
set(MAAT_FRAME_MAJOR_VERSION 2)
set(MAAT_FRAME_MINOR_VERSION 7)
set(MAAT_FRAME_MINOR_VERSION 8)
set(MAAT_FRAME_PATCH_VERSION 0)
set(MAAT_FRAME_VERSION ${MAAT_FRAME_MAJOR_VERSION}.${MAAT_FRAME_MINOR_VERSION}.${MAAT_FRAME_PATCH_VERSION})

27
src/entry/Maat_command.cpp
View File

@@ -143,9 +143,11 @@ long long redis_server_time(redisContext* ctx)
long long server_time=0;
redisReply* data_reply=NULL;
data_reply=_wrap_redisCommand(ctx,"TIME");
assert(data_reply->type==REDIS_REPLY_ARRAY);
if(data_reply->type==REDIS_REPLY_ARRAY)
{
server_time=atoll(data_reply->element[0]->str);
freeReplyObject(data_reply);
}
return server_time;
}
enum MAAT_TABLE_TYPE type_region2table(const struct Maat_region_t* p)
@@ -1434,7 +1436,10 @@ void check_maat_expiration(redisContext *ctx, void *logger)
long long server_time=0;
server_time=redis_server_time(ctx);
if(!server_time)
{
return;
}
data_reply=_wrap_redisCommand(ctx, "ZRANGEBYSCORE %s -inf %lld",mr_expire_sset,server_time);
if(data_reply->type!=REDIS_REPLY_ARRAY||data_reply->elements==0)
{
@@ -1473,7 +1478,10 @@ void cleanup_update_status(redisContext *ctx, void *logger)
long long server_time=0, version_upper_bound=0,version_lower_bound=0,version_num=0,entry_num=0;
server_time=redis_server_time(ctx);
if(!server_time)
{
return;
}
reply=_wrap_redisCommand(ctx,"MULTI");
freeReplyObject(reply);
redisAppendCommand(ctx, "ZRANGEBYSCORE %s -inf %lld",mr_version_sset,server_time-MAAT_REDIS_SYNC_TIME);
@@ -2103,6 +2111,10 @@ int Maat_cmd_set_lines(Maat_feather_t feather,const struct Maat_line_t** line_ru
{
return -1;
}
server_time=redis_server_time(write_ctx);
if(!server_time)
{
return -1;
}
s_rule=(struct serial_rule_t *)calloc(sizeof(struct serial_rule_t),line_num);
for(i=0;i<line_num;i++)
@@ -2392,7 +2404,10 @@ int Maat_cmd_commit(Maat_feather_t feather)
serial_rule_num+=calculate_serial_rule_num(p, &new_region_num, &new_group_num);
p=p->next;
}
_feather->server_time=redis_server_time(write_ctx);
_feather->server_time=redis_server_time(write_ctx);
if(!_feather->server_time)
{
goto error_out;
}
if(_feather->AUTO_NUMBERING_ON==1)
{
@@ -2612,6 +2627,10 @@ static int _Maat_command_set_one_line(struct _Maat_feather_t* _feather, enum MAA
static int _Maat_command_set_one_line(struct _Maat_feather_t* _feather, enum MAAT_OPERATION op, int id, const char* table_name, const char* line)
{
redisContext* write_ctx=get_redis_ctx_for_write(_feather);
_feather->server_time=redis_server_time(write_ctx);
if(!_feather->server_time)
{
return -1;
}
struct serial_rule_t s_rule;
set_serial_rule(&s_rule, op, id, 0, table_name, line, 0);

38
src/entry/Maat_rule.cpp
View File

@@ -33,7 +33,7 @@
#include "stream_fuzzy_hash.h"
#include "gram_index_engine.h"
int MAAT_FRAME_VERSION_2_7_20190629=1;
int MAAT_FRAME_VERSION_2_8_20190728=1;
int is_valid_expr_type(enum MAAT_EXPR_TYPE expr_type)
@@ -3169,26 +3169,39 @@ void vector_print(igraph_vector_t *v) {
}
printf("\n");
}
static size_t effective_vertices_count(igraph_vector_t *vids)
{
size_t i=0;
int tmp_vid=0;
for(i=0; i<(size_t)igraph_vector_size(vids); i++)
{
tmp_vid=(int) VECTOR(*vids)[i];
if(tmp_vid<0)
{
break;
}
}
return i;
}
void walk_group_hash(const uchar * key, uint size, void * data, void * user)
{
struct Maat_group_inner* group_rule=(struct Maat_group_inner*)data;
struct Maat_group_inner* parent_group=NULL;
struct Maat_scanner* scanner=(struct Maat_scanner*)user;
int tmp_vid=0;
igraph_vector_t vids;
igraph_vector_init(&vids, 0);
igraph_vector_t *vids=&(scanner->dfs_vids);
igraph_dfs(&(scanner->group_graph), group_rule->vertex_id, IGRAPH_OUT,
0, &vids, NULL, NULL, NULL, NULL, NULL, NULL);
0, vids, NULL, NULL, NULL, NULL, NULL, NULL);
long int i=0;
long long* temp_group_ids=ALLOC(long long, igraph_vector_size(&vids));
size_t top_group_cnt=0;
for(i=0; i<igraph_vector_size(&vids); i++)
size_t i=0, top_group_cnt=0;
size_t parent_group_cnt=effective_vertices_count(vids);
long long* temp_group_ids=ALLOC(long long, parent_group_cnt);
for(i=0; i<(size_t)igraph_vector_size(vids); i++)
{
tmp_vid=(int) VECTOR(vids)[i];
tmp_vid=(int) VECTOR(*vids)[i];
if(tmp_vid<0)
{
break;
@@ -3200,6 +3213,7 @@ void walk_group_hash(const uchar * key, uint size, void * data, void * user)
top_group_cnt++;
}
}
pthread_mutex_lock(&(group_rule->mutex));
free(group_rule->top_groups);
group_rule->top_group_cnt=top_group_cnt;
@@ -3211,7 +3225,6 @@ void walk_group_hash(const uchar * key, uint size, void * data, void * user)
scanner->most_popular_sub_group=group_rule->group_id;
}
pthread_mutex_unlock(&(group_rule->mutex));
igraph_vector_destroy(&vids);
free(temp_group_ids);
temp_group_ids=NULL;
return;
@@ -3219,7 +3232,10 @@ void walk_group_hash(const uchar * key, uint size, void * data, void * user)
void find_group_paths(struct Maat_scanner* scanner)
{
scanner->group_graph_vcount=igraph_vcount(&scanner->group_graph);
igraph_vector_init(&(scanner->dfs_vids), scanner->group_graph_vcount);
MESA_htable_iterate(scanner->group_hash, walk_group_hash, scanner);
igraph_vector_destroy(&scanner->dfs_vids);
return;
}
void do_scanner_update(struct Maat_scanner* scanner, MESA_lqueue_head garbage_q, int scan_thread_num, void* logger)

3
src/inc_internal/Maat_rule_internal.h
View File

@@ -249,6 +249,9 @@ struct Maat_scanner
MESA_htable_handle vertex_id2group;
igraph_t group_graph;
igraph_integer_t group_graph_vcount;
igraph_vector_t dfs_vids;
int grp_vertex_id_generator;
int most_popular_sub_group;
unsigned long long max_presented_top_group_cnt;

3
test/CMakeLists.txt
View File

@@ -7,6 +7,9 @@ target_link_libraries(test_igraph igraph-static)
add_executable(test_maatframe test_maatframe.cpp)
target_link_libraries(test_maatframe maat_frame_shared gtest)
add_executable(perf_test_maatframe perf_test_maatframe.cpp)
target_link_libraries(perf_test_maatframe maat_frame_shared gtest)
configure_file(table_info.conf table_info.conf COPYONLY)
configure_file(t2_tableinfo.conf t2_tableinfo.conf COPYONLY)
configure_file(maat_json.json maat_json.json COPYONLY)

205
test/perf_test_maatframe.cpp Normal file
View File

@@ -0,0 +1,205 @@
#include "Maat_rule.h"
#include "stream_fuzzy_hash.h"
#include "Maat_command.h"
#include <MESA/MESA_handle_logger.h>
#include <gtest/gtest.h>
int test_add_expr_command_copy(Maat_feather_t feather,const char* region_table,int config_id, int timeout,int label_id, const char* keywords)
{
struct Maat_cmd_t* cmd=NULL;
struct Maat_rule_t rule;
char huge_serv_def[1024*2];
memset(huge_serv_def,'s',sizeof(huge_serv_def));
struct Maat_region_t region;
int group_num=1,ret=0;
memset(&rule,0,sizeof(rule));
rule.config_id=config_id;
strcpy(rule.service_defined,"maat_command");
//MUST acqire by function, because Maat_cmd_t has some hidden members.
cmd=Maat_create_cmd(&rule, group_num);
cmd->expire_after=timeout;
cmd->label_id=label_id;
memset(&region,0,sizeof(region));
region.region_type=REGION_EXPR;
region.table_name=region_table;
region.expr_rule.district=NULL;
region.expr_rule.keywords=keywords;
region.expr_rule.expr_type=EXPR_TYPE_AND;
region.expr_rule.match_method=MATCH_METHOD_SUB;
region.expr_rule.hex_bin=UNCASE_PLAIN;
Maat_cmd_set_opt(cmd, MAAT_RULE_SERV_DEFINE, huge_serv_def, sizeof(huge_serv_def));
Maat_add_region2cmd(cmd, 0, &region);
//use pipeline model.
ret=Maat_cmd_append(feather, cmd, MAAT_OP_ADD);
if(ret<0)
{
printf("Add Maat command %d failed.\n",rule.config_id);
Maat_free_cmd(cmd);
return 0;
}
//cmd has been saved in feather, so free cmd before commit is allowed.
Maat_free_cmd(cmd);
return 0;
}
void wait_for_cmd_effective_copy(Maat_feather_t feather, long long version_before)
{
long long version_after=version_before;
int is_updating=1;
long long wating_us=0, sleep_us=1000*100;
while(is_updating||version_before==version_after)
{
Maat_read_state(feather,MAAT_STATE_IN_UPDATING, &is_updating, sizeof(is_updating));
Maat_read_state(feather,MAAT_STATE_VERSION, &version_after, sizeof(version_after));
usleep(sleep_us);//waiting for commands go into effect
wating_us+=sleep_us;
}
// printf("wait for %lld ms\n", wating_us/1000);
}
class MaatCMDPerfTest : public testing::Test
{
protected:
static void SetUpTestCase()
{
const char* test_maat_redis_ip="127.0.0.1";
unsigned short test_maat_redis_port=6379;
int g_iThreadNum=4;
const char* table_info_path="./table_info.conf";
int scan_interval_ms=500;
int effective_interval_ms=0;
logger=MESA_create_runtime_log_handle("test_maat_redis.log",0);
_shared_feather=Maat_feather(g_iThreadNum, table_info_path, logger);
Maat_set_feather_opt(_shared_feather,MAAT_OPT_INSTANCE_NAME,"perf", strlen("perf")+1);
Maat_set_feather_opt(_shared_feather, MAAT_OPT_REDIS_IP, test_maat_redis_ip, strlen(test_maat_redis_ip)+1);
Maat_set_feather_opt(_shared_feather, MAAT_OPT_REDIS_PORT, &test_maat_redis_port, sizeof(test_maat_redis_port));
Maat_set_feather_opt(_shared_feather, MAAT_OPT_SCANDIR_INTERVAL_MS,&scan_interval_ms, sizeof(scan_interval_ms));
//Set a short intevral for testing.
Maat_set_feather_opt(_shared_feather, MAAT_OPT_EFFECT_INVERVAL_MS,&effective_interval_ms, sizeof(effective_interval_ms));
const char* foregin_dir="./foreign_files/";
Maat_set_feather_opt(_shared_feather, MAAT_OPT_FOREIGN_CONT_DIR, foregin_dir, strlen(foregin_dir)+1);
int linger_timeout=2;
Maat_set_feather_opt(_shared_feather, MAAT_OPT_FOREIGN_CONT_LINGER, &linger_timeout, sizeof(linger_timeout));
Maat_cmd_flushDB(_shared_feather);
Maat_initiate_feather(_shared_feather);
}
static void TearDownTestCase()
{
Maat_burn_feather(_shared_feather);
MESA_destroy_runtime_log_handle(logger);
}
// Some expensive resource shared by all tests.
static Maat_feather_t _shared_feather;
static void *logger;
};
Maat_feather_t MaatCMDPerfTest::_shared_feather;
void* MaatCMDPerfTest::logger;
//Following tests must be coded/tested at last, for they stalled the maat update thread and interrupt other tests.
TEST_F(MaatCMDPerfTest, SetExpr100K)
{
const int CMD_EXPR_NUM=100*1000;
const char* table_name="HTTP_URL";
const char* keywords1="Hiredis";
const char* keywords2="C Client";
char escape_buff1[256],escape_buff2[256];
char keywords[256];
int label_id=5210, config_id=0,ret=0, output_id_cnt=0;
Maat_feather_t feather=MaatCMDPerfTest::_shared_feather;
long long version_before=0;
ret=Maat_read_state(feather,MAAT_STATE_VERSION, &version_before, sizeof(version_before));
Maat_str_escape(escape_buff1, sizeof(escape_buff1),keywords1);
Maat_str_escape(escape_buff2, sizeof(escape_buff2),keywords2);
snprintf(keywords,sizeof(keywords),"%s&%s",escape_buff1,escape_buff2);
config_id=(int)Maat_cmd_incrby(feather, "TEST_SEQ", CMD_EXPR_NUM);
int i=0;
for(i=0; i<CMD_EXPR_NUM;i++)
{
test_add_expr_command_copy(feather,table_name,config_id-i, 0, label_id, keywords);
}
ret=Maat_cmd_commit(feather);
EXPECT_TRUE(ret>=0);
wait_for_cmd_effective_copy(feather, version_before);
struct Maat_cmd_t* cmd=NULL;
struct Maat_rule_t rule;
memset(&rule,0,sizeof(rule));
int *output_ids=(int*)malloc(sizeof(int)*CMD_EXPR_NUM);
output_id_cnt=Maat_cmd_select(feather,label_id, output_ids, CMD_EXPR_NUM);
EXPECT_EQ(output_id_cnt, CMD_EXPR_NUM);
for(i=0; i<CMD_EXPR_NUM;i++)
{
memset(&rule,0,sizeof(rule));
rule.config_id=output_ids[i];
cmd=Maat_create_cmd(&rule, 0);
ret=Maat_cmd_append(feather, cmd, MAAT_OP_DEL);
EXPECT_EQ(ret, 0);
Maat_free_cmd(cmd);
}
ret=Maat_cmd_commit(feather);
EXPECT_EQ(ret, CMD_EXPR_NUM);
free(output_ids);
}
TEST_F(MaatCMDPerfTest, SetLinesOneMillion)
{
const int TEST_CMD_LINE_NUM=1000*1000;
Maat_feather_t feather=MaatCMDPerfTest::_shared_feather;
struct Maat_line_t **p_line=(struct Maat_line_t **)calloc(sizeof(struct Maat_line_t *), TEST_CMD_LINE_NUM);
struct Maat_line_t *line_rule=(struct Maat_line_t *)calloc(sizeof(struct Maat_line_t), TEST_CMD_LINE_NUM);
int i=0;
const char* line="1\t192.168.0.1\t4444444444\t1";
int seq=(int)Maat_cmd_incrby(feather,"TEST_PLUG_SEQ", TEST_CMD_LINE_NUM);
for(i=0;i<TEST_CMD_LINE_NUM;i++)
{
line_rule[i].label_id=0;
line_rule[i].rule_id=seq-i;
line_rule[i].table_name="QD_ENTRY_INFO";
line_rule[i].table_line=line;
// asprintf(&(line_rule[i].table_line),"1\t192.168.0.1\t%d\t1",100+i);
line_rule[i].expire_after=0;
p_line[i]=line_rule+i;
}
long long version_before=0;
Maat_read_state(feather,MAAT_STATE_VERSION, &version_before, sizeof(version_before));
Maat_cmd_set_lines(feather,(const struct Maat_line_t **)p_line,TEST_CMD_LINE_NUM, MAAT_OP_ADD);
wait_for_cmd_effective_copy(feather, version_before);
for(i=0;i<TEST_CMD_LINE_NUM;i++)
{
line_rule[i].table_line=NULL;
}
Maat_cmd_set_lines(feather, (const struct Maat_line_t **)p_line,TEST_CMD_LINE_NUM, MAAT_OP_DEL);
free(p_line);
free(line_rule);
return;
}
int main(int argc, char ** argv)
{
int ret=0;
::testing::InitGoogleTest(&argc, argv);
ret=RUN_ALL_TESTS();
return ret;
}

99
test/test_maatframe.cpp
View File

@@ -1491,8 +1491,15 @@ TEST(VirtualTable, Test1)
EXPECT_EQ(ret, 1);
EXPECT_EQ(result[0].config_id, 160);
Maat_clean_status(&mid);
const char* should_not_hit="2018-10-05 is a keywords of table KEYWORDS_TABLE. Should not hit.";
mid=NULL;
ret=Maat_full_scan_string(g_feather, table_id, CHARSET_GBK, should_not_hit, strlen(should_not_hit),
result, NULL, 4, &mid, 0);
EXPECT_EQ(ret, -2);
Maat_clean_status(&mid);
return;
}
@@ -2245,97 +2252,7 @@ TEST_F(MaatCmdTest, SetLines)
return;
}
/*
//Following tests must be coded/tested at last, for they stalled the maat update thread and interrupt other tests.
TEST_F(MaatCmdTest, SetExprOneMillion)
{
const int CMD_EXPR_NUM=1000*1000;
const char* table_name="HTTP_URL";
const char* keywords1="Hiredis";
const char* keywords2="C Client";
char escape_buff1[256],escape_buff2[256];
char keywords[256];
int label_id=5210, config_id=0,ret=0, output_id_cnt=0;
Maat_feather_t feather=MaatCmdTest::_shared_feather;
long long version_before=0;
ret=Maat_read_state(feather,MAAT_STATE_VERSION, &version_before, sizeof(version_before));
Maat_str_escape(escape_buff1, sizeof(escape_buff1),keywords1);
Maat_str_escape(escape_buff2, sizeof(escape_buff2),keywords2);
snprintf(keywords,sizeof(keywords),"%s&%s",escape_buff1,escape_buff2);
config_id=(int)Maat_cmd_incrby(feather, "TEST_SEQ", CMD_EXPR_NUM);
int i=0;
for(i=0; i<CMD_EXPR_NUM;i++)
{
test_add_expr_command(feather,table_name,config_id-i, 0, label_id, keywords);
}
ret=Maat_cmd_commit(feather);
EXPECT_TRUE(ret>=0);
wait_for_cmd_effective(feather, version_before);
struct Maat_cmd_t* cmd=NULL;
struct Maat_rule_t rule;
memset(&rule,0,sizeof(rule));
int *output_ids=(int*)malloc(sizeof(int)*CMD_EXPR_NUM);
output_id_cnt=Maat_cmd_select(feather,label_id, output_ids, CMD_EXPR_NUM);
EXPECT_EQ(output_id_cnt, CMD_EXPR_NUM);
for(i=0; i<CMD_EXPR_NUM;i++)
{
memset(&rule,0,sizeof(rule));
rule.config_id=output_ids[i];
cmd=Maat_create_cmd(&rule, 0);
ret=Maat_cmd_append(feather, cmd, MAAT_OP_DEL);
EXPECT_EQ(ret, 0);
Maat_free_cmd(cmd);
}
ret=Maat_cmd_commit(feather);
EXPECT_EQ(ret, CMD_EXPR_NUM);
free(output_ids);
}
TEST_F(MaatCmdTest, SetLinesOneMillion)
{
const int TEST_CMD_LINE_NUM=1000*1000;
Maat_feather_t feather=MaatCmdTest::_shared_feather;
struct Maat_line_t **p_line=(struct Maat_line_t **)calloc(sizeof(struct Maat_line_t *), TEST_CMD_LINE_NUM);
struct Maat_line_t *line_rule=(struct Maat_line_t *)calloc(sizeof(struct Maat_line_t), TEST_CMD_LINE_NUM);
int i=0;
const char* line="1\t192.168.0.1\t4444444444\t1";
int seq=(int)Maat_cmd_incrby(feather,"TEST_PLUG_SEQ", TEST_CMD_LINE_NUM);
for(i=0;i<TEST_CMD_LINE_NUM;i++)
{
line_rule[i].label_id=0;
line_rule[i].rule_id=seq-i;
line_rule[i].table_name="QD_ENTRY_INFO";
line_rule[i].table_line=line;
// asprintf(&(line_rule[i].table_line),"1\t192.168.0.1\t%d\t1",100+i);
line_rule[i].expire_after=0;
p_line[i]=line_rule+i;
}
long long version_before=0;
Maat_read_state(feather,MAAT_STATE_VERSION, &version_before, sizeof(version_before));
Maat_cmd_set_lines(feather,(const struct Maat_line_t **)p_line,TEST_CMD_LINE_NUM, MAAT_OP_ADD);
wait_for_cmd_effective(feather, version_before);
for(i=0;i<TEST_CMD_LINE_NUM;i++)
{
line_rule[i].table_line=NULL;
}
Maat_cmd_set_lines(feather, (const struct Maat_line_t **)p_line,TEST_CMD_LINE_NUM, MAAT_OP_DEL);
free(p_line);
free(line_rule);
return;
}
*/
int g_test_update_paused=0;
void pause_update_test_entry_cb(int table_id,const char* table_line,void* u_para)
{

4
tools/maat_redis_tool.cpp
View File

@@ -336,6 +336,10 @@ int main(int argc, char * argv[])
printf("Serialize %s to %d lines, write temp file to %s .\n",json_file, total_line_cnt, tmp_iris_path);
struct serial_rule_t * s_rule=(struct serial_rule_t *)calloc(sizeof(struct serial_rule_t),total_line_cnt);
server_time=redis_server_time(ctx);
if(!server_time)
{
printf("Get Redis Time failed.\n");
}
if(timeout>0)
{
absolute_expire_time=server_time+timeout;