connect: Try to reconnect to proxy

Try to reconnect to proxy if connection is lost (e.g. proxy died).
Timeout is defined either during configure phase (--with-timeout) or
by passing --rtimeout command line parameter.

This patch is needed to implement high availability of cc-proxy.

Fixes #53.

Signed-off-by: Dmitry Voytik <dmitry.voytik@huawei.com>

Author: Dmitry Voytik

configure.ac

@@ -61,4 +61,10 @@ AS_IF([test $CPPCHECK = 1],
    AC_SUBST(CPPCHECK_PATH)]
 )
 
+AC_ARG_WITH([timeout], [AS_HELP_STRING([--with-timeout],
+            [reconnection timeout to proxy in seconds])])
+AS_IF([test x"$with_timeout" != "x"],
+      [AC_DEFINE_UNQUOTED([RECONNECT_TIMEOUT_S], [$with_timeout],
+                          [reconnection timeout to proxy in seconds])])
+
 AC_OUTPUT

src/shim.c

@@ -37,12 +37,23 @@
 #include <sys/socket.h>
 #include <unistd.h>
 #include <netdb.h>
+#include <time.h>
 
 #include "config.h"
 #include "utils.h"
 #include "log.h"
 #include "shim.h"
 
+// How many seconds shim tries to reconnect to proxy before it exits
+#ifndef RECONNECT_TIMEOUT_S
+#define RECONNECT_TIMEOUT_S 10
+#endif
+
+// Period between attempts to reconnect to proxy in milliseconds
+#ifndef RECONNECT_POLL_MS
+#define RECONNECT_POLL_MS 333
+#endif
+
 /* globals */
 
 struct pollfd poll_fds[MAX_POLL_FDS] = {{-1}};
@@ -337,31 +348,34 @@ send_connect_command(struct cc_shim *shim)
 	return ret;
 }
 
+bool reconnect_to_proxy(struct cc_shim *shim);
+
 /*!
  * Read message received from proxy.
  *
- * \param fd File descriptor to read the data from.
+ * \param shim \ref cc_shim
  * \param buf Buffer to store the data in.
  * \param size Size in bytes to be read.
  *
  * \return true on success, false otherwise
  */
