同步发送日志调整为异步发送日志feature-aync-sendlog-ringbuff

3 files changed, 581 insertions, 367 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index d48f4a5..6b371b7 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -2,7 +2,25 @@ cmake_minimum_required(VERSION 2.8...3.10)
 
 add_definitions(-fPIC)
 
-set(SRC tsg_entry.cpp tsg_rule.cpp tsg_ssl_utils.cpp tsg_send_log.cpp tsg_statistic.cpp tsg_ssh_utils.cpp tsg_gtp_signaling.cpp tsg_action.cpp tsg_leaky_bucket.cpp tsg_dns.cpp tsg_icmp.cpp tsg_tamper.cpp tsg_bridge.cpp tsg_sync_state.cpp tsg_variable.cpp tsg_proxy.cpp mpack.c tsg_stat.cpp tsg_ssl_ja3_fingerprint.cpp)
+set(SRC tsg_entry.cpp 
+        tsg_rule.cpp 
+        tsg_ssl_utils.cpp 
+        tsg_send_log.cpp 
+        tsg_statistic.cpp 
+        tsg_ssh_utils.cpp 
+        tsg_gtp_signaling.cpp 
+        tsg_action.cpp 
+        tsg_leaky_bucket.cpp 
+        tsg_dns.cpp tsg_icmp.cpp 
+        tsg_tamper.cpp tsg_bridge.cpp 
+        tsg_sync_state.cpp 
+        tsg_variable.cpp 
+        tsg_proxy.cpp 
+        mpack.c 
+        tsg_stat.cpp 
+        tsg_ssl_ja3_fingerprint.cpp
+        ${PROJECT_SOURCE_DIR}/deps/ringbuf/ringbuf.c
+    )
 
 include_directories(${CMAKE_SOURCE_DIR}/inc)
 include_directories(/opt/MESA/include/MESA/)
diff --git a/src/tsg_send_log.cpp b/src/tsg_send_log.cpp
index 62bb7e2..3d8bbee 100644
--- a/src/tsg_send_log.cpp
+++ b/src/tsg_send_log.cpp
@@ -56,6 +56,21 @@ const id2field_t tld_type[TLD_TYPE_MAX]={{TLD_TYPE_UNKNOWN,	TLD_TYPE_UNKNOWN, 	"
 
 extern "C" int MESA_get_dev_ipv4(const char *device, int *ip_add);
 
+char *tsg_string_dup(char *string)
+{
+	if(string==NULL)
+	{
+		return NULL;
+	}
+
+	size_t len=strlen(string);
+	char *dup=(char *)malloc(len+1);
+	memcpy(dup, string, len);
+	dup[len]='\0';
+
+	return dup;
+}
+
 static int string_cat(char *dst, int dst_len, char *src)
 {
 	if(dst==NULL || dst_len<=0 || src==NULL)
@@ -1195,48 +1210,6 @@ int TLD_convert_json(struct TLD_handle_t *_handle, char *buff, unsigned int buff
 	return 1;
 }
 
-static int set_mail_eml(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream)
-{
-	struct tsg_conn_sketch_notify_data *notify_mail=(struct tsg_conn_sketch_notify_data *)session_conn_sketch_notify_data_get(a_stream);
-	if(notify_mail!=NULL && notify_mail->pdata.mail_eml_filename!=NULL && notify_mail->protocol==PROTO_MAIL)
-	{
-		TLD_delete(_handle, _instance->id2field[LOG_COMMON_MAIL_EML_FILE].name);
-		TLD_append(_handle, _instance->id2field[LOG_COMMON_MAIL_EML_FILE].name, (void *)notify_mail->pdata.mail_eml_filename, TLD_TYPE_STRING);
-		return 1;
-	}
-
-	return 0;
-}
-
-static int set_s3_filename(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream)
-{
-	struct business_notify_data *bnd_label=(struct business_notify_data *)session_business_data_get(a_stream);
-	if(bnd_label==NULL || bnd_label->pdata==NULL)
-	{
-		return 0;
-	}
-	
-	switch(bnd_label->proto)
-	{
-		case	PROTO_HTTP:
-			TLD_delete(_handle, _instance->id2field[LOG_COMMON_HTTP_REQUEST_S3_FILE].name);
-			TLD_delete(_handle, _instance->id2field[LOG_COMMON_HTTP_RESPONSE_S3_FILE].name);
-
-			if(bnd_label->s3_http==NULL)
-			{
-				break;
-			}
-
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_HTTP_REQUEST_S3_FILE].name, bnd_label->s3_http->request_filename, TLD_TYPE_STRING);
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_HTTP_RESPONSE_S3_FILE].name, bnd_label->s3_http->response_filename, TLD_TYPE_STRING);
-			break;
-		default:
-			break;
-	}
-
-	return 1;
-}
-
 int set_nat_linkinfo(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream)
 {
 	if(_instance->send_nat_linkinfo==0 || a_stream==NULL)
@@ -1578,14 +1551,11 @@ static int is_hitted_shunt(const struct streaminfo *a_stream)
 	return 0;
 }
 
