Repository: zalando/patroni
Pull request #2796 (2023-08-01 10:30:41)

Refactor docs about migration to Patroni
This PR is an attempt of refactoring the docs about migration to Patroni.

These are a few enhancements that we propose through this PR:

* Docs used to mention the procedure can only be performed in a single-node cluster. We changed that so the procedure considers a cluster composed of primary and standbys;
* Teach how to deal with pre-existing replication slots;
* Explain how to create the user for `pg_rewind`, if user intends to enable `use_pg_rewind`.

References: PAT-143.

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

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.859%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/61701571/badge)](https://coveralls.io/builds/61701571) |
| :-- | --: |
| Change from base [Build 5715554178](https://coveralls.io/builds/61676509): |  0.0% |
| Covered Lines: | 12772 |
| Relevant Lines: | 12790 |

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


Patch:
diff --git a/docs/dynamic_configuration.rst b/docs/dynamic_configuration.rst
index 5486bf6e..f5bb4ee5 100644
--- a/docs/dynamic_configuration.rst
+++ b/docs/dynamic_configuration.rst
@@ -46,9 +46,9 @@ In order to change the dynamic configuration you can use either ``patronictl edi
    -  **archive\_cleanup\_command**: cleanup command for standby leader
    -  **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader
 
--  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
 
-   -  **my\_slot\_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots.
+   -  **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni.
 
       -  **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
       -  **database**: the database name where logical slots should be created.
diff --git a/docs/existing_data.rst b/docs/existing_data.rst
index 7c78f6e3..cb07bfa9 100644
--- a/docs/existing_data.rst
+++ b/docs/existing_data.rst
@@ -10,18 +10,58 @@ To deploy a Patroni cluster without using a pre-existing PostgreSQL instance, se
 Procedure
 ---------
 
-A Patroni cluster can be started with a data directory from a single-node PostgreSQL database. This is achieved by following closely these steps:
+You can find below an overview of steps for converting an existing Postgres cluster to a Patroni managed cluster. In the steps we assume all nodes that are part of the existing cluster are currently up and running, and that you *do not* intend to change Postgres configuration while the migration is ongoing. The steps:
 
-1. Manually start PostgreSQL daemon
-2. Create Patroni superuser and replication users as defined in the :ref:`authentication <postgresql_settings>` section of the Patroni configuration. If this user is created in SQL, the following queries achieve this:
+#. Create the Postgres users as explained for :ref:`authentication <postgresql_settings>` section of the Patroni configuration. You can find sample SQL commands to create the users in the code block below, in which you need to replace the usernames and passwords as per your environment. If you already have the relevant users, then you can skip this step.
 
-.. code-block:: sql
+   .. code-block:: sql
 
-   CREATE USER $PATRONI_SUPERUSER_USERNAME WITH SUPERUSER ENCRYPTED PASSWORD '$PATRONI_SUPERUSER_PASSWORD';
-   CREATE USER $PATRONI_REPLICATION_USERNAME WITH REPLICATION ENCRYPTED PASSWORD '$PATRONI_REPLICATION_PASSWORD';
+      -- Patroni superuser
+      -- Replace PATRONI_SUPERUSER_USERNAME and PATRONI_SUPERUSER_PASSWORD accordingly
+      CREATE USER PATRONI_SUPERUSER_USERNAME WITH SUPERUSER ENCRYPTED PASSWORD 'PATRONI_SUPERUSER_PASSWORD';
 
-3. Start Patroni (e.g. ``patroni /etc/patroni/patroni.yml``). It automatically detects that PostgreSQL daemon is already running but its configuration might be out-of-date.
-4. Ask Patroni to restart the node with ``patronictl restart cluster-name node-name``. This step is only required if PostgreSQL configuration is out-of-date.
+      -- Patroni replication user
+      -- Replace PATRONI_REPLICATION_USERNAME and PATRONI_REPLICATION_PASSWORD accordingly
+      CREATE USER PATRONI_REPLICATION_USERNAME WITH REPLICATION ENCRYPTED PASSWORD 'PATRONI_REPLICATION_PASSWORD';
+
+      -- Patroni rewind user, if you intend to enable use_pg_rewind in your Patroni configuration
+      -- Replace PATRONI_REWIND_USERNAME and PATRONI_REWIND_PASSWORD accordingly
+      CREATE USER PATRONI_REWIND_USERNAME WITH ENCRYPTED PASSWORD 'PATRONI_REWIND_PASSWORD';
+      GRANT EXECUTE ON function pg_catalog.pg_ls_dir(text, boolean, boolean) TO PATRONI_REWIND_USERNAME;
+      GRANT EXECUTE ON function pg_catalog.pg_stat_file(text, boolean) TO PATRONI_REWIND_USERNAME;
+      GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text) TO PATRONI_REWIND_USERNAME;
+      GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text, bigint, bigint, boolean) TO PATRONI_REWIND_USERNAME;
+
+#. Perform the following steps on all Postgres nodes. Perform all steps on one node before proceeding with the next node. Start with the primary node, then proceed with each standby node:
+
+   #. If you are running Postgres through systemd, then disable the Postgres systemd unit. This is performed as Patroni manages starting and stopping the Postgres daemon.
+
+   #. Create a YAML configuration file for Patroni.
+
+      * **Note (specific for the primary node):** If you have replication slots being used for replication between cluster members, then it is recommended that you enable ``use_slots`` and configure the existing replication slots as permanent via the ``slots`` configuration item. Be aware that Patroni automatically creates replication slots for replication between members, and drops replication slots that it does not recognize, when ``use_slots`` is enabled. The idea of using permanent slots here is to allow your existing slots to persist while the migration to Patroni is in progress. See :ref:`YAML Configuration Settings <yaml_configuration>` for details.
+
+   #. Start Patroni using the ``patroni`` systemd service unit. It automatically detects that Postgres is already running and starts monitoring the instance.
+
+#. Hand over Postgres "start up procedure" to Patroni. In order to do that you need to restart the cluster members through ``patronictl restart cluster-name member-name`` command. For minimal downtime you might want to split this step into:
+
+   #. Immediate restart of the standby nodes.
+   #. Scheduled restart of the primary node within a maintenance window.
+
+#. If you configured permanent slots in step ``1.2.``, then you should remove them from ``slots`` configuration through ``patronictl edit-config cluster-name member-name`` command once the ``restart_lsn`` of the slots created by Patroni is able to catch up with the ``restart_lsn`` of the original slots for the corresponding members. By removing the slots from ``slots`` configuration you will allow Patroni to drop the original slots from your cluster once they are not needed anymore. You can find below an example query to check the ``restart_lsn`` of a couple slots, so you can compare them:
+
+   .. code-block:: sql
+
+      -- Assume original_slot_for_member_x is the name of the slot in your original
+      -- cluster for replicating changes to member X, and slot_for_member_x is the
+      -- slot created by Patroni for that purpose. You need restart_lsn of
+      -- slot_for_member_x to be >= restart_lsn of original_slot_for_member_x
+      SELECT slot_name,
+             restart_lsn
+      FROM pg_replication_slots
+      WHERE slot_name IN (
+          'original_slot_for_member_x',
+          'slot_for_member_x'
+      )
 
 .. _major_upgrade:
 
