Config validator implemented (#1314)

patroni/__init__.py

@@ -169,15 +169,21 @@ class Patroni(object):
 def patroni_main():
     import argparse
     from patroni.config import Config, ConfigParseError
+    from patroni.validator import schema
 
     parser = argparse.ArgumentParser()
     parser.add_argument('--version', action='version', version='%(prog)s {0}'.format(__version__))
+    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()
     try:
-        conf = Config(args.configfile)
+        if args.validate_config:
+            conf = Config(args.configfile, validator=schema)
+            sys.exit()
+        else:
+            conf = Config(args.configfile)
     except ConfigParseError as e:
         if e.value:
             print(e.value)

patroni/postgresql/__init__.py

@@ -17,7 +17,7 @@ from patroni.postgresql.misc import parse_history, postgres_major_version_to_int
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.postgresql.slots import SlotsHandler
 from patroni.exceptions import PostgresConnectionException
-from patroni.utils import Retry, RetryFailedError, polling_loop
+from patroni.utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty
 from threading import current_thread, Lock
 
 
@@ -266,9 +266,7 @@ class Postgresql(object):
     def data_directory_empty(self):
         if self.pg_control_exists():
             return False
-        if not os.path.exists(self._data_dir):
+        return data_directory_is_empty(self._data_dir)
-            return True
-        return all(os.name != 'nt' and (n.startswith('.') or n == 'lost+found') for n in os.listdir(self._data_dir))
 
     def replica_method_options(self, method):
         return deepcopy(self.config.get(method, {}))

patroni/utils.py

@@ -442,3 +442,9 @@ def validate_directory(d, msg="{} {}"):
             raise PatroniException(msg.format(d, "the directory is not writable"))
     else:
         raise PatroniException(msg.format(d, "is not a directory"))
+
+
+def data_directory_is_empty(data_dir):
+    if not os.path.exists(data_dir):
+        return True
+    return all(os.name != 'nt' and (n.startswith('.') or n == 'lost+found') for n in os.listdir(data_dir))

patroni/validator.py

@@ -0,0 +1,379 @@
+#!/usr/bin/env python3
+import os
+import socket
+import re
+import subprocess
+
+from patroni.utils import split_host_port, data_directory_is_empty
+from patroni.ctl import find_executable
+from patroni.dcs import dcs_modules
+from patroni.exceptions import ConfigParseError
+from six import string_types
+
+
+def data_directory_empty(data_dir):
+    if os.path.isfile(os.path.join(data_dir, "global", "pg_control")):
+        return False
+    return data_directory_is_empty(data_dir)
+
+
+def validate_connect_address(address):
+    try:
+        host, _ = split_host_port(address, None)
+    except (ValueError, TypeError):
+        raise ConfigParseError("contains a wrong value")
+    if host in ["127.0.0.1", "0.0.0.0", "*", "::1"]:
+        raise ConfigParseError('must not contain "127.0.0.1", "0.0.0.0", "*", "::1"')
+    return True
+
+
+def validate_host_port(host_port, listen=False, multiple_hosts=False):
+    try:
+        hosts, port = split_host_port(host_port, None)
+    except (ValueError, TypeError):
+        raise ConfigParseError("contains a wrong value")
+    else:
+        if multiple_hosts:
+            hosts = hosts.split(",")
+        else:
+            hosts = [hosts]
+        for host in hosts:
+            proto = socket.getaddrinfo(host, "", 0, socket.SOCK_STREAM, 0, socket.AI_PASSIVE)
+            s = socket.socket(proto[0][0], socket.SOCK_STREAM)
+            try:
+                if s.connect_ex((host, port)) == 0:
+                    if listen:
+                        raise ConfigParseError("Port {} is already in use.".format(port))
+                elif not listen:
+                    raise ConfigParseError("{} is not reachable".format(host_port))
+            except socket.gaierror as e:
+                raise ConfigParseError(e)
+            finally:
+                s.close()
+    return True
+
+
+def comma_separated_host_port(string):
+    assert all([validate_host_port(s.strip()) for s in string.split(",")]), "didn't pass the validation"
+    return True
+
+
+def validate_host_port_listen(host_port):
+    return validate_host_port(host_port, listen=True)
+
+
+def validate_host_port_listen_multiple_hosts(host_port):
+    return validate_host_port(host_port, listen=True, multiple_hosts=True)
+
+
+def is_ipv4_address(ip):
+    try:
+        socket.inet_aton(ip)
+    except Exception:
+        raise ConfigParseError("Is not a valid ipv4 address")
+    return True
+
+
+def is_ipv6_address(ip):
+    try:
+        socket.inet_pton(socket.AF_INET6, ip)
+    except Exception:
+        raise ConfigParseError("Is not a valid ipv6 address")
+    return True
+
+
+def get_major_version(bin_dir=None):
+    if not bin_dir:
+        binary = 'postgres'
+    else:
+        binary = os.path.join(bin_dir, 'postgres')
+    version = subprocess.check_output([binary, '--version']).decode()
+    version = re.match(r'^[^\s]+ [^\s]+ (\d+)(\.(\d+))?', version)
+    return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1)
+
+
+def validate_data_dir(data_dir):
+    if not data_dir:
+        raise ConfigParseError("is an empty string")
+    elif os.path.exists(data_dir) and not os.path.isdir(data_dir):
+        raise ConfigParseError("is not a directory")
+    elif not data_directory_empty(data_dir):
+        if not os.path.exists(os.path.join(data_dir, "PG_VERSION")):
+            raise ConfigParseError("doesn't look like a valid data directory")
+        else:
+            with open(os.path.join(data_dir, "PG_VERSION"), "r") as version:
+                pgversion = version.read().strip()
+            waldir = ("pg_wal" if float(pgversion) >= 10 else "pg_xlog")
+            if not os.path.isdir(os.path.join(data_dir, waldir)):
+                raise ConfigParseError("data dir for the cluster is not empty, but doesn't contain"
+                                       " \"{}\" directory".format(waldir))
+            bin_dir = schema.data.get("postgresql", {}).get("bin_dir", None)
+            major_version = get_major_version(bin_dir)
+            if pgversion != major_version:
+                raise ConfigParseError("data_dir directory postgresql version ({}) doesn't match"
+                                       "with 'postgres --version' output ({})".format(pgversion, major_version))
+    return True
+
+
+class Result(object):
+    def __init__(self, status, error="didn't pass validation", level=0, path="", data=""):
+        self.status = status
+        self.path = path
+        self.data = data
+        self.level = level
+        self._error = error
+        if not self.status:
+            self.error = error
+        else:
+            self.error = None
+
+    def __repr__(self):
+        return self.path + (" " + str(self.data) + " " + self._error if self.error else "")
+
+
+class Case(object):
+    def __init__(self, schema):
+        self._schema = schema
+
+
+class Or(object):
+    def __init__(self, *args):
+        self.args = args
+
+
+class Optional(object):
+    def __init__(self, name):
+        self.name = name
+
+
+class Directory(object):
+    def __init__(self, contains=None, contains_executable=None):
+        self.contains = contains
+        self.contains_executable = contains_executable
+
+    def validate(self, name):
+        if not name:
+            yield Result(False, "is an empty string")
+        elif not os.path.exists(name):
+            yield Result(False, "Directory '{}' does not exist.".format(name))
+        elif not os.path.isdir(name):
+            yield Result(False, "'{}' is not a directory.".format(name))
+        else:
+            if self.contains:
+                for path in self.contains:
+                    if not os.path.exists(os.path.join(name, path)):
+                        yield Result(False, "'{}' does not contain '{}'".format(name, path))
+            if self.contains_executable:
+                for program in self.contains_executable:
+                    if not find_executable(program, name):
+                        yield Result(False, "'{}' does not contain '{}'".format(name, program))
+
+
+class Schema(object):
+    def __init__(self, validator):
+        self.validator = validator
+
+    def __call__(self, data):
+        for i in self.validate(data):
+            if not i.status:
+                print(i)
+
+    def validate(self, data):
+        self.data = data
+        if isinstance(self.validator, string_types):
+            yield Result(isinstance(self.data, string_types), "is not a string", level=1, data=self.data)
+        elif issubclass(type(self.validator), type):
+            validator = self.validator
+            if self.validator == str:
+                validator = string_types
+            yield Result(isinstance(self.data, validator),
+                         "is not {}".format(_get_type_name(self.validator)), level=1, data=self.data)
+        elif callable(self.validator):
+            if hasattr(self.validator, "expected_type"):
+                if not isinstance(data, self.validator.expected_type):
+                    yield Result(False, "is not {}"
+                                 .format(_get_type_name(self.validator.expected_type)), level=1, data=self.data)
+                    return
+            try:
+                self.validator(data)
+                yield Result(True, data=self.data)
+            except Exception as e:
+                yield Result(False, "didn't pass validation: {}".format(e), data=self.data)
+        elif isinstance(self.validator, dict):
+            if not len(self.validator):
+                yield Result(isinstance(self.data, dict), "is not a dictionary", level=1, data=self.data)
+        elif isinstance(self.validator, list):
+            if not isinstance(self.data, list):
+                yield Result(isinstance(self.data, list), "is not a list", level=1, data=self.data)
+                return
+        for i in self.iter():
+            yield i
+
+    def iter(self):
+        if isinstance(self.validator, dict):
+            if not isinstance(self.data, dict):
+                yield Result(False, "is not a dictionary.", level=1)
+            else:
+                for i in self.iter_dict():
+                    yield i
+        elif isinstance(self.validator, list):
+            if len(self.data) == 0:
+                yield Result(False, "is an empty list", data=self.data)
+            if len(self.validator) > 0:
+                for key, value in enumerate(self.data):
+                    for v in Schema(self.validator[0]).validate(value):
+                        yield Result(v.status, v.error,
+                                     path=(str(key) + ("." + v.path if v.path else "")), level=v.level, data=value)
+        elif isinstance(self.validator, Directory):
+            for v in self.validator.validate(self.data):
+                yield v
+        elif isinstance(self.validator, Or):
+            for i in self.iter_or():
+                yield i
+
+    def iter_dict(self):
+        for key in self.validator.keys():
+            for d in self._data_key(key):
+                if d not in self.data and not isinstance(key, Optional):
+                    yield Result(False, "is not defined.", path=d)
+                elif d not in self.data and isinstance(key, Optional):
+                    continue
+                else:
+                    validator = self.validator[key]
+                    if isinstance(key, Or) and isinstance(self.validator[key], Case):
+                        validator = self.validator[key]._schema[d]
+                    for v in Schema(validator).validate(self.data[d]):
+                        yield Result(v.status, v.error,
+                                     path=(d + ("." + v.path if v.path else "")), level=v.level, data=v.data)
+
+    def iter_or(self):
+        results = []
+        for a in self.validator.args:
+            r = []
+            for v in Schema(a).validate(self.data):
+                r.append(v)
+            if any([x.status for x in r]) and not all([x.status for x in r]):
+                results += filter(lambda x: not x.status, r)
+            else:
+                results += r
+        if not any([x.status for x in results]):
+            max_level = 3
+            for v in sorted(results, key=lambda x: x.level):
+                if v.level > max_level:
+                    break
+                max_level = v.level
+                yield Result(v.status, v.error, path=v.path, level=v.level, data=v.data)
+
+    def _data_key(self, key):
+        if isinstance(self.data, dict) and isinstance(key, str):
+            yield key
+        elif isinstance(key, Optional):
+            yield key.name
+        elif isinstance(key, Or):
+            if any([i in self.data for i in key.args]):
+                for i in key.args:
+                    if i in self.data:
+                        yield i
+            else:
+                for i in key.args:
+                    yield i
+
+
+def _get_type_name(python_type):
+    return {str: 'a string', int: 'and integer', float: 'a number', bool: 'a boolean',
+            list: 'an array', dict: 'a dictionary', string_types: "a string"}.get(
+                    python_type, getattr(python_type, __name__, "unknown type"))
+
+
+def assert_(condition, message="Wrong value"):
+    assert condition, message
+
+
+userattributes = {"username": "", Optional("password"): ""}
+available_dcs = [m.split(".")[-1] for m in dcs_modules()]
+comma_separated_host_port.expected_type = string_types
+validate_connect_address.expected_type = string_types
+validate_host_port_listen.expected_type = string_types
+validate_host_port_listen_multiple_hosts.expected_type = string_types
+validate_data_dir.expected_type = string_types
+
+schema = Schema({
+  "name": str,
+  "scope": str,
+  "restapi": {
+    "listen": validate_host_port_listen,
+    "connect_address": validate_connect_address
+  },
+  Optional("bootstrap"): {
+    "dcs": {
+        Optional("ttl"): int,
+        Optional("loop_wait"): int,
+        Optional("retry_timeout"): int,
+        Optional("maximum_lag_on_failover"): int
+        },
+    "pg_hba": [str],
+    "initdb": [Or(str, dict)]
+  },
+  Or(*available_dcs): Case({
+      "consul": {
+          Or("host", "url"): Case({
+              "host": validate_host_port,
+              "url": str})
+          },
+      "etcd": {
+          Or("host", "hosts", "srv", "url", "proxy"): Case({
+              "host": validate_host_port,
+              "hosts": Or(comma_separated_host_port, [validate_host_port]),
+              "srv": str,
+              "url": str,
+              "proxy": str})
+         },
+      "exhibitor": {
+          "hosts": [str],
+          "port": lambda i: assert_(int(i) <= 65535),
+          Optional("pool_interval"): int
+          },
+      "zookeeper": {
+          "hosts": Or(comma_separated_host_port, [validate_host_port]),
+          },
+      "kubernetes": {
+          "labels": {},
+          Optional("namespace"): str,
+          Optional("scope_label"): str,
+          Optional("role_label"): str,
+          Optional("use_endpoints"): bool,
+          Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address),
+          Optional("ports"): [{"name": str, "port": int}],
+          },
+      }),
+  "postgresql": {
+    "listen": validate_host_port_listen_multiple_hosts,
+    "connect_address": validate_connect_address,
+    "authentication": {
+      "replication": userattributes,
+      "superuser": userattributes,
+      "rewind":  userattributes
+    },
+    "data_dir": validate_data_dir,
+    Optional("bin_dir"): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup",
+                                                        "postgres", "pg_isready"]),
+    Optional("parameters"): {
+      Optional("unix_socket_directories"): lambda s: assert_(all([isinstance(s, string_types), len(s)]))
+    },
+    Optional("pg_hba"): [str],
+    Optional("pg_ident"): [str],
+    Optional("pg_ctl_timeout"): int,
+    Optional("use_pg_rewind"): bool
+  },
+  Optional("watchdog"): {
+    Optional("mode"): lambda m: assert_(m in ["off", "automatic", "required"]),
+    Optional("device"): str
+  },
+  Optional("tags"): {
+    Optional("nofailover"): bool,
+    Optional("clonefrom"): bool,
+    Optional("noloadbalance"): bool,
+    Optional("replicatefrom"): str,
+    Optional("nosync"): bool
+  }
+})

