MINOR: master/cli: support bidirectional communications with workers

Some rare commands in the worker require to keep their input open and
terminate when it's closed ("show events -w", "wait"). Others maintain
a per-session context ("set anon on"). But in its default operation
mode, the master CLI passes commands one at a time to the worker, and
closes the CLI's input channel so that the command can immediately
close upon response. This effectively prevents these two specific cases
from being used.

Here the approach that we take is to introduce a bidirectional mode to
connect to the worker, where everything sent to the master is immediately
forwarded to the worker (including the raw command), allowing to queue
multiple commands at once in the same session, and to continue to watch
the input to detect when the client closes. It must be a client's choice
however, since doing so means that the client cannot batch many commands
at once to the master process, but must wait for these commands to complete
before sending new ones. For this reason we use the prefix "@@<pid>" for
this. It works exactly like "@" except that it maintains the channel
open during the whole execution. Similarly to "@<pid>" with no command,
"@@<pid>" will simply open an interactive CLI session to the worker, that
will be ended by "quit" or by closing the connection. This can be convenient
for the user, and possibly for clients willing to dedicate a connection to
the worker.

diff --git a/doc/management.txt b/doc/management.txt
index e026cb6c52012ac17d1ab7b4ece7d071032d5efb..962e2a4c14131c008e2353907662a452351d30a9 100644 (file)
--- a/doc/management.txt
+++ b/doc/management.txt
@@ -4360,7 +4360,44 @@ Example:
   CLI session. Similarly, a few rare commands ("show events", "wait") actively
   monitor the CLI for input or closure and are immediately interrupted when the
   CLI is closed. These commands will not work as expected through the master
-  CLI because the command's input is closed after each command.
+  CLI because the command's input is closed after each command. For such rare
+  casesn the "@@" variant below might be more suited.
+
+@@<[!]pid> [command...]
+  This prefix or command is very similar to the "@" prefix documented above
+  except that it enters the worker process, delivers the whole command line
+  into it as-is and stays there until the command finishes. Semi-colons are
+  delivered as well, allowing to execute a full pipelined command in a worker
+  process. The connection with the work remains open until the list of commands
+  completes. Any data sent after the commands will be forwarded to the worker
+  process' CLI and may be consumed by the commands being executed and will be
+  lost for the master process' CLI, offering a truly bidirectional connection
+  with the worker process. As such, users of such commands must be very careful
+  to wait for the command's completion before sending new commands to the
+  master CLI.
+
+  Instead of executing a single command, it is also possible to open a fully
+  interactive session on the worker process by not specifying any command
+  (i.e. "@@1" on its own line). This session can be terminated either by
+  closing the connection or by quitting the worker process (using the "quit"
+  command).
+
+  Examples:
+    # gracefully close connections and delete a server once idle (wait max 10s)
+    $ socat -t 11 /var/run/haproxy-master.sock - <<< \
+       "@@1 disable server app2/srv36; \
+       wait 10000 srv-removable app2/srv36; \
+       del server app2/srv36"
+
+    # forcefully close connections and quickly delete a server
+    $ socat /var/run/haproxy-master.sock - <<< \
+       "@@1 disable server app2/srv36; \
+       shutdown sessions server app2/srv36; \
+       wait 100 srv-removable app2/srv36; \
+       del server app2/srv36"
+
+    # show messages arriving to this ring in real time ("tail -f" equivalent)
+    $ (echo "show events buf0 -w"; read) | socat /var/run/haproxy-master.sock -
 
 expert-mode [on|off]
   This command activates the "expert-mode" for every worker accessed from the

