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.
diff --git a/src/haproxy.c b/src/haproxy.c
index b2bf558..ef52e31 100644
--- a/src/haproxy.c
+++ b/src/haproxy.c
@@ -2841,6 +2841,8 @@
/* 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 @@
/* 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;
diff --git a/src/mworker-prog.c b/src/mworker-prog.c
new file mode 100644
index 0000000..340f888
--- /dev/null
+++ b/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);
diff --git a/src/mworker.c b/src/mworker.c
index d92358d..d847ee9 100644
--- a/src/mworker.c
+++ b/src/mworker.c
@@ -100,7 +100,7 @@
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 @@
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 @@
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 @@
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);