-bool read_wire_data(int fd, uint8_t *buf, size_t size)
+bool read_wire_data(struct cc_shim *shim, uint8_t *buf, size_t size)
 {
 	ssize_t ret;
 	size_t offset = 0;
 
-	if ( fd < 0 || ! buf ) {
+	if ( shim->proxy_sock_fd < 0 || ! buf ) {
 		return false;
 	}
 
 	while(offset < size) {
-		ret = recv(fd, buf+offset, size-offset, 0);
+		ret = recv(shim->proxy_sock_fd, buf+offset, size-offset, 0);
 		if (ret == 0) {
 			shim_debug("Received EOF on file descriptor\n");
-			// TODO: Exit for now, add logic to try to reconnect
-			// to proxy.
-			exit(EXIT_FAILURE);
+			if (! reconnect_to_proxy(shim)) {
+				exit(EXIT_FAILURE);
+			}
+			return false;
 		} else if (ret < 0) {
 			shim_error("Failed to read from fd: %s\n",
 				strerror(errno));
@@ -403,7 +417,7 @@ read_frame(struct cc_shim *shim)
 		abort();
 	}
 
-	if (! read_wire_data(shim->proxy_sock_fd, buf, size)) {
+	if (! read_wire_data(shim, buf, size)) {
 		goto error;
 	}
 
@@ -434,8 +448,7 @@ read_frame(struct cc_shim *shim)
 		abort();
 	}
 
-	if (! read_wire_data(shim->proxy_sock_fd, buf + size,
-			header_size_in_bytes - size)) {
+	if (! read_wire_data(shim, buf + size, header_size_in_bytes - size)) {
 		shim_error("Error while reading frame from proxy at %s\n",
 				shim->proxy_address);
 		goto error;
@@ -453,8 +466,7 @@ read_frame(struct cc_shim *shim)
 		}
 
 		memset(buf, 0, fr->header.payload_len + 1);
-		if (! read_wire_data(shim->proxy_sock_fd, buf, 
-				fr->header.payload_len)) {
+		if (! read_wire_data(shim, buf, fr->header.payload_len)) {
 			goto error;
 		}
 		fr->payload = buf;
@@ -876,7 +888,7 @@ connect_to_proxy(struct cc_shim *shim)
 
 		if (connect(sockfd, (struct sockaddr *)&remote,
 				sizeof(struct sockaddr_un)) == -1) {
-			shim_error("Error while connecting to proxy "
+			shim_warning("Error while connecting to proxy "
 				"with address %s: %s\n", shim->proxy_address,
 				 strerror(errno));
 			goto out;
@@ -954,6 +966,51 @@ connect_to_proxy(struct cc_shim *shim)
 	return false;
 }
 
+inline void sleep_ms(int ms)
+{
+	struct timespec ts = { 0, ms * 1000000L };
+	nanosleep(&ts, NULL);
+}
+
+/*
+ * Try to re-establish tcp/unix connection with proxy in shim->timeout seconds.
+ *
+ * \param shim \ref cc_shim
+ *
+ * \return true on success, false on failure
+ */
+bool reconnect_to_proxy(struct cc_shim *shim)
+{
+	shim_warning("Reconnecting to cc-proxy (timeout %d s)\n",
+			shim->timeout);
+	int time = 0;
+
+try_again:
+	sleep_ms(RECONNECT_POLL_MS);
+	time += RECONNECT_POLL_MS;
+	if (time >= shim->timeout * 1000) {
+		shim_error("Failed to reconnect to cc-proxy (timeout %d s)\n",
+				shim->timeout);
+		return false;
+	}
+
+	close(shim->proxy_sock_fd);
+	if (! connect_to_proxy(shim)) {
+		goto try_again;
+	}
+
+	shim_debug("Sending connect command\n");
+	if (! send_connect_command(shim)) {
+		shim_error("Could not send connect command to cc-proxy\n");
+		goto try_again;
+	}
+	// Update poll_fds because connect_to_proxy(shim) might have updated
+	// shim->proxy_sock_fd
+	add_pollfd(poll_fds, PROXY_SOCK_INDEX, shim->proxy_sock_fd,
+			POLLIN | POLLPRI);
+	return true;
+}
+
 /*
  * Print version information.
  */
@@ -971,6 +1028,7 @@ print_usage(void) {
         printf("  -c,  --container-id   Container id\n");
         printf("  -d,  --debug          Enable debug output\n");
         printf("  -t,  --token          Connection token passed by cc-proxy\n");
+        printf("  -r,  --rtimeout       Reconnection timeout to cc-proxy in seconds\n");
         printf("  -u,  --uri            Connection uri. Supported schemes are tcp: and unix:\n");
         printf("  -v,  --version        Show version\n");
         printf("  -h,  --help           Display this help message\n");
@@ -983,6 +1041,7 @@ main(int argc, char **argv)
 		.container_id   =  NULL,
 		.proxy_sock_fd  = -1,
 		.token          =  NULL,
+		.timeout = RECONNECT_TIMEOUT_S,
 		.proxy_address  =  NULL,
 		.proxy_port     =  -1,
 	};
@@ -999,20 +1058,27 @@ main(int argc, char **argv)
 		{"container-id", required_argument, 0, 'c'},
 		{"debug", no_argument, 0, 'd'},
 		{"help", no_argument, 0, 'h'},
+		{"rtimeout", required_argument, 0, 'r'},
 		{"token", required_argument, 0, 't'},
 		{"uri", required_argument, 0, 'u'},
 		{"version", no_argument, 0, 'v'},
 		{ 0, 0, 0, 0},
 	};
 
-	while ((c = getopt_long(argc, argv, "c:dht:u:v", prog_opts, NULL))!= -1) {
+	while ((c = getopt_long(argc, argv, "c:dhr:t:u:v", prog_opts, NULL))!= -1) {
 		switch (c) {
 			case 'c':
 				shim.container_id = strdup(optarg);
 				break;
 			case 't':
 				shim.token = strdup(optarg);
 				break;
+			case 'r':
+				shim.timeout = atoi(optarg);
+				if (shim.timeout <= 0) {
+					shim.timeout = RECONNECT_TIMEOUT_S;
+				}
+				break;
 			case 'd':
 				debug = true;
 				break;

src/shim.h

@@ -23,6 +23,7 @@ struct cc_shim {
 	char       *container_id;
 	int         proxy_sock_fd;
 	char       *token;
+	int        timeout;		/* reconnection timeout to proxy */
 	char       *proxy_address;
 	int         proxy_port;
 };