diff --git a/include/haproxy/stream-t.h b/include/haproxy/stream-t.h
index 1f72dc4ed30b1f1197275eabc8ca689eee52a16a..7f44202ab87acd0a1d369f2204eddca191eaf317 100644 (file)
--- a/include/haproxy/stream-t.h
+++ b/include/haproxy/stream-t.h
@@ -132,8 +132,9 @@ static forceinline char *strm_show_flags(char *buf, size_t len, const char *deli
 
 
 /* flags for the proxy of the master CLI */
-/* 0x0001.. to 0x8000 are reserved for ACCESS_* flags from cli-t.h */
+/* 0x0001.. to 0x4000 are reserved for ACCESS_* flags from cli-t.h */
 
+#define PCLI_F_BIDIR    0x08000 /* communicate with worker in bidirectional mode (forward till close) */
 #define PCLI_F_PROMPT   0x10000
 #define PCLI_F_PAYLOAD  0x20000
 #define PCLI_F_RELOAD   0x40000 /* this is the "reload" stream, quits after displaying reload status */

diff --git a/src/cli.c b/src/cli.c
index 181e81b9102d985491e6a664fc11a8758c1bf916..28619dd470ae6ccb5e0a9a1ffe5abe0d4b876137 100644 (file)
--- a/src/cli.c
+++ b/src/cli.c
@@ -2876,6 +2876,70 @@ int pcli_find_and_exec_kw(struct stream *s, char **args, int argl, char **errmsg
        return 0;
 }
 
+/* parse <str> up to <end> for stream <s> and request channel <req>, searching
+ * for the '@@' prefix. If found, the pid is returned in <next_pid> and the
+ * function returns a positive value representing the number of bytes
+ * processed. If the prefix is found but the pid couldn't be parsed, <0 is
+ * returned with an error placed into <errmsg>. If nothing is found, zero is
+ * returned. In any case, <str> is advanced by the number of chars to be
+ * skipped.
+ */
+int pcli_find_bidir_prefix(struct stream *s, struct channel *req, char **str, const char *end, char **errmsg, int *next_pid)
+{
+       char *p = *str;
+       int ret = 0;
+
+       /* skip leading spaces / tabs*/
+       while (p < end && (*p == '\t' || *p == ' '))
+               p++;
+
+       /* check for '@@' prefix */
+       if (p + 2 <= end && p[0] == '@' && p[1] == '@') {
+               const char *pid_str = p;
+               int target_pid;
+
+               p += 2; // skip '@@'
+
+               /* find end of process designation, then zero out all following LWS
+                * to stop on the beginning of the command if any.
+                */
+               while (p < end && !(*p == '\t' || *p == '\n' || *p == '\r' || *p == ' '))
+                       p++;
+
+               while (p < end && (*p == '\t' || *p == '\n' || *p == '\r' || *p == ' '))
+                       *(p++) = 0;
+
+               target_pid = pcli_prefix_to_pid(pid_str + 1);
+               if (target_pid == -1) {
+                       memprintf(errmsg, "Can't find the target PID matching the prefix '%s'\n", pid_str);
+                       ret = -1;
+                       goto leave;
+               }
+
+               /* bidirectional connection to this worker */
+               s->pcli_flags |= PCLI_F_BIDIR;
+               *next_pid = target_pid;
+
+               /* skip '@@pid' and LWS */
+               b_del(&req->buf, p - *str);
+
+               /* forward what remains */
+               ret = end - p;
+
+               /* without any command, simply enter the worker in interactive mode */
+               if (!ret) {
+                       const char *cmd = "prompt;";
+                       ci_insert(req, 0, cmd, strlen(cmd));
+                       ret += strlen(cmd);
+               }
+       }
+
+       /* update with already parsed contents */
+ leave:
+       *str = p;
+       return ret;
+}
+
 /*
  * Parse the CLI request:
  *  - It does basically the same as the cli_io_handler, but as a proxy
@@ -2910,6 +2974,13 @@ int pcli_parse_request(struct stream *s, struct channel *req, char **errmsg, int
        p = str;
 
        if (!(s->pcli_flags & PCLI_F_PAYLOAD)) {
+               /* look for the '@@' prefix and intercept it if found */
+               ret = pcli_find_bidir_prefix(s, req, &p, end, errmsg, next_pid);
+               if (ret != 0) // success or failure
+                       goto end;
+
+               reql = p - str;
+               p = str;
 
                /* Looks for the end of one command */
                while (p+reql < end) {
@@ -3093,6 +3164,7 @@ int pcli_wait_for_request(struct stream *s, struct channel *req, int an_bit)
        if (s->res.analysers & AN_RES_WAIT_CLI)
                return 0;
 
+       s->pcli_flags &= ~PCLI_F_BIDIR; // only for one connection
        if ((s->pcli_flags & ACCESS_LVL_MASK) == ACCESS_LVL_NONE)
                s->pcli_flags |= strm_li(s)->bind_conf->level & ACCESS_LVL_MASK;
 
@@ -3127,12 +3199,18 @@ read_again:
                int target_pid;
                /* enough data */
 
-               /* forward only 1 command */
-               channel_forward(req, to_forward);
+               /* forward only 1 command for '@' or everything for '@@' */
+               if (!(s->pcli_flags & PCLI_F_BIDIR))
+                       channel_forward(req, to_forward);
+               else
+                       channel_forward_forever(req);
 
                if (!(s->pcli_flags & PCLI_F_PAYLOAD)) {
-                       /* we send only 1 command per request, and we write close after it */
-                       sc_schedule_shutdown(s->scb);
+                       /* we send only 1 command per request, and we write
+                        * close after it when not in full-duplex mode.
+                        */
+                       if (!(s->pcli_flags & PCLI_F_BIDIR))
+                               sc_schedule_shutdown(s->scb);
                } else {
                        pcli_write_prompt(s);
                }
@@ -3331,6 +3409,9 @@ int pcli_wait_for_response(struct stream *s, struct channel *rep, int an_bit)
 
                s->req.flags &= ~(CF_AUTO_CONNECT|CF_STREAMER|CF_STREAMER_FAST|CF_WROTE_DATA);
                s->res.flags &= ~(CF_STREAMER|CF_STREAMER_FAST|CF_WRITE_EVENT|CF_WROTE_DATA|CF_READ_EVENT);
+               s->req.to_forward = 0;
+               s->res.to_forward = 0;
+               s->pcli_flags &= ~PCLI_F_BIDIR;
                s->flags &= ~(SF_DIRECT|SF_ASSIGNED|SF_BE_ASSIGNED|SF_FORCE_PRST|SF_IGNORE_PRST);
                s->flags &= ~(SF_CURR_SESS|SF_REDIRECTABLE|SF_SRV_REUSED);
                s->flags &= ~(SF_ERR_MASK|SF_FINST_MASK|SF_REDISP);