TSG-21854 TFE使用fieldstat4序列化Manipulation Policy的metric并输出到kafka

1 files changed, 21 insertions, 7 deletions

diff --git a/plugin/business/doh/src/doh.cpp b/plugin/business/doh/src/doh.cpp
index 15032b4..ca99c44 100644
--- a/plugin/business/doh/src/doh.cpp
+++ b/plugin/business/doh/src/doh.cpp
@@ -824,11 +824,11 @@ int doh_on_data(const struct tfe_stream *stream, const struct tfe_http_session *
 void doh_send_metric_log(const struct tfe_stream * stream, struct doh_ctx *ctx, unsigned int thread_id)
 {
 	size_t c2s_byte_num = 0, s2c_byte_num =0;
-	struct tfe_fieldstat_metric_t *fieldstat = tfe_get_fieldstat_handle();
+	struct tfe_fieldstat_easy_t *fieldstat = tfe_get_fieldstat_handle();
 
-	fieldstat->tags[thread_id][TAG_VSYS_ID].value_int = ctx->result->vsys_id;
-	fieldstat->tags[thread_id][TAG_RULE_ID].value_int = ctx->result->config_id;
-	fieldstat->tags[thread_id][TAG_ACTION].value_int = 48;
+	fieldstat->tags[thread_id][TAG_VSYS_ID].value_longlong = ctx->result->vsys_id;
+	fieldstat->tags[thread_id][TAG_RULE_ID].value_longlong = ctx->result->config_id;
+	fieldstat->tags[thread_id][TAG_ACTION].value_longlong = 48;
 	fieldstat->tags[thread_id][TAG_SUB_ACTION].value_str = "redirect";
 
 	tfe_stream_info_get(stream, INFO_FROM_DOWNSTREAM_RX_OFFSET, &c2s_byte_num, sizeof(c2s_byte_num));
@@ -860,9 +860,23 @@ void doh_send_metric_log(const struct tfe_stream * stream, struct doh_ctx *ctx,
 		out_bytes = c2s_byte_num;
 	}
 
-	tfe_fieldstat_metric_incrby(fieldstat, fieldstat->column_array[COLUMN_HIT_COUNT], 1, fieldstat->tags[thread_id], TAG_MAX - 1, thread_id);
-	tfe_fieldstat_metric_incrby(fieldstat, fieldstat->column_array[COLUMN_IN_BYTES], in_bytes, fieldstat->tags[thread_id], TAG_MAX - 1, thread_id);
-	tfe_fieldstat_metric_incrby(fieldstat, fieldstat->column_array[COLUMN_OUT_BYTES], out_bytes, fieldstat->tags[thread_id], TAG_MAX - 1, thread_id);
+	tfe_fieldstat_easy_incrby(fieldstat, fieldstat->counter_array[COLUMN_HIT_COUNT], 1, fieldstat->tags[thread_id], TAG_MAX - 1, thread_id);
+	tfe_fieldstat_easy_incrby(fieldstat, fieldstat->counter_array[COLUMN_IN_BYTES], in_bytes, fieldstat->tags[thread_id], TAG_MAX - 1, thread_id);
+	tfe_fieldstat_easy_incrby(fieldstat, fieldstat->counter_array[COLUMN_OUT_BYTES], out_bytes, fieldstat->tags[thread_id], TAG_MAX - 1, thread_id);
+
+	char **payload = NULL;
+    size_t payload_len = 0;
+
+	fieldstat_easy_output_array(fieldstat->fseasy, &payload, &payload_len);
+	if (payload)
+	{
+		for (size_t i = 0; i < payload_len; i++)
+		{
+			kafka_send(tfe_get_kafka_handle(), TOPIC_RULE_HITS, payload[i], strlen(payload[i]));
+			FREE(&payload[i]);
+		}
+		FREE(&payload);
+	}
 	return;
 }