增加发送SESSION-RECORD、INTERIM-SESSION-RECORD、TRANSACTION-RECORD日志的开关v6.0.14

6 files changed, 182 insertions, 9 deletions

diff --git a/bin/main.conf b/bin/main.conf
index a0877bf..6b8e5ed 100644
--- a/bin/main.conf
+++ b/bin/main.conf
@@ -18,6 +18,12 @@ SEND_APP_ID_SWITCH=1
 SEND_NAT_LINKINFO_SWITCH=0
 RAPIDJSON_CHUNK_CAPACITY=8192
 SEND_INTERCEPT_LOG=1
+SEND_INTERIM_RECORD=1
+SEND_TRANSCATION_RECORD=1
+TCP_MIN_PKTS=3
+TCP_MIN_BYTES=5
+UDP_MIN_PKTS=3
+UDP_MIN_BYTES=5
 
 [SECURITY_HITS]
 CYCLE=1000
diff --git a/src/tsg_send_log.cpp b/src/tsg_send_log.cpp
index 8bf86ff..2a95733 100644
--- a/src/tsg_send_log.cpp
+++ b/src/tsg_send_log.cpp
@@ -1808,6 +1808,56 @@ int TLD_append_streaminfo(struct tsg_log_instance_t *instance, struct TLD_handle
 	return 0;
 }
 
+static int session_record_limit(struct tsg_log_instance_t *_instance, const struct streaminfo *a_stream, enum LOG_TYPE log_type)
+{
+	if(tsg_session_record_switch_get()==0)
+	{
+		return 1;
+	}
+	
+	if(log_type==LOG_TYPE_SESSION_RECORD)
+	{
+		if(a_stream==NULL || a_stream->pdetail==NULL)
+		{
+			return 0;
+		}
+
+		switch(a_stream->type)
+		{
+			case	STREAM_TYPE_TCP:
+				if((a_stream->ptcpdetail->clientbytes + a_stream->ptcpdetail->serverbytes < (unsigned int)_instance->tcp_min_log_bytes) ||
+					(a_stream->ptcpdetail->clientpktnum + a_stream->ptcpdetail->serverpktnum < (unsigned int)_instance->tcp_min_log_pkts))
+				{
+					return 1;
+				}
+				break;
+			case	STREAM_TYPE_UDP:
+				if((a_stream->pudpdetail->clientbytes + a_stream->pudpdetail->serverbytes < (unsigned int)_instance->udp_min_log_bytes) ||
+					(a_stream->pudpdetail->clientpktnum + a_stream->pudpdetail->serverpktnum < (unsigned int)_instance->udp_min_log_pkts))
+				{
+					return 1;
+				}
+				break;
+			default:
+				break;
+		}
+		
+		return 0;
+	}
+	
+	if(log_type==LOG_TYPE_INTERIM_SESSION_RECORD && _instance->send_interim_log==0)
+	{
+		return 1;
+	}
+	
+	if(log_type==LOG_TYPE_TRANSACTION_RECORD && _instance->send_transcation_log==0)
+	{
+		return 1;
+	}
+
+	return 0;
+}
+
 int log_common_fields_new(const char *filename, id2field_t *id2field, struct topic_stat **service2topic, int *max_service)
 {
 	int i=0,flag=0;
@@ -1939,9 +1989,9 @@ struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile, screen_stat_ha
 	
 	_instance->sum_line_id=FS_register(_instance->fs2_handle, FS_STYLE_LINE, FS_CALC_SPEED, "SUM");
 
-	MESA_load_profile_int_def(conffile, "TSG_LOG", "LOG_LEVEL",&(_instance->level), 30);	
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "LOG_LEVEL",&(_instance->level), 30);
 	MESA_load_profile_string_def(conffile, "TSG_LOG", "LOG_PATH", log_path, sizeof(log_path), "./log/tsglog");
-	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_USER_REGION", &(_instance->send_user_region), 0);		
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_USER_REGION", &(_instance->send_user_region), 0);
 	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_DATA_CENTER_SWITCH", &(_instance->send_data_center), 0);
 	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_APP_ID_SWITCH", &(_instance->send_app_id), 0);
 	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_INTERCEPT_LOG", &(_instance->send_intercept_log), 0);
