Fix #299: add support for service readiness notification

This patch adds service readiness notification to support daemons
employing systemd and s6 notification.  Complementing the native
Finit readiness support using PID files that exist already.

The two have slightly different ways of implementing readiness:

 - https://www.freedesktop.org/software/systemd/man/sd_notify.html
 - https://skarnet.org/software/s6/notifywhenup.html

Finit now provides both a NOTIFY_SOCKET environemnt variable, for
systemd, and a way to start s6 daemons with a descriptor argument.

For details on the syntax, see the `service` documentation.

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>

src/Makefile.am

@@ -42,6 +42,7 @@ logit_CFLAGS        += $(lite_CFLAGS)
 logit_LDADD          = $(lite_LIBS)
 
 finit_SOURCES      = api.c	cgroup.c	cgroup.h	\
+		     client.c	client.h			\
 		     cond.c	cond-w.c	cond.h		\
 		     conf.c	conf.h				\
 		     exec.c	finit.c		finit.h		\

src/api.c

@@ -509,6 +509,20 @@ static void api_cb(uev_t *w, void *arg, int events)
 			result = do_signal(rq.data, sizeof(rq.data), rq.runlevel);
 			break;
 
+		case INIT_CMD_NOTIFY_SOCKET:
+			svc = svc_find_by_pid(rq.runlevel);
+			if (!svc) {
+				errx(1, "Unknown PID, cannot register notify socket");
+				result = 1;
+				break;
+			}
+
+			if (uev_io_init(ctx, &svc->notify_watcher, service_notify_cb, svc, sd, UEV_READ)) {
+				err(1, "Falied initializing %s readiness notifier", svc_ident(svc, NULL, 0));
+				break;
+			}
+			return;	/* Don't close sd, used for notify */
+
 		default:
 			dbg("Unsupported cmd: %d", rq.cmd);
 			break;

src/client.c

@@ -69,10 +69,15 @@ int client_disconnect(void)
 	return rc;
 }
 
-int client_send(struct init_request *rq, ssize_t len)
+/* Only valid after client_connect() */
+int client_socket(void)
+{
+	return sd;
+}
+
+int client_request(struct init_request *rq, ssize_t len)
 {
 	struct pollfd pfd = { 0 };
-	int result = 255;
 	int rc;
 
 	if (client_connect() == -1)
@@ -90,6 +95,16 @@ int client_send(struct init_request *rq, ssize_t len)
 		return -1;
 	}
 
+	/*
+	 * Exception, only command that don't need to wait for a
+	 * response from Finit.  Registers the client socket as a
+	 * readiness notification socket with Finit
+	 */
+	if (rq->cmd == INIT_CMD_NOTIFY_SOCKET) {
+		rq->cmd = INIT_CMD_ACK;
+		return 0;
+	}
+
 	pfd.fd = sd;
 	pfd.events = POLLIN | POLLERR | POLLHUP;
 	if ((rc = poll(&pfd, 1, 2000)) <= 0) {
@@ -108,13 +123,32 @@ int client_send(struct init_request *rq, ssize_t len)
 	}
 
 	if (rq->cmd == INIT_CMD_NACK)
-		result = 1;
-	else
-		result = 0;
-exit:
+		return -1;
+
+	return 0;
+}
+
+int client_send(struct init_request *rq, ssize_t len)
+{
+	int rc = 0;
+
+	if (client_request(rq, len))
+		rc = 255;
+
 	client_disconnect();
 
-	return result;
+	return rc;
+}
+
+int client_command(int cmd)
+{
+	struct init_request rq = {
+		.magic = INIT_MAGIC,
+		.runlevel = getpid(),
+		.cmd = cmd,
+	};
+
+	return client_request(&rq, sizeof(rq));
 }
 
 svc_t *client_svc_iterator(int first)
@@ -151,7 +185,7 @@ error:
 	return NULL;
 }
 
