Repository: zalando/patroni
Pull request #2692 (2023-05-24 16:12:08)

Add support for custom Postgres binary names
When using a custom Postgres distribution it may be the case that the Postgres binaries are compiled with different names other than the ones used by the community Postgres distribution.

With that in mind we implemented a new set of settings for Patroni, so the user is able to override the default binary names with custom binary names through the new section postgresql.bin_name in the local configuration.

References: PAT-17.

coveralls: ## Pull Request Test Coverage Report for [Build 5070989280](https://coveralls.io/builds/60280473)

* **21** of **21**   **(100.0%)**  changed or added relevant lines in **2** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.0002%**) to **99.849%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/60280473/badge)](https://coveralls.io/builds/60280473) |
| :-- | --: |
| Change from base [Build 5066956626](https://coveralls.io/builds/60263590): |  0.0002% |
| Covered Lines: | 12565 |
| Relevant Lines: | 12584 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst
index d7e9670b..d2f8a49b 100644
--- a/docs/ENVIRONMENT.rst
+++ b/docs/ENVIRONMENT.rst
@@ -135,7 +135,7 @@ PostgreSQL
 -  **PATRONI\_POSTGRESQL\_PROXY\_ADDRESS**: IP address + port through which a connection pool (e.g. pgbouncer) running next to Postgres is accessible. The value is written to the member key in DCS as ``proxy_url`` and could be used/useful for service discovery.
 -  **PATRONI\_POSTGRESQL\_DATA\_DIR**: The location of the Postgres data directory, either existing or to be initialized by Patroni.
 -  **PATRONI\_POSTGRESQL\_CONFIG\_DIR**: The location of the Postgres configuration directory, defaults to the data directory. Must be writable by Patroni.
--  **PATRONI\_POSTGRESQL\_BIN_DIR**: Path to PostgreSQL binaries. (pg_ctl, pg_rewind, pg_basebackup, postgres) The  default value is an empty string meaning that PATH environment variable will be used to find the executables.
+-  **PATRONI\_POSTGRESQL\_BIN_DIR**: Path to PostgreSQL binaries. (pg_ctl, initdb, pg_controldata, pg_basebackup, postgres, pg_isready, pg_rewind) The  default value is an empty string meaning that PATH environment variable will be used to find the executables.
 -  **PATRONI\_POSTGRESQL\_PGPASS**: path to the `.pgpass <https://www.postgresql.org/docs/current/static/libpq-pgpass.html>`__ password file. Patroni creates this file before executing pg\_basebackup and under some other circumstances. The location must be writable by Patroni.
 -  **PATRONI\_REPLICATION\_USERNAME**: replication username; the user will be created during initialization. Replicas will use this user to access the replication source via streaming replication
 -  **PATRONI\_REPLICATION\_PASSWORD**: replication password; the user will be created during initialization.
diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst
index 598143a7..f58c9b0a 100644
--- a/docs/yaml_configuration.rst
+++ b/docs/yaml_configuration.rst
@@ -258,7 +258,16 @@ PostgreSQL
       own config item. See :ref:`custom replica creation methods documentation <custom_replica_creation>` for further explanation.
    -  **data\_dir**: The location of the Postgres data directory, either :ref:`existing <existing_data>` or to be initialized by Patroni.
    -  **config\_dir**: The location of the Postgres configuration directory, defaults to the data directory. Must be writable by Patroni.
-   -  **bin\_dir**: (optional) Path to PostgreSQL binaries (pg_ctl, pg_rewind, pg_basebackup, postgres). If not provided or is an empty string, PATH environment variable will be used to find the executables.
+   -  **bin\_dir**: (optional) Path to PostgreSQL binaries (pg_ctl, initdb, pg_controldata, pg_basebackup, postgres, pg_isready, pg_rewind). If not provided or is an empty string, PATH environment variable will be used to find the executables.
+   -  **bin\_name**: (optional) Make it possible to override Postgres binary names, if you are using a custom Postgres distribution:
+
+      - **pg\_ctl**: (optional) Custom name for ``pg_ctl`` binary.
+      - **initdb**: (optional) Custom name for ``initdb`` binary.
+      - **pg\controldata**: (optional) Custom name for ``pg_controldata`` binary.
+      - **pg\_basebackup**: (optional) Custom name for ``pg_basebackup`` binary.
+      - **postgres**: (optional) Custom name for ``postgres`` binary.
+      - **pg\_isready**: (optional) Custom name for ``pg_isready`` binary.
+      - **pg\_rewind**: (optional) Custom name for ``pg_rewind`` binary.
    -  **listen**: IP address + port that Postgres listens to; must be accessible from other nodes in the cluster, if you're using streaming replication. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.
    -  **use\_unix\_socket**: specifies that Patroni should prefer to use unix sockets to connect to the cluster. Default value is ``false``. If ``unix_socket_directories`` is defined, Patroni will use the first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that the default value should be used and omit ``host`` from the connection parameters.
    -  **use\_unix\_socket\_repl**: specifies that Patroni should prefer to use unix sockets for replication user cluster connection. Default value is ``false``. If ``unix_socket_directories`` is defined, Patroni will use the first suitable value from it to connect to the cluster and fallback to tcp if nothing is suitable. If ``unix_socket_directories`` is not specified in ``postgresql.parameters``, Patroni will assume that the default value should be used and omit ``host`` from the connection parameters.
diff --git a/features/Dockerfile b/features/Dockerfile
index e2f5dbb9..86c6041d 100644
--- a/features/Dockerfile
+++ b/features/Dockerfile
@@ -4,14 +4,17 @@ ARG PG_MAJOR
 ARG PGHOME=/home/postgres
 ARG LC_ALL=C.UTF-8
 ARG LANG=C.UTF-8
+ARG BASE_IMAGE=postgres
 
-FROM postgres:${PG_MAJOR}
+FROM ${BASE_IMAGE}:${PG_MAJOR}
 
 ARG PGHOME
 ARG LC_ALL
 ARG LANG
 
 ENV PGHOME="$PGHOME"
+ENV PG_USER="${PG_USER:-postgres}"
+ENV PG_GROUP="${PG_GROUP:-$PG_USER}"
 ENV LC_ALL="$LC_ALL"
 ENV LANG="$LANG"
 
@@ -41,7 +44,7 @@ RUN set -ex \
     \
     && mkdir -p "$PGHOME" \
     && sed -i "s|/var/lib/postgresql.*|$PGHOME:/bin/bash|" /etc/passwd \
-    && chown -R postgres:postgres /var/log /home/postgres \
+    && chown -R "$PG_USER:$PG_GROUP" /var/log /home/postgres \
     \
     # Download etcd \
     && curl -sL "$ETCDURL/etcd-v$ETCDVERSION-linux-$(dpkg --print-architecture).tar.gz" \
@@ -67,18 +70,18 @@ trap 'copy_output' SIGTERM
 # so we can tell etcd we're ok with running an unsupported architecture.
 export ETCD_UNSUPPORTED_ARCH=$(go env GOARCH)
 cd /src
-runuser -u postgres -- \\
+runuser -u "\$PG_USER" -- \\
     find . ! -readable 2>/dev/null \\
     | sed 's|^./||' >/tmp/copy_exclude.lst \\
     || true
-runuser -u postgres -- \\
+runuser -u "\$PG_USER" -- \\
     rsync -a \\
     --exclude=.tox \\
     --exclude="features/output*" \\
     --exclude-from="/tmp/copy_exclude.lst" \\
     . "\$PGHOME/src/"
 cd "\$PGHOME/src"
-runuser -u postgres -w ETCD_UNSUPPORTED_ARCH -- "\$@" &
+runuser -u "\$PG_USER" -w ETCD_UNSUPPORTED_ARCH -- "\$@" &
 wait $!
 # SIGINT whilst child proc is running is not seen by trap so we run a copy here instead of using
 # trap copy_output SIGINT EXIT
diff --git a/patroni/__main__.py b/patroni/__main__.py
index 9c7f5cea..2a669d3f 100644
--- a/patroni/__main__.py
+++ b/patroni/__main__.py
@@ -1,11 +1,13 @@
 import logging
 import os
 import signal
+import sys
 import time
 
+from argparse import Namespace
 from typing import Any, Dict, Optional, TYPE_CHECKING
 
-from patroni.daemon import AbstractPatroniDaemon, abstract_main
+from patroni.daemon import AbstractPatroniDaemon, abstract_main, get_base_arg_parser
 
 if TYPE_CHECKING:  # pragma: no cover
     from .config import Config
@@ -133,21 +135,40 @@ class Patroni(AbstractPatroniDaemon):
             logger.exception('Exception during Ha.shutdown')
 
 
-def patroni_main() -> None:
+def patroni_main(configfile: str) -> None:
     from multiprocessing import freeze_support
-    from patroni.validator import schema
 
     freeze_support()
-    abstract_main(Patroni, schema)
+    abstract_main(Patroni, configfile)
+
+
+def process_arguments() -> Namespace:
+    parser = get_base_arg_parser()
+    parser.add_argument('--validate-config', action='store_true', help='Run config validator and exit')
+    args = parser.parse_args()
+
+    if args.validate_config:
+        from patroni.validator import schema
+        from patroni.config import Config, ConfigParseError
+
+        try:
+            Config(args.configfile, validator=schema)
+            sys.exit()
+        except ConfigParseError as e:
+            sys.exit(e.value)
+
+    return args
 
 
 def main() -> None:
     from patroni import check_psycopg
 
+    args = process_arguments()
+
     check_psycopg()
 
     if os.getpid() != 1:
-        return patroni_main()
+        return patroni_main(args.configfile)
 
     # Patroni started with PID=1, it looks like we are in the container
     from types import FrameType
@@ -180,7 +201,7 @@ def main() -> None:
     signal.signal(signal.SIGTERM, passtochild)
 
     import multiprocessing
-    patroni = multiprocessing.Process(target=patroni_main)
+    patroni = multiprocessing.Process(target=patroni_main, args=(args.configfile,))
     patroni.start()
     pid = patroni.pid
     patroni.join()
diff --git a/patroni/daemon.py b/patroni/daemon.py
index 88f6b133..e160ce46 100644
--- a/patroni/daemon.py
+++ b/patroni/daemon.py
@@ -6,6 +6,7 @@ Currently it is only used for the main "Thread" of ``patroni`` and ``patroni_raf
 from __future__ import print_function
 
 import abc
+import argparse
 import os
 import signal
 import sys
@@ -15,7 +16,22 @@ from typing import Any, Optional, Type, TYPE_CHECKING
 
 if TYPE_CHECKING:  # pragma: no cover
     from .config import Config
-    from .validator import Schema
+
+
+def get_base_arg_parser() -> argparse.ArgumentParser:
+    """Create a basic argument parser with the arguments used for both patroni and raft controller daemon.
+
+    :returns: 'argparse.ArgumentParser' object
+    """
+    from .config import Config
+    from .version import __version__
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--version', action='version', version='%(prog)s {0}'.format(__version__))
+    parser.add_argument('configfile', nargs='?', default='',
+                        help='Patroni may also read the configuration from the {0} environment variable'
+                        .format(Config.PATRONI_CONFIG_VARIABLE))
+    return parser
 
 
 class AbstractPatroniDaemon(abc.ABC):
@@ -141,41 +157,17 @@ class AbstractPatroniDaemon(abc.ABC):
         self.logger.shutdown()
 
 
-def abstract_main(cls: Type[AbstractPatroniDaemon], validator: Optional['Schema'] = None) -> None:
+def abstract_main(cls: Type[AbstractPatroniDaemon], configfile: str) -> None:
     """Create the main entry point of a given daemon process.
 
-    Expose a basic argument parser, parse the command-line arguments, and run the given daemon process.
-
     :param cls: a class that should inherit from :class:`AbstractPatroniDaemon`.
-    :param validator: used to validate the daemon configuration schema, if requested by the user through
-        ``--validate-config`` CLI option.
+    :param configfile:
     """
-    import argparse
-
     from .config import Config, ConfigParseError
-    from .version import __version__
-
-    parser = argparse.ArgumentParser()
-    parser.add_argument('--version', action='version', version='%(prog)s {0}'.format(__version__))
-    if validator:
-        parser.add_argument('--validate-config', action='store_true', help='Run config validator and exit')
-    parser.add_argument('configfile', nargs='?', default='',
-                        help='Patroni may also read the configuration from the {0} environment variable'
-                        .format(Config.PATRONI_CONFIG_VARIABLE))
-    args = parser.parse_args()
-    validate_config = validator and args.validate_config
     try:
-        if validate_config:
-            Config(args.configfile, validator=validator)
-            sys.exit()
-
-        config = Config(args.configfile)
+        config = Config(configfile)
     except ConfigParseError as e:
-        if e.value:
-            print(e.value, file=sys.stderr)
-        if not validate_config:
-            parser.print_help()
-        sys.exit(1)
+        sys.exit(e.value)
 
     controller = cls(config)
     try:
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 1ff4a60a..a61c6b59 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -965,25 +965,26 @@ class AbstractDCS(abc.ABC):
         return self._last_failsafe
 
     @abc.abstractmethod
-    def _update_leader(self) -> bool:
+    def _update_leader(self, leader: Leader) -> bool:
         """Update leader key (or session) ttl
 
-        :returns: `!True` if leader key (or session) has been updated successfully.
+        :param leader: a reference to a current leader key object
+        :returns: `!True` if leader key (or session) has been updated successfully
 
         You have to use CAS (Compare And Swap) operation in order to update leader key,
         for example for etcd `prevValue` parameter must be used.
         If update fails due to DCS not being accessible or because it is not able to
         process requests (hopefuly temporary), the ~DCSError exception should be raised."""
 
-    def update_leader(self, last_lsn: Optional[int], slots: Optional[Dict[str, int]] = None,
-                      failsafe: Optional[Dict[str, str]] = None) -> bool:
+    def update_leader(self, leader: Leader, last_lsn: Optional[int],
+                      slots: Optional[Dict[str, int]] = None, failsafe: Optional[Dict[str, str]] = None) -> bool:
         """Update leader key (or session) ttl and optime/leader
 
         :param last_lsn: absolute WAL LSN in bytes
         :param slots: dict with permanent slots confirmed_flush_lsn
         :returns: `!True` if leader key (or session) has been updated successfully."""
 
-        ret = self._update_leader()
+        ret = self._update_leader(leader)
         if ret and last_lsn:
             status: Dict[str, Any] = {self._OPTIME: last_lsn}
             if slots:
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 2383a964..5327be0f 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -549,13 +549,11 @@ class Consul(AbstractDCS):
         except InvalidSession:
             logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt')
             self._session = None
-            retry.deadline = retry.stoptime - time.time()
+            retry.ensure_deadline(0)
 
             retry(self._do_refresh_session)
 
-            retry.deadline = retry.stoptime - time.time()
-            if retry.deadline < 1:
-                raise ConsulError('_do_attempt_to_acquire_leader timeout')
+            retry.ensure_deadline(1, ConsulError('_do_attempt_to_acquire_leader timeout'))
 
             return retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
 
@@ -564,9 +562,7 @@ class Consul(AbstractDCS):
         retry = self._retry.copy()
         self._run_and_handle_exceptions(self._do_refresh_session, retry=retry)
 
-        retry.deadline = retry.stoptime - time.time()
-        if retry.deadline < 1:
-            raise ConsulError('attempt_to_acquire_leader timeout')
+        retry.ensure_deadline(1, ConsulError('attempt_to_acquire_leader timeout'))
 
         ret = self._run_and_handle_exceptions(self._do_attempt_to_acquire_leader, retry, retry=None)
         if not ret:
@@ -608,28 +604,20 @@ class Consul(AbstractDCS):
             raise ReturnFalseException
 
     @catch_return_false_exception
-    def _update_leader(self) -> bool:
+    def _update_leader(self, leader: Leader) -> bool:
         retry = self._retry.copy()
 
         self._run_and_handle_exceptions(self._do_refresh_session, True, retry=retry)
 
-        if self._session:
-            cluster = self.cluster
-            leader_session = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
-            if leader_session != self._session:
-                retry.deadline = retry.stoptime - time.time()
-                if retry.deadline < 1:
-                    raise ConsulError('update_leader timeout')
-                logger.warning('Recreating the leader key due to session mismatch')
-                if cluster and cluster.leader:
-                    self._run_and_handle_exceptions(self._client.kv.delete, self.leader_path,
-                                                    cas=cluster.leader.version)
-
-                retry.deadline = retry.stoptime - time.time()
-                if retry.deadline < 0.5:
-                    raise ConsulError('update_leader timeout')
-                self._run_and_handle_exceptions(self._client.kv.put, self.leader_path,
-                                                self._name, acquire=self._session)
+        if self._session and leader.session != self._session:
+            retry.ensure_deadline(1, ConsulError('update_leader timeout'))
+
+            logger.warning('Recreating the leader key due to session mismatch')
+            self._run_and_handle_exceptions(self._client.kv.delete, self.leader_path, cas=leader.version)
+
+            retry.ensure_deadline(0.5, ConsulError('update_leader timeout'))
+
+            self._run_and_handle_exceptions(self._client.kv.put, self.leader_path, self._name, acquire=self._session)
 
         return bool(self._session)
 
@@ -663,8 +651,7 @@ class Consul(AbstractDCS):
         retry = self._retry.copy()
         ret = retry(self._client.kv.put, self.sync_path, value, cas=version)
         if ret:  # We have no other choise, only read after write :(
-            retry.deadline = retry.stoptime - time.time()
-            if retry.deadline < 0.5:
+            if not retry.ensure_deadline(0.5):
                 return False
             _, ret = self.retry(self._client.kv.get, self.sync_path)
             if ret and (ret.get('Value') or b'').decode('utf-8') == value:
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index 190b79fc..3e7f0681 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -798,7 +798,7 @@ class Etcd(AbstractEtcd):
         return bool(self._client.set(self.failsafe_path, value))
 
     @catch_return_false_exception
-    def _update_leader(self) -> bool:
+    def _update_leader(self, leader: Leader) -> bool:
         return bool(self._run_and_handle_exceptions(self._do_update_leader, retry=None))
 
     @catch_etcd_errors
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index 55923df3..b89e36a5 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -16,7 +16,7 @@ from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
 
 from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\
-    TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
+    TimelineHistory, catch_return_false_exception, citus_group_re
 from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry
 from ..exceptions import DCSError, PatroniException
 from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT
@@ -343,9 +343,13 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
     def lease_keepalive(self, ID: str, retry: Optional[Retry] = None) -> Optional[str]:
         return self.call_rpc('/lease/keepalive', {'ID': ID}, retry).get('result', {}).get('TTL')
 
-    def txn(self, compare: Dict[str, Any], success: Dict[str, Any], retry: Optional[Retry] = None) -> Dict[str, Any]:
-        ret = self.call_rpc('/kv/txn', {'compare': [compare], 'success': [success]}, retry)
-        return ret if ret.get('succeeded') else {}
+    def txn(self, compare: Dict[str, Any], success: Dict[str, Any],
+            failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
+        fields = {'compare': [compare], 'success': [success]}
+        if failure:
+            fields['failure'] = [failure]
+        ret = self.call_rpc('/kv/txn', fields, retry)
+        return ret if failure or ret.get('succeeded') else {}
 
     @_handle_auth_errors
     def put(self, key: str, value: str, lease: Optional[str] = None, create_revision: Optional[str] = None,
@@ -360,7 +364,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
         else:
             return self.call_rpc('/kv/put', fields, retry)
         compare['key'] = fields['key']
-        return self.txn(compare, {'request_put': fields}, retry)
+        return self.txn(compare, {'request_put': fields}, retry=retry)
 
     @_handle_auth_errors
     def deleterange(self, key: str, range_end: Union[bytes, str, None] = None,
@@ -369,7 +373,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
         if mod_revision is None:
             return self.call_rpc('/kv/deleterange', fields, retry)
         compare = {'target': 'MOD', 'mod_revision': mod_revision, 'key': fields['key']}
-        return self.txn(compare, {'request_delete_range': fields}, retry)
+        return self.txn(compare, {'request_delete_range': fields}, retry=retry)
 
     def deleteprefix(self, key: str, retry: Optional[Retry] = None) -> Dict[str, Any]:
         return self.deleterange(key, prefix_range_end(key), retry=retry)
@@ -603,7 +607,13 @@ class PatroniEtcd3Client(Etcd3Client):
 
         if self._kv_cache:
             value = delete = None
-            if method == '/kv/txn' and ret.get('succeeded'):
+            # For the 'failure' case we only support a second (nested) transaction that attempts to
+            # update/delete the same keys. Anything more complex than that we don't need and therefore it doesn't
+            # make sense to write a universal response analyzer and we can just check expected JSON path.
+            if method == '/kv/txn'\
+                    and (ret.get('succeeded') or 'failure' in fields and 'request_txn' in fields['failure'][0]
+                         and ret.get('responses', [{'response_txn': {'succeeded': False}}])[0]
+                         .get('response_txn', {}).get('succeeded')):
                 on_success = fields['success'][0]
                 value = on_success.get('request_put')
                 delete = on_success.get('request_delete_range')
@@ -813,19 +823,17 @@ class Etcd3(AbstractEtcd):
             return retry(*args, **kwargs)
 
         try:
-            return _retry(self._client.put, self.leader_path, self._name, self._lease, '0')
+            return _retry(self._client.put, self.leader_path, self._name, self._lease, create_revision='0')
         except LeaseNotFound:
             logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt')
             self._lease = None
-            retry.deadline = retry.stoptime - time.time()
+            retry.ensure_deadline(0)
 
             _retry(self._do_refresh_lease)
 
-            retry.deadline = retry.stoptime - time.time()
-            if retry.deadline < 1:
-                raise Etcd3Error('_do_attempt_to_acquire_leader timeout')
+            retry.ensure_deadline(1, Etcd3Error('_do_attempt_to_acquire_leader timeout'))
 
-            return _retry(self._client.put, self.leader_path, self._name, self._lease, '0')
+            return _retry(self._client.put, self.leader_path, self._name, self._lease, create_revision='0')
 
     @catch_return_false_exception
     def attempt_to_acquire_leader(self) -> bool:
@@ -837,9 +845,7 @@ class Etcd3(AbstractEtcd):
 
         self._run_and_handle_exceptions(self._do_refresh_lease, retry=_retry)
 
-        retry.deadline = retry.stoptime - time.time()
-        if retry.deadline < 1:
-            raise Etcd3Error('attempt_to_acquire_leader timeout')
+        retry.ensure_deadline(1, Etcd3Error('attempt_to_acquire_leader timeout'))
 
         ret = self._run_and_handle_exceptions(self._do_attempt_to_acquire_leader, retry, retry=None)
         if not ret:
@@ -867,7 +873,7 @@ class Etcd3(AbstractEtcd):
         return bool(self._client.put(self.failsafe_path, value))
 
     @catch_return_false_exception
-    def _update_leader(self) -> bool:
+    def _update_leader(self, leader: Leader) -> bool:
         retry = self._retry.copy()
 
         def _retry(*args: Any, **kwargs: Any) -> Any:
@@ -876,24 +882,24 @@ class Etcd3(AbstractEtcd):
 
         self._run_and_handle_exceptions(self._do_refresh_lease, True, retry=_retry)
 
-        if self._lease:
-            cluster = self.cluster
-            leader_lease = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
-            if leader_lease != self._lease:
-                retry.deadline = retry.stoptime - time.time()
-                if retry.deadline < 1:
-                    raise Etcd3Error('update_leader timeout')
-
-                try:
-                    self._run_and_handle_exceptions(self._client.put, self.leader_path,
-                                                    self._name, self._lease, '0', retry=_retry)
-                except ReturnFalseException:
-                    pass
+        if self._lease and leader.session != self._lease:
+            retry.ensure_deadline(1, Etcd3Error('update_leader timeout'))
+
+            fields = {'key': base64_encode(self.leader_path), 'value': base64_encode(self._name), 'lease': self._lease}
+            # First we try to update lease on existing leader key "hoping" that we still owning it
+            compare1 = {'key': fields['key'], 'target': 'VALUE', 'value': fields['value']}
+            request_put = {'request_put': fields}
+            # If the first comparison failed we will try to create the new leader key in a transaction
+            compare2 = {'key': fields['key'], 'target': 'CREATE', 'create_revision': '0'}
+            request_txn = {'request_txn': {'compare': [compare2], 'success': [request_put]}}
+            ret = self._run_and_handle_exceptions(self._client.txn, compare1, request_put, request_txn, retry=_retry)
+            return ret.get('succeeded', False)\
+                or ret.get('responses', [{}])[0].get('response_txn', {}).get('succeeded', False)
         return bool(self._lease)
 
     @catch_etcd_errors
     def initialize(self, create_new: bool = True, sysid: str = ""):
-        return self.retry(self._client.put, self.initialize_path, sysid, None, '0' if create_new else None)
+        return self.retry(self._client.put, self.initialize_path, sysid, create_revision='0' if create_new else None)
 
     @catch_etcd_errors
     def _delete_leader(self) -> bool:
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index 3630ae1e..0d457ba4 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -1129,7 +1129,7 @@ class Kubernetes(AbstractDCS):
         """Unused"""
         raise NotImplementedError  # pragma: no cover
 
-    def _update_leader(self) -> bool:
+    def _update_leader(self, leader: Leader) -> bool:
         """Unused"""
         raise NotImplementedError  # pragma: no cover
 
@@ -1153,8 +1153,7 @@ class Kubernetes(AbstractDCS):
             raise KubernetesError(e)
 
         # if we are here, that means update failed with 409
-        retry.deadline = retry.stoptime - time.time()
-        if retry.deadline < 1:
+        if not retry.ensure_deadline(1):
             return False  # No time for retry. Tell ha.py that we have to demote due to failed update.
 
         # Try to get the latest version directly from K8s API instead of relying on async cache
@@ -1168,8 +1167,7 @@ class Kubernetes(AbstractDCS):
 
         self._kinds.set(self.leader_path, kind)
 
-        retry.deadline = retry.stoptime - time.time()
-        if retry.deadline < 0.5:
+        if not retry.ensure_deadline(0.5):
             return False
 
         kind_annotations = kind and kind.metadata.annotations or {}
@@ -1182,8 +1180,8 @@ class Kubernetes(AbstractDCS):
         return bool(_run_and_handle_exceptions(self._patch_or_create, self.leader_path, annotations,
                                                kind_resource_version, ips=ips, retry=_retry))
 
-    def update_leader(self, last_lsn: Optional[int], slots: Optional[Dict[str, int]] = None,
-                      failsafe: Optional[Dict[str, str]] = None) -> bool:
+    def update_leader(self, leader: Leader, last_lsn: Optional[int],
+                      slots: Optional[Dict[str, int]] = None, failsafe: Optional[Dict[str, str]] = None) -> bool:
         kind = self._kinds.get(self.leader_path)
         kind_annotations = kind and kind.metadata.annotations or {}
 
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index c068f7f0..c4286f1e 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -419,7 +419,7 @@ class Raft(AbstractDCS):
     def _write_failsafe(self, value: str) -> bool:
         return self._sync_obj.set(self.failsafe_path, value, timeout=1) is not False
 
-    def _update_leader(self) -> bool:
+    def _update_leader(self, leader: Leader) -> bool:
         ret = self._sync_obj.set(self.leader_path, self._name, ttl=self._ttl,
                                  handle_raft_error=False, prevValue=self._name) is not False
         if not ret and self._sync_obj.get(self.leader_path) is None:
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 5ec9cd04..ed2b71ed 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -443,10 +443,8 @@ class ZooKeeper(AbstractDCS):
     def _write_failsafe(self, value: str) -> bool:
         return self._set_or_create(self.failsafe_path, value) is not False
 
-    def _update_leader(self) -> bool:
-        cluster = self.cluster
-        session = cluster and isinstance(cluster.leader, Leader) and cluster.leader.session
-        if self._client.client_id and self._client.client_id[0] != session:
+    def _update_leader(self, leader: Leader) -> bool:
+        if self._client.client_id and self._client.client_id[0] != leader.session:
             logger.warning('Recreating the leader ZNode due to ownership mismatch')
             try:
                 self._client.retry(self._client.delete, self.leader_path)
diff --git a/patroni/ha.py b/patroni/ha.py
index 1190ad1c..3108f787 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -239,8 +239,10 @@ class Ha(object):
                 slots = self.state_handler.slots()
             except Exception:
                 logger.exception('Exception when called state_handler.last_operation()')
+        if TYPE_CHECKING:  # pragma: no cover
+            assert self.cluster.leader is not None
         try:
-            ret = self.dcs.update_leader(last_lsn, slots, self._failsafe_config())
+            ret = self.dcs.update_leader(self.cluster.leader, last_lsn, slots, self._failsafe_config())
         except DCSError:
             raise
         except Exception:
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index 305aa4ce..6d2d18af 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -227,8 +227,17 @@ class Postgresql(object):
         return 0
 
     def pgcommand(self, cmd: str) -> str:
-        """Returns path to the specified PostgreSQL command"""
-        return os.path.join(self._bin_dir, cmd)
+        """Return path to the specified PostgreSQL command.
+
+        .. note::
+            If ``postgresql.bin_name.*cmd*`` was configured by the user then that binary name is used, otherwise the
+            default binary name *cmd* is used.
+
+        :param cmd: the Postgres binary name to get path to.
+
+        :returns: path to Postgres binary named *cmd*.
+        """
+        return os.path.join(self._bin_dir, (self.config.get('bin_name', {}) or {}).get(cmd, cmd))
 
     def pg_ctl(self, cmd: str, *args: str, **kwargs: Any) -> bool:
         """Builds and executes pg_ctl command
diff --git a/patroni/raft_controller.py b/patroni/raft_controller.py
index 2f9f7858..a9d7424c 100644
--- a/patroni/raft_controller.py
+++ b/patroni/raft_controller.py
@@ -1,7 +1,7 @@
 import logging
 
 from .config import Config
-from .daemon import AbstractPatroniDaemon, abstract_main
+from .daemon import AbstractPatroniDaemon, abstract_main, get_base_arg_parser
 from .dcs.raft import KVStoreTTL
 
 logger = logging.getLogger(__name__)
@@ -27,4 +27,7 @@ class RaftController(AbstractPatroniDaemon):
 
 
 def main() -> None:
-    abstract_main(RaftController)
+    parser = get_base_arg_parser()
+    args = parser.parse_args()
+
+    abstract_main(RaftController, args.configfile)
diff --git a/patroni/utils.py b/patroni/utils.py
index f6cfa919..eb02c561 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -531,6 +531,21 @@ class Retry(object):
         """Get the current stop time."""
         return self._cur_stoptime or 0
 
+    def ensure_deadline(self, timeout: float, raise_ex: Optional[Exception] = None) -> bool:
+        """Calculates, sets, and checks the remaining deadline time.
+
+        :param timeout: if the *deadline* is smaller than the provided *timeout* value raise *raise_ex* exception
+        :param raise_ex: the exception object that will be raised if the *deadline* is smaller than provided *timeout*
+        :returns: `False` if *deadline* is smaller than a provided *timeout* and *raise_ex* isn't set. Otherwise `True`
+        :raises Exception: if calculated deadline is smaller than provided *timeout*
+        """
+        self.deadline = self.stoptime - time.time()
+        if self.deadline < timeout:
+            if raise_ex:
+                raise raise_ex
+            return False
+        return True
+
     def __call__(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
         """Call a function *func* with arguments ``*args`` and ``*kwargs`` in a loop.
 
diff --git a/patroni/validator.py b/patroni/validator.py
index 9a4b117c..f780f9e0 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -176,6 +176,16 @@ def is_ipv6_address(ip: str) -> bool:
     return True
 
 
+def get_bin_name(bin_name: str) -> str:
+    """Get the value of ``postgresql.bin_name[*bin_name*]`` configuration option.
+
+    :param bin_name: a key to be retrieved from ``postgresql.bin_name`` configuration.
+
+    :returns: value of ``postgresql.bin_name[*bin_name*]``, if present, otherwise *bin_name*.
+    """
+    return (schema.data.get('postgresql', {}).get('bin_name', {}) or {}).get(bin_name, bin_name)
+
+
 def get_major_version(bin_dir: OptionalType[str] = None) -> str:
     """Get the major version of PostgreSQL.
 
@@ -191,9 +201,9 @@ def get_major_version(bin_dir: OptionalType[str] = None) -> str:
         * Returns `15` for PostgreSQL 15.2
     """
     if not bin_dir:
-        binary = 'postgres'
+        binary = get_bin_name('postgres')
     else:
-        binary = os.path.join(bin_dir, 'postgres')
+        binary = os.path.join(bin_dir, get_bin_name('postgres'))
     version = subprocess.check_output([binary, '--version']).decode()
     version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version)
     if TYPE_CHECKING:  # pragma: no cover
@@ -242,6 +252,38 @@ def validate_data_dir(data_dir: str) -> bool:
     return True
 
 
+def validate_binary_name(bin_name: str) -> bool:
+    """Validate the value of ``postgresql.binary_name[*bin_name*]`` configuration option.
+
+    If ``postgresql.bin_dir`` is set and the value of the *bin_name* meets these conditions:
+
+        * The path join of ``postgresql.bin_dir`` plus the *bin_name* value exists; and
+        * The path join as above is executable
+
+    If ``postgresql.bin_dir`` is not set, then validate that the value of *bin_name* meets this
+    condition:
+
+        * Is found in the system PATH using ``which``
+
+    :param bin_name: the value of the ``postgresql.bin_name[*bin_name*]``
+
+    :returns: ``True`` if the conditions are true
+
+    :raises :class:`patroni.exceptions.ConfigParserError`: if:
+        * *bin_name* is not set; or
+        * the path join of the ``postgresql.bin_dir`` plus *bin_name* does not exist; or
+        * the path join as above is not executable; or
+        * the *bin_name* cannot be found in the system PATH
+
+    """
+    if not bin_name:
+        raise ConfigParseError("is an empty string")
+    bin_dir = schema.data.get('postgresql', {}).get('bin_dir', None)
+    if not shutil.which(bin_name, path=bin_dir):
+        raise ConfigParseError(f"does not contain '{bin_name}' in '{bin_dir or '$PATH'}'")
+    return True
+
+
 class Result(object):
     """Represent the result of a given validation that was performed.
 
@@ -406,6 +448,31 @@ class Directory(object):
             yield from self._check_executables(path=name)
 
 
+class BinDirectory(Directory):
+    """Check if a Postgres binary directory contains the expected files.
+
+    It is a subclass of :class:`Directory` with an extended capability: translating ``BINARIES`` according to configured
+    ``postgresql.bin_name``, if any.
+
+    :cvar BINARIES: list of executable files that should exist directly under a given Postgres binary directory.
+    """
+
+    # ``pg_rewind`` is not in the list because its usage by Patroni is optional. Also, it is not available by default on
+    # Postgres 9.3 and 9.4, versions which Patroni supports.
+    BINARIES = ["pg_ctl", "initdb", "pg_controldata", "pg_basebackup", "postgres", "pg_isready"]
+
+    def validate(self, name: str) -> Iterator[Result]:
+        """Check if the expected executables can be found under *name* binary directory.
+
+        :param name: path to the base directory against which executables will be validated. Check against PATH if
+            *name* is not provided.
+
+        :yields: objects with the error message related to the failure, if any check fails.
+        """
+        self.contains_executable: List[str] = [get_bin_name(binary) for binary in self.BINARIES]
+        yield from super().validate(name)
+
+
 class Schema(object):
     """Define a configuration schema.
 
@@ -700,6 +767,7 @@ class IntValidator(object):
     :ivar base_unit: the base unit to convert the value to before checking if it's within `min` and `max` range.
     :ivar raise_assert: if an ``assert`` call should be performed regarding expected type and valid range.
     """
+
     expected_type = int
 
     def __init__(self, min: OptionalType[int] = None, max: OptionalType[int] = None,
@@ -736,6 +804,10 @@ class IntValidator(object):
 
 
 def validate_watchdog_mode(value: Any) -> None:
+    """Validate ``watchdog.mode`` configuration option.
+
+    :param value: value of ``watchdog.mode`` to be validated.
+    """
     assert_(isinstance(value, (str, bool)), "expected type is not a string")
     assert_(value in (False, "off", "automatic", "required"))
 
@@ -748,6 +820,7 @@ setattr(validate_connect_address, 'expected_type', str)
 setattr(validate_host_port_listen, 'expected_type', str)
 setattr(validate_host_port_listen_multiple_hosts, 'expected_type', str)
 setattr(validate_data_dir, 'expected_type', str)
+setattr(validate_binary_name, 'expected_type', str)
 validate_etcd = {
     Or("host", "hosts", "srv", "srv_suffix", "url", "proxy"): Case({
         "host": validate_host_port,
@@ -823,8 +896,16 @@ schema = Schema({
             Optional("rewind"): userattributes
         },
         "data_dir": validate_data_dir,
-        Optional("bin_dir", ""): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup",
-                                                                "postgres", "pg_isready"]),
+        Optional("bin_name"): {
+            Optional("pg_ctl"): validate_binary_name,
+            Optional("initdb"): validate_binary_name,
+            Optional("pg_controldata"): validate_binary_name,
+            Optional("pg_basebackup"): validate_binary_name,
+            Optional("postgres"): validate_binary_name,
+            Optional("pg_isready"): validate_binary_name,
+            Optional("pg_rewind"): validate_binary_name,
+        },
+        Optional("bin_dir", ""): BinDirectory(),
         Optional("parameters"): {
             Optional("unix_socket_directories"): str
         },
diff --git a/tox.ini b/tox.ini
index 72353d4f..71104357 100644
--- a/tox.ini
+++ b/tox.ini
@@ -113,10 +113,13 @@ labels =
 setenv =
     {[common]postgres_matrix}
     DOCKER_BUILDKIT = 1
+passenv =
+    BASE_IMAGE
 commands =
     docker build . \
         --tag patroni-dev:{env:PG_MAJOR} \
         --build-arg PG_MAJOR \
+        --build-arg BASE_IMAGE={env:BASE_IMAGE:postgres} \
         --file features/Dockerfile
 allowlist_externals = docker
 


Test patch:
diff --git a/tests/test_consul.py b/tests/test_consul.py
index 165d2382..ebbbb90e 100644
--- a/tests/test_consul.py
+++ b/tests/test_consul.py
@@ -176,23 +176,24 @@ class TestConsul(unittest.TestCase):
     @patch.object(consul.Consul.Session, 'renew')
     @patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException))
     def test_update_leader(self, mock_renew):
+        leader = self.c.get_cluster().leader
         self.c._session = 'fd4f44fe-2cac-bba5-a60b-304b51ff39b8'
         with patch.object(consul.Consul.KV, 'delete', Mock(return_value=True)):
             with patch.object(consul.Consul.KV, 'put', Mock(return_value=True)):
-                self.assertTrue(self.c.update_leader(12345, failsafe={'foo': 'bar'}))
+                self.assertTrue(self.c.update_leader(leader, 12345, failsafe={'foo': 'bar'}))
             with patch.object(consul.Consul.KV, 'put', Mock(side_effect=ConsulException)):
-                self.assertFalse(self.c.update_leader(12345))
-            with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 100, 200, 300])):
-                self.assertRaises(ConsulError, self.c.update_leader, 12345)
+                self.assertFalse(self.c.update_leader(leader, 12345))
+            with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 100, 200, 300])):
+                self.assertRaises(ConsulError, self.c.update_leader, leader, 12345)
         with patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
-            self.assertRaises(ConsulError, self.c.update_leader, 12345)
+            self.assertRaises(ConsulError, self.c.update_leader, leader, 12345)
         with patch.object(consul.Consul.KV, 'delete', Mock(side_effect=ConsulException)):
-            self.assertFalse(self.c.update_leader(12347))
+            self.assertFalse(self.c.update_leader(leader, 12347))
         mock_renew.side_effect = RetryFailedError('')
         self.c._last_session_refresh = 0
-        self.assertRaises(ConsulError, self.c.update_leader, 12346)
+        self.assertRaises(ConsulError, self.c.update_leader, leader, 12346)
         mock_renew.side_effect = ConsulException
-        self.assertFalse(self.c.update_leader(12347))
+        self.assertFalse(self.c.update_leader(leader, 12347))
 
     @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True))
     def test_delete_leader(self):
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index 6699127a..b5add201 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -296,14 +296,15 @@ class TestEtcd(unittest.TestCase):
         self.etcd.write_leader_optime('0')
 
     def test_update_leader(self):
-        self.assertTrue(self.etcd.update_leader(None, failsafe={'foo': 'bar'}))
+        leader = self.etcd.get_cluster().leader
+        self.assertTrue(self.etcd.update_leader(leader, None, failsafe={'foo': 'bar'}))
         with patch.object(etcd.Client, 'write',
                           Mock(side_effect=[etcd.EtcdConnectionFailed, etcd.EtcdClusterIdChanged, Exception])):
-            self.assertRaises(EtcdError, self.etcd.update_leader, None)
-            self.assertFalse(self.etcd.update_leader(None))
-            self.assertRaises(EtcdError, self.etcd.update_leader, None)
+            self.assertRaises(EtcdError, self.etcd.update_leader, leader, None)
+            self.assertFalse(self.etcd.update_leader(leader, None))
+            self.assertRaises(EtcdError, self.etcd.update_leader, leader, None)
         with patch.object(etcd.Client, 'write', Mock(side_effect=etcd.EtcdKeyNotFound)):
-            self.assertFalse(self.etcd.update_leader(None))
+            self.assertFalse(self.etcd.update_leader(leader, None))
 
     def test_initialize(self):
         self.assertFalse(self.etcd.initialize())
diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py
index 6f56d0e6..a737f199 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -235,16 +235,20 @@ class TestEtcd3(BaseTestEtcd3):
             self.etcd3.touch_member({})
 
     def test__update_leader(self):
+        leader = self.etcd3.get_cluster().leader
         self.etcd3._lease = None
-        self.etcd3.update_leader('123', failsafe={'foo': 'bar'})
+        with patch.object(Etcd3Client, 'txn', Mock(return_value={'succeeded': True})):
+            self.etcd3.update_leader(leader, '123', failsafe={'foo': 'bar'})
         self.etcd3._last_lease_refresh = 0
-        self.etcd3.update_leader('124')
+        self.etcd3.update_leader(leader, '124')
         with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)),\
                 patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
-            self.assertRaises(Etcd3Error, self.etcd3.update_leader, '126')
+            self.assertRaises(Etcd3Error, self.etcd3.update_leader, leader, '126')
+        self.etcd3._lease = leader.session
+        self.etcd3.update_leader(leader, '124')
         self.etcd3._last_lease_refresh = 0
         with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(side_effect=Unknown)):
-            self.assertFalse(self.etcd3.update_leader('125'))
+            self.assertFalse(self.etcd3.update_leader(leader, '125'))
 
     def test_take_leader(self):
         self.assertFalse(self.etcd3.take_leader())
diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py
index fbefb448..cfed1559 100644
--- a/tests/test_kubernetes.py
+++ b/tests/test_kubernetes.py
@@ -340,35 +340,37 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
 
     @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test_update_leader(self, mock_patch_namespaced_endpoints):
-        self.assertIsNotNone(self.k.update_leader('123', failsafe={'foo': 'bar'}))
+        leader = self.k.get_cluster().leader
+        self.assertIsNotNone(self.k.update_leader(leader, '123', failsafe={'foo': 'bar'}))
         args = mock_patch_namespaced_endpoints.call_args[0]
         self.assertEqual(args[2].subsets[0].addresses[0].target_ref.resource_version, '10')
         self.k._kinds._object_cache['test'].subsets[:] = []
-        self.assertIsNotNone(self.k.update_leader('123'))
+        self.assertIsNotNone(self.k.update_leader(leader, '123'))
         self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1'
-        self.assertFalse(self.k.update_leader('123'))
+        self.assertFalse(self.k.update_leader(leader, '123'))
 
     @patch.object(k8s_client.CoreV1Api, 'read_namespaced_endpoints', create=True)
     @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True)
     def test__update_leader_with_retry(self, mock_patch, mock_read):
+        leader = self.k.get_cluster().leader
         mock_read.return_value = mock_read_namespaced_endpoints()
         mock_patch.side_effect = k8s_client.rest.ApiException(502, '')
-        self.assertFalse(self.k.update_leader('123'))
+        self.assertFalse(self.k.update_leader(leader, '123'))
         mock_patch.side_effect = RetryFailedError('')
-        self.assertRaises(KubernetesError, self.k.update_leader, '123')
+        self.assertRaises(KubernetesError, self.k.update_leader, leader, '123')
         mock_patch.side_effect = k8s_client.rest.ApiException(409, '')
         with patch('time.time', Mock(side_effect=[0, 100, 200, 0, 0, 0, 0, 100, 200])):
-            self.assertFalse(self.k.update_leader('123'))
-            self.assertFalse(self.k.update_leader('123'))
-        self.assertFalse(self.k.update_leader('123'))
+            self.assertFalse(self.k.update_leader(leader, '123'))
+            self.assertFalse(self.k.update_leader(leader, '123'))
+        self.assertFalse(self.k.update_leader(leader, '123'))
         mock_patch.side_effect = [k8s_client.rest.ApiException(409, ''), mock_namespaced_kind()]
         mock_read.return_value.metadata.resource_version = '2'
         self.assertIsNotNone(self.k._update_leader_with_retry({}, '1', []))
         mock_patch.side_effect = k8s_client.rest.ApiException(409, '')
         mock_read.side_effect = RetryFailedError('')
-        self.assertRaises(KubernetesError, self.k.update_leader, '123')
+        self.assertRaises(KubernetesError, self.k.update_leader, leader, '123')
         mock_read.side_effect = Exception
-        self.assertFalse(self.k.update_leader('123'))
+        self.assertFalse(self.k.update_leader(leader, '123'))
 
     @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints',
                   Mock(side_effect=[k8s_client.rest.ApiException(500, ''),
diff --git a/tests/test_patroni.py b/tests/test_patroni.py
index ad8e775c..cb40b232 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -15,7 +15,7 @@ from patroni.exceptions import DCSError
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
 from patroni import check_psycopg
-from patroni.__main__ import Patroni, main as _main, patroni_main
+from patroni.__main__ import Patroni, main as _main
 from threading import Thread
 
 from . import psycopg_connect, SleepException
@@ -52,14 +52,14 @@ class TestPatroni(unittest.TestCase):
 
     @patch('sys.argv', ['patroni.py'])
     def test_no_config(self):
-        self.assertRaises(SystemExit, patroni_main)
+        self.assertRaises(SystemExit, _main)
 
     @patch('sys.argv', ['patroni.py', '--validate-config', 'postgres0.yml'])
     @patch('socket.socket.connect_ex', Mock(return_value=1))
     def test_validate_config(self):
-        self.assertRaises(SystemExit, patroni_main)
+        self.assertRaises(SystemExit, _main)
         with patch.object(config.Config, '__init__', Mock(return_value=None)):
-            self.assertRaises(SystemExit, patroni_main)
+            self.assertRaises(SystemExit, _main)
 
     @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()]))
     @patch('sys.frozen', Mock(return_value=True), create=True)
@@ -94,11 +94,11 @@ class TestPatroni(unittest.TestCase):
         with patch('subprocess.call', Mock(return_value=1)):
             with patch.object(Patroni, 'run', Mock(side_effect=SleepException)):
                 os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
-                self.assertRaises(SleepException, patroni_main)
+                self.assertRaises(SleepException, _main)
             with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
                 with patch('patroni.ha.Ha.is_paused', Mock(return_value=True)):
                     os.environ['PATRONI_POSTGRESQL_DATA_DIR'] = 'data/test0'
-                    patroni_main()
+                    _main()
 
     @patch('os.getpid')
     @patch('multiprocessing.Process')
diff --git a/tests/test_raft.py b/tests/test_raft.py
index 2029a61e..ab68c2e6 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -146,8 +146,8 @@ class TestRaft(unittest.TestCase):
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
         self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}'))
-        raft.get_cluster()
-        self.assertTrue(raft.update_leader('1', failsafe={'foo': 'bat'}))
+        leader = raft.get_cluster().leader
+        self.assertTrue(raft.update_leader(leader, '1', failsafe={'foo': 'bat'}))
         self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}'))
         self.assertTrue(raft._sync_obj.set(raft.status_path, '{'))
         raft.get_citus_coordinator()
diff --git a/tests/test_validator.py b/tests/test_validator.py
index cec1f140..4c195a0e 100644
--- a/tests/test_validator.py
+++ b/tests/test_validator.py
@@ -271,3 +271,38 @@ class TestValidator(unittest.TestCase):
         errors = schema2(config_2)
         output = "\n".join(errors)
         self.assertEqual(['some_dir'], parse_output(output))
+
+    def test_validate_binary_name(self, mock_out, mock_err):
+        r = copy.copy(required_binaries)
+        r.remove('postgres')
+        r.append('fake-postgres')
+        binaries.extend(r)
+        c = copy.deepcopy(config)
+        c["postgresql"]["bin_name"] = {"postgres": "fake-postgres"}
+        del c["postgresql"]["bin_dir"]
+        errors = schema(c)
+        output = "\n".join(errors)
+        self.assertEqual(['raft.bind_addr', 'raft.self_addr'], parse_output(output))
+
+    def test_validate_binary_name_missing(self, mock_out, mock_err):
+        r = copy.copy(required_binaries)
+        r.remove('postgres')
+        binaries.extend(r)
+        c = copy.deepcopy(config)
+        c["postgresql"]["bin_name"] = {"postgres": "fake-postgres"}
+        del c["postgresql"]["bin_dir"]
+        errors = schema(c)
+        output = "\n".join(errors)
+        self.assertEqual(['postgresql.bin_dir', 'postgresql.bin_name.postgres', 'raft.bind_addr', 'raft.self_addr'],
+                         parse_output(output))
+
+    def test_validate_binary_name_empty_string(self, mock_out, mock_err):
+        r = copy.copy(required_binaries)
+        binaries.extend(r)
+        c = copy.deepcopy(config)
+        c["postgresql"]["bin_name"] = {"postgres": ""}
+        del c["postgresql"]["bin_dir"]
+        errors = schema(c)
+        output = "\n".join(errors)
+        self.assertEqual(['postgresql.bin_dir', 'postgresql.bin_name.postgres', 'raft.bind_addr', 'raft.self_addr'],
+                         parse_output(output))
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index c2f3f583..baf9ad10 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -250,14 +250,15 @@ class TestZooKeeper(unittest.TestCase):
             self.zk.take_leader()
 
     def test_update_leader(self):
-        self.assertFalse(self.zk.update_leader(12345))
+        leader = self.zk.get_cluster().leader
+        self.assertFalse(self.zk.update_leader(leader, 12345))
         with patch.object(MockKazooClient, 'delete', Mock(side_effect=RetryFailedError)):
-            self.assertRaises(ZooKeeperError, self.zk.update_leader, 12345)
+            self.assertRaises(ZooKeeperError, self.zk.update_leader, leader, 12345)
         with patch.object(MockKazooClient, 'delete', Mock(side_effect=NoNodeError)):
-            self.assertTrue(self.zk.update_leader(12345, failsafe={'foo': 'bar'}))
+            self.assertTrue(self.zk.update_leader(leader, 12345, failsafe={'foo': 'bar'}))
             with patch.object(MockKazooClient, 'create', Mock(side_effect=[RetryFailedError, Exception])):
-                self.assertRaises(ZooKeeperError, self.zk.update_leader, 12345)
-                self.assertFalse(self.zk.update_leader(12345))
+                self.assertRaises(ZooKeeperError, self.zk.update_leader, leader, 12345)
+                self.assertFalse(self.zk.update_leader(leader, 12345))
 
     @patch.object(Cluster, 'min_version', PropertyMock(return_value=(2, 0)))
     def test_write_leader_optime(self):