bfdd: support connecting to BFD data plane

Add option to connect to a data plane server instead of receiving
connections.

Signed-off-by: Rafael Zalamena <rzalamena@opensourcerouting.org>

3 files changed, 229 insertions, 15 deletions

diff --git a/bfdd/bfd.h b/bfdd/bfd.h
index d6e762c52..7c537b40d 100644
--- a/bfdd/bfd.h
+++ b/bfdd/bfd.h
@@ -762,10 +762,11 @@ int ptm_bfd_notify(struct bfd_session *bs, uint8_t notify_state);
 /**
  * Initialize BFD data plane infrastructure for distributed BFD implementation.
  *
- * \param sa listening socket address.
- * \param salen listening socket address structure length.
+ * \param sa socket address.
+ * \param salen socket address structure length.
+ * \param client `true` means connecting socket, `false` listening socket.
  */
-void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen);
+void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen, bool client);
 
 /**
  * Attempts to delegate the BFD session liveness detection to hardware.
diff --git a/bfdd/bfdd.c b/bfdd/bfdd.c
index 5fda85e81..b8a059708 100644
--- a/bfdd/bfdd.c
+++ b/bfdd/bfdd.c
@@ -204,6 +204,7 @@ static void
 distributed_bfd_init(const char *arg)
 {
 	char *sptr, *saux;
+	bool is_client = false;
 	size_t slen;
 	socklen_t salen;
 	char addr[64];
@@ -235,11 +236,17 @@ distributed_bfd_init(const char *arg)
 	memset(&sa, 0, sizeof(sa));
 
 	/* Fill the address information. */
-	if (strcmp(type, "unix") == 0) {
+	if (strcmp(type, "unix") == 0 || strcmp(type, "unixc") == 0) {
+		if (strcmp(type, "unixc") == 0)
+			is_client = true;
+
 		salen = sizeof(sa.sun);
 		sa.sun.sun_family = AF_UNIX;
 		strlcpy(sa.sun.sun_path, addr, sizeof(sa.sun.sun_path));
-	} else if (strcmp(type, "ipv4") == 0) {
+	} else if (strcmp(type, "ipv4") == 0 || strcmp(type, "ipv4c") == 0) {
+		if (strcmp(type, "ipv4c") == 0)
+			is_client = true;
+
 		salen = sizeof(sa.sin);
 		sa.sin.sin_family = AF_INET;
 
@@ -255,7 +262,10 @@ distributed_bfd_init(const char *arg)
 		if (inet_pton(AF_INET, addr, &sa.sin.sin_addr) != 1)
 			errx(1, "%s: inet_pton: invalid address %s", __func__,
 			     addr);
-	} else if (strcmp(type, "ipv6") == 0) {
+	} else if (strcmp(type, "ipv6") == 0 || strcmp(type, "ipv6c") == 0) {
+		if (strcmp(type, "ipv6c") == 0)
+			is_client = true;
+
 		salen = sizeof(sa.sin6);
 		sa.sin6.sin6_family = AF_INET6;
 
@@ -295,7 +305,7 @@ distributed_bfd_init(const char *arg)
 	}
 
 	/* Initialize BFD data plane listening socket. */
-	bfd_dplane_init((struct sockaddr *)&sa, salen);
+	bfd_dplane_init((struct sockaddr *)&sa, salen, is_client);
 }
 
 static void bg_init(void)
diff --git a/bfdd/dplane.c b/bfdd/dplane.c
index f68fd2cb2..b8f0aadd9 100644
--- a/bfdd/dplane.c
+++ b/bfdd/dplane.c
@@ -22,6 +22,7 @@
 #include <zebra.h>
 
 #include <netinet/in.h>
+#include <netinet/tcp.h>
 #include <sys/socket.h>
 #include <sys/un.h>
 
@@ -35,6 +36,7 @@
 #include <time.h>
 
 #include "lib/hook.h"
+#include "lib/network.h"
 #include "lib/printfrr.h"
 #include "lib/stream.h"
 #include "lib/thread.h"