-static svc_t *do_cmd(int cmd, const char *arg)
+static svc_t *do_find(int cmd, const char *arg)
 {
 	struct init_request rq = {
 		.magic = INIT_MAGIC,
@@ -182,13 +216,12 @@ error:
 
 svc_t *client_svc_find(const char *arg)
 {
-	return do_cmd(INIT_CMD_SVC_FIND, arg);
+	return do_find(INIT_CMD_SVC_FIND, arg);
 }
 
-
 svc_t *client_svc_find_by_cond(const char *arg)
 {
-	return do_cmd(INIT_CMD_SVC_FIND_BYC, arg);
+	return do_find(INIT_CMD_SVC_FIND_BYC, arg);
 }
 
 /**

src/client.h

@@ -29,8 +29,12 @@
 
 int    client_connect          (void);
 int    client_disconnect       (void);
+int    client_socket           (void);
 
+int    client_request          (struct init_request *rq, ssize_t len);
 int    client_send             (struct init_request *rq, ssize_t len);
+int    client_command          (int cmd);
+
 svc_t *client_svc_iterator     (int first);
 svc_t *client_svc_find         (const char *arg);
 svc_t *client_svc_find_by_cond (const char *arg);

src/finit.h

@@ -96,6 +96,7 @@
 #define INIT_CMD_SVC_FIND       131
 #define INIT_CMD_SVC_FIND_BYC   132
 #define INIT_CMD_SIGNAL         133
+#define INIT_CMD_NOTIFY_SOCKET  200 /* For readiness notification socket */
 #define INIT_CMD_NACK           254
 #define INIT_CMD_ACK            255
 

src/private.h

@@ -38,6 +38,7 @@ int       api_init         (uev_ctx_t *ctx);
 int       api_exit         (void);
 
 void      service_monitor  (pid_t lost, int status);
+void      service_notify_cb(uev_t *w, void *arg, int events);
 
 const char *plugin_hook_str(hook_point_t no);
 int       plugin_exists    (hook_point_t no);

src/service.c

@@ -40,6 +40,7 @@
 #include <wordexp.h>
 
 #include "cgroup.h"
+#include "client.h"
 #include "conf.h"
 #include "cond.h"
 #include "finit.h"
@@ -589,14 +590,37 @@ static int service_start(svc_t *svc)
 	sigprocmask(SIG_BLOCK, &nmask, &omask);
 
 	pid = service_fork(svc);
-	if (pid == 0) {
+	if (pid < 0) {
+		result = -1;
+		goto fail;
+	}
+	if (pid > 1) {
+		svc->pid = pid;
+		svc->start_time = jiffies();
+	} else if (pid == 0) {
 		char *args[MAX_NUM_SVC_ARGS + 1];
 		int status;
 
+		if (!svc_is_tty(svc))
+			redirect(svc);
+
 		if (!svc_is_sysv(svc)) {
 			wordexp_t we = { 0 };
 			int rc;
 
+			if (svc->notify) {
+				if (client_command(INIT_CMD_NOTIFY_SOCKET)) {
+					err(1, "%s: failed creating notify socket", svc_ident(svc, NULL, 0));
+					client_disconnect();
+					svc->notify = 0; /* does not change parent, restarting may work. */
+				} else {
+					char val[20];
+
+					snprintf(val, sizeof(val), "%d", client_socket());
+					setenv("NOTIFY_SOCKET", val, 1);
+				}
+			}
+
 			if ((rc = wordexp(svc->cmd, &we, 0))) {
 				errx(1, "%s: failed wordexp(%s): %d", svc_ident(svc, NULL, 0), svc->cmd, rc);
 			nomem:
@@ -613,6 +637,19 @@ static int service_start(svc_t *svc)
 				if (len == 0)
 					break;
 
+				if (svc->notify) {
+					char *ptr = strstr(arg, "%n");
+
+					if (ptr) {
+						len = snprintf(str, sizeof(str), "%d", client_socket());
+						if (len > 0 && len <= 2) {
+							ptr[0] = ' ';
+							ptr[1] = ' ';
+							memcpy(ptr, str, len);
+						}
+					}
+				}
+
 				/*
 				 * Escape forbidden characters in wordexp()
 				 * but allowed in Finit run/task stanzas,
@@ -678,8 +715,6 @@ static int service_start(svc_t *svc)
 		if (pid < 1)
 			logit(LOG_ERR, "failed setsid(), pid %d: %s", pid, strerror(errno));
 
-		if (!svc_is_tty(svc))
-			redirect(svc);
 		sig_unblock();
 
 		if (svc_is_runtask(svc))
@@ -702,9 +737,6 @@ static int service_start(svc_t *svc)
 	if (!svc_is_sysv(svc))
 		logit(LOG_CONSOLE | LOG_NOTICE, "Starting %s[%d]", svc_ident(svc, NULL, 0), pid);
 
-	svc->pid = pid;
-	svc->start_time = jiffies();
-
 	switch (svc->type) {
 	case SVC_TYPE_RUN:
 		svc->status = complete(svc->cmd, pid);
@@ -728,6 +760,7 @@ static int service_start(svc_t *svc)
 		break;
 	}
 
+fail:
 	sigprocmask(SIG_SETMASK, &omask, NULL);
 	if (do_progress)
 		print_result(result);
@@ -779,6 +812,13 @@ static void service_cleanup(svc_t *svc)
 		logit(LOG_CRIT, "Failed removing service %s pidfile %s",
 		      svc_ident(svc, NULL, 0), fn);
 
+	/* Ensure we don't have any notify socket lingering */
+	if (svc->notify && svc->notify_watcher.fd > 0) {
+		uev_io_stop(&svc->notify_watcher);
+		close(svc->notify_watcher.fd);
+		svc->notify_watcher.fd = 0;
+	}
+
 	/* No longer running, update books. */
 	if (svc_is_tty(svc) && svc->pid > 1)
 		utmp_set_dead(svc->pid); /* Set DEAD_PROCESS UTMP entry */
@@ -1021,6 +1061,15 @@ static void parse_log(svc_t *svc, char *arg)
 	}
 }
 
+static int parse_notify(char *arg)
+{
+	if (!strcmp(arg, "systemd"))
+		return 1;
+	if (!strcmp(arg, "s6"))
+		return 2;
+	return 0;		/* unsupported/disabled */
+}
+
 static void parse_env(svc_t *svc, char *env)
 {
 	if (!env)
@@ -1267,6 +1316,7 @@ int service_register(int type, char *cfg, struct rlimit rlimit[], char *file)
 	char *id = NULL, *env = NULL, *cgroup = NULL;
 	char *pre_script = NULL, *post_script = NULL;
 	char *ready_script = NULL;
+	char *notify = NULL;
 	struct tty tty = { 0 };
 	char *dev = NULL;
 	int respawn = 0;
@@ -1332,6 +1382,8 @@ int service_register(int type, char *cfg, struct rlimit rlimit[], char *file)
 			pid = cmd;
 		else if (MATCH_CMD(cmd, "name:", arg))
 			name = cmd;
+		else if (MATCH_CMD(cmd, "notify:", arg))
+			notify = arg;
 		else if (MATCH_CMD(cmd, "type:forking", arg))
 			forking = 1;
 		else if (MATCH_CMD(cmd, "manual:yes", arg))
@@ -1523,6 +1575,17 @@ int service_register(int type, char *cfg, struct rlimit rlimit[], char *file)
 		parse_script("ready", ready_script, svc->ready_script, sizeof(svc->ready_script));
 	if (log)
 		parse_log(svc, log);
+	if (notify) {
+		int type = parse_notify(notify);
+
+		if (type <= 0)
+			goto disable_notify;
+		svc->notify = type;
+	} else if (svc->notify) {
+	disable_notify:
+		/* XXX: close existing socket */
+		svc->notify = 0;
+	}
 	if (desc)
 		strlcpy(svc->desc, desc, sizeof(svc->desc));
 	else if (type == SVC_TYPE_TTY)
@@ -2008,6 +2071,12 @@ restart:
 			char condstr[MAX_COND_LEN];
 
 			dbg("%s: stopped, cleaning up timers and conditions ...", svc_ident(svc, NULL, 0));
+			if (svc->notify && svc->notify_watcher.fd > 0) {
+				uev_io_stop(&svc->notify_watcher);
+				close(svc->notify_watcher.fd);
+				svc->notify_watcher.fd = 0;
+			}
+
 			service_timeout_cancel(svc);
 			cond_clear(mkcond(svc, condstr, sizeof(condstr)));
 
@@ -2144,6 +2213,11 @@ restart:
 
 				svc_mark_clean(svc);
 			}
+			if (svc->notify == 2 && svc->notify_watcher.fd == 0) {
+				char buf[120];
+				snprintf(buf, sizeof(buf), "service/%s/ready", svc_ident(svc, NULL, 0));
+				cond_set(buf);
+			}
 			break;
 		}
 		break;
@@ -2274,6 +2348,43 @@ int service_completed(void)
 	return 1;
 }
 
+/*
+ * Called when a service sends readiness notification, or when
+ * the service closes its end of the IPC connection.
+ */
+void service_notify_cb(uev_t *w, void *arg, int events)
+{
+	svc_t *svc = (svc_t *)arg;
+	char buf[120];
+	ssize_t len;
+
+	if (UEV_ERROR == events) {
+		warn("Spurious problem with %s notify callback, restarting.", svc_ident(svc, NULL, 0));
+		uev_io_start(w);
+		return;
+	}
+
+	len = read(w->fd, buf, sizeof(buf) - 1);
+	if (len == -1) {
+		warn("Failed reading notification from %s", svc_ident(svc, NULL, 0));
+		return;
+	}
+
+	buf[len] = 0;
+
+	/* systemd and s6, respectively.  The latter then closes the socket */
+	if (!strcmp(buf, "READY=1\n") || !strcmp(buf, "\n")) {
+		snprintf(buf, sizeof(buf), "service/%s/ready", svc_ident(svc, NULL, 0));
+		cond_set(buf);
+		/* s6 applications close their socket after notification */
+		if (svc->notify == 2) {
+			uev_io_stop(w);
+			close(w->fd);
+			w->fd = 0;
+		}
+	}
+}
+
 /*
  * Every five¹ minutes we sweep over all services, skipping crashed or
  * otherwise no longer running ones.  Decrement non-zero crash counters

src/svc.h

@@ -192,6 +192,12 @@ typedef struct svc {
 	uev_t          timer;
 	void           (*timer_cb)(struct svc *svc);
 
+	/*
+	 * Readiness notification socket: systemd, s6
+	 */
+	int	       notify;	       /* 0: none, 1: systemd, 2: s6 */
+	uev_t	       notify_watcher; /* i/o watcher */
+
 	/* time at svc_del(), used by gc timer */
 	struct timespec gc;
 } svc_t;