@@ -1951,6 +2001,13 @@ struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile, screen_stat_ha
 	MESA_load_profile_int_def(conffile, "TSG_LOG", "VSYSTEM_ID", &(_instance->vsystem_id), 1);
 	MESA_load_profile_int_def(conffile, "SYSTEM","UNKNOWN_APP_ID", &_instance->unknown_app_id, 4);
 
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_INTERIM_RECORD", &(_instance->send_interim_log), 1);
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_TRANSCATION_RECORD", &(_instance->send_transcation_log), 1);
+	MESA_load_profile_int_def(conffile, "TSG_LOG","TCP_MIN_PKTS", &_instance->tcp_min_log_pkts, 3);
+	MESA_load_profile_int_def(conffile, "TSG_LOG","TCP_MIN_BYTES", &_instance->tcp_min_log_bytes, 5);
+	MESA_load_profile_int_def(conffile, "TSG_LOG","UDP_MIN_PKTS", &_instance->udp_min_log_pkts, 3);
+	MESA_load_profile_int_def(conffile, "TSG_LOG","UDP_MIN_BYTES", &_instance->udp_min_log_bytes, 5);
+
 	_instance->logger=MESA_create_runtime_log_handle(log_path, _instance->level);
 	if(_instance->logger==NULL)
 	{
@@ -2282,7 +2339,7 @@ int tsg_send_log(struct tsg_log_instance_t *instance, struct TLD_handle_t *handl
 			}
 			// no break;
 		case	LOG_TYPE_INTERIM_SESSION_RECORD:
-			if(tsg_session_record_switch_get()==0)
+			if(session_record_limit(_instance, a_stream, log_type))
 			{
 				break;
 			}
@@ -2302,7 +2359,6 @@ int tsg_send_log(struct tsg_log_instance_t *instance, struct TLD_handle_t *handl
 			send_log_by_type(_instance, _handle, a_stream, log_type, thread_id);
 			break;
 		default:
-			TLD_cancel(handle);
 			return 0;
 	}
 	
diff --git a/src/tsg_send_log_internal.h b/src/tsg_send_log_internal.h
index bbadd9e..1307fb5 100644
--- a/src/tsg_send_log_internal.h
+++ b/src/tsg_send_log_internal.h
@@ -203,6 +203,12 @@ struct tsg_log_instance_t
 	int max_service;
 	int vsystem_id;
 	int unknown_app_id;
+	int tcp_min_log_pkts;
+	int tcp_min_log_bytes;
+	int udp_min_log_pkts;
+	int udp_min_log_bytes;
+	int send_interim_log;
+	int send_transcation_log;
 	int send_user_region;
 	int send_app_id;
 	int send_intercept_log;
diff --git a/test/src/CMakeLists.txt b/test/src/CMakeLists.txt
index c3905d8..fc60733 100644
--- a/test/src/CMakeLists.txt
+++ b/test/src/CMakeLists.txt
@@ -22,7 +22,7 @@ target_link_libraries(gtest_bridge gtest-static ctemplate-static cjson MESA_prof
 add_executable(gtest_action ${PROJECT_SOURCE_DIR}/src/tsg_action.cpp ${PROJECT_SOURCE_DIR}/src/tsg_leaky_bucket.cpp ${PROJECT_SOURCE_DIR}/src/tsg_dns.cpp ${PROJECT_SOURCE_DIR}/src/tsg_icmp.cpp ${PROJECT_SOURCE_DIR}/src/tsg_tamper.cpp ${PROJECT_SOURCE_DIR}/src/tsg_variable.cpp gtest_common.cpp gtest_action.cpp)
 target_link_libraries(gtest_action gtest-static ctemplate-static cjson MESA_prof_load MESA_handle_logger maat4 MESA_field_stat2)
 
-add_executable(gtest_sendlog ${PROJECT_SOURCE_DIR}/src/tsg_send_log.cpp ${PROJECT_SOURCE_DIR}/src/tsg_variable.cpp gtest_common.cpp gtest_sendlog.cpp)
+add_executable(gtest_sendlog ${PROJECT_SOURCE_DIR}/src/tsg_send_log.cpp ${PROJECT_SOURCE_DIR}/src/tsg_variable.cpp gtest_common.cpp gtest_kafka.cpp gtest_sendlog.cpp)
 target_link_libraries(gtest_sendlog gtest-static ctemplate-static cjson MESA_prof_load MESA_handle_logger maat4 rdkafka MESA_field_stat2)
 
 set(TSG_MASTER_SRC ${PROJECT_SOURCE_DIR}/src/tsg_entry.cpp 
diff --git a/test/src/gtest_kafka.h b/test/src/gtest_kafka.h
index 45dcbb0..f4015de 100644
--- a/test/src/gtest_kafka.h
+++ b/test/src/gtest_kafka.h
@@ -1,3 +1,6 @@
 #pragma once
 
+extern int rd_kafka_get_sendlog_cnt(void);
+extern void rd_kafka_clean_sendlog_cnt(void);
+extern const char *rd_kafka_get_sendlog_payload(int idx);
 
diff --git a/test/src/gtest_sendlog.cpp b/test/src/gtest_sendlog.cpp
index b7d1214..b13cf1f 100644
--- a/test/src/gtest_sendlog.cpp
+++ b/test/src/gtest_sendlog.cpp
@@ -2,14 +2,22 @@
 #include <string.h>

 #include <unistd.h>

 

+#include <MESA/field_stat2.h>

+#include <MESA/stream.h>

+

 #include "tsg_rule.h"

+#include "tsg_send_log.h"

+#include "tsg_send_log_internal.h"

 

 #include "gtest_common.h"

+#include "gtest_kafka.h"

 

 #include <gtest/gtest.h>

 

 struct maat *g_tsg_maat_feather;

 

+extern struct tsg_log_instance_t *tsg_sendlog_init(const char * conffile, screen_stat_handle_t fs2_handle);

+

 char *tsg_device_tag_get(void)

 {

 	return NULL;

@@ -27,7 +35,7 @@ int tsg_location_type_get(void)
 

 int tsg_session_record_switch_get(void)

 {

-	return 0;

+	return 1;

 }

 

 void *session_mac_linkinfo_get(const struct streaminfo * a_stream)

@@ -127,14 +135,108 @@ int tsg_set_policy_flow(const struct streaminfo * a_stream, struct maat_rule * p
 	return 0;

 }

 

-TEST(MasterTest, SetVlan)

+TEST(Master, SendInterimRecord)

+{

+	struct streaminfo a_stream={0};

+	struct tcpdetail pdetail={NULL, 0, 0, 3, 50, 3, 50, 0, 1};

+	a_stream.ptcpdetail=&pdetail;

+	a_stream.type=STREAM_TYPE_TCP;

+

+	struct maat_rule rules={0, 0, 0, 2, 0, 1};

+

+	struct TLD_handle_t * handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_INTERIM_SESSION_RECORD, &rules, 1, 0);

+	int sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(1, sendlog_cnt);

+	EXPECT_STREQ("{\"common_server_port\":0,\"common_client_port\":0,\"common_stream_dir\":0,\"common_address_type\":0,\"common_stream_trace_id\":\"5\",\"common_sled_ip\":\"0.0.0.0\",\"common_t_vsys_id\":1,\"common_vsys_id\":1}",rd_kafka_get_sendlog_payload(0));

+	rd_kafka_clean_sendlog_cnt();

+	EXPECT_EQ(0, rd_kafka_get_sendlog_cnt());

+

+	g_tsg_log_instance->send_interim_log=0;

+	handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_INTERIM_SESSION_RECORD, &rules, 1, 0);

+	sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(0, sendlog_cnt);

+}

+

+TEST(Master, SendTranscationRecord)

+{

+	struct streaminfo a_stream={0};

+	struct tcpdetail pdetail={NULL, 0, 0, 3, 50, 3, 50, 0, 1};

+	a_stream.ptcpdetail=&pdetail;

+	a_stream.type=STREAM_TYPE_TCP;

+

+	struct maat_rule rules={0, 0, 0, 2, 0, 1};

+

+	struct TLD_handle_t * handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_TRANSACTION_RECORD, &rules, 1, 0);

+	int sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(1, sendlog_cnt);

+	EXPECT_STREQ("{\"common_server_port\":0,\"common_client_port\":0,\"common_stream_dir\":0,\"common_address_type\":0,\"common_stream_trace_id\":\"5\",\"common_sled_ip\":\"0.0.0.0\",\"common_t_vsys_id\":1,\"common_vsys_id\":1}",rd_kafka_get_sendlog_payload(0));

+	rd_kafka_clean_sendlog_cnt();

+	EXPECT_EQ(0, rd_kafka_get_sendlog_cnt());

+

+	g_tsg_log_instance->send_transcation_log=0;

+	handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_TRANSACTION_RECORD, &rules, 1, 0);

+	sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(0, sendlog_cnt);

+}