tests/test_validator.py

@@ -0,0 +1,221 @@
+import unittest
+import os
+import socket
+import copy
+from mock import Mock, patch, mock_open
+from patroni.validator import schema
+from six import StringIO
+
+config = {
+    "name": "string",
+    "scope": "string",
+    "restapi": {
+        "listen": "127.0.0.2:800",
+        "connect_address": "127.0.0.2:800"
+    },
+    "bootstrap": {
+        "dcs": {
+            "ttl": 1000,
+            "loop_wait": 1000,
+            "retry_timeout": 1000,
+            "maximum_lag_on_failover": 1000
+            },
+        "pg_hba": ["string"],
+        "initdb": ["string", {"key":"value"}]
+    },
+    "consul": {
+        "host": "127.0.0.1:5000"
+    },
+    "etcd": {
+        "hosts": "127.0.0.1:2379,127.0.0.1:2380"
+    },
+    "exhibitor": {
+        "hosts": ["string"],
+        "port": 4000,
+        "pool_interval": 1000
+    },
+    "zookeeper": {
+        "hosts":  "127.0.0.1:3379,127.0.0.1:3380"
+    },
+    "kubernetes": {
+        "namespace": "string",
+        "labels": {},
+        "scope_label": "string",
+        "role_label": "string",
+        "use_endpoints": False,
+        "pod_ip": "127.0.0.1",
+        "ports": [{"name": "string", "port": 1000}],
+    },
+    "postgresql": {
+        "listen": "127.0.0.2,::1:543",
+        "connect_address": "127.0.0.2:543",
+        "authentication": {
+            "replication": {"username": "user"},
+            "superuser": {"username": "user"},
+            "rewind": {"username": "user"},
+        },
+        "data_dir": "/tmp/data_dir",
+        "bin_dir": "/tmp/bin_dir",
+        "parameters": {
+            "unix_socket_directories": "."
+        },
+        "pg_hba": [u"string"],
+        "pg_ident": ["string"],
+        "pg_ctl_timeout": 1000,
+        "use_pg_rewind": False
+    },
+    "watchdog": {
+        "mode": "off",
+        "device": "string"
+    },
+    "tags": {
+      "nofailover": False,
+      "clonefrom": False,
+      "noloadbalance": False,
+      "nosync": False
+    }
+}
+
+directories = []
+files = []
+
+def isfile_side_effect(arg):
+    return arg in files
+
+
+def isdir_side_effect(arg):
+    return arg in directories
+
+
+def exists_side_effect(arg):
+    return isfile_side_effect(arg) or isdir_side_effect(arg)
+
+
+def connect_side_effect(host_port):
+    _, port = host_port
+    if port < 1000:
+        return 1
+    elif port < 10000:
+        return 0
+    else:
+        raise socket.gaierror()
+
+
+def parse_output(output):
+    result = []
+    for s in output.split("\n"):
+        x = s.split(" ")[0]
+        if x and x not in result:
+            result.append(x)
+    result.sort()
+    return result
+
+
+@patch('socket.socket.connect_ex', Mock(side_effect=connect_side_effect))
+@patch('os.path.exists', Mock(side_effect=exists_side_effect))
+@patch('os.path.isdir', Mock(side_effect=isdir_side_effect))
+@patch('os.path.isfile', Mock(side_effect=isfile_side_effect))
+@patch('sys.stderr', new_callable=StringIO)
+@patch('sys.stdout', new_callable=StringIO)
+class TestValidator(unittest.TestCase):
+
+    def setUp(self):
+        del files[:]
+        del directories[:]
+
+    def test_empty_config(self, mock_out, mock_err):
+        schema({})
+        output = mock_out.getvalue()
+        self.assertEqual(['consul', 'etcd', 'exhibitor', 'kubernetes', 'name', 'postgresql', 'restapi', 'scope', 'zookeeper'], parse_output(output))
+
+    def test_complete_config(self, mock_out, mock_err):
+        schema(config)
+        output = mock_out.getvalue()
+        self.assertEqual(['postgresql.bin_dir'], parse_output(output))
+
+    def test_bin_dir_is_file(self, mock_out, mock_err):
+        files.append(config["postgresql"]["data_dir"])
+        files.append(config["postgresql"]["bin_dir"])
+        c = copy.deepcopy(config)
+        c["restapi"]["connect_address"] = False
+        c["etcd"]["hosts"] = ["127.0.0.1:2379","1244.0.0.1:2379","127.0.0.1:invalidport"]
+        c["kubernetes"]["pod_ip"] = "127.0.0.1111"
+        schema(c)
+        output = mock_out.getvalue()
+        self.assertEqual(['etcd.hosts.1', 'etcd.hosts.2', 'kubernetes.pod_ip', 'postgresql.bin_dir',
+                          'postgresql.data_dir', 'restapi.connect_address'] , parse_output(output))
+
+    def test_bin_dir_is_empty(self, mock_out, mock_err):
+        directories.append(config["postgresql"]["data_dir"])
+        directories.append(config["postgresql"]["bin_dir"])
+        files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
+        c = copy.deepcopy(config)
+        c["restapi"]["connect_address"] = "127.0.0.1"
+        c["kubernetes"]["pod_ip"] = "::1"
+        c["consul"]["host"] = "127.0.0.1:50000"
+        c["etcd"]["host"] = "127.0.0.1:237"
+        c["postgresql"]["listen"] = "127.0.0.1:5432"
+        with patch('patroni.validator.open', mock_open(read_data='9')):
+            schema(c)
+        output = mock_out.getvalue()
+        self.assertEqual(['consul.host', 'etcd.host', 'postgresql.bin_dir', 'postgresql.data_dir',
+                          'postgresql.listen', 'restapi.connect_address'], parse_output(output))
+
+    @patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 12.1"))
+    def test_data_dir_contains_pg_version(self, mock_out, mock_err):
+        directories.append(config["postgresql"]["data_dir"])
+        directories.append(config["postgresql"]["bin_dir"])
+        directories.append(os.path.join(config["postgresql"]["data_dir"], "pg_wal"))
+        files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
+        files.append(os.path.join(config["postgresql"]["data_dir"], "PG_VERSION"))
+        files.append(os.path.join(config["postgresql"]["bin_dir"], "pg_ctl"))
+        files.append(os.path.join(config["postgresql"]["bin_dir"], "initdb"))
+        files.append(os.path.join(config["postgresql"]["bin_dir"], "pg_controldata"))
+        files.append(os.path.join(config["postgresql"]["bin_dir"], "pg_basebackup"))
+        files.append(os.path.join(config["postgresql"]["bin_dir"], "postgres"))
+        files.append(os.path.join(config["postgresql"]["bin_dir"], "pg_isready"))
+        with patch('patroni.validator.open', mock_open(read_data='12')):
+            schema(config)
+        output = mock_out.getvalue()
+        self.assertEqual([], parse_output(output))
+
+    @patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 12.1"))
+    def test_pg_version_missmatch(self, mock_out, mock_err):
+        directories.append(config["postgresql"]["data_dir"])
+        directories.append(config["postgresql"]["bin_dir"])
+        directories.append(os.path.join(config["postgresql"]["data_dir"], "pg_wal"))
+        files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
+        files.append(os.path.join(config["postgresql"]["data_dir"], "PG_VERSION"))
+        c = copy.deepcopy(config)
+        c["etcd"]["hosts"] = []
+        del c["postgresql"]["bin_dir"]
+        with patch('patroni.validator.open', mock_open(read_data='11')):
+            schema(c)
+        output = mock_out.getvalue()
+        self.assertEqual(['etcd.hosts', 'postgresql.data_dir'], parse_output(output))
+
+    @patch('subprocess.check_output', Mock(return_value=b"postgres (PostgreSQL) 12.1"))
+    def test_pg_wal_doesnt_exist(self, mock_out, mock_err):
+        directories.append(config["postgresql"]["data_dir"])
+        directories.append(config["postgresql"]["bin_dir"])
+        files.append(os.path.join(config["postgresql"]["data_dir"], "global", "pg_control"))
+        files.append(os.path.join(config["postgresql"]["data_dir"], "PG_VERSION"))
+        c = copy.deepcopy(config)
+        del c["postgresql"]["bin_dir"]
+        with patch('patroni.validator.open', mock_open(read_data='11')):
+            schema(c)
+        output = mock_out.getvalue()
+        self.assertEqual(['postgresql.data_dir'], parse_output(output))
+
+
+    def test_data_dir_is_empty_string(self, mock_out, mock_err):
+        directories.append(config["postgresql"]["data_dir"])
+        directories.append(config["postgresql"]["bin_dir"])
+        c = copy.deepcopy(config)
+        c["kubernetes"] = False
+        c["postgresql"]["pg_hba"] = ""
+        c["postgresql"]["data_dir"] = ""
+        c["postgresql"]["bin_dir"] = ""
+        schema(c)
+        output = mock_out.getvalue()
+        self.assertEqual(['kubernetes', 'postgresql.bin_dir', 'postgresql.data_dir', 'postgresql.pg_hba'], parse_output(output))