From: W.C.A. Wijngaards <wouter@nlnetlabs.nl>
Date: Fri, 31 Jan 2020 16:07:40 +0000 (+0100)
Subject: dnstap io, connect and write over TCP.
X-Git-Tag: 1.11.0rc1~120^2~46
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=1420d59949b3bfd63b3f860a48d73030f2ea41a9;p=thirdparty%2Funbound.git

dnstap io, connect and write over TCP.
---

diff --git a/dnstap/dnstap.c b/dnstap/dnstap.c
index 810f3200a..2787c73ec 100644
--- a/dnstap/dnstap.c
+++ b/dnstap/dnstap.c
@@ -149,7 +149,10 @@ dt_create(const char *socket_path, unsigned num_workers,
 		free(env);
 		return NULL;
 	}
-	dt_io_thread_apply_cfg(env->dtio, cfg);
+	if(!dt_io_thread_apply_cfg(env->dtio, cfg)) {
+		free(env);
+		return NULL;
+	}
 	dt_apply_cfg(env, cfg);
 	return env;
 }
diff --git a/dnstap/dtstream.c b/dnstap/dtstream.c
index b734fc9c1..445de18b8 100644
--- a/dnstap/dtstream.c
+++ b/dnstap/dtstream.c
@@ -47,6 +47,7 @@
 #include "util/config_file.h"
 #include "util/ub_event.h"
 #include "util/net_help.h"
+#include "services/outside_network.h"
 #ifdef HAVE_SYS_UN_H
 #include <sys/un.h>
 #endif
@@ -218,13 +219,25 @@ void dt_io_thread_delete(struct dt_io_thread* dtio)
 		item = nextitem;
 	}
 	free(dtio->socket_path);
+	free(dtio->ip_str);
 	free(dtio);
 }
 
-void dt_io_thread_apply_cfg(struct dt_io_thread* dtio, struct config_file *cfg)
+int dt_io_thread_apply_cfg(struct dt_io_thread* dtio, struct config_file *cfg)
 {
-	dtio->upstream_is_unix = 1;
-	dtio->socket_path = strdup(cfg->dnstap_socket_path);
+	/*
+	dtio->upstream_is_tcp = 1;
+	dtio->ip_str = strdup("127.0.0.1@1234");
+	*/
+	if(cfg->dnstap_socket_path && cfg->dnstap_socket_path[0]) {
+		dtio->socket_path = strdup(cfg->dnstap_socket_path);
+		if(!dtio->socket_path) {
+			log_err("malloc failure");
+			return 0;
+		}
+		dtio->upstream_is_unix = 1;
+	}
+	return 1;
 }
 
 int dt_io_thread_register_queue(struct dt_io_thread* dtio,
@@ -477,6 +490,8 @@ static int dtio_check_nb_connect(struct dt_io_thread* dtio)
 #endif
 	if(error != 0) {
 		char* to = dtio->socket_path;
+		if(!to) to = dtio->ip_str;
+		if(!to) to = "";
 #ifndef USE_WINSOCK
 		log_err("dnstap io: failed to connect to \"%s\": %s",
 			to, strerror(error));
@@ -487,7 +502,12 @@ static int dtio_check_nb_connect(struct dt_io_thread* dtio)
 		return -1; /* error, close it */
 	}
 
-	verbose(VERB_ALGO, "dnstap io: connected to \"%s\"", dtio->socket_path);
+	if(dtio->ip_str)
+		verbose(VERB_DETAIL, "dnstap io: connected to %s",
+			dtio->ip_str);
+	else if(dtio->socket_path)
+		verbose(VERB_DETAIL, "dnstap io: connected to \"%s\"",
+			dtio->socket_path);
 	dtio_reconnect_clear(dtio);
 	dtio->check_nb_connect = 0;
 	return 1; /* everything okay */
@@ -682,6 +702,9 @@ static int dtio_check_close(struct dt_io_thread* dtio)
 	while(1) {
 		r = recv(dtio->fd, (void*)buf, sizeof(buf), 0);
 		if(r == -1) {
+			char* to = dtio->socket_path;
+			if(!to) to = dtio->ip_str;
+			if(!to) to = "";
 #ifndef USE_WINSOCK
 			if(errno == EINTR || errno == EAGAIN)
 				return 1; /* try later */
@@ -696,12 +719,17 @@ static int dtio_check_close(struct dt_io_thread* dtio)
 				return 1; /* try later */
 			}
 #endif
-			log_err("dnstap io: output recv: %s", strerror(errno));
+			if(dtio->reconnect_timeout > DTIO_RECONNECT_TIMEOUT_MIN && verbosity < 4)
+				break; /* no log retries on low verbosity */
+			log_err("dnstap io: output closed, recv %s: %s", to,
+				strerror(errno));
 			/* and close below */
 			break;
 		}
 		if(r == 0) {
-			verbose(VERB_ALGO, "dnstap io: output closed by the other side");
+			if(dtio->reconnect_timeout > DTIO_RECONNECT_TIMEOUT_MIN && verbosity < 4)
+				break; /* no log retries on low verbosity */
+			verbose(VERB_DETAIL, "dnstap io: output closed by the other side");
 			/* and close below */
 			break;
 		}
@@ -1131,11 +1159,10 @@ static int dtio_control_start_send(struct dt_io_thread* dtio)
 	return 1;
 }
 