-static int set_xxxx_from_user_region(struct TLD_handle_t *_handle, struct tsg_log_instance_t *_instance, struct maat_rule *p_result, int thread_seq)
+static int set_xxxx_from_user_region(struct TLD_handle_t *_handle, struct tsg_log_instance_t *_instance, struct maat_rule *p_result)
 {
 	cJSON *item=NULL;
 	cJSON *object=NULL;
 
-	TLD_delete(_handle, _instance->id2field[LOG_COMMON_USER_REGION].name);
-	TLD_delete(_handle, _instance->id2field[LOG_COMMON_SUB_ACTION].name);
-
 	if(p_result->action!=TSG_ACTION_NONE)
 	{
 		struct maat_compile *compile=(struct maat_compile *)matched_rule_cites_security_compile(g_tsg_maat_feather, p_result->rule_id);
@@ -1630,40 +1600,6 @@ int set_application_behavior(struct tsg_log_instance_t *_instance, struct TLD_ha
 	return 1;
 }
 
-int set_policy_action_para_exec_result(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream, struct maat_rule *p_result)
-{
-	struct tsg_notify_execution_result *execution_result=(struct tsg_notify_execution_result *)session_mirrored_and_capture_packets_exec_result_get(a_stream);
-	if(execution_result==NULL)
-	{
-		return 0;
-	}
-
-	TLD_delete(_handle, _instance->id2field[LOG_COMMON_MIRRORED_PKTS].name);
-	TLD_delete(_handle, _instance->id2field[LOG_COMMON_MIRRORED_BYTES].name);
-	TLD_delete(_handle, _instance->id2field[LOG_COMMON_PACKET_CAPTURE_FILE].name);
-	
-	for(int i=0; i<execution_result->stat_mirrored_cnt; i++)
-	{
-		if(execution_result->stat_mirrored[i].compile_id==p_result->rule_id)
-		{
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_MIRRORED_PKTS].name, (void *)(execution_result->stat_mirrored[i].packets), TLD_TYPE_LONG);
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_MIRRORED_BYTES].name, (void *)(execution_result->stat_mirrored[i].bytes), TLD_TYPE_LONG);
-			break;
-		}
-	}
-
-	for(int i=0; i<execution_result->capture_result_cnt; i++)
-	{		
-		if(execution_result->capture_result[i].compile_id==p_result->rule_id)
-		{	
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_PACKET_CAPTURE_FILE].name, (void *)(execution_result->capture_result[i].packet_path), TLD_TYPE_STRING);
-			break;
-		}
-	}
-	
-	return 1;
-}
-
 int set_session_attributes(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream)
 {
 	struct session_runtime_attribute *srt_attribute=(struct session_runtime_attribute *)session_runtime_attribute_get(a_stream);
@@ -1867,24 +1803,6 @@ static int session_record_limit(struct tsg_log_instance_t *_instance, const stru
 	return 0;
 }
 
-int append_common_field(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream)
-{
-	TLD_append_streaminfo(_instance, _handle, a_stream);
-	TLD_append(_handle, _instance->id2field[LOG_COMMON_SLED_IP].name, (void *)(_instance->local_ip_str), TLD_TYPE_STRING);
-	if(strlen(g_tsg_para.device_sn)>0)
-	{
-		TLD_append(_handle, _instance->id2field[LOG_COMMON_DEVICE_ID].name, (void *)(g_tsg_para.device_sn), TLD_TYPE_STRING);
-	}
-
-	TLD_append(_handle, _instance->id2field[LOG_COMMON_DATA_CENTER].name, (void *)tsg_data_center_get(), TLD_TYPE_STRING);
-	TLD_append(_handle, _instance->id2field[LOG_COMMON_DEVICE_TAG].name, (void *)tsg_device_tag_get(), TLD_TYPE_STRING);
-	TLD_append(_handle, _instance->id2field[LOG_COMMON_TRAFFIC_VSYSTEM_ID].name, (void *)(long)_instance->vsystem_id, TLD_TYPE_LONG);
-
-	set_application_behavior(_instance, _handle, a_stream);
-
-	return 1;
-}
-
 int log_common_fields_new(const char *filename, id2field_t *id2field, struct topic_stat **service2topic, int *max_service)
 {
 	int i=0,flag=0;
@@ -1987,221 +1905,271 @@ int log_common_fields_new(const char *filename, id2field_t *id2field, struct top
 	return 0;
 }
 
-struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile)
+void tsg_enforce_rule_result_free(struct enforce_rule_result *enforce_rule, size_t n_enforce_rule)
 {
-	char override_sled_ip[32]={0};
-	char kafka_errstr[1024]={0};
-	unsigned int local_ip_nr=0;
-	rd_kafka_conf_t *rdkafka_conf = NULL;
-	char broker_list[1024]={0};
-	struct tsg_log_instance_t *_instance=NULL;
-	char common_field_file[128]={0};
-	char log_path[128]={0};
-
-	_instance=(struct tsg_log_instance_t *)calloc(1, sizeof(struct tsg_log_instance_t));
+	for(size_t i=0; i<n_enforce_rule && enforce_rule!=NULL; i++)
+	{
+		if(enforce_rule[i].mail_eml_filename!=NULL)
+		{
+			free(enforce_rule[i].mail_eml_filename);
+			enforce_rule[i].mail_eml_filename=NULL;
+		}
 
-	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_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);
-	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_NAT_LINKINFO_SWITCH", &(_instance->send_nat_linkinfo), 0);
-	MESA_load_profile_int_def(conffile, "TSG_LOG", "RAPIDJSON_CHUNK_CAPACITY", &(_instance->rapidjson_chunk_capacity), 2048);
+		if(enforce_rule[i].http_request_filename!=NULL)
+		{
+			free(enforce_rule[i].http_request_filename);
+			enforce_rule[i].http_request_filename=NULL;
+		}
 
-	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);
+		if(enforce_rule[i].http_response_filename!=NULL)
+		{
+			free(enforce_rule[i].http_response_filename);
+			enforce_rule[i].http_response_filename=NULL;
+		}
 
-	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)
-	{
-		printf("MESA_create_runtime_log_handle failed ..., path: %s level: %d", log_path, _instance->level);
-		free(_instance);
-		_instance=NULL;
-		return NULL;
+		if(enforce_rule[i].packet_capture_file!=NULL)
+		{
+			free(enforce_rule[i].packet_capture_file);
+			enforce_rule[i].packet_capture_file=NULL;
+		}
 	}
-	
-	MESA_load_profile_int_def(conffile, "TSG_LOG", "MODE",&(_instance->mode), 0);
-	if(_instance->mode==CLOSE)
+}
+
+std::string tsg_rapidjson_serialize(Document *document, rapidjson::StringBuffer& sbuff) 
+{
+    sbuff.Clear();
+    rapidjson::Writer<rapidjson::StringBuffer> writer(sbuff);
+    document->Accept(writer);
+    return std::string(sbuff.GetString());
+}
+
+void *tsg_log_ringbuf_consume(void *arg)
+{
+	struct logger_ringbuf_schema *ringbuf=(struct logger_ringbuf_schema *)arg;
+	rapidjson::StringBuffer sbuff(0, 1024*16);
+
+	while(1)
 	{
-		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, "Disable tsg_send_log");
-		return _instance;
-	}
-	
-	MESA_load_profile_int_def(conffile, "TSG_LOG", "RECOVERY_INTERVEL_S", &(_instance->recovery_interval), 30);	
+		size_t offset=0;
+		size_t scratch_sz=ringbuf_consume(ringbuf->ring, &offset);
+		if(scratch_sz==0)
+		{
+			tsg_stat_sendlog_update(ringbuf->consume_row_idx, LOG_STATUS_FAIL, 1);
+			continue;
+		}
+		
+		struct logger_scratch *scratch=(struct logger_scratch *)(ringbuf->buff+offset);
 
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "COMMON_FIELD_FILE", common_field_file, sizeof(common_field_file), NULL);
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "BROKER_LIST", broker_list, sizeof(broker_list), NULL);
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "SASL_USERNAME", _instance->sasl_username, sizeof(_instance->sasl_username), ""); //admin
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "SASL_PASSWD", _instance->sasl_passwd, sizeof(_instance->sasl_passwd), "");
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "COMPRESSION_TYPE", _instance->compression, sizeof(_instance->compression), ""); //snappy
+		TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_VSYSTEM_ID].name);
 
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "SEND_QUEUE_MAX_MESSAGE", _instance->send_queue_max_msg, sizeof(_instance->send_queue_max_msg), "1000000");
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "REFRESH_INTERVAL_MS", _instance->refresh_interval_ms, sizeof(_instance->refresh_interval_ms), "600000");
-	MESA_load_profile_string_def(conffile, "TSG_LOG", "REQUIRE_ACK", _instance->require_ack, sizeof(_instance->require_ack), "1");
+		for(int i=0; i<scratch->n_enforce_rule && scratch->enforce_rule!=NULL; i++)
+		{
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_POLICY_ID].name, (void *)(long)(scratch->enforce_rule[i].rule.rule_id), TLD_TYPE_LONG);
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_SERVICE].name, (void *)(long)(scratch->enforce_rule[i].rule.service_id), TLD_TYPE_LONG);
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_VSYSTEM_ID].name, (void *)(long)scratch->enforce_rule[i].rule.vsys_id, TLD_TYPE_LONG);
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_ACTION].name, (void *)(long)(scratch->enforce_rule[i].rule.action), TLD_TYPE_LONG);
 