@@ -30,14 +70,14 @@ Major Upgrade of PostgreSQL Version
 
 The only possible way to do a major upgrade currently is:
 
-1. Stop Patroni
-2. Upgrade PostgreSQL binaries and perform `pg_upgrade <https://www.postgresql.org/docs/current/pgupgrade.html>`_ on the primary node
-3. Update patroni.yml
-4. Remove the initialize key from DCS or wipe complete cluster state from DCS. The second one could be achieved by running ``patronictl remove <cluster-name>``. It is necessary because pg_upgrade runs initdb which actually creates a new database with a new PostgreSQL system identifier.
-5. If you wiped the cluster state in the previous step, you may wish to copy patroni.dynamic.json from old data dir to the new one.  It will help you to retain some PostgreSQL parameters you had set before.
-6. Start Patroni on the primary node.
-7. Upgrade PostgreSQL binaries, update patroni.yml and wipe the data_dir on standby nodes.
-8. Start Patroni on the standby nodes and wait for the replication to complete.
+#. Stop Patroni
+#. Upgrade PostgreSQL binaries and perform `pg_upgrade <https://www.postgresql.org/docs/current/pgupgrade.html>`_ on the primary node
+#. Update patroni.yml
+#. Remove the initialize key from DCS or wipe complete cluster state from DCS. The second one could be achieved by running ``patronictl remove <cluster-name>``. It is necessary because pg_upgrade runs initdb which actually creates a new database with a new PostgreSQL system identifier.
+#. If you wiped the cluster state in the previous step, you may wish to copy patroni.dynamic.json from old data dir to the new one.  It will help you to retain some PostgreSQL parameters you had set before.
+#. Start Patroni on the primary node.
+#. Upgrade PostgreSQL binaries, update patroni.yml and wipe the data_dir on standby nodes.
+#. Start Patroni on the standby nodes and wait for the replication to complete.
 
 Running pg_upgrade on standby nodes is not supported by PostgreSQL. If you know what you are doing, you can try the rsync procedure described in https://www.postgresql.org/docs/current/pgupgrade.html instead of wiping data_dir on standby nodes. The safest way is however to let Patroni replicate the data for you.
 
