MEDIUM: mworker-prog: implement program for master-worker

This patch implements the external binary support in the master worker.

To configure an external process, you need to use the program section,
for example:

	program dataplane-api
		command ./dataplane_api

Those processes are launched at the same time as the workers.

During a reload of HAProxy, those processes are dealing with the same
sequence as a worker:

  - the master is re-executed
  - the master sends a USR1 signal to the program
  - the master launches a new instance of the program

During a stop, or restart, a SIGTERM is sent to the program.

Makefile

@@ -791,7 +791,7 @@ OBJS = src/proto_http.o src/cfgparse-listen.o src/proto_htx.o src/stream.o    \
        src/xxhash.o src/hpack-enc.o src/h2.o src/freq_ctr.o src/lru.o         \
        src/protocol.o src/arg.o src/hpack-huff.o src/hdr_idx.o src/base64.o   \
        src/hash.o src/mailers.o src/activity.o src/http_msg.o src/version.o   \
-       src/mworker.o
+       src/mworker.o src/mworker-prog.o
 
 EBTREE_OBJS = $(EBTREE_DIR)/ebtree.o $(EBTREE_DIR)/eb32sctree.o \
               $(EBTREE_DIR)/eb32tree.o $(EBTREE_DIR)/eb64tree.o \

include/proto/mworker.h

@@ -33,5 +33,7 @@ void mworker_cleanlisteners();
 
 int mworker_child_nb();
 
+int mworker_ext_launch_all();
+
 
 #endif /* PROTO_MWORKER_H_ */

include/types/global.h