@@ -52,6 +54,19 @@ DEFINE_MTYPE_STATIC(BFDD, BFDD_DPLANE_CTX, "Data plane client allocated memory")
 struct bfd_dplane_ctx {
 	/** Client file descriptor. */
 	int sock;
+	/** Is this a connected or accepted? */
+	bool client;
+	/** Is the socket still connecting? */
+	bool connecting;
+	/** Client/server address. */
+	union {
+		struct sockaddr sa;
+		struct sockaddr_in sin;
+		struct sockaddr_in6 sin6;
+		struct sockaddr_un sun;
+	} addr;
+	/** Address length. */
+	socklen_t addrlen;
 	/** Data plane current last used ID. */
 	uint16_t last_id;
 
@@ -63,6 +78,8 @@ struct bfd_dplane_ctx {
 	struct thread *inbufev;
 	/** Output event data. */
 	struct thread *outbufev;
+	/** Connection event. */
+	struct thread *connectev;
 
 	/** Amount of bytes read. */
 	uint64_t in_bytes;
@@ -89,6 +106,8 @@ struct bfd_dplane_ctx {
  */
 typedef void (*bfd_dplane_expect_cb)(struct bfddp_message *msg, void *arg);
 
+static int bfd_dplane_client_connect(struct thread *t);
+static bool bfd_dplane_client_connecting(struct bfd_dplane_ctx *bdc);
 static void bfd_dplane_ctx_free(struct bfd_dplane_ctx *bdc);
 static int _bfd_dplane_add_session(struct bfd_dplane_ctx *bdc,
 				   struct bfd_session *bs);
@@ -310,7 +329,14 @@ static ssize_t bfd_dplane_flush(struct bfd_dplane_ctx *bdc)
 
 static int bfd_dplane_write(struct thread *t)
 {
-	bfd_dplane_flush(THREAD_ARG(t));
+	struct bfd_dplane_ctx *bdc = THREAD_ARG(t);
+
+	/* Handle connection stage. */
+	if (bdc->connecting && bfd_dplane_client_connecting(bdc))
+		return 0;
+
+	bfd_dplane_flush(bdc);
+
 	return 0;
 }
 
@@ -395,6 +421,10 @@ static int bfd_dplane_enqueue(struct bfd_dplane_ctx *bdc, const void *buf,
 {
 	size_t rlen;
 
+	/* Handle not connected yet client. */
+	if (bdc->client && bdc->sock == -1)
+		return -1;
+
 	/* Not enough space. */
 	if (buflen > STREAM_WRITEABLE(bdc->outbuf)) {
 		bdc->out_fullev++;
@@ -626,6 +656,11 @@ static struct bfd_dplane_ctx *bfd_dplane_ctx_new(int sock)
 	bdc->sock = sock;
 	bdc->inbuf = stream_new(BFD_DPLANE_CLIENT_BUF_SIZE);
 	bdc->outbuf = stream_new(BFD_DPLANE_CLIENT_BUF_SIZE);
+
+	/* If not socket ready, skip read and session registration. */
+	if (sock == -1)
+		return bdc;
+
 	thread_add_read(master, bfd_dplane_read, bdc, sock, &bdc->inbufev);
 
 	/* Register all unattached sessions. */
@@ -654,6 +689,25 @@ static void bfd_dplane_ctx_free(struct bfd_dplane_ctx *bdc)
 		zlog_debug("%s: terminating data plane client %d", __func__,
 			   bdc->sock);
 
+	/* Client mode has special treatment. */
+	if (bdc->client) {
+		/* Disable connection event if any. */
+		THREAD_OFF(bdc->connectev);
+
+		/* Normal treatment on shutdown. */
+		if (bglobal.bg_shutdown)
+			goto free_resources;
+
+		/* Attempt reconnection. */
+		socket_close(&bdc->sock);
+		THREAD_OFF(bdc->inbufev);
+		THREAD_OFF(bdc->outbufev);
+		thread_add_timer(master, bfd_dplane_client_connect, bdc, 3,
+				 &bdc->connectev);
+		return;
+	}
+
+free_resources:
 	/* Remove from the list of attached data planes. */
 	TAILQ_REMOVE(&bglobal.bg_dplaneq, bdc, entry);
 
@@ -810,6 +864,146 @@ reschedule_and_return:
 	return 0;
 }
 
+/*
+ * Data plane connecting socket.
+ */
+static void _bfd_dplane_client_bootstrap(struct bfd_dplane_ctx *bdc)
+{
+	bdc->connecting = false;
+
+	/* Clean up buffers. */
+	stream_reset(bdc->inbuf);
+	stream_reset(bdc->outbuf);
+
+	/* Ask for read notifications. */
+	thread_add_read(master, bfd_dplane_read, bdc, bdc->sock, &bdc->inbufev);
+
+	/* Remove all sessions then register again to send them all. */
+	bfd_key_iterate(_bfd_session_unregister_dplane, bdc);
+	bfd_key_iterate(_bfd_session_register_dplane, bdc);
+}
+
+static bool bfd_dplane_client_connecting(struct bfd_dplane_ctx *bdc)
+{
+	int rv;
+	socklen_t rvlen = sizeof(rv);
+
+	/* Make sure `errno` is reset, then test `getsockopt` success. */
+	errno = 0;
+	if (getsockopt(bdc->sock, SOL_SOCKET, SO_ERROR, &rv, &rvlen) == -1)
+		rv = -1;
+
+	/* Connection successful. */
+	if (rv == 0) {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: connected to server: %d", __func__,
+				   bdc->sock);
+
+		_bfd_dplane_client_bootstrap(bdc);
+		return false;
+	}
+
+	switch (rv) {
+	case EINTR:
+	case EAGAIN:
+	case EALREADY:
+	case EINPROGRESS:
+		/* non error, wait more. */
+		return true;
+
+	default:
+		zlog_warn("%s: connection failed: %s", __func__,
+			  strerror(errno));
+		bfd_dplane_ctx_free(bdc);
+		return true;
+	}
+}
+
+static int bfd_dplane_client_connect(struct thread *t)
+{
+	struct bfd_dplane_ctx *bdc = THREAD_ARG(t);
+	int rv, sock;
+	socklen_t rvlen = sizeof(rv);
+
+	/* Allocate new socket. */
+	sock = socket(bdc->addr.sa.sa_family, SOCK_STREAM, 0);
+	if (sock == -1) {
+		zlog_warn("%s: failed to initialize socket: %s", __func__,
+			  strerror(errno));
+		goto reschedule_connect;
+	}
+
+	/* Set non blocking socket. */
+	set_nonblocking(sock);
+
+	/* Set 'no delay' (disables nagle algorithm) for IPv4/IPv6. */
+	rv = 1;
+	if (bdc->addr.sa.sa_family != AF_UNIX
+	    && setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &rv, rvlen) == -1)
+		zlog_warn("%s: TCP_NODELAY: %s", __func__, strerror(errno));
+
+	/* Attempt to connect. */
+	rv = connect(sock, &bdc->addr.sa, bdc->addrlen);
+	if (rv == -1 && (errno != EINPROGRESS && errno != EAGAIN)) {
+		zlog_warn("%s: data plane connection failed: %s", __func__,
+			  strerror(errno));
+		goto reschedule_connect;
+	}
+
+	bdc->sock = sock;
+	if (rv == -1) {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: server connection in progress: %d",
+				   __func__, sock);
+
+		/* If we are not connected yet, ask for write notifications. */
+		bdc->connecting = true;
+		thread_add_write(master, bfd_dplane_write, bdc, bdc->sock,
+				 &bdc->outbufev);
+	} else {
+		if (bglobal.debug_dplane)
+			zlog_debug("%s: server connection: %d", __func__, sock);
+
+		/* Otherwise just start accepting data. */
+		_bfd_dplane_client_bootstrap(bdc);
+	}
+
+	return 0;
+
+reschedule_connect:
+	THREAD_OFF(bdc->inbufev);
+	THREAD_OFF(bdc->outbufev);
+	socket_close(&sock);
+	thread_add_timer(master, bfd_dplane_client_connect, bdc, 3,
+			 &bdc->connectev);
+	return 0;
+}
+
+static void bfd_dplane_client_init(const struct sockaddr *sa, socklen_t salen)
+{
+	struct bfd_dplane_ctx *bdc;
+
+	/* Allocate context and copy address for reconnection. */
+	bdc = bfd_dplane_ctx_new(-1);
+	if (salen <= sizeof(bdc->addr)) {
+		memcpy(&bdc->addr, sa, salen);
+		bdc->addrlen = sizeof(bdc->addr);
+	} else {
+		memcpy(&bdc->addr, sa, sizeof(bdc->addr));
+		bdc->addrlen = sizeof(bdc->addr);
+		zlog_warn("%s: server address truncated (from %d to %d)",
+			  __func__, salen, bdc->addrlen);
+	}
+
+	bdc->client = true;
+
+	thread_add_timer(master, bfd_dplane_client_connect, bdc, 0,
+			 &bdc->connectev);
+
+	/* Insert into data plane lists. */
+	TAILQ_INSERT_TAIL(&bglobal.bg_dplaneq, bdc, entry);
+}
+
 /**
  * Termination phase of the distributed BFD infrastructure: free all allocated
  * resources.
@@ -835,12 +1029,27 @@ static int bfd_dplane_finish_late(void)
 /*
  * Data plane exported functions.
  */
-void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen)
+void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen, bool client)
 {
 	int sock;
 
 	zlog_info("initializing distributed BFD");
 
+	/* Initialize queue header. */
+	TAILQ_INIT(&bglobal.bg_dplaneq);
+
+	/* Initialize listening socket. */
+	bglobal.bg_dplane_sock = -1;
+
+	/* Observe shutdown events. */
+	hook_register(frr_fini, bfd_dplane_finish_late);
+
+	/* Handle client mode. */
+	if (client) {
+		bfd_dplane_client_init(sa, salen);
+		return;
+	}
+
 	/*
 	 * Data plane socket creation:
 	 * - Set REUSEADDR option for taking over previously open socket.
@@ -883,12 +1092,6 @@ void bfd_dplane_init(const struct sockaddr *sa, socklen_t salen)
 	bglobal.bg_dplane_sock = sock;
 	thread_add_read(master, bfd_dplane_accept, &bglobal, sock,
 			&bglobal.bg_dplane_sockev);
-
-	/* Initialize queue header. */
-	TAILQ_INIT(&bglobal.bg_dplaneq);
-
-	/* Observe shutdown events. */
-	hook_register(frr_fini, bfd_dplane_finish_late);
 }
 
 int bfd_dplane_add_session(struct bfd_session *bs)