diff --git a/patroni/config.py b/patroni/config.py
index facf16fd..d2770517 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -13,6 +13,7 @@ from . import PATRONI_ENV_PREFIX
 from .collections import CaseInsensitiveDict
 from .dcs import ClusterConfig, Cluster
 from .exceptions import ConfigParseError
+from .file_perm import pg_perm
 from .postgresql.config import ConfigHandler
 from .utils import deep_compare, parse_bool, parse_int, patch_config
 
@@ -275,11 +276,13 @@ class Config(object):
         if self._cache_needs_saving:
             tmpfile = fd = None
             try:
+                pg_perm.set_permissions_from_data_directory(self._data_dir)
                 (fd, tmpfile) = tempfile.mkstemp(prefix=self.__CACHE_FILENAME, dir=self._data_dir)
                 with os.fdopen(fd, 'w') as f:
                     fd = None
                     json.dump(self.dynamic_configuration, f)
                 tmpfile = shutil.move(tmpfile, self._cache_file)
+                os.chmod(self._cache_file, pg_perm.file_create_mode)
                 self._cache_needs_saving = False
             except Exception:
                 logger.exception('Exception when saving file: %s', self._cache_file)
diff --git a/patroni/ctl.py b/patroni/ctl.py
index 312c0cec..b1d63d2b 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -36,7 +36,7 @@ from collections import defaultdict
 from contextlib import contextmanager
 from prettytable import ALL, FRAME, PrettyTable
 from urllib.parse import urlparse
-from typing import Any, Dict, Generator, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
     from psycopg2 import cursor