@@ -187,6 +187,9 @@ struct mworker_proc {
 	int pid;
 	char type;  /* m(aster), w(orker)  */
 	/* 3 bytes hole here */
+	char *id;
+	char **command;
+	char *path;
 	int ipc_fd[2]; /* 0 is master side, 1 is worker side */
 	int relative_pid;
 	int reloads;

src/haproxy.c

@@ -2841,6 +2841,8 @@ int main(int argc, char **argv)
 
 		/* the father launches the required number of processes */
 		if (!(global.mode & MODE_MWORKER_WAIT)) {
+			if (global.mode & MODE_MWORKER)
+				mworker_ext_launch_all();
 			for (proc = 0; proc < global.nbproc; proc++) {
 				ret = fork();
 				if (ret < 0) {
@@ -2862,7 +2864,7 @@ int main(int argc, char **argv)
 					/* find the right mworker_proc */
 					list_for_each_entry(child, &proc_list, list) {
 						if (child->relative_pid == relative_pid &&
-						    child->reloads == 0) {
+						    child->reloads == 0 && child->type == 'w') {
 							child->timestamp = now.tv_sec;
 							child->pid = ret;
 							break;

src/mworker-prog.c

@@ -0,0 +1,225 @@
+/*
+ * Master Worker - program
+ *
+ * Copyright HAProxy Technologies - William Lallemand <wlallemand@haproxy.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version
+ * 2 of the License, or (at your option) any later version.
+ *
+ */
+
+#define _GNU_SOURCE
+
+#include <sys/types.h>
+#include <errno.h>
+#include <grp.h>
+#include <stdio.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <common/cfgparse.h>
+#include <common/errors.h>
+#include <common/initcall.h>
+
+#include <proto/log.h>
+#include <proto/mworker.h>
+
+static int use_program = 0; /* do we use the program section ? */
+
+/*
+ * Launch every programs
+ */
+int mworker_ext_launch_all()
+{
+	int ret;
+	struct mworker_proc *child;
+
+	if (!use_program)
+		return 0;
+
+	/* find the right mworker_proc */
+	list_for_each_entry(child, &proc_list, list) {
+		if (child->reloads == 0 && child->type == 'e') {
+			child->timestamp = now.tv_sec;
+
+			ret = fork();
+			if (ret < 0) {
+				ha_alert("Cannot fork program '%s'.\n", child->id);
+				exit(EXIT_FAILURE); /* there has been an error */
+			} else if (ret > 0) { /* parent */
+				child->pid = ret;
+				ha_notice("New program '%s' (%d) forked\n", child->id, ret);
+				continue;
+			} else if (ret == 0) {
+				/* In child */
+				mworker_unblock_signals();
+				mworker_cleanlisteners();
+				mworker_cleantasks();
+
+				execvp(child->command[0], child->command);
+
+				ha_alert("Cannot execute %s: %s\n", child->command[0], strerror(errno));
+				exit(EXIT_FAILURE);
+			}
+		}
+	}
+
+	return 0;
+
+}
+
+
+/* Configuration */
+
+int cfg_parse_program(const char *file, int linenum, char **args, int kwm)
+{
+	static struct mworker_proc *ext_child = NULL;
+	struct mworker_proc *child;
+	int err_code = 0;
+
+	if (!strcmp(args[0], "program")) {
+		if (alertif_too_many_args(1, file, linenum, args, &err_code)) {
+			err_code |= ERR_ABORT;
+			goto error;
+		}
+
+		if (!*args[1]) {
+			ha_alert("parsing [%s:%d] : '%s' expects an <id> argument\n",
+				 file, linenum, args[0]);
+			err_code |= ERR_ALERT | ERR_ABORT;
+			goto error;
+		}
+
+		ext_child = calloc(1, sizeof(*ext_child));
+		if (!ext_child) {
+			ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
+			err_code |= ERR_ALERT | ERR_ABORT;
+			goto error;
+		}
+
+		ext_child->type = 'e'; /* external process */
+		ext_child->command = NULL;
+		ext_child->path = NULL;
+		ext_child->id = NULL;
+		ext_child->pid = -1;
+		ext_child->relative_pid = -1;
+		ext_child->reloads = 0;
+		ext_child->timestamp = -1;
+		ext_child->ipc_fd[0] = -1;
+		ext_child->ipc_fd[1] = -1;
+		LIST_INIT(&ext_child->list);
+
+		list_for_each_entry(child, &proc_list, list) {
+			if (child->reloads == 0 && child->type == 'e') {
+				if (!strcmp(args[1], child->id)) {
+					ha_alert("parsing [%s:%d]: '%s' program section already exists in the configuration.\n", file, linenum, args[1]);
+					err_code |= ERR_ALERT | ERR_ABORT;
+					goto error;
+				}
+			}
+		}
+
+		ext_child->id = strdup(args[1]);
+		if (!ext_child->id) {
+			ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
+			err_code |= ERR_ALERT | ERR_ABORT;
+			goto error;
+		}
+
+		LIST_ADDQ(&proc_list, &ext_child->list);
+
+	} else if (!strcmp(args[0], "command")) {
+		int arg_nb = 0;
+		int i = 0;
+
+		if (*(args[1]) == 0) {
+			ha_alert("parsing [%s:%d]: '%s' expects a command with optional arguments separated in words.\n", file, linenum, args[0]);
+			err_code |= ERR_ALERT | ERR_FATAL;
+			goto error;
+		}
+
+		while (*args[arg_nb+1])
+			arg_nb++;
+
+		ext_child->command = calloc(arg_nb+1, sizeof(*ext_child->command));
+
+		if (!ext_child->command) {
+			ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
+			err_code |= ERR_ALERT | ERR_ABORT;
+			goto error;
+		}
+
+		while (i < arg_nb) {
+			ext_child->command[i] = strdup(args[i+1]);
+			if (!ext_child->command[i]) {
+				ha_alert("parsing [%s:%d] : out of memory.\n", file, linenum);
+				err_code |= ERR_ALERT | ERR_ABORT;
+				goto error;
+			}
+			i++;
+		}
+		ext_child->command[i] = NULL;
+
+	} else {
+		ha_alert("parsing [%s:%d] : unknown keyword '%s' in '%s' section\n", file, linenum, args[0], "program");
+		err_code |= ERR_ALERT | ERR_FATAL;
+		goto error;
+	}
+
+	use_program = 1;
+
+	return err_code;
+
+error:
+	LIST_DEL(&ext_child->list);
+	if (ext_child->command) {
+		int i;
+
+		for (i = 0; ext_child->command[i]; i++) {
+			if (ext_child->command[i]) {
+				free(ext_child->command[i]);
+				ext_child->command[i] = NULL;
+			}
+		}
+		free(ext_child->command);
+		ext_child->command = NULL;
+	}
+	if (ext_child->id) {
+		free(ext_child->id);
+		ext_child->id = NULL;
+	}
+
+	free(ext_child);
+	ext_child = NULL;
+
+	return err_code;
+
+}
+
+int cfg_program_postparser()
+{
+	int err_code = 0;
+	struct mworker_proc *child;
+
+	list_for_each_entry(child, &proc_list, list) {
+		if (child->reloads == 0 && child->type == 'e') {
+			if (child->command == NULL) {
+				ha_alert("The program section '%s' lacks a command to launch.\n", child->id);
+				err_code |= ERR_ALERT | ERR_FATAL;
+			}
+		}
+	}
+
+	if (use_program && !(global.mode & MODE_MWORKER)) {
+		ha_alert("Can't use a 'program' section without master worker mode.\n");
+		err_code |= ERR_ALERT | ERR_FATAL;
+	}
+
+	return err_code;
+}
+
+
+REGISTER_CONFIG_SECTION("program", cfg_parse_program, NULL);
+REGISTER_CONFIG_POSTPARSER("program", cfg_program_postparser);

src/mworker.c

@@ -100,7 +100,7 @@ void mworker_proc_list_to_env()
 
 	list_for_each_entry(child, &proc_list, list) {
 		if (child->pid > -1)
-			memprintf(&msg, "%s|type=%c;fd=%d;pid=%d;rpid=%d;reloads=%d;timestamp=%d", msg ? msg : "", child->type, child->ipc_fd[0], child->pid, child->relative_pid, child->reloads, child->timestamp);
+			memprintf(&msg, "%s|type=%c;fd=%d;pid=%d;rpid=%d;reloads=%d;timestamp=%d;id=%s", msg ? msg : "", child->type, child->ipc_fd[0], child->pid, child->relative_pid, child->reloads, child->timestamp, child->id ? child->id : "");
 	}
 	if (msg)
 		setenv("HAPROXY_PROCESSES", msg, 1);
@@ -145,12 +145,17 @@ void mworker_env_to_proc_list()
 				child->reloads = atoi(subtoken+8) + 1;
 			} else if (strncmp(subtoken, "timestamp=", 10) == 0) {
 				child->timestamp = atoi(subtoken+10);
+			} else if (strncmp(subtoken, "id=", 3) == 0) {
+				child->id = strdup(subtoken+3);
 			}
 		}
-		if (child->pid)
+		if (child->pid) {
 			LIST_ADDQ(&proc_list, &child->list);
-		else
+		} else {
+			free(child->id);
 			free(child);
+
+		}
 	}
 
 	unsetenv("HAPROXY_PROCESSES");
@@ -238,16 +243,18 @@ void mworker_catch_sigchld(struct sig_handler *sh)
 
 		if (!childfound) {
 			/* We didn't find the PID in the list, that shouldn't happen but we can emit a warning */
-			ha_warning("Worker %d exited with code %d (%s)\n", exitpid, status, (status >= 128) ? strsignal(status - 128) : "Exit");
+			ha_warning("Process %d exited with code %d (%s)\n", exitpid, status, (status >= 128) ? strsignal(status - 128) : "Exit");
 		} else {
-			/* check if exited child was is a current child */
+			/* check if exited child is a current child */
 			if (child->reloads == 0) {
 				if (child->type == 'w')
 					ha_alert("Current worker #%d (%d) exited with code %d (%s)\n", child->relative_pid, exitpid, status, (status >= 128) ? strsignal(status - 128) : "Exit");
+				else if (child->type == 'e')
+					ha_alert("Current program '%s' (%d) exited with code %d (%s)\n", child->id, exitpid, status, (status >= 128) ? strsignal(status - 128) : "Exit");
 
 				if (status != 0 && status != 130 && status != 143
 				    && !(global.tune.options & GTUNE_NOEXIT_ONFAILURE)) {
-					ha_alert("exit-on-failure: killing every workers with SIGTERM\n");
+					ha_alert("exit-on-failure: killing every processes with SIGTERM\n");
 					if (exitcode < 0)
 						exitcode = status;
 					mworker_kill(SIGTERM);
@@ -256,6 +263,8 @@ void mworker_catch_sigchld(struct sig_handler *sh)
 				if (child->type == 'w') {
 					ha_warning("Former worker #%d (%d) exited with code %d (%s)\n", child->relative_pid, exitpid, status, (status >= 128) ? strsignal(status - 128) : "Exit");
 					delete_oldpid(exitpid);
+				} else if (child->type == 'e') {
+					ha_warning("Former program '%s' (%d) exited with code %d (%s)\n", child->id, exitpid, status, (status >= 128) ? strsignal(status - 128) : "Exit");
 				}
 			}
 			free(child);