-/** open the output file descriptor */
-static void dtio_open_output(struct dt_io_thread* dtio)
+/** open the output file descriptor for af_local */
+static int dtio_open_output_local(struct dt_io_thread* dtio)
 {
 #ifdef HAVE_SYS_UN_H
-	struct ub_event* ev;
 	struct sockaddr_un s;
 	dtio->fd = socket(AF_LOCAL, SOCK_STREAM, 0);
 	if(dtio->fd == -1) {
@@ -1146,7 +1173,7 @@ static void dtio_open_output(struct dt_io_thread* dtio)
 		log_err("dnstap io: failed to create socket: %s",
 			wsa_strerror(WSAGetLastError()));
 #endif
-		return;
+		return 0;
 	}
 	memset(&s, 0, sizeof(s));
 #ifdef HAVE_STRUCT_SOCKADDR_UN_SUN_LEN
@@ -1173,8 +1200,83 @@ static void dtio_open_output(struct dt_io_thread* dtio)
 		closesocket(dtio->fd);
 #endif
 		dtio->fd = -1;
-		dtio_reconnect_enable(dtio);
-		return;
+		return 0;
+	}
+	return 1;
+#else
+	log_err("cannot create af_local socket");
+	return 0;
+#endif /* HAVE_SYS_UN_H */
+}
+
+/** open the output file descriptor for af_inet and af_inet6 */
+static int dtio_open_output_tcp(struct dt_io_thread* dtio)
+{
+	struct sockaddr_storage addr;
+	socklen_t addrlen;
+	memset(&addr, 0, sizeof(addr));
+	addrlen = (socklen_t)sizeof(addr);
+
+	if(!extstrtoaddr(dtio->ip_str, &addr, &addrlen)) {
+		log_err("could not parse IP '%s'", dtio->ip_str);
+		return 0;
+	}
+	dtio->fd = socket(addr.ss_family, SOCK_STREAM, 0);
+	if(dtio->fd == -1) {
+#ifndef USE_WINSOCK
+		log_err("can't create socket: %s", strerror(errno));
+#else
+		log_err("can't create socket: %s",
+			wsa_strerror(WSAGetLastError()));
+#endif
+		return 0;
+	}
+	fd_set_nonblock(dtio->fd);
+	if(connect(dtio->fd, (struct sockaddr*)&addr, addrlen) == -1) {
+		if(errno == EINPROGRESS)
+			return 1; /* wait until connect done*/
+#ifndef USE_WINSOCK
+		if(tcp_connect_errno_needs_log(
+			(struct sockaddr *)&addr, addrlen)) {
+			log_err("dnstap io: failed to connect to %s: %s",
+				dtio->ip_str, strerror(errno));
+		}
+#else
+		if(WSAGetLastError() == WSAEINPROGRESS ||
+			WSAGetLastError() == WSAEWOULDBLOCK)
+			return 1; /* wait until connect done*/
+		if(tcp_connect_errno_needs_log(
+			(struct sockaddr *)&addr, addrlen)) {
+			log_err("dnstap io: failed to connect to %s: %s",
+				dtio->ip_str, wsa_strerror(WSAGetLastError()));
+		}
+#endif
+
+#ifndef USE_WINSOCK
+		close(dtio->fd);
+#else
+		closesocket(dtio->fd);
+#endif
+		dtio->fd = -1;
+		return 0;
+	}
+	return 1;
+}
+
+/** open the output file descriptor */
+static void dtio_open_output(struct dt_io_thread* dtio)
+{
+	struct ub_event* ev;
+	if(dtio->upstream_is_unix) {
+		if(!dtio_open_output_local(dtio)) {
+			dtio_reconnect_enable(dtio);
+			return;
+		}
+	} else if(dtio->upstream_is_tcp || dtio->upstream_is_tls) {
+		if(!dtio_open_output_tcp(dtio)) {
+			dtio_reconnect_enable(dtio);
+			return;
+		}
 	}
 	dtio->check_nb_connect = 1;
 
@@ -1209,7 +1311,6 @@ static void dtio_open_output(struct dt_io_thread* dtio)
 		dtio_reconnect_enable(dtio);
 		return;
 	}
-#endif /* HAVE_SYS_UN_H */
 }
 
 /** perform the setup of the writer thread on the established event_base */
diff --git a/dnstap/dtstream.h b/dnstap/dtstream.h
index f2c13fde4..8fa2352f4 100644
--- a/dnstap/dtstream.h
+++ b/dnstap/dtstream.h
@@ -228,8 +228,9 @@ void dt_io_thread_delete(struct dt_io_thread* dtio);
  * Apply config to the dtio thread
  * @param dtio: io thread, not yet started.
  * @param cfg: config file struct.
+ * @return false on malloc failure.
  */
-void dt_io_thread_apply_cfg(struct dt_io_thread* dtio,
+int dt_io_thread_apply_cfg(struct dt_io_thread* dtio,
 	struct config_file *cfg);
 
 /**
diff --git a/testcode/testbound.c b/testcode/testbound.c
index 4405231c0..fdbc0b92d 100644
--- a/testcode/testbound.c
+++ b/testcode/testbound.c
@@ -577,3 +577,8 @@ void wsvc_cron_cb(void* ATTR_UNUSED(arg))
 }
 #endif /* UB_ON_WINDOWS */
 
+int tcp_connect_errno_needs_log(struct sockaddr* ATTR_UNUSED(addr),
+	socklen_t ATTR_UNUSED(addrlen))
+{
+	return 1;
+}