@@ -1816,7 +1816,7 @@ def resume(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait: b
 
 
 @contextmanager
-def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Generator[str, None, None]:
+def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Iterator[str]:
     """Create a temporary file with specified contents that persists for the context.
 
     :param contents: binary string that will be written to the file.
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 29a4d766..4bcac3ce 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -555,36 +555,50 @@ class TimelineHistory(NamedTuple):
         return TimelineHistory(version, value, lines)
 
 
-class Cluster(NamedTuple):
-    """Immutable object (namedtuple) which represents PostgreSQL cluster.
+class Cluster(NamedTuple('Cluster',
+                         [('initialize', Optional[str]),
+                          ('config', Optional[ClusterConfig]),
+                          ('leader', Optional[Leader]),
+                          ('last_lsn', int),
+                          ('members', List[Member]),
+                          ('failover', Optional[Failover]),
+                          ('sync', SyncState),
+                          ('history', Optional[TimelineHistory]),
+                          ('slots', Optional[Dict[str, int]]),
+                          ('failsafe', Optional[Dict[str, str]]),
+                          ('workers', Dict[int, 'Cluster'])])):
+    """Immutable object (namedtuple) which represents PostgreSQL or Citus cluster.
+
+    .. note::
+        We are using an old-style attribute declaration here because otherwise it is not possible to override `__new__`
+        method. Without it the *workers* by default gets always the same :class:`dict` object that could be mutated.
+
     Consists of the following fields:
-    :param initialize: shows whether this cluster has initialization key stored in DC or not.
-    :param config: global dynamic configuration, reference to `ClusterConfig` object
-    :param leader: `Leader` object which represents current leader of the cluster
-    :param last_lsn: int or long object containing position of last known leader LSN.
-        This value is stored in the `/status` key or `/optime/leader` (legacy) key
-    :param members: list of Member object, all PostgreSQL cluster members including leader
-    :param failover: reference to `Failover` object
-    :param sync: reference to `SyncState` object, last observed synchronous replication state.
-    :param history: reference to `TimelineHistory` object
-    :param slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}
-    :param failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
-    :param workers: workers of the Citus cluster, optional. Format: {int(group): Cluster()}
+
+    :ivar initialize: shows whether this cluster has initialization key stored in DC or not.
+    :ivar config: global dynamic configuration, reference to `ClusterConfig` object.
+    :ivar leader: :class:`Leader` object which represents current leader of the cluster.
+    :ivar last_lsn: :class:int object containing position of last known leader LSN.
+                     This value is stored in the `/status` key or `/optime/leader` (legacy) key.
+    :ivar members: list of:class:` Member` objects, all PostgreSQL cluster members including leader
+    :ivar failover: reference to :class:`Failover` object.
+    :ivar sync: reference to :class:`SyncState` object, last observed synchronous replication state.
+    :ivar history: reference to `TimelineHistory` object.
+    :ivar slots: state of permanent logical replication slots on the primary in the format: {"slot_name": int}.
+    :ivar failsafe: failsafe topology. Node is allowed to become the leader only if its name is found in this list.
+    :ivar workers: dictionary of workers of the Citus cluster, optional. Each key is an :class:`int` representing
+                   the group, and the corresponding value is a :class:`Cluster` instance.
     """
-    initialize: Optional[str]
-    config: Optional[ClusterConfig]
-    leader: Optional[Leader]
-    last_lsn: int
-    members: List[Member]
-    failover: Optional[Failover]
-    sync: SyncState
-    history: Optional[TimelineHistory]
-    slots: Optional[Dict[str, int]]
-    failsafe: Optional[Dict[str, str]]
-    workers: Dict[int, 'Cluster'] = {}
+
+    def __new__(cls, *args: Any, **kwargs: Any):
+        """Make workers argument optional and set it to an empty dict object."""
+        if len(args) < len(cls._fields) and 'workers' not in kwargs:
+            kwargs['workers'] = {}
+        return super(Cluster, cls).__new__(cls, *args, **kwargs)
 
     @staticmethod
     def empty() -> 'Cluster':
+        """Produce an empty :class:`Cluster` instance."""
         return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None)
 
     def is_empty(self):
