Repository: zalando/patroni (python)
Issue: Can not run archive command containing shell operators
### What happened?

if use shell operators(`&&` or `|`) in `archive_command`, `archive_ready_wals` will fail because it use `psutil.Popen(shelex(cmd))` to run the command, which assume cmd to be a single process.

Just for example, set archive_command='sleep 20 && /bin/false':
```
2023-07-18 18:32:34,071 INFO: Trying to archive 00000001000000000000002F: sleep 20 && /bin/false
sleep: invalid time interval ‘&&’
sleep: invalid time interval ‘/bin/false’
Try 'sleep --help' for more information.
2023-07-18 18:32:34,075 INFO: Failed to archive WAL segment 00000001000000000000002F
```


### How can we reproduce it (as minimally and precisely as possible)?

-

### What did you expect to happen?

Using shell operators in `archive_command` is a common practice since the doc definition of `archive_command` is a shell command, such as:
```
archive_command = 'test ! -f /mnt/server/archivedir/%f && cp %p /mnt/server/archivedir/%f'  # Unix
```

The implementation should not be too complicated:
1. Do not use `shlex` to parse the command into token sequences when calling.
2. Add `shell=True` parameter when calling psutil.Popen().

I am happy to provide help.

### Patroni/PostgreSQL/DCS version

-   Patroni version: 3.0.4
-   PostgreSQL version: 14
-   DCS (and its version): kubernetes


### Patroni configuration file

```yaml
-
```


### patronictl show-config

```yaml
-
```


### Patroni log files

```shell
-
```


### PostgreSQL log files

```shell
-
```


### Have you tried to use GitHub issue search?

- [X] Yes

### Anything else we need to know?

_No response_
Pull request description: Run archive_command through shell
Resolve #2764 

Patch:
diff --git a/patroni/api.py b/patroni/api.py
index 836a8673..8d991d5f 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -463,6 +463,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
         * ``patroni_dcs_last_seen``: epoch timestamp when DCS was last contacted successfully;
         * ``patroni_pending_restart``: ``1`` if this PostgreSQL node is pending a restart, else ``0``;
         * ``patroni_is_paused``: ``1`` if Patroni is in maintenance node, else ``0``.
+
+        For PostgreSQL v9.6+ the response will also have the following:
+
+        * ``patroni_postgres_streaming``: 1 if Postgres is streaming from another node, else ``0``;
+        * ``patroni_postgres_in_archive_recovery``: ``1`` if Postgres isn't streaming and
+                                                    there is ``restore_command`` available, else ``0``.
         """
         postgres = self.get_postgresql_status(True)
         patroni = self.server.patroni
diff --git a/patroni/ha.py b/patroni/ha.py
index 1148802a..f704500e 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -473,7 +473,8 @@ class Ha(object):
 
         # timeout > 0 indicates that we still have the leader lock, and it was just updated
         if timeout\
-                and data.get('Database cluster state') in ('in production', 'shutting down', 'shut down')\
+                and data.get('Database cluster state') in ('in production', 'in crash recovery',
+                                                           'shutting down', 'shut down')\
                 and self.state_handler.state == 'crashed'\
                 and self.state_handler.role in ('primary', 'master')\
                 and not self.state_handler.config.recovery_conf_exists():
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index f70ad065..ff6fc751 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -370,7 +370,7 @@ class Rewind(object):
                 # it is the author of archive_command, who is responsible
                 # for not overriding the WALs already present in archive
                 logger.info('Trying to archive %s: %s', wal, cmd)
-                if self._postgresql.cancellable.call(shlex.split(cmd)) == 0:
+                if self._postgresql.cancellable.call([cmd], shell=True) == 0:
                     new_name = os.path.join(status_dir, wal + '.done')
                     try:
                         shutil.move(old_name, new_name)


Test patch:
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index b5858d8e..a4fafcff 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -239,7 +239,7 @@ class TestRewind(BaseTestPostgresql):
 
         with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])):
             # successful archive_command call
-            with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)):
+            with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)) as mock_subprocess_call:
                 get_guc_value_res = [
                     'on', 'command %f',
                     'always', 'command %f',
@@ -252,6 +252,10 @@ class TestRewind(BaseTestPostgresql):
                                           '000000000000000000000000', 'command 000000000000000000000000'),
                                          mock_logger_info.call_args[0])
                         mock_logger_info.reset_mock()
+                        mock_subprocess_call.assert_called_once()
+                        self.assertEqual(mock_subprocess_call.call_args.args[0], ['command 000000000000000000000000'])
+                        self.assertEqual(mock_subprocess_call.call_args.kwargs['shell'], True)
+                        mock_subprocess_call.reset_mock()
 
             # failed archive_command call
             with patch.object(CancellableSubprocess, 'call', Mock(return_value=1)):