+

+TEST(Master, SendSessionRecord)

 {

-	//int ret=set_vlan(NULL, NULL, NULL, 0, NULL, LOG_COMMON_TUNNELS_VLAN_SRC_ID);

-	//EXPECT_EQ(1, ret);

+	struct streaminfo a_stream={0};

+	struct tcpdetail pdetail={NULL, 0, 0, 3, 50, 3, 50, 0, 1};

+	a_stream.ptcpdetail=&pdetail;

+	a_stream.type=STREAM_TYPE_TCP;

+

+	struct maat_rule rules={0, 0, 0, 2, 0, 1};

+

+	struct TLD_handle_t * handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_SESSION_RECORD, &rules, 1, 0);

+	int sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(1, sendlog_cnt);

+	EXPECT_STREQ("{\"common_server_port\":0,\"common_client_port\":0,\"common_stream_dir\":0,\"common_address_type\":0,\"common_stream_trace_id\":\"5\",\"common_sled_ip\":\"0.0.0.0\",\"common_t_vsys_id\":1,\"common_vsys_id\":1}",rd_kafka_get_sendlog_payload(0));

+	rd_kafka_clean_sendlog_cnt();

+	EXPECT_EQ(0, rd_kafka_get_sendlog_cnt());

+

+	//pkts=3, bytes=0