diff --git a/patroni/file_perm.py b/patroni/file_perm.py
new file mode 100644
index 00000000..ed9c4e67
--- /dev/null
+++ b/patroni/file_perm.py
@@ -0,0 +1,95 @@
+"""Helper object that helps with figuring out file and directory permissions based on permissions of PGDATA.
+
+:var logger: logger of this module.
+:var pg_perm: instance of the :class:`__FilePermissions` object.
+"""
+import logging
+import os
+import stat
+
+logger = logging.getLogger(__name__)
+
+
+class __FilePermissions:
+    """Helper class for managing permissions of directories and files under PGDATA.
+
+    Execute :meth:`set_permissions_from_data_directory` to figure out which permissions should be used for files and
+    directories under PGDATA based on permissions of PGDATA root directory.
+    """
+
+    # Mode mask for data directory permissions that only allows the owner to
+    # read/write directories and files -- mask 077.
+    __PG_MODE_MASK_OWNER = stat.S_IRWXG | stat.S_IRWXO
+
+    # Mode mask for data directory permissions that also allows group read/execute -- mask 027.
+    __PG_MODE_MASK_GROUP = stat.S_IWGRP | stat.S_IRWXO
+
+    # Default mode for creating directories -- mode 700.
+    __PG_DIR_MODE_OWNER = stat.S_IRWXU
+
+    # Mode for creating directories that allows group read/execute -- mode 750.
+    __PG_DIR_MODE_GROUP = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP
+
+    # Default mode for creating files -- mode 600.
+    __PG_FILE_MODE_OWNER = stat.S_IRUSR | stat.S_IWUSR
+
+    # Mode for creating files that allows group read -- mode 640.
+    __PG_FILE_MODE_GROUP = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP
+
+    def __init__(self) -> None:
+        """Create a :class:`__FilePermissions` object and set default permissions."""
+        self.__set_owner_permissions()
+        self.__set_umask()
+
+    def __set_umask(self) -> None:
+        """Set umask value based on calculations.
+
+        .. note::
+            Should only be called once either :meth:`__set_owner_permissions`
+            or :meth:`__set_group_permissions` has been executed.
+        """
+        try:
+            os.umask(self.__pg_mode_mask)
+        except Exception as e:
+            logger.error('Can not set umask to %03o: %r', self.__pg_mode_mask, e)
+
+    def __set_owner_permissions(self) -> None:
+        """Make directories/files accessible only by the owner."""
+        self.__pg_dir_create_mode = self.__PG_DIR_MODE_OWNER
+        self.__pg_file_create_mode = self.__PG_FILE_MODE_OWNER
+        self.__pg_mode_mask = self.__PG_MODE_MASK_OWNER
+
+    def __set_group_permissions(self) -> None:
+        """Make directories/files accessible by the owner and readable by group."""
+        self.__pg_dir_create_mode = self.__PG_DIR_MODE_GROUP
+        self.__pg_file_create_mode = self.__PG_FILE_MODE_GROUP
+        self.__pg_mode_mask = self.__PG_MODE_MASK_GROUP
+
+    def set_permissions_from_data_directory(self, data_dir: str) -> None:
+        """Set new permissions based on provided *data_dir*.
+
+        :param data_dir: reference to PGDATA to calculate permissions from.
+        """
+        try:
+            st = os.stat(data_dir)
+            if (st.st_mode & self.__PG_DIR_MODE_GROUP) == self.__PG_DIR_MODE_GROUP:
+                self.__set_group_permissions()
+            else:
+                self.__set_owner_permissions()
+        except Exception as e:
+            logger.error('Can not check permissions on %s: %r', data_dir, e)
+        else:
+            self.__set_umask()
+
+    @property
+    def dir_create_mode(self) -> int:
+        """Directory permissions."""
+        return self.__pg_dir_create_mode
+
+    @property
+    def file_create_mode(self) -> int:
+        """File permissions."""
+        return self.__pg_file_create_mode
+
+
+pg_perm = __FilePermissions()
diff --git a/patroni/ha.py b/patroni/ha.py
index 2445aa02..994db00b 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -527,10 +527,17 @@ class Ha(object):
         return msg
 
     def _get_node_to_follow(self, cluster: Cluster) -> Union[Leader, Member, None]:
-        # determine the node to follow. If replicatefrom tag is set,
-        # try to follow the node mentioned there, otherwise, follow the leader.
-        if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)):
+        """Determine the node to follow.
+
+        :param cluster: the currently known cluster state from DCS.
+
+        :returns: the node which we should be replicating from.
+        """
+        # The standby leader or when there is no standby leader we want to follow
+        # the remote member, except when there is no standby leader in pause.
+        if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()):
             node_to_follow = self.get_remote_member()
+        # If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader.
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
             node_to_follow = cluster.get_member(self.patroni.replicatefrom)
         else:
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index 04fab0a2..e9844710 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -12,7 +12,7 @@ from datetime import datetime
 from dateutil import tz
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
-from typing import Any, Callable, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from .bootstrap import Bootstrap
 from .callback_executor import CallbackAction, CallbackExecutor
@@ -999,7 +999,7 @@ class Postgresql(object):
 
     @contextmanager
     def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432,
-                                          **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+                                          **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = self.config.replication.copy()
         conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
                            connect_timeout=3, replication=1, options='-c statement_timeout=2000')
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 3a61a31d..6d1fe384 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -6,14 +6,16 @@ import socket
 import stat
 import time
 