-	MESA_load_profile_string_def(conffile, "SYSTEM", "TCP_LABEL", _instance->tcp_label, sizeof(_instance->tcp_label), "tcp_flow_stat");
-	MESA_load_profile_string_def(conffile, "SYSTEM", "UDP_LABEL", _instance->udp_label, sizeof(_instance->udp_label), "udp_flow_stat");
+			if(scratch->enforce_rule[i].mirrored_packets>0)
+			{
+				TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_MIRRORED_PKTS].name, (void *)(scratch->enforce_rule[i].mirrored_packets), TLD_TYPE_LONG);
+			}
 
-	_instance->tcp_flow_project_id=project_customer_register(_instance->tcp_label, "struct");
-	_instance->udp_flow_project_id=project_customer_register(_instance->udp_label, "struct");
-	if(_instance->tcp_flow_project_id<0 || _instance->udp_flow_project_id<0)
-	{
-		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, 
-									"project_customer_register is error, tcp_label: %s udp_label: %s, please check etc/project.conf",
-									_instance->tcp_label,
-									_instance->udp_label
-								);
-	}
+			if(scratch->enforce_rule[i].mirrored_bytes>0)
+			{
+				TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_MIRRORED_BYTES].name, (void *)(scratch->enforce_rule[i].mirrored_bytes), TLD_TYPE_LONG);
+			}
 
-	MESA_load_profile_string_def(conffile, "SYSTEM", "OVERRIDE_SLED_IP", override_sled_ip, sizeof(override_sled_ip), "OVERRIDE_SLED_IP");
-	char *sled_ip=getenv(override_sled_ip);
-	if(sled_ip==NULL)
-	{
-		char nic_name[32]={0};
-		MESA_load_profile_string_def(conffile, "SYSTEM", "NIC_NAME", nic_name, sizeof(nic_name), "lo");	
-		int ret=MESA_get_dev_ipv4(nic_name, (int *)&local_ip_nr);
-		if(ret<0)
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_MAIL_EML_FILE].name, (void *)(scratch->enforce_rule[i].mail_eml_filename), TLD_TYPE_STRING);
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_HTTP_REQUEST_S3_FILE].name, (void *)(scratch->enforce_rule[i].http_request_filename), TLD_TYPE_STRING);
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_HTTP_RESPONSE_S3_FILE].name, (void *)(scratch->enforce_rule[i].http_response_filename), TLD_TYPE_STRING);
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_PACKET_CAPTURE_FILE].name, (void *)(scratch->enforce_rule[i].packet_capture_file), TLD_TYPE_STRING);
+
+			set_xxxx_from_user_region(scratch->handle, scratch->instance, &(scratch->enforce_rule[i].rule));
+			
+			std::string sbuff_str=tsg_rapidjson_serialize(scratch->handle->document, sbuff);
+			tsg_send_payload(scratch->instance, scratch->enforce_rule[i].log_type, (char *)sbuff_str.c_str(), sbuff_str.length(), scratch->thread_id);
+
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_POLICY_ID].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_SERVICE].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_ACTION].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_USER_REGION].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_VSYSTEM_ID].name);
+			
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_MIRRORED_PKTS].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_MIRRORED_BYTES].name);
+
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_MAIL_EML_FILE].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_HTTP_REQUEST_S3_FILE].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_HTTP_RESPONSE_S3_FILE].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_PACKET_CAPTURE_FILE].name);
+
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_USER_REGION].name);
+			TLD_delete(scratch->handle, scratch->instance->id2field[LOG_COMMON_SUB_ACTION].name);
+
+			fs3_rule_stat_update(RULE_STAT_SEND, (int)(scratch->enforce_rule[i].rule.action), 1);
+		}
+
+		if(scratch->log_type!=LOG_TYPE_MAX)
 		{
-			MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, 
-										"GET_LOCAL_IP MESA_get_dev_ipv4 is error, nic_name: %s, please check tsgconf/main.conf",
-										nic_name
-									);
-			return NULL;
+			TLD_append(scratch->handle, scratch->instance->id2field[LOG_COMMON_VSYSTEM_ID].name, (void *)(long)scratch->instance->vsystem_id, TLD_TYPE_LONG);
+			std::string sbuff_str=tsg_rapidjson_serialize(scratch->handle->document, sbuff);
+			tsg_send_payload(scratch->instance, scratch->log_type, (char *)sbuff_str.c_str(), sbuff_str.length(), scratch->thread_id);
 		}
-		inet_ntop(AF_INET,&(local_ip_nr),_instance->local_ip_str,sizeof(_instance->local_ip_str));
-	}
-	else
-	{
-		memcpy(_instance->local_ip_str, sled_ip, MIN(sizeof(_instance->local_ip_str)-1, strlen(sled_ip)));
-	}
 	
-	rdkafka_conf = rd_kafka_conf_new();	
-	rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", _instance->send_queue_max_msg, kafka_errstr, sizeof(kafka_errstr));
-	rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", _instance->refresh_interval_ms, kafka_errstr, sizeof(kafka_errstr));
-	rd_kafka_conf_set(rdkafka_conf, "request.required.acks", _instance->require_ack, kafka_errstr, sizeof(kafka_errstr));	
-	rd_kafka_conf_set(rdkafka_conf, "socket.keepalive.enable", "true", kafka_errstr, sizeof(kafka_errstr));	
-	rd_kafka_conf_set(rdkafka_conf, "bootstrap.servers", broker_list, kafka_errstr, sizeof(kafka_errstr));
-	
-	if(strlen(_instance->compression)>0)
-	{
-		rd_kafka_conf_set(rdkafka_conf, "compression.codec", _instance->compression, kafka_errstr, sizeof(kafka_errstr));
-	}
+		tsg_stat_sendlog_update(ringbuf->consume_row_idx, LOG_STATUS_SUCCESS, 1);
 
-	if(strlen(_instance->sasl_username)> 0 && strlen(_instance->sasl_passwd)>0)
-	{
-		rd_kafka_conf_set(rdkafka_conf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr));
-		rd_kafka_conf_set(rdkafka_conf, "sasl.mechanisms", "PLAIN", kafka_errstr, sizeof(kafka_errstr));
-		rd_kafka_conf_set(rdkafka_conf, "sasl.username", _instance->sasl_username, kafka_errstr, sizeof(kafka_errstr));
-		rd_kafka_conf_set(rdkafka_conf, "sasl.password", _instance->sasl_passwd, kafka_errstr, sizeof(kafka_errstr));
+		TLD_cancel(scratch->handle);
+		if(scratch->enforce_rule!=NULL)
+		{
+			tsg_enforce_rule_result_free(scratch->enforce_rule, scratch->n_enforce_rule);
+			free(scratch->enforce_rule);
+		}
+		
+		memset(scratch, 0, sizeof(struct logger_scratch));
+		ringbuf_release(ringbuf->ring, sizeof(struct logger_scratch));
 	}
