TSG-16531 PacketAdapter适配容器环境，使用mrzcpd收包，通过RAW Socket注RST包v2.0.0-20230810

6 files changed, 574 insertions, 560 deletions

diff --git a/platform/src/inject_pkt.c b/platform/src/inject_pkt.c
deleted file mode 100644
index 9e336c0..0000000
--- a/platform/src/inject_pkt.c
+++ /dev/null
@@ -1,53 +0,0 @@
-#include "inject_pkt.h"
-
-int inject_ipv4_pkt(char *ip4_addr, uint8_t *data, uint32_t len)
-{
-    int fd = 0;
-    struct sockaddr_in saddr4 = {0};
-
-    saddr4.sin_family = PF_INET;
-    saddr4.sin_addr.s_addr = inet_addr(ip4_addr);
-
-    fd = socket(PF_INET, SOCK_RAW, IPPROTO_RAW);
-    if (fd == -1)
-    {
-        LOG_ERROR("Failed at socket(PF_INET, SOCK_RAW), %d: %s", errno, strerror(errno));
-        return -1;
-    }
-
-    if (sendto(fd, data, len, 0, (struct sockaddr *)&saddr4, sizeof(saddr4)) == -1)
-    {
-        LOG_ERROR("Failed at send(), %d: %s", errno, strerror(errno));
-        close(fd);
-        return -1;
-    }
-
-    close(fd);
-    return 0;
-}
-
-int inject_ipv6_pkt(char *ip6_addr, uint8_t *data, uint32_t len)
-{
-    int fd = 0;
-    struct sockaddr_in6 saddr6 = {0};
-
-    saddr6.sin6_family = PF_INET6;
-    inet_pton(AF_INET6, ip6_addr, &saddr6.sin6_addr);
-
-    fd = socket(PF_INET6, SOCK_RAW, IPPROTO_RAW);
-    if (fd == -1)
-    {
-        LOG_ERROR("Failed at socket(PF_INET6, SOCK_RAW), %d: %s", errno, strerror(errno));
-        return -1;
-    }
-
-    if (sendto(fd, data, len, 0, (struct sockaddr *)&saddr6, sizeof(saddr6)) == -1)
-    {
-        LOG_ERROR("Failed at send(), %d: %s", errno, strerror(errno));
-        close(fd);
-        return -1;
-    }
-
-    close(fd);
-    return 0;
-}
\ No newline at end of file
diff --git a/platform/src/packet_adapter.c b/platform/src/packet_adapter.c
deleted file mode 100644
index 43ead6c..0000000
--- a/platform/src/packet_adapter.c
+++ /dev/null
@@ -1,506 +0,0 @@
-#include "decode_ipv4.h"
-#include "decode_ipv6.h"
-#include "decode_tcp.h"
-#include "decode_udp.h"
-#include "decode_gtp.h"
-#include "inject_pkt.h"
-#include "system.h"
-
-#include <linux/netfilter.h> // for NF_ACCEPT
-#include <libnetfilter_queue/libnetfilter_queue.h>
-
-#ifdef GIT_VERSION
-static __attribute__((__used__)) const char *Packet_Adapter_Version = GIT_VERSION;
-#else
-static __attribute__((__used__)) const char *Packet_Adapter_Version = "Unknown";
-#endif
-
-typedef struct pkt_info_s
-{
-	uint32_t id;	   // unique ID of packet in queue
-	uint16_t protocol; // hw protocol
-	uint8_t hook;	   // netfilter hook
-	u_int32_t mark;
-	u_int32_t indev;
-	u_int32_t outdev;
-	u_int32_t phys_indev;
-	u_int32_t phys_outdev;
-
-	uint8_t *payload;
-	uint32_t payload_len;
-
-	char src_addr[512];
-} pkt_info_t;
-
-typedef struct union_info_s
-{
-	ipv4_info_t ipv4;
-	ipv6_info_t ipv6;
-	tcp_info_t tcp;
-	udp_info_t udp;
-} union_info_t;
-
-typedef struct pkt_paser_s
-{
-	pkt_info_t raw;
-	union_info_t external;
-	gtp_info_t gtp;
-	union_info_t internal;
-} pkt_paser_t;
-
-static int is_dump_packet_info = 0;
-
-static void dump_info(pkt_paser_t *parser)
-{
-	char buff[4096] = {0};
-	size_t size = sizeof(buff);
-	size_t len = 0;
-
-	len += snprintf(buff + len, size - len, "{");
-	len += snprintf(buff + len, size - len,
-					"\"raw_info\":{\"id\":%u,\"protocol\":%u,\"hook\":%u,\"mark\":%u,\"indev\":%u,\"outdev\":%u,\"phys_indev\":%u,\"phys_outdev\":%u,\"src_addr\":\"%s\",\"data_len\":%u}",
-					parser->raw.id,
-					parser->raw.protocol,
-					parser->raw.hook,
-					parser->raw.mark,
-					parser->raw.indev,
-					parser->raw.outdev,
-					parser->raw.phys_indev,
-					parser->raw.phys_outdev,
-					parser->raw.src_addr,
-					parser->raw.payload_len);
-
-	// external
-	if (parser->external.ipv4.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"external_ipv4\":");
-		len += dump_ipv4_info(&(parser->external.ipv4), buff + len, size - len);
-	}
-	if (parser->external.ipv6.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"external_ipv6\":");
-		len += dump_ipv6_info(&(parser->external.ipv6), buff + len, size - len);
-	}
-	if (parser->external.udp.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"external_udp\":");
-		len += dump_udp_info(&(parser->external.udp), buff + len, size - len);
-	}
-	if (parser->external.tcp.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"external_tcp\":");
-		len += dump_tcp_info(&(parser->external.tcp), buff + len, size - len);
-	}
-
-	// gtp
-	if (parser->gtp.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"gtp\":");
-		len += dump_gtp_info(&(parser->gtp), buff + len, size - len);
-	}
-
-	// internal
-	if (parser->internal.ipv4.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"internal_ipv4\":");
-		len += dump_ipv4_info(&(parser->internal.ipv4), buff + len, size - len);
-	}
-	if (parser->internal.ipv6.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"internal_ipv6\":");
-		len += dump_ipv6_info(&(parser->internal.ipv6), buff + len, size - len);
-	}
-	if (parser->internal.udp.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"internal_udp\":");
-		len += dump_udp_info(&(parser->internal.udp), buff + len, size - len);
-	}
-	if (parser->internal.tcp.hdr)
-	{
-		len += snprintf(buff + len, size - len, ",\"internal_tcp\":");
-		len += dump_tcp_info(&(parser->internal.tcp), buff + len, size - len);
-	}
-
-	len += snprintf(buff + len, size - len, "}");
-	LOG_DEBUG("%s", buff);
-}
-
-static int decode_ip_tcp_udp(union_info_t *parser, const uint8_t *data, uint32_t len)
-{
-	int next_protocol = 0;
-	uint8_t *payload = NULL;
-	uint32_t payload_len = 0;
-
-	if (len < IPV4_HEADER_LEN)
-	{
-		LOG_ERROR("Parser IP header: packet length too small %d", len);
-		return -1;
-	}
-
-	if (IP_GET_RAW_VER(data) == 4)
-	{
-		if (decode_ipv4(&(parser->ipv4), data, len) == -1)
-		{
-			return -1;
-		}
-
-		payload = parser->ipv4.payload;
-		payload_len = parser->ipv4.payload_len;
-		next_protocol = parser->ipv4.next_protocol;
-	}
-	else if (IP_GET_RAW_VER(data) == 6)
-	{
-		if (decode_ipv6(&(parser->ipv6), data, len) == -1)
-		{
-			return -1;
-		}
-		payload = parser->ipv6.payload;
-		payload_len = parser->ipv6.payload_len;
-		next_protocol = parser->ipv6.next_protocol;
-	}
-	else
-	{
-		LOG_ERROR("Unknown IP version %d", IP_GET_RAW_VER(data));
-		return -1;
-	}
-
-	if (next_protocol == IPPROTO_UDP)
-	{
-		if (decode_udp(&(parser->udp), payload, payload_len) == -1)
-		{
-			return -1;
-		}
-		return 0;
-	}
-	else if (next_protocol == IPPROTO_TCP)
-	{
-		if (decode_tcp(&(parser->tcp), payload, payload_len) == -1)
-		{
-			return -1;
-		}
-		return 0;
-	}
-	else
-	{
-		LOG_ERROR("Unknown Internal L4 next_protocol version %d", next_protocol);
-		return -1;
-	}
-}
-
-///////////////////////////////////////////////////////////////////////////////
-// NFQ API
-///////////////////////////////////////////////////////////////////////////////
-
-static int decode_pkt(pkt_info_t *packet, struct nfgenmsg *nfmsg, struct nfq_data *nfa)
-{
-	struct nfqnl_msg_packet_hdr *packet_hdr = NULL;
-	struct nfqnl_msg_packet_hw *packet_hw = NULL;
-
-	packet_hdr = nfq_get_msg_packet_hdr(nfa);
-	if (packet_hdr == NULL)
-	{
-		LOG_ERROR("Failed at nfq_get_msg_packet_hdr()");
-		return 0;
-	}
-	packet->id = ntohl(packet_hdr->packet_id);
-
-	packet->payload_len = nfq_get_payload(nfa, &packet->payload);
-	if (packet->payload_len <= 0)
-	{
-		LOG_ERROR("Failed at nfq_get_payload()");
-		return packet->id;
-	}
-	packet->protocol = ntohs(packet_hdr->hw_protocol);
-	packet->hook = packet_hdr->hook;
-
-	packet_hw = nfq_get_packet_hw(nfa);
-	if (packet_hw)
-	{
-		int i = 0;
-		int offset = 0;
-		int len = sizeof(packet->src_addr);
-		int hlen = ntohs(packet_hw->hw_addrlen);
-
-		for (i = 0; i < hlen - 1; i++)
-		{
-			offset += snprintf(packet->src_addr + offset, len - offset, "%02x:", packet_hw->hw_addr[i]);
-		}
-		snprintf(packet->src_addr + offset, len - offset, "%02x", packet_hw->hw_addr[hlen - 1]);
-	}
-
-	packet->mark = nfq_get_nfmark(nfa);
-	packet->indev = nfq_get_indev(nfa);
-	packet->outdev = nfq_get_outdev(nfa);
-	packet->phys_indev = nfq_get_physindev(nfa);
-	packet->phys_outdev = nfq_get_physoutdev(nfa);
-
-	return packet->id;
-}
-/*
- * nfmsg : message objetc that contains the packet
- * nfa : Netlink packet data handle
- */
-static int packet_handler_cb(struct nfq_q_handle *qh, struct nfgenmsg *nfmsg, struct nfq_data *nfa, void *data)
-{
-	int offest = 0;
-	pkt_paser_t parser = {0};
-	int packet_id = decode_pkt(&(parser.raw), nfmsg, nfa);
-
-	// external
-	if (decode_ip_tcp_udp(&(parser.external), parser.raw.payload, parser.raw.payload_len) == -1)
-	{
-		goto end;
-	}
-
-	if (parser.external.udp.hdr == NULL)
-	{
-		LOG_ERROR("External L4 protocol not UDP");
-		goto end;
-	}
-
-	// decode GTP
-	if (decode_gtp(&(parser.gtp), parser.external.udp.payload, parser.external.udp.payload_len) == -1)
-	{
-		return -1;
-	}
-
-	// internal
-	if (decode_ip_tcp_udp(&(parser.internal), parser.gtp.payload, parser.gtp.payload_len) == -1)
-	{
-		goto end;
-	}
-
-	/*
-	 * NF_DROP : discarded the packet
-	 * NF_ACCEPT : the packet passes, continue iterations
-	 * NF_QUEUE : inject the packet into a different queue (the target queue number is in the high 16 bits of the verdict)
-	 * NF_REPEAT : iterate the same cycle once more
-	 * NF_STOP : accept, but don't continue iterations
-	 */
-	// nfq_set_verdict()
-	// nfq_set_verdict2()
-	// nfq_set_verdict_batch()
-	// nfq_set_verdict_batch2()
-	// nfq_set_verdict_mark()
-
-	if (parser.external.ipv4.hdr)
-	{
-		offest += parser.external.ipv4.hdr_len;
-	}
-	if (parser.external.ipv6.hdr)
-	{
-		offest += parser.external.ipv6.hdr_len;
-	}
-
-	offest += parser.external.udp.hdr_len;
-	offest += parser.gtp.hdr_len;
-
-	if (is_dump_packet_info)
-	{
-		dump_info(&parser);
-		LOG_DEBUG("Offset : %d", offest);
-	}
-
-	uint8_t *inject_data = parser.raw.payload + offest;
-	uint32_t inject_data_len = parser.raw.payload_len - offest;
-
-	if (offest > 0)
-	{
-		if ((parser.external.ipv4.hdr && parser.internal.ipv4.hdr) || (parser.external.ipv6.hdr && parser.internal.ipv6.hdr))
-		{
-			return nfq_set_verdict(qh, packet_id, NF_ACCEPT, inject_data_len, inject_data);
-		}
-
-		if (parser.external.ipv4.hdr && parser.internal.ipv6.hdr)
-		{
-			if (inject_ipv6_pkt(parser.internal.ipv6.dst_addr, inject_data, inject_data_len) == -1)
-			{
-				goto end;
-			}
-			return nfq_set_verdict(qh, packet_id, NF_DROP, 0, NULL);
-		}
-
-		if (parser.external.ipv6.hdr && parser.internal.ipv4.hdr)
-		{
-			if (inject_ipv4_pkt(parser.internal.ipv4.dst_addr, inject_data, inject_data_len) == -1)
-			{
-				goto end;
-			}
-			return nfq_set_verdict(qh, packet_id, NF_DROP, 0, NULL);
-		}
-	}
-
-end:
-	return nfq_set_verdict(qh, packet_id, NF_ACCEPT, 0, NULL);
-}
-
-static void sig_handler(int signo)
-{
-	if (signo == SIGUSR1)
-	{
-		is_dump_packet_info = 1;
-		LOG_ERROR("received SIGUSR1, enable dump packet info");
-	}
-	if (signo == SIGUSR2)
-	{
-		is_dump_packet_info = 0;
-		LOG_ERROR("received SIGUSR2, disable dump packet info");
-	}
-}
-
-static void usage(char *cmd)
-{
-	fprintf(stderr, "USAGE: %s [OPTIONS]\n", cmd);
-	fprintf(stderr, "    -v       -- show version\n");
-	fprintf(stderr, "    -i id    -- set queue id\n");
-	fprintf(stderr, "    -d       -- run daemon\n");
-	fprintf(stderr, "    -p       -- dump packet info\n");
-	fprintf(stderr, "    -h       -- show help\n");
-	fprintf(stderr, "Signal: \n");
-	fprintf(stderr, "    kill -s SIGUSR1 `pidof %s`     -- enable dump packet info\n", cmd);
-	fprintf(stderr, "    kill -s SIGUSR2 `pidof %s`     -- disable dump packet info\n", cmd);
-}
-
-/*
- * doc : http://www.netfilter.org/projects/libnetfilter_queue/doxygen/html/
- * Library setup : http://www.netfilter.org/projects/libnetfilter_queue/doxygen/html/group__LibrarySetup.html
- * Queue handling : http://www.netfilter.org/projects/libnetfilter_queue/doxygen/html/group__Queue.html
- * Message parsing : http://www.netfilter.org/projects/libnetfilter_queue/doxygen/html/group__Parsing.html
- */
-int main(int argc, char **argv)
-{
-	int fd;
-	int rv;
-	int opt;
-	uint16_t queue = 1;
-	struct nfq_handle *handle = NULL;
-	struct nfq_q_handle *q_handle = NULL;
-	char buf[65535] __attribute__((aligned));
-
-	is_dump_packet_info = 0;
-
-	while ((opt = getopt(argc, argv, "vi:dph")) != -1)
-	{
-		switch (opt)
-		{
-		case 'v':
-			fprintf(stderr, "Packet Adapter Version: %s\n", Packet_Adapter_Version);
-			return 0;
-		case 'i':
-			queue = atoi(optarg);
-			if (queue < 0 || queue > 65535)
-			{
-				fprintf(stderr, "Usage: %s queueid %d out of range [0, 65535]\n", argv[0], queue);
-				return 0;
-			}
-			break;
-		case 'd':
-			run_daemon();
-			break;
-		case 'p':
-			is_dump_packet_info = 1;
-			break;
-		case 'h': /* fall through */
-		default:
-			usage(argv[0]);
-			return 0;
-		}
-	}
-
-	LOG_DEBUG("Using queue: %d", queue);
-
-	if (signal(SIGUSR1, sig_handler) == SIG_ERR)
-	{
-		LOG_ERROR("Failed at signal(SIGUSR1), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	if (signal(SIGUSR2, sig_handler) == SIG_ERR)
-	{
-		LOG_ERROR("Failed at signal(SIGUSR2), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	handle = nfq_open();
-	if (handle == NULL)
-	{
-		LOG_ERROR("Failed at nfq_open(), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	if (nfq_unbind_pf(handle, AF_INET) < 0)
-	{
-		LOG_ERROR("Failed at nfq_unbind_pf(), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	if (nfq_bind_pf(handle, AF_INET) < 0)
-	{
-		LOG_ERROR("Failed at nfq_bind_pf(), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	q_handle = nfq_create_queue(handle, queue, &packet_handler_cb, NULL);
-	if (q_handle == NULL)
-	{
-		LOG_ERROR("Failed at nfq_create_queue(), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	/*
-	 * NFQNL_COPY_NONE - noop, do not use it
-	 * NFQNL_COPY_META - copy only packet metadata
-	 * NFQNL_COPY_PACKET - copy entire packet
-	 */
-	if (nfq_set_mode(q_handle, NFQNL_COPY_PACKET, 0xffff) < 0)
-	{
-		LOG_ERROR("Failed at nfq_set_mode(NFQNL_COPY_PACKET), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	if (nfq_set_queue_maxlen(q_handle, 65535) < 0)
-	{
-		LOG_ERROR("Failed at nfq_set_queue_maxlen(65535), %d: %s", errno, strerror(errno));
-		goto error;
-	}
-
-	LOG_DEBUG("Waiting for packets...");
-
-	fd = nfq_fd(handle);
-	for (;;)
-	{
-		if ((rv = recv(fd, buf, sizeof(buf), 0)) >= 0)
-		{
-			nfq_handle_packet(handle, buf, rv);
-			continue;
-		}
-		/* 
-		 * if your application is too slow to digest the packets that
-		 * are sent from kernel-space, the socket buffer that we use
-		 * to enqueue packets may fill up returning ENOBUFS. Depending
-		 * on your application, this error may be ignored. Please, see
-		 * the doxygen documentation of this library on how to improve
-		 * this situation.
-		 */
-		if (rv < 0 && errno == ENOBUFS)
-		{
-			LOG_ERROR("Losing packets !!!");
-			continue;
-		}
-
-		LOG_ERROR("Failed at recv(), %d: %s", errno, strerror(errno));
-	}
-
-error:
-	if (q_handle)
-	{
-		nfq_destroy_queue(q_handle);
-	}
-
-	if (handle)
-	{
-		nfq_close(handle);
-	}
-
-	return 0;
-}
diff --git a/platform/src/packet_adapter.cpp b/platform/src/packet_adapter.cpp
new file mode 100644
index 0000000..6fa94f0
--- /dev/null
+++ b/platform/src/packet_adapter.cpp
@@ -0,0 +1,240 @@
+#include <errno.h>
+#include <string.h>
+#include <signal.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <sys/prctl.h>
+
+#include "log.h"
+#include "system.h"
+#include "packet_io.h"
+#include "packet_stat.h"
+#include "packet_handle.h"
+
+#define LOG_MAIN "PacketAdapter"
+
+#ifdef GIT_VERSION
+static __attribute__((__used__)) const char *Packet_Adapter_Version = GIT_VERSION;
+#else
+static __attribute__((__used__)) const char *Packet_Adapter_Version = "Unknown";
+#endif
+
+/******************************************************************************
+ * Struct
+ ******************************************************************************/
+
+struct thread
+{
+    int index;
+    pthread_t tid;
+    struct runtime_ctx *runtime;
+};
+
+struct runtime_ctx
+{
+    int enable_debug;
+    int need_stop;
+
+    struct metrics metrics;
+    struct packet_io *handle;
+    struct packet_stat *stat;
+    struct thread threads[MAX_THREAD_NUM];
+};
+
+/******************************************************************************
+ * Static
+ ******************************************************************************/
+
+struct runtime_ctx static_runtime_ctx = {0};
+struct runtime_ctx *runtime = &static_runtime_ctx;
+
+/******************************************************************************
+ * API
+ ******************************************************************************/
+
+static enum action packet_handle_callback(const char *data, int len, void *args)
+{
+    struct metrics *metrics = (struct metrics *)args;
+    packet_handle(data, len, metrics);
+
+    return ACTION_BYPASS;
+}
+
+static void *worker_thread_cycle(void *arg)
+{
+    struct thread *thread = (struct thread *)arg;
+    struct runtime_ctx *runtime = thread->runtime;
+    struct packet_io *handle = runtime->handle;
+
+    char thread_name[16];
+    snprintf(thread_name, sizeof(thread_name), "pkt-adapter:%d", thread->index);
+    prctl(PR_SET_NAME, (unsigned long long)thread_name, NULL, NULL, NULL);
+
+    if (packet_io_thread_init(handle, thread->index) != 0)
+    {
+        goto error_out;
+    }
+
+    LOG_INFO("%s: worker thread %d is running", LOG_MAIN, thread->index);
+    while (!runtime->need_stop)
+    {
+        if (packet_io_thread_polling(handle, thread->index) == 0)
+        {
+            packet_io_thread_wait(handle, thread->index, -1);
+        }
+    }
+
+error_out:
+    LOG_ERROR("%s: worker thread %d exiting", LOG_MAIN, thread->index);
+    return (void *)NULL;
+}
+
+static void signal_handler(int signo)
+{
+    if (signo == SIGUSR1)
+    {
+        runtime->enable_debug = 1;
+        LOG_ERROR("%s: received SIGUSR1, enable debug", LOG_MAIN);
+    }
+    if (signo == SIGUSR2)
+    {
+        runtime->enable_debug = 0;
+        LOG_ERROR("%s: received SIGUSR2, disable debug", LOG_MAIN);
+    }
+    if (signo == SIGHUP)
+    {
+        LOG_RELOAD();
+        LOG_ERROR("%s: received SIGHUP, reload zlog.conf", LOG_MAIN);
+    }
+    if (signo == SIGINT)
+    {
+        runtime->need_stop = 1;
+        LOG_ERROR("%s: received SIGINT, exit !!!", LOG_MAIN);
+    }
+    if (signo == SIGQUIT)
+    {
+        runtime->need_stop = 1;
+        LOG_ERROR("%s: received SIGQUIT, exit !!!", LOG_MAIN);
+    }
+    if (signo == SIGTERM)
+    {
+        runtime->need_stop = 1;
+        LOG_ERROR("%s: received SIGTERM, exit !!!", LOG_MAIN);
+    }
+}
+
+static void usage(char *cmd)
+{
+    fprintf(stderr, "USAGE: %s [OPTIONS]\n", cmd);
+    fprintf(stderr, "    -v       -- show version\n");
+    fprintf(stderr, "    -d       -- run daemon\n");
+    fprintf(stderr, "    -h       -- show help\n");
+    fprintf(stderr, "Signal: \n");
+    fprintf(stderr, "    kill -s SIGUSR1 `pidof %s`     -- enable  debug\n", cmd);
+    fprintf(stderr, "    kill -s SIGUSR2 `pidof %s`     -- disable debug\n", cmd);
+}
+
+int main(int argc, char **argv)
+{
+    int opt;
+    const char *profile = "./conf/packet_adapter.conf";
+
+    if (LOG_INIT("./conf/zlog.conf") == -1)
+    {
+        return -1;
+    }
+
+    while ((opt = getopt(argc, argv, "vdh")) != -1)
+    {
+        switch (opt)
+        {
+        case 'v':
+            fprintf(stderr, "Packet Adapter Version: %s\n", Packet_Adapter_Version);
+            return 0;
+        case 'd':
+            run_daemon();
+            break;
+        case 'h': /* fall through */
+        default:
+            usage(argv[0]);
+            return 0;
+        }
+    }
+
+    LOG_ERROR("%s: TSG Packet Adapter Engine, Version: %s Start ...", LOG_MAIN, Packet_Adapter_Version);
+
+    if (signal(SIGUSR1, signal_handler) == SIG_ERR)
+    {
+        LOG_ERROR("%s: failed at signal(SIGUSR1), %d: %s", LOG_MAIN, errno, strerror(errno));
+        goto error;
+    }
+    if (signal(SIGUSR2, signal_handler) == SIG_ERR)
+    {
+        LOG_ERROR("%s: failed at signal(SIGUSR2), %d: %s", LOG_MAIN, errno, strerror(errno));
+        goto error;
+    }
+    if (signal(SIGHUP, signal_handler) == SIG_ERR)
+    {
+        LOG_ERROR("%s: failed at signal(SIGHUP), %d: %s", LOG_MAIN, errno, strerror(errno));
+        goto error;
+    }
+    if (signal(SIGINT, signal_handler) == SIG_ERR)
+    {
+        LOG_ERROR("%s: failed at signal(SIGINT), %d: %s", LOG_MAIN, errno, strerror(errno));
+        goto error;
+    }
+    if (signal(SIGQUIT, signal_handler) == SIG_ERR)
+    {
+        LOG_ERROR("%s: failed at signal(SIGQUIT), %d: %s", LOG_MAIN, errno, strerror(errno));
+        goto error;
+    }
+
+    if (signal(SIGTERM, signal_handler) == SIG_ERR)
+    {
+        LOG_ERROR("%s: failed at signal(SIGTERM), %d: %s", LOG_MAIN, errno, strerror(errno));
+        goto error;
+    }
+
+    runtime->stat = packet_stat_create(profile);
+    if (runtime->stat == NULL)
+    {
+        goto error;
+    }
+
+    runtime->handle = packet_io_create(profile);
+    if (runtime->handle == NULL)
+    {
+        goto error;
+    }
+
+    packet_io_set_callback(runtime->handle, packet_handle_callback, &runtime->metrics);
+    for (int i = 0; i < packet_io_thread_number(runtime->handle); i++)
+    {
+        runtime->threads[i].tid = 0;
+        runtime->threads[i].index = i;
+        runtime->threads[i].runtime = runtime;
+    }
+
+    for (int i = 0; i < packet_io_thread_number(runtime->handle); i++)
+    {
+        struct thread *thread = &runtime->threads[i];
+        if (pthread_create(&thread->tid, NULL, worker_thread_cycle, (void *)thread) < 0)
+        {
+            LOG_ERROR("%s: unable to create worker thread %d, error %d: %s", LOG_MAIN, i, errno, strerror(errno));
+            runtime->need_stop = 1;
+        }
+    }
+
+    while (!runtime->need_stop)
+    {
+        packet_stat_output(runtime->stat, &runtime->metrics);
+        sleep(packet_stat_cycle(runtime->stat));
+    }
+
+error:
+    packet_stat_destory(runtime->stat);
+    packet_io_destory(runtime->handle);
+    LOG_CLOSE();
+
+    return 0;
+}
\ No newline at end of file
diff --git a/platform/src/packet_handle.cpp b/platform/src/packet_handle.cpp
new file mode 100644
index 0000000..07f736d
--- /dev/null
+++ b/platform/src/packet_handle.cpp
@@ -0,0 +1,145 @@
+#include <stddef.h>
+
+#include "packet_parser.h"
+#include "packet_inject.h"
+#include "packet_handle.h"
+
+static void packet_inject(int next_proto, const char *data, int len, struct metrics *metrics)
+{
+    if (next_proto == 4)
+    {
+        struct ip *hdr = (struct ip *)data;
+        if (packet_inject_ipv4(&hdr->ip_dst, data, len) == 0)
+        {
+            ATOMIC_ADD(&metrics->succ_tx_v4_pkts, 1);
+            ATOMIC_ADD(&metrics->succ_tx_v4_bytes, len);
+        }
+        else
+        {
+            ATOMIC_ADD(&metrics->err_tx_v4_pkts, 1);
+            ATOMIC_ADD(&metrics->err_tx_v4_bytes, len);
+        }
+    }
+
+    if (next_proto == 6)
+    {
+        struct ip6_hdr *hdr = (struct ip6_hdr *)data;
+        if (packet_inject_ipv6(&hdr->ip6_dst, data, len) == 0)
+        {
+            ATOMIC_ADD(&metrics->succ_tx_v6_pkts, 1);
+            ATOMIC_ADD(&metrics->succ_tx_v6_bytes, len);
+        }
+        else
+        {
+            ATOMIC_ADD(&metrics->err_tx_v6_pkts, 1);
+            ATOMIC_ADD(&metrics->err_tx_v6_bytes, len);
+        }
+    }
+}
+
+static void packet_handle_error(struct metrics *metrics, int n_pkts, int n_bytes)
+{
+    ATOMIC_ADD(&metrics->rx_err_pkts, n_pkts);
+    ATOMIC_ADD(&metrics->rx_err_bytes, n_bytes);
+}
+
+// return 1: is  gtp
+// return 0: not gtp
+static int packet_handle_gtp(struct packet_parser *handler, struct metrics *metrics)
+{
+    const struct layer_record *gtp_layer = packet_parser_get_most_outer(handler, LAYER_TYPE_GTPV1_U);
+    if (gtp_layer == NULL)
+    {
+        return 0;
+    }
+
+    if (gtp_layer->hdr_offset + gtp_layer->hdr_len >= handler->packet_len)
+    {
+        packet_handle_error(metrics, 1, handler->packet_len);
+        return 1;
+    }
+
+    const char *inject_data = (const char *)handler->packet_data + gtp_layer->hdr_offset + gtp_layer->hdr_len;
+    int inject_len = gtp_layer->pld_len;
+    uint8_t next_proto = gtp_next_proto((const char *)handler->packet_data + gtp_layer->hdr_offset);
+
+    if (next_proto != 4 && next_proto != 6)
+    {
+        packet_handle_error(metrics, 1, handler->packet_len);
+        return 1;
+    }
+    else
+    {
+        packet_inject(next_proto, inject_data, inject_len, metrics);
+        return 1;
+    }
+}
+
+// return 1: is  l3
+// return 0: not l3
+static int packet_handle_l3(struct packet_parser *handler, struct metrics *metrics)
+{
+    const struct layer_record *l3_layer = packet_parser_get_most_outer(handler, LAYER_TYPE_L3);
+    if (l3_layer == NULL)
+    {
+        return 0;
+    }
+
+    if (l3_layer->hdr_offset >= handler->packet_len)
+    {
+        packet_handle_error(metrics, 1, handler->packet_len);
+        return 1;
+    }
+
+    const char *inject_data = (const char *)handler->packet_data + l3_layer->hdr_offset;
+    int inject_len = l3_layer->hdr_len + l3_layer->pld_len;
+    uint8_t next_proto = 0;
+
+    if (l3_layer->type == LAYER_TYPE_IPV4)
+    {
+        next_proto = 4;
+    }
+    else if (l3_layer->type == LAYER_TYPE_IPV6)
+    {
+        next_proto = 6;
+    }
+    else
+    {
+        packet_handle_error(metrics, 1, handler->packet_len);
+        return 1;
+    }
+
+    packet_inject(next_proto, inject_data, inject_len, metrics);
+    return 1;
+}
+
+void packet_handle(const char *data, int len, struct metrics *metrics)
+{
+    ATOMIC_ADD(&metrics->rx_pkts, 1);
+    ATOMIC_ADD(&metrics->rx_bytes, len);
+
+    if (data == NULL || len <= 0)
+    {
+        packet_handle_error(metrics, 1, len);
+        return;
+    }
+
+    struct packet_parser handler;
+    uint64_t packet_id = ATOMIC_READ(&metrics->rx_pkts);
+    packet_parser_init(&handler);
+    packet_parser_parse(&handler, data, len, packet_id);
+
+    // Handle GTP
+    if (packet_handle_gtp(&handler, metrics) == 1)
+    {
+        return;
+    }
+
+    // Handle L3
+    if (packet_handle_l3(&handler, metrics) == 1)
+    {
+        return;
+    }
+
+    packet_handle_error(metrics, 1, len);
+}
diff --git a/platform/src/packet_stat.cpp b/platform/src/packet_stat.cpp
new file mode 100644
index 0000000..d9d3e40
--- /dev/null
+++ b/platform/src/packet_stat.cpp
@@ -0,0 +1,181 @@
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <MESA/field_stat2.h>
+#include <MESA/MESA_prof_load.h>
+
+#include "log.h"
+#include "utils.h"
+
+enum STAT_FIELD
+{
+    STAT_RX_PKT,
+    STAT_RX_B,
+
+    STAT_RX_ERR_PKT,
+    STAT_RX_ERR_B,
+
+    STAT_SUCC_TX_V4_PKT,
+    STAT_SUCC_TX_V4_B,
+
+    STAT_SUCC_TX_V6_PKT,
+    STAT_SUCC_TX_V6_B,
+
+    STAT_ERR_TX_V4_PKT,
+    STAT_ERR_TX_V4_B,
+
+    STAT_ERR_TX_V6_PKT,
+    STAT_ERR_TX_V6_B,
+
+    STAT_MAX,
+};
+
+static const char *stat_map[] =
+    {
+        [STAT_RX_PKT] = "nf_rx_pkt",
+        [STAT_RX_B] = "nf_rx_B",
+
+        [STAT_RX_ERR_PKT] = "rx_err_pkt",
+        [STAT_RX_ERR_B] = "rx_err_B",
+
+        [STAT_SUCC_TX_V4_PKT] = "succ_tx_4_pkt",
+        [STAT_SUCC_TX_V4_B] = "succ_tx_4_B",
+
+        [STAT_SUCC_TX_V6_PKT] = "succ_tx_6_pkt",
+        [STAT_SUCC_TX_V6_B] = "succ_tx_6_B",
+
+        [STAT_ERR_TX_V4_PKT] = "err_tx_4_pkt",
+        [STAT_ERR_TX_V4_B] = "err_tx_4_B",
+
+        [STAT_ERR_TX_V6_PKT] = "err_tx_6_pkt",
+        [STAT_ERR_TX_V6_B] = "err_tx_6_B",
+
+        [STAT_MAX] = NULL};
+
+struct stat_config
+{
+    char output_file[256];
+    char statsd_server[32];
+    int statsd_port;
+    int statsd_format;
+    int statsd_cycle;
+
+    int prometheus_listen_port;
+    char prometheus_listen_url[256];
+};
+
+struct packet_stat
+{
+    struct stat_config config;
+    screen_stat_handle_t fs_handle;
+    int fs_id[512];
+};
+
+static void packet_stat_config(const char *profile, struct stat_config *config)
+{
+    MESA_load_profile_string_def(profile, "STAT", "output_file", config->output_file, sizeof(config->output_file), "log/packet_adapter.fs2");
+    MESA_load_profile_string_def(profile, "STAT", "statsd_server", config->statsd_server, sizeof(config->statsd_server), "127.0.0.1");
+    MESA_load_profile_int_def(profile, "STAT", "statsd_port", &(config->statsd_port), 8100);
+    MESA_load_profile_int_def(profile, "STAT", "statsd_format", &(config->statsd_format), 1); // FS_OUTPUT_STATSD=1, FS_OUTPUT_INFLUX_LINE=2
+    MESA_load_profile_int_def(profile, "STAT", "statsd_cycle", &(config->statsd_cycle), 1);
+
+    MESA_load_profile_int_def(profile, "STAT", "prometheus_listen_port", &(config->prometheus_listen_port), 9001);
+    MESA_load_profile_string_def(profile, "STAT", "prometheus_listen_url", config->prometheus_listen_url, sizeof(config->prometheus_listen_url), "/packet_prometheus");
+
+    if (config->statsd_format != 1 && config->statsd_format != 2)
+    {
+        config->statsd_format = 1;
+    }
+
+    LOG_DEBUG("STAT->output_file            : %s", config->output_file);
+    LOG_DEBUG("STAT->statsd_server          : %s", config->statsd_server);
+    LOG_DEBUG("STAT->statsd_port            : %d", config->statsd_port);
+    LOG_DEBUG("STAT->statsd_format          : %d", config->statsd_format);
+    LOG_DEBUG("STAT->statsd_cycle           : %d", config->statsd_cycle);
+    LOG_DEBUG("STAT->prometheus_listen_port : %d", config->prometheus_listen_port);
+    LOG_DEBUG("STAT->prometheus_listen_url  : %s", config->prometheus_listen_url);
+}
+
+struct packet_stat *packet_stat_create(const char *profile)
+{
+    struct packet_stat *handle = (struct packet_stat *)calloc(1, sizeof(struct packet_stat));
+    assert(handle != NULL);
+
+    packet_stat_config(profile, &handle->config);
+
+    FS_library_set_prometheus_port(handle->config.prometheus_listen_port);
+    FS_library_set_prometheus_url_path(handle->config.prometheus_listen_url);
+    FS_library_init();
+
+    int value = 0;
+    handle->fs_handle = FS_create_handle(); // TODO memleak no free() API
+    FS_set_para(handle->fs_handle, APP_NAME, "packet_adapter", 13);
+    FS_set_para(handle->fs_handle, OUTPUT_DEVICE, handle->config.output_file, strlen(handle->config.output_file));
+    value = 1;
+    FS_set_para(handle->fs_handle, OUTPUT_PROMETHEUS, &value, sizeof(value));
+    value = 1;
+    FS_set_para(handle->fs_handle, PRINT_MODE, &value, sizeof(value));
+    value = 0;
+    FS_set_para(handle->fs_handle, CREATE_THREAD, &value, sizeof(value));
+
+    if (strlen(handle->config.statsd_server) > 0 && handle->config.statsd_port != 0)
+    {
+        FS_set_para(handle->fs_handle, STATS_SERVER_IP, handle->config.statsd_server, strlen(handle->config.statsd_server));
+        FS_set_para(handle->fs_handle, STATS_SERVER_PORT, &(handle->config.statsd_port), sizeof(handle->config.statsd_port));
+        FS_set_para(handle->fs_handle, STATS_FORMAT, &handle->config.statsd_format, sizeof(handle->config.statsd_format));
+    }
+
+    for (int i = 0; i < STAT_MAX; i++)
+    {
+        handle->fs_id[i] = FS_register(handle->fs_handle, FS_STYLE_FIELD, FS_CALC_CURRENT, stat_map[i]);
+    }
+
+    FS_start(handle->fs_handle);
+
+    return handle;
+}
+
+void packet_stat_destory(struct packet_stat *handle)
+{
+    if (handle)
+    {
+        FS_library_destroy();
+        free(handle);
+        handle = NULL;
+    }
+}
+
+void packet_stat_output(struct packet_stat *handle, struct metrics *metrics)
+{
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_RX_PKT], 0, FS_OP_SET, ATOMIC_READ(&(metrics->rx_pkts)));
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_RX_B], 0, FS_OP_SET, ATOMIC_READ(&(metrics->rx_bytes)));
+
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_RX_ERR_PKT], 0, FS_OP_SET, ATOMIC_READ(&(metrics->rx_err_pkts)));
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_RX_ERR_B], 0, FS_OP_SET, ATOMIC_READ(&(metrics->rx_err_bytes)));
+
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_SUCC_TX_V4_PKT], 0, FS_OP_SET, ATOMIC_READ(&(metrics->succ_tx_v4_pkts)));
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_SUCC_TX_V4_B], 0, FS_OP_SET, ATOMIC_READ(&(metrics->succ_tx_v4_bytes)));
+
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_SUCC_TX_V6_PKT], 0, FS_OP_SET, ATOMIC_READ(&(metrics->succ_tx_v6_pkts)));
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_SUCC_TX_V6_B], 0, FS_OP_SET, ATOMIC_READ(&(metrics->succ_tx_v6_bytes)));
+
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_ERR_TX_V4_PKT], 0, FS_OP_SET, ATOMIC_READ(&(metrics->err_tx_v4_pkts)));
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_ERR_TX_V4_B], 0, FS_OP_SET, ATOMIC_READ(&(metrics->err_tx_v4_bytes)));
+
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_ERR_TX_V6_PKT], 0, FS_OP_SET, ATOMIC_READ(&(metrics->err_tx_v6_pkts)));
+    FS_operate(handle->fs_handle, handle->fs_id[STAT_ERR_TX_V6_B], 0, FS_OP_SET, ATOMIC_READ(&(metrics->err_tx_v6_bytes)));
+
+    FS_passive_output(handle->fs_handle);
+}
+
+int packet_stat_cycle(struct packet_stat *handle)
+{
+    if (handle)
+    {
+        return handle->config.statsd_cycle;
+    }
+    else
+    {
+        return 0;
+    }
+}
\ No newline at end of file
diff --git a/platform/src/system.c b/platform/src/system.cpp
index 76141de..7cbdf0e 100644
--- a/platform/src/system.c
+++ b/platform/src/system.cpp
@@ -1,4 +1,11 @@
-#include "system.h"
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <stdlib.h>
+
+#include "log.h"
 
 int run_daemon(void)
 {