+from contextlib import contextmanager
 from urllib.parse import urlparse, parse_qsl, unquote
 from types import TracebackType
-from typing import Any, Collection, Dict, List, Optional, Union, Tuple, Type, TYPE_CHECKING
+from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING
 
 from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException
+from ..file_perm import pg_perm
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
 from ..validator import IntValidator, EnumValidator
 
@@ -367,6 +369,30 @@ class ConfigHandler(object):
             configuration.append('pg_ident.conf')
         return configuration
 
+    def set_file_permissions(self, filename: str) -> None:
+        """Set permissions of file *filename* according to the expected permissions if it resides under PGDATA.
+
+        .. note::
+            Do nothing if the file is not under PGDATA.
+
+        :param filename: path to a file which permissions might need to be adjusted.
+        """
+        if is_subpath(self._postgresql.data_dir, filename):
+            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
+            os.chmod(filename, pg_perm.file_create_mode)
+
+    @contextmanager
+    def config_writer(self, filename: str) -> Iterator[ConfigWriter]:
+        """Create :class:`ConfigWriter` object and set permissions on a *filename*.
+
+        :param filename: path to a config file.
+
+        :yields: :class:`ConfigWriter` object.
+        """
+        with ConfigWriter(filename) as writer:
+            yield writer
+        self.set_file_permissions(filename)
+
     def save_configuration_files(self, check_custom_bootstrap: bool = False) -> bool:
         """
             copy postgresql.conf to postgresql.conf.backup to be able to retrieve configuration files
@@ -380,6 +406,7 @@ class ConfigHandler(object):
                     backup_file = os.path.join(self._postgresql.data_dir, f + '.backup')
                     if os.path.isfile(config_file):
                         shutil.copy(config_file, backup_file)
+                        self.set_file_permissions(backup_file)
             except IOError:
                 logger.exception('unable to create backup copies of configuration files')
         return True
@@ -393,9 +420,11 @@ class ConfigHandler(object):
                 if not os.path.isfile(config_file):
                     if os.path.isfile(backup_file):
                         shutil.copy(backup_file, config_file)
+                        self.set_file_permissions(config_file)
                     # Previously we didn't backup pg_ident.conf, if file is missing just create empty
                     elif f == 'pg_ident.conf':
                         open(config_file, 'w').close()
+                        self.set_file_permissions(config_file)
         except IOError:
             logger.exception('unable to restore configuration files from backup')
 
@@ -409,7 +438,7 @@ class ConfigHandler(object):
         if self._postgresql.enforce_hot_standby_feedback:
             configuration['hot_standby_feedback'] = 'on'
 
-        with ConfigWriter(self._postgresql_conf) as f:
+        with self.config_writer(self._postgresql_conf) as f:
             include = self._config.get('custom_conf') or self._postgresql_base_conf_name
             f.writeline("include '{0}'\n".format(ConfigWriter.escape(include)))
             for name, value in sorted((configuration).items()):
@@ -439,6 +468,7 @@ class ConfigHandler(object):
         if not self.hba_file and not self._config.get('pg_hba'):
             with open(self._pg_hba_conf, 'a') as f:
                 f.write('\n{}\n'.format('\n'.join(config)))
+            self.set_file_permissions(self._pg_hba_conf)
         return True
 
     def replace_pg_hba(self) -> Optional[bool]:
@@ -458,14 +488,14 @@ class ConfigHandler(object):
                                   self.local_replication_address['host'], self.local_replication_address['port'],
                                   0, socket.SOCK_STREAM, socket.IPPROTO_TCP)})
 
-            with ConfigWriter(self._pg_hba_conf) as f:
+            with self.config_writer(self._pg_hba_conf) as f:
                 for address, t in addresses.items():
                     f.writeline((
                         '{0}\treplication\t{1}\t{3}\ttrust\n'
                         '{0}\tall\t{2}\t{3}\ttrust'
                     ).format(t, self.replication['username'], self._superuser.get('username') or 'all', address))
         elif not self.hba_file and self._config.get('pg_hba'):
-            with ConfigWriter(self._pg_hba_conf) as f:
+            with self.config_writer(self._pg_hba_conf) as f:
                 f.writelines(self._config['pg_hba'])
             return True
 
@@ -478,7 +508,7 @@ class ConfigHandler(object):
         """
 
         if not self.ident_file and self._config.get('pg_ident'):