-	
-	if(!(_instance->kafka_handle=rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
+
+	pthread_exit(NULL);
+}
+
+void tsg_log_ringbuf_produce(struct tsg_log_instance_t *instance, struct TLD_handle_t *handle, enum LOG_TYPE log_type, struct enforce_rule_result *enforce_rule, size_t n_enforce_rule, int thread_id)
+{
+	int i=(thread_id%instance->ringbuf_num);
+	struct logger_ringbuf_schema *ringbuf=&(instance->ringbuf[i]);
+	ssize_t offset=ringbuf_acquire(ringbuf->ring, ringbuf->worker[thread_id], sizeof(struct logger_scratch));
+	if(offset==-1)
 	{
-		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, "KAFKA_INIT rd_kafka_new is error");
-		return NULL;
+		TLD_cancel(handle);
+		tsg_enforce_rule_result_free(enforce_rule, n_enforce_rule);
+		tsg_stat_sendlog_update(ringbuf->produce_row_idx, LOG_STATUS_FAIL, 1);
+		return;
 	}
 
-	log_common_fields_new(common_field_file, _instance->id2field, &(_instance->service2topic), &(_instance->max_service));
+	struct logger_scratch *scratch=(struct logger_scratch *)(ringbuf->buff+offset);
+	scratch->handle=handle;
+	scratch->instance=instance;
+	scratch->log_type=log_type;
+	scratch->thread_id=thread_id;
 
-	if(_instance->service2topic!=NULL)
+	if(enforce_rule!=NULL && n_enforce_rule>0)
 	{
-		_instance->sum_stat_row_id=tsg_stat_sendlog_row_init("sum");
-		for(int i=0; i<_instance->max_service; i++)
-		{
-			if(_instance->service2topic[i].type==TLD_TYPE_MAX && strlen(_instance->service2topic[i].name)>0)
-			{
-				register_topic(_instance, &( _instance->service2topic[i]));
-			}
-		}
+		scratch->n_enforce_rule=MIN(n_enforce_rule, LOGGER_SCRATCH_RULE_MAX);
+		scratch->enforce_rule=(struct enforce_rule_result *)calloc(scratch->n_enforce_rule, sizeof(struct enforce_rule_result));
+		memcpy(scratch->enforce_rule, enforce_rule, sizeof(struct enforce_rule_result)*scratch->n_enforce_rule);
 	}
 	else
 	{
-		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, "KAFKA_INIT log_common_fields_new is error, please check %s", common_field_file);
+		scratch->n_enforce_rule=0;
+		scratch->enforce_rule=NULL;
 	}
 	
-	return _instance;
+	ringbuf_produce(ringbuf->ring, ringbuf->worker[thread_id]);
+	tsg_stat_sendlog_update(ringbuf->produce_row_idx, LOG_STATUS_SUCCESS, 1);
 }
-void tsg_sendlog_destroy(struct tsg_log_instance_t * instance)
+
+void tsg_rule_log_enable_additional_set(const struct streaminfo *a_stream, struct enforce_rule_result *enforce_rule, size_t n_enforce_rule)
 {
-	if(instance==NULL)
-	{
-		return ;
-	}
-	
-	if(instance->mode!=CLOSE)
+	struct tsg_notify_execution_result *execution_result=(struct tsg_notify_execution_result *)session_mirrored_and_capture_packets_exec_result_get(a_stream);
+
+	for(size_t i=0;i<n_enforce_rule; i++)
 	{
-		for(int i=0; i<instance->max_service; i++)
+		if(enforce_rule[i].rule.do_log==LOG_ABORT)
 		{
-			if(instance->service2topic[i].type!=TLD_TYPE_MAX || i==1)  //i=1 equal i=0, service id of security event is 0 and 1
- 			{
-				continue;
-			}
-			
-			if(instance->service2topic[i].topic_rkt!=NULL)
-			{
-				rd_kafka_topic_destroy(instance->service2topic[i].topic_rkt);
-			}
+			continue;
+		}
 
-			if(instance->service2topic[i].drop_start!=NULL)
+		if(execution_result!=NULL)
+		{
+			for(int j=0; j<execution_result->stat_mirrored_cnt; j++)
 			{
-				free(instance->service2topic[i].drop_start);
-				instance->service2topic[i].drop_start=NULL;
+				if(execution_result->stat_mirrored[j].compile_id==enforce_rule[i].rule.rule_id)
+				{
+					enforce_rule[i].mirrored_bytes=execution_result->stat_mirrored[j].bytes;
+					enforce_rule[i].mirrored_packets=execution_result->stat_mirrored[j].packets;
+					break;
+				}
 			}
 
-			if(instance->service2topic[i].send_log_percent!=NULL)
-			{
-				free(instance->service2topic[i].send_log_percent);
-				instance->service2topic[i].send_log_percent=NULL;
+			for(int j=0; j<execution_result->capture_result_cnt; j++)
+			{		
+				if(execution_result->capture_result[j].compile_id!=enforce_rule[i].rule.rule_id)
+				{
+					continue;
+				}
+
+				if(execution_result->capture_result[j].packet_path!=NULL)
+				{
+					enforce_rule[i].packet_capture_file=tsg_string_dup(execution_result->capture_result[j].packet_path);
+				}
+				
+				break;
 			}
 		}
-		
-		//rd_kafka_destroy_flags(instance->kafka_handle, 4);
-		rd_kafka_destroy(instance->kafka_handle);
-
-		free(instance->service2topic);
-		instance->service2topic=NULL;
-	}
 
-	
-	MESA_destroy_runtime_log_handle(instance->logger);
-	instance->logger=NULL;
+		if(enforce_rule[i].rule.do_log==LOG_NOFILE)
+		{
+			continue;
+		}
 
-	free(instance);
-	instance=NULL;
-	return ;
+		struct business_notify_data *bnd_label=(struct business_notify_data *)session_business_data_get(a_stream);
+		if(bnd_label!=NULL && bnd_label->pdata!=NULL && bnd_label->proto==PROTO_HTTP)
+		{
+			enforce_rule[i].http_request_filename=tsg_string_dup(bnd_label->s3_http->request_filename);
+			enforce_rule[i].http_response_filename=tsg_string_dup(bnd_label->s3_http->response_filename);
+			continue;
+		}
+		
+		struct tsg_conn_sketch_notify_data *notify_mail=(struct tsg_conn_sketch_notify_data *)session_conn_sketch_notify_data_get(a_stream);
+		if(notify_mail!=NULL && notify_mail->pdata.mail_eml_filename!=NULL && notify_mail->protocol==PROTO_MAIL)
+		{
+			enforce_rule[i].mail_eml_filename=tsg_string_dup(notify_mail->pdata.mail_eml_filename);
+		}
+	}
 }
 