+	pdetail={NULL, 0, 0, 0, 0, 3, 0, 0, 1};

+	a_stream.ptcpdetail=&pdetail;

+	a_stream.type=STREAM_TYPE_TCP;


+	handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_SESSION_RECORD, &rules, 1, 0);

+	sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(0, sendlog_cnt);

+

+	//pkts=2, bytes=1500

+	pdetail={NULL, 0, 0, 0, 0, 2, 1500, 0, 1};

+	a_stream.ptcpdetail=&pdetail;

+	a_stream.type=STREAM_TYPE_TCP;


+	handle=TLD_create(0);

+	tsg_send_log(g_tsg_log_instance, handle, &a_stream, LOG_TYPE_SESSION_RECORD, &rules, 1, 0);

+	sendlog_cnt=rd_kafka_get_sendlog_cnt();

+	EXPECT_EQ(0, sendlog_cnt);

 }

 

 int main(int argc, char *argv[])

 {

+	screen_stat_handle_t fs2_handle=FS_create_handle();

+

+	int value=0,cycle=30;

+	value=1;//Rewrite

+	FS_set_para(fs2_handle, PRINT_MODE, &value, sizeof(value));

+	value=1;//Do not create stat thread

+	FS_set_para(fs2_handle, CREATE_THREAD, &value, sizeof(value));

+	FS_set_para(fs2_handle, STAT_CYCLE, &cycle, sizeof(cycle));

+	FS_set_para(fs2_handle, APP_NAME, (char *)"test", strlen((char *)"test")+1);

+

+	value=FS_OUTPUT_INFLUX_LINE;

+	FS_set_para(fs2_handle, STATS_FORMAT, &value, sizeof(value));

+

+	g_tsg_log_instance=tsg_sendlog_init("./tsgconf/main.conf", fs2_handle);

+	FS_start(fs2_handle);

+	

 	testing::InitGoogleTest(&argc, argv);

 	return RUN_ALL_TESTS();

 }