dplaneserver: Add dplaneserver to test dplane_fpm_pb.

`dplaneserver` is a socket server-side program that emulates fpm to receive data from zebra and decodes it in protobuf format. It's used to test dplane_fpm_pb. Signed-off-by: Hongyu Li <[email protected]>
FRRouting · Aug 25, 2023 · 9f3539b · 9f3539b
1 parent e7d4dce
commit 9f3539b
Show file tree

Hide file tree

Showing 6 changed files with 481 additions and 0 deletions.
diff --git a/Makefile.am b/Makefile.am
@@ -181,6 +181,7 @@ include zebra/subdir.am
 include watchfrr/subdir.am
 include qpb/subdir.am
 include fpm/subdir.am
+include dplaneserver/subdir.am
 include grpc/subdir.am
 include tools/subdir.am
 
@@ -272,6 +273,7 @@ EXTRA_DIST += \
 	doc/user/Makefile \
 	eigrpd/Makefile \
 	fpm/Makefile \
+	dplaneserver/Makefile \
 	grpc/Makefile \
 	isisd/Makefile \
 	ldpd/Makefile \

diff --git a/dplaneserver/Makefile b/dplaneserver/Makefile
@@ -0,0 +1,10 @@
+all: ALWAYS
+	@$(MAKE) -s -C .. dplaneserver/dplaneserver
+%: ALWAYS
+	@$(MAKE) -s -C .. dplaneserver/$@
+
+Makefile:
+	#nothing
+ALWAYS:
+.PHONY: ALWAYS makefiles
+.SUFFIXES:
diff --git a/dplaneserver/dplaneserver.c b/dplaneserver/dplaneserver.c
@@ -0,0 +1,284 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ * Server socket program to simulate fpm using protobuf
+ * Copyright (C) 2023 Alibaba, Inc.
+ *                    Hongyu Li
+ */
+#include "dplaneserver.h"
+#include "fpm/fpm.pb-c.h"
+
+#define FPM_HEADER_SIZE 4
+DEFINE_MGROUP(DPLANESERVER, "dplaneserver");
+DEFINE_MTYPE(DPLANESERVER, DPLANE_BUFFER, "dplaneserver communication");
+struct Dplaneserver_data dplaneserver_data = { .bufSize = 2048,
+					     .messageBuffer = NULL,
+					     .pos = 0,
+					     .server_socket = 0,
+					     .connection_socket = 0,
+					     .connected = false,
+					     .server_up = false };
+enum fpm_msg_op fm_op;
+
+void process_fpm_msg(struct fpm_msg_hdr_t *fpm_hdr)
+{
+	size_t msg_len = fpm_msg_len(fpm_hdr);
+	Fpm__Message *msg;
+
+	msg = fpm__message__unpack(NULL, msg_len - FPM_HEADER_SIZE,
+				   (uint8_t *)fpm_msg_data(fpm_hdr));
+	if (msg) {
+		fm_op = msg->type;
+		switch (fm_op) {
+		case FPM_OP_ROUTE_INSTALL:
+			if (!msg->add_route) {
+				zlog_err("%s: ROUTE_INSTALL info doesn't exist",
+					 __func__);
+				break;
+			}
+			process_route_install_msg(msg->add_route);
+			break;
+		/* Un-handled at this time */
+		case FPM_OP_ROUTE_DELETE:
+			break;
+		}
+		fpm__message__free_unpacked(msg, NULL);
+	} else {
+		zlog_err("%s: unpack fpm message failed", __func__);
+		return;
+	}
+}
+
+int dplaneserver_init(void)
+{
+	struct sockaddr_in server_addr;
+
+	dplaneserver_data.server_socket = socket(PF_INET, SOCK_STREAM,
+						IPPROTO_TCP);
+	if (dplaneserver_data.server_socket < 0) {
+		zlog_err("%s: Can not open socket", __func__);
+		return -1;
+	}
+
+	if (sockopt_reuseaddr(dplaneserver_data.server_socket) == -1) {
+		zlog_err("%s: Can not set socket opt", __func__);
+		return -1;
+	}
+
+	// bind port
+	memset(&server_addr, 0, sizeof(server_addr));
+	if (is_ipv6)
+		server_addr.sin_family = AF_INET6;
+	else
+		server_addr.sin_family = AF_INET;
+	server_addr.sin_port = htons(FPM_DEFAULT_PORT);
+	server_addr.sin_addr.s_addr = FPM_DEFAULT_IP;
+
+	if (bind(dplaneserver_data.server_socket,
+		 (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
+		zlog_err("%s: bing socket to address failed", __func__);
+		return -1;
+	}
+
+	if (listen(dplaneserver_data.server_socket, 10) == -1) {
+		zlog_err("%s: listen socket failed", __func__);
+		return -1;
+	}
+
+	dplaneserver_data.server_up = true;
+	dplaneserver_data.messageBuffer = XMALLOC(MTYPE_DPLANE_BUFFER,
+						 dplaneserver_data.bufSize);
+
+	if (IS_DPLANE_SERVER_DEBUG)
+		zlog_debug("%s: connect socket successfully", __func__);
+	return 0;
+}
+
+void dplaneserver_exit(void)
+{
+	XFREE(MTYPE_DPLANE_BUFFER, dplaneserver_data.messageBuffer);
+	if (dplaneserver_data.connected)
+		close(dplaneserver_data.connection_socket);
+	if (dplaneserver_data.server_up)
+		close(dplaneserver_data.server_socket);
+}
+
+int dplaneserver_read_data(void)
+{
+	struct fpm_msg_hdr_t *fpm_hdr;
+	size_t msg_len;
+	size_t start = 0, left;
+	ssize_t read_len;
+
+	read_len = read(dplaneserver_data.connection_socket,
+			dplaneserver_data.messageBuffer + dplaneserver_data.pos,
+			dplaneserver_data.bufSize - dplaneserver_data.pos);
+
+	if (read_len == 0) {
+		if (IS_DPLANE_SERVER_DEBUG)
+			zlog_debug("%s: socket connection closed", __func__);
+		return -2;
+	}
+	if (read_len < 0) {
+		zlog_err("%s: socket connection read error", __func__);
+		return -1;
+	}
+	dplaneserver_data.pos += (uint32_t)read_len;
+	while (true) {
+		fpm_hdr = (struct fpm_msg_hdr_t *)(dplaneserver_data.messageBuffer +
+					    start);
+		left = dplaneserver_data.pos - start;
+		if (left < FPM_MSG_HDR_LEN)
+			break;
+		/* fpm_msg_len includes header size */
+		msg_len = fpm_msg_len(fpm_hdr);
+		if (left < msg_len)
+			break;
+		if (!fpm_msg_ok(fpm_hdr, left)) {
+			zlog_err("%s: fpm message header check failed");
+			return -1;
+		}
+		process_fpm_msg(fpm_hdr);
+		start += msg_len;
+	}
+	/* update msg buffer*/
+	memmove(dplaneserver_data.messageBuffer,
+		dplaneserver_data.messageBuffer + start,
+		dplaneserver_data.pos - start);
+	dplaneserver_data.pos = dplaneserver_data.pos - (uint32_t)start;
+	return 0;
+}
+
+int dplaneserver_poll(void)
+{
+	struct pollfd poll_fd_set[MAX_CLIENTS + 1];
+
+	memset(poll_fd_set, 0, sizeof(poll_fd_set));
+	poll_fd_set[0].fd = dplaneserver_data.server_socket;
+	poll_fd_set[0].events = POLLIN;
+	while (true) {
+		// poll for events
+		int nready = poll(poll_fd_set, MAX_CLIENTS + 1, -1);
+
+		if (nready == -1) {
+			zlog_err("%s: failed to poll socket", __func__);
+			return -1;
+		}
+		if (poll_fd_set[0].revents & POLLIN) {
+			struct sockaddr_in client_addr;
+			int i;
+			socklen_t addr_len = sizeof(client_addr);
+			int client_fd = accept(dplaneserver_data.server_socket,
+					       (struct sockaddr *)&client_addr,
+					       &addr_len);
+
+			if (client_fd == -1) {
+				zlog_err("%s: failed to accept client connection",
+					 __func__);
+				continue;
+			}
+			// add new connection to poll fd set
+			for (i = 1; i <= MAX_CLIENTS; i++) {
+				if (poll_fd_set[i].fd == 0) {
+					if (IS_DPLANE_SERVER_DEBUG)
+						zlog_debug("%s: a new client has connected",
+							   __func__);
+					poll_fd_set[i].fd = client_fd;
+					poll_fd_set[i].events = POLLIN;
+					dplaneserver_data.connection_socket =
+						client_fd;
+					dplaneserver_data.connected = true;
+					break;
+				}
+			}
+			if (i > MAX_CLIENTS) {
+				close(client_fd);
+				continue;
+			}
+		}
+		// check for events on client sockets
+		for (int i = 1; i <= MAX_CLIENTS; i++) {
+			if (poll_fd_set[i].fd == 0)
+				continue;
+			if (poll_fd_set[i].revents & POLLIN) {
+				int res = dplaneserver_read_data();
+				/* if func return -1 or -2 it means errors occur*/
+				if (res)
+					return res;
+			}
+			if (poll_fd_set[i].revents &
+			    (POLLERR | POLLHUP | POLLNVAL)) {
+				if (IS_DPLANE_SERVER_DEBUG)
+					zlog_debug("%s: socket POLLERR | POLLHUP | POLLNVAL event happened",
+						   __func__);
+				close(poll_fd_set[i].fd);
+				poll_fd_set[i].fd = 0;
+			}
+		}
+	}
+}
+
+void process_route_install_msg(Fpm__AddRoute *msg)
+{
+	char buf[4096] = { 0 };
+
+	if (!msg->key) {
+		zlog_err("%s: ROUTE_INSTALL route key doesn't exist", __func__);
+		return;
+	}
+	if (!msg->key->prefix) {
+		zlog_err("%s: ROUTE_INSTALL prefix doesn't exist", __func__);
+		return;
+	}
+	if (IS_DPLANE_SERVER_DEBUG)
+		zlog_debug("%s: msg address family:%d",
+			  __func__, msg->address_family);
+	if (msg->address_family == AF_INET) {
+		inet_ntop(AF_INET, msg->key->prefix->bytes.data, buf,
+			  sizeof(buf));
+		if (IS_DPLANE_SERVER_DEBUG)
+			zlog_debug("%s: key ipv4 prefix:%pI4", __func__, buf);
+	} else if (msg->address_family == AF_INET6) {
+		inet_ntop(AF_INET6, msg->key->prefix->bytes.data, buf,
+			  sizeof(buf));
+		if (IS_DPLANE_SERVER_DEBUG)
+			zlog_debug("%s: key ipv6 prefix:%pI6", __func__, buf);
+	} else {
+		zlog_err("%s: not ipv4 or ipv6 address family", __func__);
+		return;
+	}
+	if (IS_DPLANE_SERVER_DEBUG)
+		zlog_debug("%s: key length:%d", __func__, msg->key->prefix->length);
+
+	json_object *json = json_object_new_object();
+
+	json_object_int_add(json, "vrfId", (int64_t)(msg->vrf_id));
+	json_object_int_add(json, "addressFamily",
+			    (int64_t)(msg->address_family));
+	json_object_int_add(json, "metric", (int64_t)(msg->metric));
+	json_object_int_add(json, "subAddressFamily",
+			    (int64_t)(msg->sub_address_family));
+	json_object_int_add(json, "hasRouteType",
+			    (int64_t)(msg->has_route_type));
+	json_object_int_add(json, "routeType", (int64_t)(msg->route_type));
+	if (msg->key) {
+		json_object_string_add(json, "prefix", buf);
+		json_object_int_add(json, "prefixLength",
+				    (int64_t)(msg->key->prefix->length));
+	}
+	if (output_file_path) {
+		FILE *fp = fopen(output_file_path, "a+");
+
+		if (!fp) {
+			zlog_err("%s open output json file failed:%s", __func__,
+				 output_file_path);
+		} else {
+			fprintf(fp, "%s\n",
+				json_object_to_json_string_ext(json,
+							       JSON_C_TO_STRING_PRETTY));
+			fclose(fp);
+		}
+	} else {
+		zlog_err("%s output json file doesn't exist", __func__);
+	}
+	json_object_free(json);
+}
diff --git a/dplaneserver/dplaneserver.h b/dplaneserver/dplaneserver.h
@@ -0,0 +1,74 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ * Server socket program to simulate fpm using protobuf
+ * Copyright (C) 2023 Alibaba, Inc.
+ *                    Hongyu Li
+ */
+#ifndef _DPLANESERVER_H
+#define _DPLANESERVER_H
+
+#ifdef HAVE_CONFIG_H
+#include "config.h" /* Include this explicitly */
+#endif
+
+#include <arpa/inet.h>
+#include <stdio.h>
+#include <netinet/in.h>
+#include <sys/poll.h>
+#include <errno.h>
+#include <net/if.h>
+#include <unistd.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include "zlog.h"
+#include "fpm/fpm.h"
+#include "lib/json.h"
+#include "fpm/fpm.pb-c.h"
+#include "lib/memory.h"
+#include "lib/sockunion.h"
+
+extern char *output_file_path;
+extern struct Dplaneserver_data dplaneserver_data;
+extern bool is_ipv6;
+extern bool debug_mode;
+
+#define MAX_CLIENTS 10
+#define BUFFER_SIZE 1024
+#define DPLANE_SERVER_DEBUG 0x01
+#define IS_DPLANE_SERVER_DEBUG (debug_mode & DPLANE_SERVER_DEBUG)
+
+#define FPM_DEFAULT_PORT 2620
+#ifndef FPM_DEFAULT_IP
+#define FPM_DEFAULT_IP (htonl(INADDR_LOOPBACK))
+#endif
+#ifndef INADDR_LOOPBACK
+#define INADDR_LOOPBACK 0x7f000001 /* Internet address 127.0.0.1.  */
+#endif
+
+DECLARE_MTYPE(DPLANE_BUFFER);
+
+struct Dplaneserver_data {
+	unsigned int bufSize;
+	char *messageBuffer;
+	unsigned int pos;
+	int server_socket;
+	int connection_socket;
+	bool connected;
+	bool server_up;
+};
+
+enum fpm_msg_op {
+	FPM_OP_NONE = 0,
+
+	/* Route update */
+	FPM_OP_ROUTE_INSTALL,
+	FPM_OP_ROUTE_DELETE,
+};
+
+int dplaneserver_init(void);
+void dplaneserver_exit(void);
+int dplaneserver_poll(void);
+int dplaneserver_read_data(void);
+void process_route_install_msg(Fpm__AddRoute *msg);
+
+#endif