-int send_log_by_type(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream, LOG_TYPE log_type, int thread_id)
+void tsg_rule_log_enable_type_set(struct enforce_rule_result *enforce_rule, size_t n_enforce_rule, enum LOG_TYPE log_type)
 {
-	int ret=update_percent(_instance, log_type, LOG_STATUS_DROP, thread_id);
-	if(ret==1)
+	for(size_t i=0;i<n_enforce_rule; i++)
 	{
-		MASTER_LOG(_instance->logger, RLOG_LV_DEBUG, LOG_MODULE_SENDLOG, 
-									"drop log: log_type=%d send_log_percent: %d addr=%s",
-									log_type,
-									_instance->service2topic[log_type].send_log_percent[thread_id],
-									(a_stream==NULL ? "" : printaddr(&(a_stream->addr), thread_id))
-								);
+		enforce_rule[i].log_type=log_type;
 	}
+}
+
+void tsg_rule_log_enable_select(struct maat_rule *rules, size_t n_rules, struct enforce_rule_result *enforce_rule, size_t n_enforce_rule, size_t *enforce_rule_offset)
+{
+	int repeat_rule=0;
+	for(size_t i=0;i<n_rules; i++)
+	{
+		if(rules[i].do_log==LOG_ABORT)
+		{
+			continue;
+		}
 		
-	StringBuffer sb(0, _instance->rapidjson_chunk_capacity);
-	Writer<StringBuffer> writer(sb);
-	_handle->document->Accept(writer);
+		for(size_t j=0; j<(*enforce_rule_offset); j++)
+		{
+			if(enforce_rule[j].rule.rule_id==rules[i].rule_id)
+			{
+				repeat_rule=1;
+				continue;
+			}
+		}
 
-	tsg_send_payload(_instance, log_type, (char *)sb.GetString(), sb.GetSize(), thread_id);
-	
-	return 0;
+		if(repeat_rule==1)
+		{
+			repeat_rule=0;
+			continue;
+		}
+
+		if(n_enforce_rule < (*enforce_rule_offset)+1)
+		{
+			break;
+		}
+
+		enforce_rule[(*enforce_rule_offset)++].rule=rules[i];
+	}
 }
 
