diff --git a/docs/replica_bootstrap.rst b/docs/replica_bootstrap.rst index 233c2c5b..f5359e7b 100644 --- a/docs/replica_bootstrap.rst +++ b/docs/replica_bootstrap.rst @@ -76,13 +76,14 @@ scripts to clone a new replica. Those are configured in the ``postgresql`` confi no_master: 1 envdir: {{WALE_ENV_DIR}} use_iam: 1 + basebackup: + max-rate: '100M' The ``create_replica_method`` defines available replica creation methods and the order of executing them. Patroni will -stop on the first one that returns 0. The basebackup is the built-in method and doesn't require any configuration. The -rest of the methods should define a separate section in the configuration file, listing the command to execute and any -custom parameters that should be passed to that command. All parameters will be passed in a ``--name=value`` format. -Besides user-defined parameters, Patroni supplies a couple of cluster-specific ones: +stop on the first one that returns 0. Each method should define a separate section in the configuration file, listing the command +to execute and any custom parameters that should be passed to that command. All parameters will be passed in a +``--name=value`` format. Besides user-defined parameters, Patroni supplies a couple of cluster-specific ones: --scope Which cluster this replica belongs to @@ -98,4 +99,32 @@ A special ``no_master`` parameter, if defined, allows Patroni to call the replic running master or replicas. In that case, an empty string will be passed in a connection string. This is useful for restoring the formerly running cluster from the binary backup. +A ``basebackup`` method is a special case: it will be used if ``create_replica_method`` is empty, although it is possible +to list it explicitly among the ``create_replica_method`` methods. This method initializes a new replica with the +``pg_basebackup``, the base backup is taken from the master unless there are replicas with ``clonefrom`` tag, in which case one +of such replicas will be used as the origin for pg_basebackup. It works without any configuration; however, it is +possible to specify a ``basebackup`` configuration section. Same rules as with the other method configuration apply, +namely, only long (with --) options should be specified there. Not all parameters make sense, if you override a connection +string or provide an option to created tar-ed or compressed base backups, patroni won't be able to make a replica out +of it. There is no validation performed on the names or values of the parameters passed to the ``basebackup`` section. +You can specify basebackup parameters as either a map (key-value pairs) or a list of elements, where each element +could be either a key-value pair or a single key (for options that does not receive any values, for instance, ``--verbose``). +Consider those 2 examples: + +.. code:: YAML + + postgresql: + basebackup: + max-rate: '100M' + checkpoint: 'fast' + +and + +.. code:: YAML + + postgresql: + basebackup: + - verbose + - max-rate: '100M' + If all replica creation methods fail, Patroni will try again all methods in order during the next event loop cycle. diff --git a/patroni/postgresql.py b/patroni/postgresql.py index f31e4828..fc473f00 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -498,28 +498,44 @@ class Postgresql(object): return not os.path.exists(self._data_dir) or os.listdir(self._data_dir) == [] @staticmethod - def initdb_allowed_option(name): - if name in ['pgdata', 'nosync', 'pwfile', 'sync-only']: - raise Exception('{0} option for initdb is not allowed'.format(name)) - return True + def process_user_options(tool, options, not_allowed_options, error_handler): + user_options = [] - def get_initdb_options(self, config): - options = [] - for o in config: - if isinstance(o, string_types) and self.initdb_allowed_option(o): - options.append('--{0}'.format(o)) - elif isinstance(o, dict): - keys = list(o.keys()) - if len(keys) != 1 or not isinstance(keys[0], string_types) or not self.initdb_allowed_option(keys[0]): - raise Exception('Invalid option: {0}'.format(o)) - options.append('--{0}={1}'.format(keys[0], o[keys[0]])) - else: - raise Exception('Unknown type of initdb option: {0}'.format(o)) - return options + def option_is_allowed(name): + ret = name not in not_allowed_options + if not ret: + error_handler('{0} option for {1} is not allowed'.format(name, tool)) + return ret + + if isinstance(options, dict): + for k, v in options.items(): + if k and v: + user_options.append('--{0}={1}'.format(k, v)) + elif isinstance(options, list): + for opt in options: + if isinstance(opt, string_types) and option_is_allowed(opt): + user_options.append('--{0}'.format(opt)) + elif isinstance(opt, dict): + keys = list(opt.keys()) + if len(keys) != 1 or not isinstance(opt[keys[0]], string_types) or not option_is_allowed(keys[0]): + error_handler('Error when parsing {0} key-value option {1}: only one key-value is allowed' + ' and value should be a string'.format(tool, opt[keys[0]])) + user_options.append('--{0}={1}'.format(keys[0], opt[keys[0]])) + else: + error_handler('Error when parsing {0} option {1}: value should be string value' + ' or a single key-value pair'.format(tool, opt)) + else: + error_handler('{0} options must be list ot dict'.format(tool)) + return user_options def _initdb(self, config): self.set_state('initalizing new cluster') - options = self.get_initdb_options(config.get('initdb') or []) + not_allowed_options = ('pgdata', 'nosync', 'pwfile', 'sync-only', 'version') + + def error_handler(e): + raise Exception(e) + + options = self.process_user_options('initdb', config.get('initdb') or [], not_allowed_options, error_handler) pwfile = None if self._superuser: @@ -659,7 +675,7 @@ class Postgresql(object): break # if the method is basebackup, then use the built-in if replica_method == "basebackup": - ret = self.basebackup(connstring, env) + ret = self.basebackup(connstring, env, self.config.get(replica_method, {})) if ret == 0: logger.info("replica has been created using basebackup") # if basebackup succeeds, exit with success @@ -672,7 +688,7 @@ class Postgresql(object): method_config = {} # user-defined method; check for configuration # not required, actually - if replica_method in self.config: + if self.config.get(replica_method, {}): method_config = self.config[replica_method].copy() # look to see if the user has supplied a full command path # if not, use the method name as the command @@ -1601,23 +1617,27 @@ $$""".format(name, ' '.join(options)), name, password, password) logger.exception('Could not remove data directory %s', self._data_dir) self.move_data_directory() - def basebackup(self, conn_url, env): + def basebackup(self, conn_url, env, options): # creates a replica data dir using pg_basebackup. # this is the default, built-in create_replica_method # tries twice, then returns failure (as 1) # uses "stream" as the xlog-method to avoid sync issues + # supports additional user-supplied options, those are not validated maxfailures = 2 ret = 1 + not_allowed_options = ('pgdata', 'format', 'wal-method', 'xlog-method', 'gzip', + 'version', 'compress', 'dbname', 'host', 'port', 'username', 'password') + user_options = self.process_user_options('basebackup', options, not_allowed_options, logger.error) + for bbfailures in range(0, maxfailures): with self._cancellable_lock: if self._is_cancelled: break if not self.data_directory_empty(): self.remove_data_directory() - try: ret = self.cancellable_subprocess_call([self._pgcommand('pg_basebackup'), '--pgdata=' + self._data_dir, - '-X', 'stream', '--dbname=' + conn_url], env=env) + '-X', 'stream', '--dbname=' + conn_url] + user_options, env=env) if ret == 0: break else: diff --git a/postgres1.yml b/postgres1.yml index 9c1d141b..683bc965 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -75,6 +75,9 @@ postgresql: password: zalando parameters: unix_socket_directories: '.' + basebackup: + - verbose + - max-rate: 100M tags: nofailover: false noloadbalance: false diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index b6f4d2d5..7fab3d65 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -210,12 +210,11 @@ class TestPostgresql(unittest.TestCase): def tearDown(self): shutil.rmtree('data') - def test_get_initdb_options(self): - self.assertEquals(self.p.get_initdb_options([{'encoding': 'UTF8'}, 'data-checksums']), - ['--encoding=UTF8', '--data-checksums']) - self.assertRaises(Exception, self.p.get_initdb_options, [{'pgdata': 'bar'}]) - self.assertRaises(Exception, self.p.get_initdb_options, [{'foo': 'bar', 1: 2}]) - self.assertRaises(Exception, self.p.get_initdb_options, [1]) + def test__initdb(self): + self.assertRaises(Exception, self.p.bootstrap, {'initdb': [{'pgdata': 'bar'}]}) + self.assertRaises(Exception, self.p.bootstrap, {'initdb': [{'foo': 'bar', 1: 2}]}) + self.assertRaises(Exception, self.p.bootstrap, {'initdb': [1]}) + self.assertRaises(Exception, self.p.bootstrap, {'initdb': 1}) @patch('os.path.exists', Mock(return_value=True)) @patch('os.unlink', Mock()) @@ -428,6 +427,29 @@ class TestPostgresql(unittest.TestCase): del self.p.config['wale'] self.assertEquals(self.p.create_replica(self.leader), 0) + self.p.config['create_replica_method'] = ['basebackup'] + self.p.config['basebackup'] = [{'max_rate': '100M'}, 'no-sync'] + self.assertEquals(self.p.create_replica(self.leader), 0) + + self.p.config['basebackup'] = [{'max_rate': '100M', 'compress': '9'}] + with mock.patch('patroni.postgresql.logger.error', new_callable=Mock()) as mock_logger: + self.p.create_replica(self.leader) + mock_logger.assert_called_once() + self.assertTrue("only one key-value is allowed and value should be a string" in mock_logger.call_args[0][0], + "not matching {0}".format(mock_logger.call_args[0][0])) + + self.p.config['basebackup'] = [42] + with mock.patch('patroni.postgresql.logger.error', new_callable=Mock()) as mock_logger: + self.p.create_replica(self.leader) + mock_logger.assert_called_once() + self.assertTrue("value should be string value or a single key-value pair" in mock_logger.call_args[0][0], + "not matching {0}".format(mock_logger.call_args[0][0])) + + self.p.config['basebackup'] = {"foo": "bar"} + self.assertEquals(self.p.create_replica(self.leader), 0) + + self.p.config['create_replica_method'] = ['wale', 'basebackup'] + del self.p.config['basebackup'] mock_cancellable_subprocess_call.return_value = 1 self.assertEquals(self.p.create_replica(self.leader), 1) @@ -445,7 +467,7 @@ class TestPostgresql(unittest.TestCase): def test_basebackup(self): self.p.cancel() - self.p.basebackup(None, None) + self.p.basebackup(None, None, {'foo': 'bar'}) @patch.object(Postgresql, 'is_running', Mock(return_value=True)) def test_sync_replication_slots(self): @@ -463,8 +485,10 @@ class TestPostgresql(unittest.TestCase): cluster.members.extend([alias1, alias2]) self.p.sync_replication_slots(cluster) errorlog_mock.assert_called_once() - assert "test-3" in errorlog_mock.call_args[0][1] - assert "test.3" in errorlog_mock.call_args[0][1] + self.assertTrue("test-3" in errorlog_mock.call_args[0][1], + "non matching {0}".format(errorlog_mock.call_args[0][1])) + self.assertTrue("test.3" in errorlog_mock.call_args[0][1], + "non matching {0}".format(errorlog_mock.call_args[0][1])) @patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)) def test__query(self):