-            with ConfigWriter(self._pg_ident_conf) as f:
+            with self.config_writer(self._pg_ident_conf) as f:
                 f.writelines(self._config['pg_ident'])
             return True
 
@@ -800,9 +830,11 @@ class ConfigHandler(object):
         if self._postgresql.major_version >= 120000:
             if parse_bool(recovery_params.pop('standby_mode', None)):
                 open(self._standby_signal, 'w').close()
+                self.set_file_permissions(self._standby_signal)
             else:
                 self._remove_file_if_exists(self._standby_signal)
                 open(self._recovery_signal, 'w').close()
+                self.set_file_permissions(self._recovery_signal)
 
             def restart_required(name: str) -> bool:
                 if self._postgresql.major_version >= 140000:
@@ -813,8 +845,7 @@ class ConfigHandler(object):
             self._current_recovery_params = CaseInsensitiveDict({n: [v, restart_required(n), self._postgresql_conf]
                                                                  for n, v in recovery_params.items()})
         else:
-            with ConfigWriter(self._recovery_conf) as f:
-                os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD)
+            with self.config_writer(self._recovery_conf) as f:
                 self._write_recovery_params(f, recovery_params)
 
     def remove_recovery_conf(self) -> None:
@@ -843,6 +874,7 @@ class ConfigHandler(object):
         if overwrite:
             try:
                 with open(self._auto_conf, 'w') as f:
+                    self.set_file_permissions(self._auto_conf)
                     for raw_line in lines:
                         f.write(raw_line)
             except Exception:
diff --git a/patroni/postgresql/connection.py b/patroni/postgresql/connection.py
index 6a556983..277a4889 100644
--- a/patroni/postgresql/connection.py
+++ b/patroni/postgresql/connection.py
@@ -2,7 +2,7 @@ import logging
 
 from contextlib import contextmanager
 from threading import Lock
-from typing import Any, Dict, Generator, Union, TYPE_CHECKING
+from typing import Any, Dict, Iterator, Union, TYPE_CHECKING
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Connection as Connection3, Cursor
     from psycopg2 import connection, cursor
@@ -44,7 +44,7 @@ class Connection(object):
 
 
 @contextmanager