-int send_event_log(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream, LOG_TYPE log_type, struct maat_rule *rules, size_t n_rules, int thread_id)
+void tsg_hits_rule_metrics_set(const struct streaminfo *a_stream, struct maat_rule *rules, size_t n_rules, int thread_id)
 {
 	int repeat_cnt=0;
 	int policy_id[MAX_RESULT_NUM]={0};
@@ -2210,12 +2178,6 @@ int send_event_log(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_h
 	{
 		if(is_multi_hit_same_policy(&(rules[i]), policy_id, &repeat_cnt))
 		{
-			MASTER_LOG(_instance->logger, RLOG_LV_DEBUG, LOG_MODULE_SENDLOG, 
-										"tsg same log:cfg_id=%d service=%d addr=%s",
-										rules[i].rule_id,
-										rules[i].service_id,
-										(a_stream==NULL ? "" : printaddr(&(a_stream->addr), thread_id))
-									);
 			continue;
 		}
 
@@ -2223,86 +2185,26 @@ int send_event_log(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_h
 		{
 			tsg_set_policy_flow(a_stream, &(rules[i]), thread_id);
 		}
-
-		switch(rules[i].do_log)
-		{
-			case	LOG_ABORT:
-				MASTER_LOG(_instance->logger, RLOG_LV_DEBUG, LOG_MODULE_SENDLOG, 
-											"tsg abort log:cfg_id=%d service=%d addr=%s",
-											rules[i].rule_id,
-											rules[i].service_id,
-											(a_stream==NULL ? "" : printaddr(&(a_stream->addr), thread_id))
-										);
-
-				fs3_rule_stat_update(RULE_STAT_ABORT, (int)rules[i].action, 1);
-				continue;
-				break;
-			case	LOG_ALL:
-				if(rules[i].action==TSG_ACTION_MONITOR)
-				{
-					set_s3_filename(_instance, _handle, a_stream);
-					set_mail_eml(_instance, _handle, a_stream);
-				}
-				break;
-			case	LOG_NOFILE:
-				if(rules[i].action==TSG_ACTION_MONITOR)
-				{
-					TLD_delete(_handle, _instance->id2field[LOG_COMMON_MAIL_EML_FILE].name);
-					TLD_delete(_handle, _instance->id2field[LOG_COMMON_HTTP_REQUEST_S3_FILE].name);
-					TLD_delete(_handle, _instance->id2field[LOG_COMMON_HTTP_RESPONSE_S3_FILE].name);
-				}
-				break;
-			default:
-				break;
-		}
-		
-		TLD_append(_handle, _instance->id2field[LOG_COMMON_POLICY_ID].name, (void *)(long)(rules[i].rule_id), TLD_TYPE_LONG);
-		TLD_append(_handle, _instance->id2field[LOG_COMMON_SERVICE].name, (void *)(long)(rules[i].service_id), TLD_TYPE_LONG);
-		TLD_append(_handle, _instance->id2field[LOG_COMMON_VSYSTEM_ID].name, (void *)(long)rules[i].vsys_id, TLD_TYPE_LONG);
-		TLD_append(_handle, _instance->id2field[LOG_COMMON_ACTION].name, (void *)(long)(rules[i].action), TLD_TYPE_LONG);
-
-		set_policy_action_para_exec_result(_instance, _handle, a_stream, &(rules[i]));
-		
-		set_xxxx_from_user_region(_handle, _instance, &(rules[i]), thread_id);
-
-		send_log_by_type(_instance, _handle, a_stream, log_type, thread_id);
-
-		TLD_delete(_handle, _instance->id2field[LOG_COMMON_POLICY_ID].name);
-		TLD_delete(_handle, _instance->id2field[LOG_COMMON_SERVICE].name);
-		TLD_delete(_handle, _instance->id2field[LOG_COMMON_ACTION].name);
-		TLD_delete(_handle, _instance->id2field[LOG_COMMON_USER_REGION].name);
-		TLD_delete(_handle, _instance->id2field[LOG_COMMON_VSYSTEM_ID].name);
-
-		fs3_rule_stat_update(RULE_STAT_SEND, (int)rules[i].action, 1);
 	}
-
-	return 0;
 }
 
-int deal_event_rules(struct tsg_log_instance_t *_instance, struct TLD_handle_t *_handle, const struct streaminfo *a_stream, LOG_TYPE log_type, TSG_SERVICE service, int *is_append_common_field, int thread_id)
+int tsg_enforce_rule_append(const struct streaminfo *a_stream, enum TSG_SERVICE service, struct enforce_rule_result *enforce_rule, size_t n_enforce_rule)
 {
-	struct matched_policy_rules *matched_rules=(struct matched_policy_rules *)session_matched_rules_get(a_stream, service);
-	if(matched_rules==NULL || matched_rules->n_rules==0)
+	size_t offset=0;
+	struct matched_policy_rules *matched_policy=(struct matched_policy_rules *)session_matched_rules_get(a_stream, service);
+	if(matched_policy!=NULL && matched_policy->n_rules>0)
 	{
-		return 0;
-	}
+		tsg_hits_rule_metrics_set(a_stream, matched_policy->rules, matched_policy->n_rules, a_stream->threadnum);
+		size_t n_rules=MIN(n_enforce_rule, matched_policy->n_rules);
 
-	if((*is_append_common_field)==0)
-	{
-		(*is_append_common_field)=1;
-		append_common_field(_instance, _handle, a_stream);
-	}
-	
-	if(service==TSG_SERVICE_INTERCEPT)
-	{
-		set_intercept_info(_instance, _handle, a_stream);
+		memset(enforce_rule, 0, sizeof(struct enforce_rule_result)*n_rules);		
+		tsg_rule_log_enable_select(matched_policy->rules, n_rules, enforce_rule, n_enforce_rule, &offset);
+
+		session_matched_rules_free(a_stream, service, (void *)matched_policy);
+		session_matched_rules_async(a_stream, service, NULL);
 	}
-	
-	send_event_log(_instance, _handle, a_stream, log_type, matched_rules->rules, matched_rules->n_rules, thread_id);
-	session_matched_rules_free(a_stream, service, (void *)matched_rules);
-	session_matched_rules_async(a_stream, service, NULL);
 
-	return 1;
+	return offset;
 }
 
 int tsg_send_log(struct tsg_log_instance_t *instance, struct TLD_handle_t *handle, const struct streaminfo *a_stream, enum LOG_TYPE log_type, struct maat_rule *rules, size_t n_rules, int thread_id)
@@ -2336,60 +2238,87 @@ int tsg_send_log(struct tsg_log_instance_t *instance, struct TLD_handle_t *handl
 		return 0;
 	}
 
-	int is_append_common_field=0;
+	TLD_append_streaminfo(_instance, _handle, a_stream);
+	TLD_append(_handle, _instance->id2field[LOG_COMMON_SLED_IP].name, (void *)(_instance->local_ip_str), TLD_TYPE_STRING);
+	if(strlen(g_tsg_para.device_sn)>0)
+	{
+		TLD_append(_handle, _instance->id2field[LOG_COMMON_DEVICE_ID].name, (void *)(g_tsg_para.device_sn), TLD_TYPE_STRING);
+	}
 
+	TLD_append(_handle, _instance->id2field[LOG_COMMON_DATA_CENTER].name, (void *)tsg_data_center_get(), TLD_TYPE_STRING);
+	TLD_append(_handle, _instance->id2field[LOG_COMMON_DEVICE_TAG].name, (void *)tsg_device_tag_get(), TLD_TYPE_STRING);
+	TLD_append(_handle, _instance->id2field[LOG_COMMON_TRAFFIC_VSYSTEM_ID].name, (void *)(long)_instance->vsystem_id, TLD_TYPE_LONG);
+	
+	set_application_behavior(_instance, _handle, a_stream);
+	
+	size_t enforce_rule_offset=0;
+	size_t n_enforce_rule=LOGGER_SCRATCH_RULE_MAX;
+	struct enforce_rule_result enforce_rule[n_enforce_rule];
+	
 	switch(log_type)
 	{
 		case	LOG_TYPE_SECURITY_EVENT:
-			append_common_field(instance, handle, a_stream);
-			send_event_log(_instance, _handle, a_stream, LOG_TYPE_SECURITY_EVENT, rules, n_rules, thread_id);
-			break;
 		case	LOG_TYPE_INTERCEPT_EVENT:
-			append_common_field(instance, handle, a_stream);
-			send_event_log(_instance, _handle, a_stream, LOG_TYPE_INTERCEPT_EVENT, rules, n_rules, thread_id);
+			memset(enforce_rule, 0, sizeof(struct enforce_rule_result)*n_enforce_rule);
+			tsg_hits_rule_metrics_set(a_stream, rules, n_rules, thread_id);
+			tsg_rule_log_enable_select(rules, n_rules, enforce_rule, n_enforce_rule, &enforce_rule_offset);
+			tsg_rule_log_enable_additional_set(a_stream, enforce_rule, enforce_rule_offset);
+			tsg_rule_log_enable_type_set(enforce_rule, enforce_rule_offset, log_type);
+			tsg_log_ringbuf_produce(_instance, _handle, log_type, enforce_rule, enforce_rule_offset, thread_id);
 			break;
 		case	LOG_TYPE_SESSION_RECORD:
 			if(_instance->send_intercept_log==1)
 			{
-				deal_event_rules(_instance, _handle, a_stream, LOG_TYPE_INTERCEPT_EVENT, TSG_SERVICE_INTERCEPT, &is_append_common_field, thread_id);
+				size_t offset=tsg_enforce_rule_append(a_stream, TSG_SERVICE_INTERCEPT, enforce_rule, n_enforce_rule);
+				if(offset>0)
+				{
+					set_intercept_info(_instance, _handle, a_stream);
+					tsg_rule_log_enable_type_set(enforce_rule, offset, LOG_TYPE_INTERCEPT_EVENT);
+					enforce_rule_offset+=offset;
+				}
 			}
 			// no break;
 		case	LOG_TYPE_TRANSACTION_RECORD:
-			deal_event_rules(_instance, _handle, a_stream, LOG_TYPE_SECURITY_EVENT, TSG_SERVICE_SECURITY, &is_append_common_field, thread_id);
+			{
+				size_t offset=tsg_enforce_rule_append(a_stream, TSG_SERVICE_SECURITY, enforce_rule+enforce_rule_offset, n_enforce_rule-enforce_rule_offset);
+				if(offset>0)
+				{
+					tsg_rule_log_enable_type_set(enforce_rule+enforce_rule_offset, offset, LOG_TYPE_SECURITY_EVENT);
+					tsg_rule_log_enable_additional_set(a_stream, enforce_rule+enforce_rule_offset, offset);
+					enforce_rule_offset+=offset;
+				}
+			}
 			// no break;
 		case	LOG_TYPE_INTERIM_SESSION_RECORD:
 			if(session_record_limit(_instance, a_stream, log_type))
 			{
+				if(enforce_rule_offset>0)
+				{
+					tsg_log_ringbuf_produce(_instance, _handle, LOG_TYPE_MAX, enforce_rule, enforce_rule_offset, thread_id);
+				}
+				else
+				{
+					TLD_cancel(_handle);
+				}
 				break;
 			}
 
-			if(is_append_common_field==0)
-			{
-				append_common_field(_instance, _handle, a_stream);
-			}
-
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_VSYSTEM_ID].name, (void *)(long)_instance->vsystem_id, TLD_TYPE_LONG);
-			
 			set_sce_profile_ids(_instance, _handle, a_stream);
 			set_shaping_profile_ids(_instance, _handle, a_stream);
 			set_shaping_rule_ids(_instance, _handle, a_stream);
 			set_nat_linkinfo(_instance, _handle, a_stream);
-			send_log_by_type(_instance, _handle, a_stream, log_type, thread_id);
+			tsg_log_ringbuf_produce(_instance, _handle, log_type, enforce_rule, enforce_rule_offset, thread_id);
 			break;
 		case	LOG_TYPE_BGP_RECORD:
 		case	LOG_TYPE_VOIP_RECORD:
 		case	LOG_TYPE_GTPC_RECORD:
 		case	LOG_TYPE_INTERNAL_RTP_RECORD:
-			append_common_field(_instance, _handle, a_stream);
-			TLD_append(_handle, _instance->id2field[LOG_COMMON_VSYSTEM_ID].name, (void *)(long)_instance->vsystem_id, TLD_TYPE_LONG);
-			send_log_by_type(_instance, _handle, a_stream, log_type, thread_id);
+			tsg_log_ringbuf_produce(_instance, _handle, log_type, NULL, 0, thread_id);
 			break;
 		default:
-			return 0;
+			break;
 	}
 	
