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

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

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;
@@ -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;
 	}
 	

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;

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 

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);
 

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)
 {
-	//int ret=set_vlan(NULL, NULL, NULL, 0, NULL, LOG_COMMON_TUNNELS_VLAN_SRC_ID);
+	struct streaminfo a_stream={0};
-	//EXPECT_EQ(1, ret);
+	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)
+{
+	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);