-def get_connection_cursor(**kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+def get_connection_cursor(**kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
     conn = psycopg.connect(**kwargs)
     with conn.cursor() as cur:
         yield cur
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index fde81a7a..82b83002 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -5,11 +5,12 @@ import shutil
 from collections import defaultdict
 from contextlib import contextmanager
 from threading import Condition, Thread
-from typing import Any, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from .connection import get_connection_cursor
 from .misc import format_lsn, fsync_dir
 from ..dcs import Cluster, Leader
+from ..file_perm import pg_perm
 from ..psycopg import OperationalError
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -253,7 +254,7 @@ class SlotsHandler(object):
                 self._schedule_load_slots = True
 
     @contextmanager
-    def get_local_connection_cursor(self, **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+    def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = self._postgresql.config.local_connect_kwargs
         conn_kwargs.update(kwargs)
         with get_connection_cursor(**conn_kwargs) as cur:
@@ -342,7 +343,7 @@ class SlotsHandler(object):
         return ret
 
     @contextmanager
-    def _get_leader_connection_cursor(self, leader: Leader) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+    def _get_leader_connection_cursor(self, leader: Leader) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials)
         conn_kwargs['dbname'] = self._postgresql.database
         with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
@@ -471,23 +472,28 @@ class SlotsHandler(object):
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
         if copy_slots and self._postgresql.stop():
+            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
             for name, value in copy_slots.items():
-                slot_dir = os.path.join(self._postgresql.slots_handler.pg_replslot_dir, name)
+                slot_dir = os.path.join(self.pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):
                     shutil.rmtree(slot_tmp_dir)
                 os.makedirs(slot_tmp_dir)
+                os.chmod(slot_tmp_dir, pg_perm.dir_create_mode)
                 fsync_dir(slot_tmp_dir)
-                with open(os.path.join(slot_tmp_dir, 'state'), 'wb') as f:
+                slot_filename = os.path.join(slot_tmp_dir, 'state')
+                with open(slot_filename, 'wb') as f:
+                    os.chmod(slot_filename, pg_perm.file_create_mode)
                     f.write(value['data'])
                     f.flush()
                     os.fsync(f.fileno())
                 if os.path.exists(slot_dir):
                     shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
+                os.chmod(slot_dir, pg_perm.dir_create_mode)
                 fsync_dir(slot_dir)
                 self._logical_slots_processing_queue[name] = None
-            fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
+            fsync_dir(self.pg_replslot_dir)
             self._postgresql.start()
 
     def schedule(self, value: Optional[bool] = None) -> None:


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
index 3f7e4049..cf798d00 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -85,6 +85,7 @@ class TestConfig(unittest.TestCase):
     @patch('os.path.exists', Mock(return_value=True))
     @patch('os.remove', Mock(side_effect=IOError))
     @patch('os.close', Mock(side_effect=IOError))
+    @patch('os.chmod', Mock())
     @patch('shutil.move', Mock(return_value=None))
     @patch('json.dump', Mock())
     def test_save_cache(self):
diff --git a/tests/test_file_perm.py b/tests/test_file_perm.py
new file mode 100644
index 00000000..e85b3c65
--- /dev/null
+++ b/tests/test_file_perm.py
@@ -0,0 +1,33 @@
+import unittest
+import stat
+
+from mock import Mock, patch
+
+from patroni.file_perm import pg_perm
+
+
+class TestFilePermissions(unittest.TestCase):
+
+    @patch('os.stat')
+    @patch('os.umask')
+    @patch('patroni.file_perm.logger.error')
+    def test_set_umask(self, mock_logger, mock_umask, mock_stat):
+        mock_umask.side_effect = Exception
+        mock_stat.return_value.st_mode = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP
+        pg_perm.set_permissions_from_data_directory('test')
+
+        # umask is called with PG_MODE_MASK_GROUP
+        self.assertEqual(mock_umask.call_args[0][0], stat.S_IWGRP | stat.S_IRWXO)
+        self.assertEqual(mock_logger.call_args[0][0], 'Can not set umask to %03o: %r')
+
+        mock_umask.reset_mock()
+        mock_stat.return_value.st_mode = stat.S_IRWXU
+        pg_perm.set_permissions_from_data_directory('test')
+        # umask is called with PG_MODE_MASK_OWNER (permissions changed from group to owner)
+        self.assertEqual(mock_umask.call_args[0][0], stat.S_IRWXG | stat.S_IRWXO)
+
+    @patch('os.stat', Mock(side_effect=FileNotFoundError))
+    @patch('patroni.file_perm.logger.error')
+    def test_set_permissions_from_data_directory(self, mock_logger):
+        pg_perm.set_permissions_from_data_directory('test')
+        self.assertEqual(mock_logger.call_args[0][0], 'Can not check permissions on %s: %r')
diff --git a/tests/test_ha.py b/tests/test_ha.py
index ae6be3e2..d3a75ed3 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -1421,6 +1421,7 @@ class TestHa(PostgresInit):
     @patch('os.open', Mock())
     @patch('os.fsync', Mock())
     @patch('os.close', Mock())
+    @patch('os.chmod', Mock())
     @patch('os.rename', Mock())
     @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
     @patch('builtins.open', mock_open())
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 036e225b..b584f2b2 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -523,8 +523,9 @@ class TestPostgresql(BaseTestPostgresql):
     def test_save_configuration_files(self):
         self.p.config.save_configuration_files()
 
-    @patch('os.path.isfile', Mock(side_effect=[False, True]))
-    @patch('shutil.copy', Mock(side_effect=IOError))
+    @patch('os.path.isfile', Mock(side_effect=[False, True, False, True]))
+    @patch('shutil.copy', Mock(side_effect=[None, IOError]))
+    @patch('os.chmod', Mock())
     def test_restore_configuration_files(self):
         self.p.config.restore_configuration_files()