-	TLD_cancel(handle);
-
 	return 0;
 }
 
@@ -2468,3 +2397,231 @@ int tsg_unknown_app_id_get(struct tsg_log_instance_t *instance)
 	return 0;
 }
 
+
+struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile)
+{
+	char override_sled_ip[32]={0};
+	char kafka_errstr[1024]={0};
+	unsigned int local_ip_nr=0;
+	rd_kafka_conf_t *rdkafka_conf = NULL;
+	char broker_list[1024]={0};
+	struct tsg_log_instance_t *_instance=NULL;
+	char common_field_file[128]={0};
+	char log_path[128]={0};
+
+	_instance=(struct tsg_log_instance_t *)calloc(1, sizeof(struct tsg_log_instance_t));
+
+	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_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);
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "SEND_NAT_LINKINFO_SWITCH", &(_instance->send_nat_linkinfo), 0);
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "RAPIDJSON_CHUNK_CAPACITY", &(_instance->rapidjson_chunk_capacity), 2048);
+
+	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)
+	{
+		printf("MESA_create_runtime_log_handle failed ..., path: %s level: %d", log_path, _instance->level);
+		free(_instance);
+		_instance=NULL;
+		return NULL;
+	}
+	
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "MODE",&(_instance->mode), 0);
+	if(_instance->mode==CLOSE)
+	{
+		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, "Disable tsg_send_log");
+		return _instance;
+	}
+	
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "RECOVERY_INTERVEL_S", &(_instance->recovery_interval), 30);	
+
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "COMMON_FIELD_FILE", common_field_file, sizeof(common_field_file), NULL);
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "BROKER_LIST", broker_list, sizeof(broker_list), NULL);
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "SASL_USERNAME", _instance->sasl_username, sizeof(_instance->sasl_username), ""); //admin
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "SASL_PASSWD", _instance->sasl_passwd, sizeof(_instance->sasl_passwd), "");
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "COMPRESSION_TYPE", _instance->compression, sizeof(_instance->compression), ""); //snappy
+
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "SEND_QUEUE_MAX_MESSAGE", _instance->send_queue_max_msg, sizeof(_instance->send_queue_max_msg), "1000000");
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "REFRESH_INTERVAL_MS", _instance->refresh_interval_ms, sizeof(_instance->refresh_interval_ms), "600000");
+	MESA_load_profile_string_def(conffile, "TSG_LOG", "REQUIRE_ACK", _instance->require_ack, sizeof(_instance->require_ack), "1");
+
+	MESA_load_profile_string_def(conffile, "SYSTEM", "TCP_LABEL", _instance->tcp_label, sizeof(_instance->tcp_label), "tcp_flow_stat");
+	MESA_load_profile_string_def(conffile, "SYSTEM", "UDP_LABEL", _instance->udp_label, sizeof(_instance->udp_label), "udp_flow_stat");
+
+	_instance->tcp_flow_project_id=project_customer_register(_instance->tcp_label, "struct");
+	_instance->udp_flow_project_id=project_customer_register(_instance->udp_label, "struct");
+	if(_instance->tcp_flow_project_id<0 || _instance->udp_flow_project_id<0)
+	{
+		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, 
+									"project_customer_register is error, tcp_label: %s udp_label: %s, please check etc/project.conf",
+									_instance->tcp_label,
+									_instance->udp_label
+								);
+	}
+
+	MESA_load_profile_string_def(conffile, "SYSTEM", "OVERRIDE_SLED_IP", override_sled_ip, sizeof(override_sled_ip), "OVERRIDE_SLED_IP");
+	char *sled_ip=getenv(override_sled_ip);
+	if(sled_ip==NULL)
+	{
+		char nic_name[32]={0};
+		MESA_load_profile_string_def(conffile, "SYSTEM", "NIC_NAME", nic_name, sizeof(nic_name), "lo");	
+		int ret=MESA_get_dev_ipv4(nic_name, (int *)&local_ip_nr);
+		if(ret<0)
+		{
+			MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, 
+										"GET_LOCAL_IP MESA_get_dev_ipv4 is error, nic_name: %s, please check tsgconf/main.conf",
+										nic_name
+									);
+			return NULL;
+		}
+		inet_ntop(AF_INET,&(local_ip_nr),_instance->local_ip_str,sizeof(_instance->local_ip_str));
+	}
+	else
+	{
+		memcpy(_instance->local_ip_str, sled_ip, MIN(sizeof(_instance->local_ip_str)-1, strlen(sled_ip)));
+	}
+	
+	rdkafka_conf = rd_kafka_conf_new();	
+	rd_kafka_conf_set(rdkafka_conf, "queue.buffering.max.messages", _instance->send_queue_max_msg, kafka_errstr, sizeof(kafka_errstr));
+	rd_kafka_conf_set(rdkafka_conf, "topic.metadata.refresh.interval.ms", _instance->refresh_interval_ms, kafka_errstr, sizeof(kafka_errstr));
+	rd_kafka_conf_set(rdkafka_conf, "request.required.acks", _instance->require_ack, kafka_errstr, sizeof(kafka_errstr));	
+	rd_kafka_conf_set(rdkafka_conf, "socket.keepalive.enable", "true", kafka_errstr, sizeof(kafka_errstr));	
+	rd_kafka_conf_set(rdkafka_conf, "bootstrap.servers", broker_list, kafka_errstr, sizeof(kafka_errstr));
+	
+	if(strlen(_instance->compression)>0)
+	{
+		rd_kafka_conf_set(rdkafka_conf, "compression.codec", _instance->compression, kafka_errstr, sizeof(kafka_errstr));
+	}
+
+	if(strlen(_instance->sasl_username)> 0 && strlen(_instance->sasl_passwd)>0)
+	{
+		rd_kafka_conf_set(rdkafka_conf, "security.protocol", "sasl_plaintext", kafka_errstr, sizeof(kafka_errstr));
+		rd_kafka_conf_set(rdkafka_conf, "sasl.mechanisms", "PLAIN", kafka_errstr, sizeof(kafka_errstr));
+		rd_kafka_conf_set(rdkafka_conf, "sasl.username", _instance->sasl_username, kafka_errstr, sizeof(kafka_errstr));
+		rd_kafka_conf_set(rdkafka_conf, "sasl.password", _instance->sasl_passwd, kafka_errstr, sizeof(kafka_errstr));
+	}
+	
+	if(!(_instance->kafka_handle=rd_kafka_new(RD_KAFKA_PRODUCER, rdkafka_conf, kafka_errstr, sizeof(kafka_errstr))))
+	{
+		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, "KAFKA_INIT rd_kafka_new is error");
+		return NULL;
+	}
+
+	log_common_fields_new(common_field_file, _instance->id2field, &(_instance->service2topic), &(_instance->max_service));
+
+	if(_instance->service2topic!=NULL)
+	{
+		_instance->sum_stat_row_id=tsg_stat_sendlog_row_init("sum");
+		for(int i=0; i<_instance->max_service; i++)
+		{
+			if(_instance->service2topic[i].type==TLD_TYPE_MAX && strlen(_instance->service2topic[i].name)>0)
+			{
+				register_topic(_instance, &( _instance->service2topic[i]));
+			}
+		}
+	}
+	else
+	{
+		MASTER_LOG(_instance->logger, RLOG_LV_FATAL, LOG_MODULE_SENDLOG, "KAFKA_INIT log_common_fields_new is error, please check %s", common_field_file);
+	}
+
+	int32_t ringbuf_size=0;
+    MESA_load_profile_int_def(conffile, "TSG_LOG", "RINGBUFF_SIZE", &(ringbuf_size), 100000);
+	MESA_load_profile_int_def(conffile, "TSG_LOG", "RINGBUFF_NUM", &(_instance->ringbuf_num), 1);
+
+	_instance->ringbuf=(struct logger_ringbuf_schema *)calloc(1, _instance->ringbuf_num * sizeof(struct logger_ringbuf_schema));
+
+	for(int i=0; i<_instance->ringbuf_num; i++)
+	{
+		struct logger_ringbuf_schema *ringbuf=&(_instance->ringbuf[i]);
+
+		int thread_num=get_thread_count();
+		size_t ringbuf_obj_size=0;
+		ringbuf_get_sizes(thread_num, &ringbuf_obj_size, NULL);
+		ringbuf->ring=(ringbuf_t *)calloc(1, ringbuf_obj_size);
+
+		ringbuf->size=ringbuf_size;
+		ringbuf->buff=(char *)calloc(1, sizeof(struct logger_scratch)*ringbuf->size);
+		ringbuf_setup(ringbuf->ring, thread_num, sizeof(struct logger_scratch)*ringbuf->size);
+		ringbuf->worker=(ringbuf_worker_t **)calloc(1, sizeof(ringbuf_worker_t *)*thread_num);
+		for(int j=0; j<thread_num; j++)
+		{
+			ringbuf->worker[j]=ringbuf_register(ringbuf->ring, j);
+		}
+
+		char ringbuf_name[32]={0};
+
+		snprintf(ringbuf_name, sizeof(ringbuf_name), "ringbuf_produce_%d", i);
+		ringbuf->produce_row_idx=tsg_stat_sendlog_row_init(ringbuf_name);
+		snprintf(ringbuf_name, sizeof(ringbuf_name), "ringbuf_consume_%d", i);
+		ringbuf->consume_row_idx=tsg_stat_sendlog_row_init(ringbuf_name);
+
+		pthread_create(&(ringbuf->pthread_id), NULL, tsg_log_ringbuf_consume, (void *)ringbuf);
+		pthread_setname_np(ringbuf->pthread_id, "SNEDLOG_RINGBUF");
+	}
+
+
+	return _instance;
+}
+void tsg_sendlog_destroy(struct tsg_log_instance_t * instance)
+{
+	if(instance==NULL)
+	{
+		return ;
+	}
+	
+	if(instance->mode!=CLOSE)
+	{
+		for(int i=0; i<instance->max_service; i++)
+		{
+			if(instance->service2topic[i].type!=TLD_TYPE_MAX || i==1)  //i=1 equal i=0, service id of security event is 0 and 1
+ 			{
+				continue;
+			}
+			
+			if(instance->service2topic[i].topic_rkt!=NULL)
+			{
+				rd_kafka_topic_destroy(instance->service2topic[i].topic_rkt);
+			}
+
+			if(instance->service2topic[i].drop_start!=NULL)
+			{
+				free(instance->service2topic[i].drop_start);
+				instance->service2topic[i].drop_start=NULL;
+			}
+
+			if(instance->service2topic[i].send_log_percent!=NULL)
+			{
+				free(instance->service2topic[i].send_log_percent);
+				instance->service2topic[i].send_log_percent=NULL;
+			}
+		}
+		
+		//rd_kafka_destroy_flags(instance->kafka_handle, 4);
+		rd_kafka_destroy(instance->kafka_handle);
+
+		free(instance->service2topic);
+		instance->service2topic=NULL;
+	}
+
+	
+	MESA_destroy_runtime_log_handle(instance->logger);
+	instance->logger=NULL;
+
+	free(instance);
+	instance=NULL;
+	return ;
+}
diff --git a/src/tsg_send_log_internal.h b/src/tsg_send_log_internal.h
index 25a9c89..51e8b91 100644
--- a/src/tsg_send_log_internal.h
+++ b/src/tsg_send_log_internal.h
@@ -5,6 +5,7 @@
 #include <librdkafka/rdkafka.h>
 #include <MESA/cJSON.h>
 #include <time.h>
+#include "ringbuf/ringbuf.h"
 
 #define	MIN_L7_PROTO_ID	100
 #define	MAX_L7_PROTO_ID	150
@@ -175,10 +176,47 @@ struct topic_stat
 	rd_kafka_topic_t *topic_rkt;
 };
 
+struct enforce_rule_result
+{
+	int log_type;  //enum LOG_TYPE
+	struct maat_rule rule;
+
+	char *mail_eml_filename;
+	char *http_request_filename;
+	char *http_response_filename;
+	
+	long mirrored_bytes;
+	long mirrored_packets;
+	char *packet_capture_file;
+};
+
+#define	LOGGER_SCRATCH_RULE_MAX	16
+struct logger_scratch
+{
+	int thread_id;
+	int n_enforce_rule;
+	int log_type; // enum LOG_TYPE
+	struct TLD_handle_t *handle;
+	struct tsg_log_instance_t *instance;
+	struct enforce_rule_result *enforce_rule;
+};
+
+struct logger_ringbuf_schema
+{
+	int produce_row_idx;
+	int consume_row_idx;
+	pthread_t pthread_id;
+	size_t size;
+	char *buff;
+	ringbuf_t *ring;
+	ringbuf_worker_t **worker;
+};
+
 struct tsg_log_instance_t
 {
 	int mode;
 	int level;
+	int ringbuf_num;
 	int max_service;
 	int vsystem_id;
 	int unknown_app_id;
@@ -210,6 +248,7 @@ struct tsg_log_instance_t
 	id2field_t id2field[LOG_COMMON_MAX];
 	rd_kafka_t *kafka_handle;
 	struct topic_stat *service2topic;
+	struct logger_ringbuf_schema *ringbuf;
 	void *logger;
 };