Add watchdog support on Linux (#343)

Ensures that system gets rebooted before TTL runs out.

Initial version. Open questions:

    Do we want to disable watchdog while we are not master?

docs/watchdog.rst

@@ -0,0 +1,34 @@
+================
+Watchdog support
+================
+
+Having multiple PostgreSQL servers running as master can result in transactions lost due to diverging timelines. This situation is also called a split-brain problem. To avoid split-brain Patroni needs to ensure PostgreSQL will not accept any transaction commits after leader key expires in the DCS. Under normal circumstances Patroni will try to achieve this by stopping PostgreSQL when leader lock update fails for any reason. However, this may fail to happen due to various reasons:
+
+- Patroni has crashed due to a bug, out-of-memory condition or by being accidentally killed by a system administrator.
+
+- Shutting down PostgreSQL is too slow.
+
+- Patroni does not get to run due to high load on the system, th VM being paused by the hypervisor, or other infrastructure issues.
+
+To guarantee correct behavior under these conditions Patroni supports watchdog devices. Watchdog devices are software or hardware mechanisms that will reset the whole system when they do not get a keepalive heartbeat within a specified timeframe.
+
+To be safe under all circumstances Patroni will set up the watchdog to expire after half of TTL. The watchdog will reset every time the high availability loop runs. This means that `ttl` must be at least twice `loop_wait` plus some safety margin. Default setup of `loop_wait=10` and `ttl=30` gives HA loop 5 seconds (ttl / 2 - loop_wait) to complete before the system gets forcefully reset. This is rather aggressive and you probably should increase `ttl` and/or reduce `loop_wait` if you decide to use a watchdog.
+
+Currently watchdogs are only supported using Linux watchdog device interface.
+
+Setting up software watchdog on Linux
+-------------------------------------
+
+Default Patroni configuration will try to use `/dev/watchdog` on Linux if it's accessible to Patroni. For most use cases using software watchdog built into the Linux kernel is secure enough.
+
+To enable software watchdog issue the following commands as root before starting Patroni:
+
+.. code-block:: bash
+
+    modprobe softdog
+    # Replace postgres with the user you will be running patroni under
+    chown postgres /dev/watchdog
+
+For testing it may be helpful to disable rebooting by adding `soft_noboot=1` to the modprobe command line. In this case the watchdog will just log a line in kernel ring buffer, visible via `dmesg`.
+
+Patroni will log information about the watchdog when it's successfully enabled.

features/environment.py

@@ -1,14 +1,18 @@
 import abc
 import consul
+import datetime
 import etcd
 import kazoo.client
 import kazoo.exceptions
 import os
+import psutil
 import psycopg2
 import shutil
+import signal
 import six
 import subprocess
 import tempfile
+import threading
 import time
 import yaml
 
@@ -74,18 +78,28 @@ class AbstractController(object):
         if self._log:
             self._log.close()
 
+    def cancel_background(self):
+        pass
 
 class PatroniController(AbstractController):
     __PORT = 5440
     PATRONI_CONFIG = '{}.yml'
     """ starts and stops individual patronis"""
 
-    def __init__(self, context, name, work_directory, output_dir, tags=None):
+    def __init__(self, context, name, work_directory, output_dir, tags=None, with_watchdog=False):
         super(PatroniController, self).__init__(context, 'patroni_' + name, work_directory, output_dir)
         PatroniController.__PORT += 1
         self._data_dir = os.path.join(work_directory, 'data', name)
         self._connstring = None
-        self._config = self._make_patroni_test_config(name, tags)
+        if with_watchdog:
+            self.watchdog = WatchdogMonitor(name, work_directory, output_dir)
+            custom_config = {'watchdog': {'driver': 'testing', 'device': self.watchdog.fifo_path, 'mode': 'required'}}
+        else:
+            self.watchdog = None
+            custom_config = None
+
+        self._config = self._make_patroni_test_config(name, tags, custom_config)
+        self._closables = []
 
         self._conn = None
         self._curs = None
@@ -109,6 +123,8 @@ class PatroniController(AbstractController):
                 yaml.safe_dump(config, w, default_flow_style=False)
 
     def _start(self):
+        if self.watchdog:
+            self.watchdog.start()
         return subprocess.Popen(['coverage', 'run', '--source=patroni', '-p', 'patroni.py', self._config],
                                 stdout=self._log, stderr=subprocess.STDOUT, cwd=self._work_directory)
 
@@ -116,6 +132,8 @@ class PatroniController(AbstractController):
         if postgres:
             return subprocess.call(['pg_ctl', '-D', self._data_dir, 'stop', '-mi', '-w'])
         super(PatroniController, self).stop(kill, timeout)
+        if self.watchdog:
+            self.watchdog.stop()
 
     def _is_accessible(self):
         cursor = self.query("SELECT 1", fail_ok=True)
@@ -123,7 +141,7 @@ class PatroniController(AbstractController):
             cursor.execute("SET synchronous_commit TO 'local'")
             return True
 
-    def _make_patroni_test_config(self, name, tags):
+    def _make_patroni_test_config(self, name, tags, custom_config):
         patroni_config_name = self.PATRONI_CONFIG.format(name)
         patroni_config_path = os.path.join(self._output_dir, patroni_config_name)
 
@@ -151,6 +169,15 @@ class PatroniController(AbstractController):
         if tags:
             config['tags'] = tags
 
+        if custom_config is not None:
+            def recursive_update(dst, src):
+                for k, v in src.items():
+                    if k in dst and isinstance(dst[k], dict):
+                        recursive_update(dst[k], v)
+                    else:
+                        dst[k] = v
+            recursive_update(config, custom_config)
+
         with open(patroni_config_path, 'w') as f:
             yaml.safe_dump(config, f, default_flow_style=False)
 
@@ -188,6 +215,87 @@ class PatroniController(AbstractController):
             time.sleep(1)
         return False
 
+    def get_watchdog(self):
+        return self.watchdog
+
+    def _get_pid(self):
+        try:
+            pidfile = os.path.join(self._data_dir, 'postmaster.pid')
+            if not os.path.exists(pidfile):
+                return None
+            return int(open(pidfile).readline().strip())
+        except:
+            return None
+
+    def database_is_running(self):
+        pid = self._get_pid()
+        if not pid:
+            return False
+        try:
+            os.kill(pid, 0)
+        except OSError:
+            return False
+        return True
+
+    def postmaster_hang(self, timeout):
+        hang = ProcessHang(self._get_pid(), timeout)
+        self._closables.append(hang)
+        hang.start()
+
+    def checkpoint_hang(self, timeout):
+        pid = self._get_pid()
+        if not pid:
+            return False
+        proc = psutil.Process(pid)
+        for child in proc.children():
+            if 'checkpoint' in child.cmdline()[0]:
+                checkpointer = child
+                break
+        else:
+            return False
+        hang = ProcessHang(checkpointer.pid, timeout)
+        self._closables.append(hang)
+        hang.start()
+        return True
+
+    def cancel_background(self):
+        for obj in self._closables:
+            obj.close()
+        self._closables = []
+
+    def terminate_backends(self):
+        pid = self._get_pid()
+        if not pid:
+            return False
+        proc = psutil.Process(pid)
+        for p in proc.children():
+            if 'process' not in p.cmdline()[0]:
+                p.terminate()
+
+class ProcessHang(object):
+
+    """A background thread implementing a cancelable process hang via SIGSTOP."""
+
+    def __init__(self, pid, timeout):
+        self._cancelled = threading.Event()
+        self._thread = threading.Thread(target=self.run)
+        self.pid = pid
+        self.timeout = timeout
+
+    def start(self):
+        self._thread.start()
+
+    def run(self):
+        os.kill(self.pid, signal.SIGSTOP)
+        try:
+            self._cancelled.wait(self.timeout)
+        finally:
+            os.kill(self.pid, signal.SIGCONT)
+
+    def close(self):
+        self._cancelled.set()
+        self._thread.join()
+
 
 class AbstractDcsController(AbstractController):
 
@@ -386,13 +494,15 @@ class PatroniPoolController(object):
     def output_dir(self):
         return self._output_dir
 
-    def start(self, name, max_wait_limit=20, tags=None):
+    def start(self, name, max_wait_limit=20, tags=None, with_watchdog=False):
         if name not in self._processes:
-            self._processes[name] = PatroniController(self._context, name, self.patroni_path, self._output_dir, tags)
+            self._processes[name] = PatroniController(self._context, name, self.patroni_path, self._output_dir, tags, with_watchdog=with_watchdog)
         self._processes[name].start(max_wait_limit)
 
     def __getattr__(self, func):
-        if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', 'add_tag_to_config']:
+        if func not in ['stop', 'query', 'write_label', 'read_label', 'check_role_has_changed_to', 'add_tag_to_config',
+                        'get_watchdog', 'database_is_running', 'checkpoint_hang', 'postmaster_hang',
+                        'terminate_backends']:
             raise AttributeError("PatroniPoolController instance has no attribute '{0}'".format(func))
 
         def wrapper(name, *args, **kwargs):
@@ -401,6 +511,7 @@ class PatroniPoolController(object):
 
     def stop_all(self):
         for ctl in self._processes.values():
+            ctl.cancel_background()
             ctl.stop()
         self._processes.clear()
 
@@ -419,6 +530,123 @@ class PatroniPoolController(object):
         return self._dcs
 
 
+class WatchdogMonitor(object):
+    """Testing harness for emulating a watchdog device as a named pipe. Because we can't easily emulate ioctl's we
+    require a custom driver on Patroni side. The device takes no action, only notes if it was pinged and/or triggered.
+    """
+    def __init__(self, name, work_directory, output_dir):
+        self.fifo_path = os.path.join(work_directory, 'data', 'watchdog.{0}.fifo'.format(name))
+        self.fifo_file = None
+        self._stop_requested = False # Relying on bool setting being atomic
+        self._thread = None
+        self.last_ping = None
+        self.was_pinged = False
+        self.was_closed = False
+        self._was_triggered = False
+        self.timeout = 60
+        self._log_file = open(os.path.join(output_dir, 'watchdog.{0}.log'.format(name)), 'w')
+        self._log("watchdog {0} initialized".format(name))
+
+    def _log(self, msg):
+        tstamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S,%f")
+        self._log_file.write("{0}: {1}\n".format(tstamp, msg))
+
+    def start(self):
+        assert self._thread is None
+        self._stop_requested = False
+        self._log("starting fifo {0}".format(self.fifo_path))
+        fifo_dir = os.path.dirname(self.fifo_path)
+        if os.path.exists(self.fifo_path):
+            os.unlink(self.fifo_path)
+        elif not os.path.exists(fifo_dir):
+            os.mkdir(fifo_dir)
+        os.mkfifo(self.fifo_path)
+        self.last_ping = time.time()
+
+        self._thread = threading.Thread(target=self.run)
+        self._thread.start()
+
+    def run(self):
+        try:
+            while not self._stop_requested:
+                self._log("opening")
+                self.fifo_file = os.open(self.fifo_path, os.O_RDONLY)
+                try:
+                    self._log("Fifo {0} connected".format(self.fifo_path))
+                    self.was_closed = False
+                    while not self._stop_requested:
+                        c = os.read(self.fifo_file, 1)
+
+                        if c == b'X':
+                            self._log("Stop requested")
+                            return
+                        elif c == b'':
+                            self._log("Pipe closed")
+                            break
+                        elif c == b'C':
+                            command = b''
+                            c = os.read(self.fifo_file, 1)
+                            while c != b'\n' and c != b'':
+                                command += c
+                                c = os.read(self.fifo_file, 1)
+                            command = command.decode('utf8')
+
+                            if command.startswith('timeout='):
+                                self.timeout = int(command.split('=')[1])
+                                self._log("timeout={0}".format(self.timeout))
+                        elif c in [b'V', b'1']:
+                            cur_time = time.time()
+                            if cur_time - self.last_ping > self.timeout:
+                                self._log("Triggered")
+                                self._was_triggered = True
+                            if c == b'V':
+                                self._log("magic close")
+                                self.was_closed = True
+                            elif c == b'1':
+                                self.was_pinged = True
+                                self._log("ping after {0} seconds".format(cur_time - (self.last_ping or cur_time)))
+                                self.last_ping = cur_time
+                        else:
+                            self._log('Unknown command {0} received from fifo'.format(c))
+                finally:
+                    self.was_closed = True
+                    self._log("closing")
+                    os.close(self.fifo_file)
+        except Exception as e:
+            self._log("Error {0}".format(e))
+        finally:
+            self._log("stopping")
+            self._log_file.flush()
+            if os.path.exists(self.fifo_path):
+                os.unlink(self.fifo_path)
+
+    def stop(self):
+        self._log("Monitor stop")
+        self._stop_requested = True
+        try:
+            if os.path.exists(self.fifo_path):
+                fd = os.open(self.fifo_path, os.O_WRONLY)
+                os.write(fd, b'X')
+                os.close(fd)
+        except Exception as e:
+            self._log("err while closing: {0}".format(str(e)))
+        if self._thread:
+            self._thread.join()
+            self._thread = None
+
+
+    def reset(self):
+        self._log("reset")
+        self.was_pinged = self.was_closed = self._was_triggered = False
+
+    @property
+    def was_triggered(self):
+        delta = time.time() - self.last_ping
+        triggered = self._was_triggered or not self.was_closed and delta > self.timeout
+        self._log("triggered={0}, {1}s left".format(triggered, self.timeout - delta))
+        return triggered
+
+
 # actions to execute on start/stop of the tests and before running invidual features
 def before_all(context):
     context.ci = 'TRAVIS_BUILD_NUMBER' in os.environ or 'BUILD_NUMBER' in os.environ

features/steps/basic_replication.py

@@ -11,7 +11,7 @@ def start_patroni(context, name):
 
 @step('I shut down {name:w}')
 def stop_patroni(context, name):
-    return context.pctl.stop(name)
+    return context.pctl.stop(name, timeout=60)
 
 
 @step('I kill {name:w}')

features/steps/patroni_api.py

@@ -111,7 +111,7 @@ def check_response(context, component, data):
         assert context.status_code == int(data),\
             "status code {0} != {1}, response: {2}".format(context.status_code, data, context.response)
     elif component == 'returncode':
-        assert context.status_code == int(data), "return code {0} != {1}".format(context.status_code, data)
+        assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code, data, context.response)
     elif component == 'text':
         assert context.response == data.strip('"'), "response {0} does not contain {1}".format(context.response, data)
     elif component == 'output':

features/steps/watchdog.py

@@ -0,0 +1,74 @@
+from behave import step, then
+import time
+
+def polling_loop(timeout, interval=1):
+    """Returns an iterator that returns values until timeout has passed. Timeout is measured from start of iteration."""
+    start_time = time.time()
+    iteration = 0
+    end_time = start_time + timeout
+    while time.time() < end_time:
+        yield iteration
+        iteration += 1
+        time.sleep(interval)
+
+
+
+@step('I start {name:w} with watchdog')
+def start_patroni_with_watchdog(context, name):
+    return context.pctl.start(name, with_watchdog=True)
+
+
+@step('{name:w} watchdog has been pinged after {timeout:d} seconds')
+def watchdog_was_pinged(context, name, timeout):
+    for _ in polling_loop(timeout):
+        if context.pctl.get_watchdog(name).was_pinged:
+            return True
+    return False
+
+
+@then('{name:w} watchdog has been closed')
+def watchdog_was_closed(context, name):
+    assert context.pctl.get_watchdog(name).was_closed
+
+
+@step('I wait for next {name:w} watchdog ping')
+def watchdog_reset_pinged(context, name):
+    context.pctl.get_watchdog(name).reset()
+
+
+@then('{name:w} watchdog is triggered after {timeout:d} seconds')
+def watchdog_was_triggered(context, name, timeout):
+    for _ in polling_loop(timeout):
+        if context.pctl.get_watchdog(name).was_triggered:
+            return True
+    assert False
+
+
+@then('{name:w} watchdog was not triggered')
+def watchdog_was_not_triggered(context, name):
+    assert not context.pctl.get_watchdog(name).was_triggered
+
+
+@step('{name:w} checkpoint takes {timeout:d} seconds')
+def checkpoint_hang(context, name, timeout):
+    assert context.pctl.checkpoint_hang(name, timeout)
+
+
+@step('{name:w} hangs for {timeout:d} seconds')
+def postmaster_hang(context, name, timeout):
+    return context.pctl.postmaster_hang(name, timeout)
+
+
+@step('I terminate {name:w} user processes')
+def terminate_backends(context, name):
+    return context.pctl.terminate_backends(name)
+
+
+@step('Sleep for {timeout:d} seconds')
+def dcs_connection_lost(context, timeout):
+    time.sleep(timeout)
+
+
+@then('{name:w} database is running')
+def database_is_running(context, name):
+    assert context.pctl.database_is_running(name)

features/watchdog.feature

@@ -0,0 +1,43 @@
+Feature: watchdog
+  Verify that watchdog gets pinged and triggered under appropriate circumstances.
+
+  Scenario: watchdog is opened, pinged and closed
+    Given I start postgres0 with watchdog
+    Then postgres0 is a leader after 10 seconds
+    And postgres0 role is the primary after 10 seconds
+    And postgres0 watchdog has been pinged after 10 seconds
+    When I shut down postgres0
+    Then postgres0 watchdog has been closed
+
+  Scenario: watchdog is updated during pause
+    Given I start postgres0 with watchdog
+    Then postgres0 role is the primary after 10 seconds
+    When I run patronictl.py pause batman
+    And I wait for next postgres0 watchdog ping
+    Then I receive a response returncode 0
+    And postgres0 watchdog has been pinged after 10 seconds
+    When I shut down postgres0
+    Then postgres0 watchdog has been closed
+    And postgres0 database is running
+
+  Scenario: watchdog is updated during shutdown checkpoint
+    Given I start postgres0 with watchdog
+    Then postgres0 role is the primary after 10 seconds
+    And Sleep for 10 seconds
+    Given I run patronictl.py resume batman
+    Then I receive a response returncode 0
+    When I start postgres1
+    Then postgres1 role is the secondary after 10 seconds
+    When postgres0 checkpoint takes 30 seconds
+    And I shut down postgres0
+    Then postgres0 watchdog was not triggered
+    And postgres1 role is the primary after 10 seconds
+
+  Scenario: watchdog is triggered if postgres stops responding
+    Given I start postgres0 with watchdog
+    Then postgres0 role is the secondary after 10 seconds
+    When I shut down postgres1
+    Then postgres0 role is the primary after 10 seconds
+    When postgres0 hangs for 30 seconds
+    And I terminate postgres0 user processes
+    Then postgres0 watchdog is triggered after 30 seconds

patroni/__init__.py

@@ -16,6 +16,7 @@ class Patroni(object):
         from patroni.ha import Ha
         from patroni.postgresql import Postgresql
         from patroni.version import __version__
+        from patroni.watchdog import Watchdog
 
         self.setup_signal_handlers()
 
@@ -26,6 +27,7 @@ class Patroni(object):
 
         self.postgresql = Postgresql(self.config['postgresql'])
         self.api = RestApiServer(self, self.config['restapi'])
+        self.watchdog = Watchdog(self.config)
         self.ha = Ha(self)
 
         self.tags = self.get_tags()
@@ -98,6 +100,7 @@ class Patroni(object):
             self.next_run = time.time()
 
     def run(self):
+        self.ha.start()
         self.api.start()
         self.next_run = time.time()
 
@@ -124,6 +127,10 @@ class Patroni(object):
         signal.signal(signal.SIGHUP, self.sighup_handler)
         signal.signal(signal.SIGTERM, self.sigterm_handler)
 
+    def shutdown(self):
+        self.api.shutdown()
+        self.ha.shutdown()
+
 
 def patroni_main():
     logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
@@ -135,12 +142,7 @@ def patroni_main():
     except KeyboardInterrupt:
         pass
     finally:
-        patroni.api.shutdown()
+        patroni.shutdown()
-        if patroni.ha.is_paused():
-            logger.info('Leader key is not deleted and Postgresql is not stopped due paused state')
-        else:
-            patroni.ha.while_not_sync_standby(lambda: patroni.postgresql.stop(checkpoint=False))
-            patroni.dcs.delete_leader()
 
 
 def pg_ctl_start(args):

patroni/async_executor.py

@@ -1,9 +1,55 @@
 import logging
-from threading import RLock, Thread
+from threading import Lock, RLock, Thread
 
 logger = logging.getLogger(__name__)
 
 
+class CriticalTask(object):
+    """Represents a critical task in a background process that we either need to cancel or get the result of.
+
+    Fields of this object may be accessed only when holding a lock on it. To perform the critical task the background
+    thread must, while holding lock on this object, check `is_cancelled` flag, run the task and mark the task as
+    complete using `complete()`.
+
+    The main thread must hold async lock to prevent the task from completing, hold lock on critical task object,
+    call cancel. If the task has completed `cancel()` will return False and `result` field will contain the result of
+    the task. When cancel returns True it is guaranteed that the background task will notice the `is_cancelled` flag.
+    """
+    def __init__(self):
+        self._lock = Lock()
+        self.is_cancelled = False
+        self.result = None
+
+    def reset(self):
+        """Must be called every time the background task is finished.
+
+        Must be called from async thread. Caller must hold lock on async executor when calling."""
+        self.is_cancelled = False
+        self.result = None
+
+    def cancel(self):
+        """Tries to cancel the task, returns True if the task has already run.
+
+        Caller must hold lock on async executor and the task when calling."""
+        if self.result is not None:
+            return False
+        self.is_cancelled = True
+        return True
+
+    def complete(self, result):
+        """Mark task as completed along with a result.
+
+        Must be called from async thread. Caller must hold lock on task when calling."""
+        self.result = result
+
+    def __enter__(self):
+        self._lock.acquire()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self._lock.release()
+
+
 class AsyncExecutor(object):
 
     def __init__(self, ha_wakeup):
@@ -11,6 +57,7 @@ class AsyncExecutor(object):
         self._thread_lock = RLock()
         self._scheduled_action = None
         self._scheduled_action_lock = RLock()
+        self.critical_task = CriticalTask()
 
     @property
     def busy(self):
@@ -43,6 +90,8 @@ class AsyncExecutor(object):
         finally:
             with self:
                 self.reset_scheduled_action()
+                with self.critical_task:
+                    self.critical_task.reset()
             if wakeup is not None:
                 self._ha_wakeup()
 

patroni/config.py

@@ -48,6 +48,9 @@ class Config(object):
             'bin_dir': '',
             'use_slots': True,
             'parameters': {p: v[0] for p, v in Postgresql.CMDLINE_OPTIONS.items()}
+        },
+        'watchdog': {
+            'mode': 'automatic',
         }
     }
 
@@ -272,7 +275,7 @@ class Config(object):
                         config['postgresql'][name].update(self._process_postgresql_parameters(value, True))
                     elif name != 'use_slots':  # replication slots must be enabled/disabled globally
                         config['postgresql'][name] = deepcopy(value)
-            elif name not in config:
+            elif name not in config or name in ['watchdog']:
                 config[name] = deepcopy(value) if value else {}
 
         # restapi server expects to get restapi.auth = 'username:password'

patroni/exceptions.py

@@ -23,3 +23,7 @@ class DCSError(PatroniException):
 
 class PostgresConnectionException(PostgresException):
     pass
+
+
+class WatchdogError(PatroniException):
+    pass

patroni/ha.py

@@ -12,8 +12,8 @@ from multiprocessing.pool import ThreadPool
 from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
 from patroni.postgresql import ACTION_ON_START
-from patroni.utils import polling_loop, tzutc
+from patroni.utils import polling_loop, null_context, tzutc
-from threading import RLock
+from threading import RLock, Event, Thread
 
 logger = logging.getLogger(__name__)
 
@@ -46,6 +46,53 @@ class _MemberStatus(namedtuple('_MemberStatus', 'member,reachable,in_recovery,wa
         return None
 
 
+class BackgroundKeepaliveSender(object):
+    """A context manager that sends keepalives every loop_wait seconds in a background thread while the context is
+    running, but only after a safepoint has been reached. After the safepoint PostgreSQL must not be allowed to
+    transition to master before the context has ended. Intended use is for long operations that run in main HA loop.
+
+    If safe event is given it must be triggered when no client can be accessing PostgreSQL as master. If this condition
+    is already guaranteed before entering the context the safe event can be omitted.
+    """
+    def __init__(self, ha, safe_event=None):
+        """
+        :param safe_event: None or threading.Event that is cleared when context is entered.
+        """
+        self.ha = ha
+        self.safe_event = safe_event
+        self._stop_event = Event()
+        self._bg_thread = Thread(target=self.run)
+        self.loop_wait = ha.dcs.loop_wait
+
+    def __enter__(self):
+        if self.safe_event is not None:
+            self.safe_event.clear()
+        self._bg_thread.start()
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        # FIXME: Do we want to handle the case where the safe event was not set?
+        # e.g. stop failed with an exception, looks like witholding keepalives is ok then
+        # We do want to avoid it when we don't have keepalives enabled, but maybe we can
+        # avoid creating the thread in the first place.
+        # if not self.safe_event.is_set():
+        #     self.safe_event.set()????
+        self._stop_event.set()
+        self._bg_thread.join()
+        # Always send at least one keepalive
+        self.ha.keepalive()
+
+    def run(self):
+        if self.safe_event is not None:
+            self.safe_event.wait()
+            logger.debug("Background keepalive safe event reached")
+        while not self._stop_event.is_set():
+            logger.debug("Sending background keepalive")
+            self.ha.keepalive()
+            if not self._stop_event.wait(self.loop_wait):
+                self.ha.keepalive_sent = False
+        logger.debug("Stopping background keepalive")
+
+
 class Ha(object):
 
     def __init__(self, patroni):
@@ -57,6 +104,7 @@ class Ha(object):
         self.recovering = False
         self._start_timeout = None
         self._async_executor = AsyncExecutor(self.wakeup)
+        self.watchdog = patroni.watchdog
 
         # Each member publishes various pieces of information to the DCS using touch_member. This lock protects
         # the state and publishing procedure to have consistent ordering and avoid publishing stale values.
@@ -64,6 +112,10 @@ class Ha(object):
         # Count of concurrent sync disabling requests. Value above zero means that we don't want to be synchronous
         # standby. Changes protected by _member_state_lock.
         self._disable_sync = 0
+        # We need to send keepalives at most once per lock update so it is guaranteed that keepalive expires before
+        # lock TTL runs out. However we want to do it as soon as we determine that it is safe to do so. This flag
+        # keeps track whether a keepalive has been sent in the current cycle.
+        self.keepalive_sent = False
 
     def is_paused(self):
         return self.cluster and self.cluster.is_paused()
@@ -77,15 +129,20 @@ class Ha(object):
         self.cluster = cluster
 
     def acquire_lock(self):
-        return self.dcs.attempt_to_acquire_leader()
+        ret = self.dcs.attempt_to_acquire_leader()
+        if ret:
+            self.keepalive()
+        return ret
 
     def update_lock(self, write_leader_optime=False):
         ret = self.dcs.update_leader()
-        if ret and write_leader_optime:
+        if ret:
-            try:
+            self.keepalive()
-                self.dcs.write_leader_optime(self.state_handler.last_operation())
+            if write_leader_optime:
-            except:
+                try:
-                pass
+                    self.dcs.write_leader_optime(self.state_handler.last_operation())
+                except:
+                    pass
         return ret
 
     def has_lock(self):
@@ -147,20 +204,22 @@ class Ha(object):
         # no initialize key and node is allowed to be master and has 'bootstrap' section in a configuration file
         elif self.cluster.initialize is None and not self.patroni.nofailover and 'bootstrap' in self.patroni.config:
             if self.dcs.initialize(create_new=True):  # race for initialization
-                try:
+                with self._background_keepalive_context(wait_for_safepoint=False):
-                    self.state_handler.bootstrap(self.patroni.config['bootstrap'])
+                    try:
-                    self.dcs.initialize(create_new=False, sysid=self.state_handler.sysid)
+                        self.state_handler.bootstrap(self.patroni.config['bootstrap'])
-                except:  # initdb or start failed
+                        self.dcs.initialize(create_new=False, sysid=self.state_handler.sysid)
-                    # remove initialization key and give a chance to other members
+                    except:  # initdb or start failed
-                    logger.info("removing initialize key after failed attempt to initialize the cluster")
+                        # remove initialization key and give a chance to other members
-                    self.dcs.cancel_initialization()
+                        logger.info("removing initialize key after failed attempt to initialize the cluster")
-                    self.state_handler.stop('immediate')
+                        self.dcs.cancel_initialization()
-                    self.state_handler.move_data_directory()
+                        self.state_handler.stop('immediate')
-                    raise
+                        self.state_handler.move_data_directory()
-                self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
+                        raise
-                self.dcs.take_leader()
+                    self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration,
-                self.load_cluster_from_dcs()
+                                                         separators=(',', ':')))
-                return 'initialized a new cluster'
+                    self.dcs.take_leader()
+                    self.load_cluster_from_dcs()
+                    return 'initialized a new cluster'
             else:
                 return 'failed to acquire initialize lock'
         else:
@@ -204,6 +263,7 @@ class Ha(object):
             node_to_follow = self._get_node_to_follow(self.cluster)
 
         self.recovering = True
+
         self._async_executor.schedule('restarting after failure')
         self._async_executor.run_async(self.state_handler.follow, (node_to_follow, timeout))
         return msg
@@ -227,6 +287,7 @@ class Ha(object):
         node_to_follow = self._get_node_to_follow(self.cluster)
 
         if self.is_paused():
+            self.keepalive()
             if not (self.state_handler.need_rewind and self.state_handler.can_rewind) or self.cluster.is_unlocked():
                 self.state_handler.set_role('master' if is_leader else 'replica')
                 if is_leader:
@@ -234,8 +295,10 @@ class Ha(object):
                 elif not node_to_follow:
                     return 'no action'
         elif is_leader:
-            self.demote('immediate')
+            self.demote('immediate-nolock')
             return demote_reason
+        else:
+            self.keepalive()
 
         if self._handle_rewind():
             return self._async_executor.scheduled_action
@@ -519,33 +582,47 @@ class Ha(object):
             graceful is used when failing over to another node due to user request. May only be called running async.
             immediate is used when we determine that we are not suitable for master and want to failover quickly
                 without regard for data durability. May only be called synchronously.
+            immediate-nolock is used when find out that we have lost the lock to be master. Need to bring down
+                PostgreSQL as quickly as possible without regard for data durability. May only be called synchronously.
         """
-        assert mode in ['offline', 'graceful', 'immediate']
+        mode_control = {
-        self.state_handler.trigger_check_diverged_lsn()
+            'offline':          dict(stop='fast', checkpoint=False, release=False, offline=True, async=False),
-        if mode != 'offline':
+            'graceful':         dict(stop='fast', checkpoint=True, release=True, offline=False, async=False),
-            if mode == 'immediate':
+            'immediate':        dict(stop='immediate', checkpoint=False, release=True, offline=False, async=True),
-                self.state_handler.stop('immediate', checkpoint=False)
+            'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async=True),
-            else:
+        }[mode]
-                self.state_handler.stop()
+
+        with self._background_keepalive_context() if mode != 'graceful' else null_context():
+            self.state_handler.trigger_check_diverged_lsn()
+            self.state_handler.stop(mode_control['stop'], checkpoint=mode_control['checkpoint'])
             self.state_handler.set_role('demoted')
-            self.release_leader_key_voluntarily()
+
-            time.sleep(2)  # Give a time to somebody to take the leader lock
+            if mode_control['release']:
-            cluster = self.dcs.get_cluster()
+                    self.release_leader_key_voluntarily()
-            node_to_follow = self._get_node_to_follow(cluster)
+                    time.sleep(2)  # Give a time to somebody to take the leader lock
-            if mode == 'immediate':
+            if mode_control['offline']:
-                # We will try to start up as a standby now. If no one takes the leader lock before we finish
+                node_to_follow, leader = None, None
-                # recovery we will try to promote ourselves.
+            else:
+                cluster = self.dcs.get_cluster()
+                node_to_follow, leader = self._get_node_to_follow(cluster), cluster.leader
+
+            # FIXME: with mode offline called from DCS exception handler and handle_long_action_in_progress
+            # there could be an async action already running, calling follow from here will lead
+            # to racy state handler state updates.
+            if mode_control['async']:
                 self._async_executor.schedule('starting after demotion')
                 self._async_executor.run_async(self.state_handler.follow, (node_to_follow,))
             else:
-                if self.state_handler.rewind_needed_and_possible(cluster.leader):
+                if self.state_handler.rewind_needed_and_possible(leader):
                     return False  # do not start postgres, but run pg_rewind on the next iteration
-                return self.state_handler.follow(node_to_follow)
+                self.state_handler.follow(node_to_follow)
+
+    def _background_keepalive_context(self, wait_for_safepoint=True):
+        if self.watchdog.is_running:
+            safe_event = self.state_handler.stop_safepoint_reached if wait_for_safepoint else None
+            return BackgroundKeepaliveSender(self, safe_event)
         else:
-            # Need to become unavailable as soon as possible, so initiate a stop here. However as we can't release
+            return null_context()
-            # the leader key we don't care about confirming the shutdown quickly and can use a regular stop.
-            self.state_handler.stop(checkpoint=False)
-            self.state_handler.follow(None)
 
     def should_run_scheduled_action(self, action_name, scheduled_at, cleanup_fn):
         if scheduled_at and not self.is_paused():
@@ -640,8 +717,7 @@ class Ha(object):
         else:
             # when we are doing manual failover there is no guaranty that new leader is ahead of any other node
             # node tagged as nofailover can be ahead of the new leader either, but it is always excluded from elections
-            check_diverged_lsn = bool(self.cluster.failover) or self.patroni.nofailover
+            if bool(self.cluster.failover) or self.patroni.nofailover:
-            if check_diverged_lsn:
                 self.state_handler.trigger_check_diverged_lsn()
                 time.sleep(2)  # Give a time to somebody to take the leader lock
 
@@ -654,6 +730,8 @@ class Ha(object):
     def process_healthy_cluster(self):
         if self.has_lock():
             if self.is_paused() and not self.state_handler.is_leader():
+                # Not a master
+                self.keepalive()
                 if self.cluster.failover and self.cluster.failover.candidate == self.state_handler.name:
                     return 'waiting to become master after promote...'
 
@@ -672,14 +750,14 @@ class Ha(object):
                 # Either there is no connection to DCS or someone else acquired the lock
                 logger.error('failed to update leader lock')
                 if self.state_handler.is_leader():
-                    self.demote('offline')
+                    self.demote('immediate-nolock')
                     return 'demoted self because failed to update leader lock in DCS'
                 else:
                     return 'not promoting because failed to update leader lock in DCS'
         else:
             logger.info('does not have lock')
         return self.follow('demoting self because i do not have the lock and i was a leader',
-                           'no action.  i am a secondary and i am following a leader', False)
+                           'no action.  i am a secondary and i am following a leader', refresh=False)
 
     def evaluate_scheduled_restart(self):
         if self._async_executor.busy:  # Restart already in progress
@@ -771,13 +849,13 @@ class Ha(object):
             # leader key (if it belong to us) rather than trying to start postgres once again.
             self.recovering = True
 
-        # No that restart is scheduled we can set timeout for startup, it will get reset
+        # Now that restart is scheduled we can set timeout for startup, it will get reset
         # once async executor runs and main loop notices PostgreSQL as up.
         timeout = restart_data.get('timeout', self.patroni.config['master_start_timeout'])
         self.set_start_timeout(timeout)
 
         # For non async cases we want to wait for restart to complete or timeout before returning.
-        do_restart = functools.partial(self.state_handler.restart, timeout)
+        do_restart = functools.partial(self.state_handler.restart, timeout, self._async_executor.critical_task)
         if self.is_synchronous_mode() and not self.has_lock():
             do_restart = functools.partial(self.while_not_sync_standby, do_restart)
 
@@ -818,15 +896,28 @@ class Ha(object):
         self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, ))
 
     def handle_long_action_in_progress(self):
-        if self.has_lock():
+        try:
-            if self.update_lock():
+            if self.has_lock() and self.update_lock():
                 return 'updated leader lock during ' + self._async_executor.scheduled_action
             else:
-                return 'failed to update leader lock during ' + self._async_executor.scheduled_action
+                # Don't have lock, make sure we are not starting up a master in the background
-        elif self.cluster.is_unlocked():
+                if self.state_handler.role == 'master':
-            return 'not healthy enough for leader race'
+                    logger.info("Demoting master during " + self._async_executor.scheduled_action)
-        else:
+                    if self._async_executor.scheduled_action == 'restart':
-            return self._async_executor.scheduled_action + ' in progress'
+                        # Restart needs a special interlocking cancel because postmaster may be just started in a
+                        # background thread and has not even written a pid file yet.
+                        with self._async_executor.critical_task as task:
+                            if not task.cancel():
+                                self.state_handler.terminate_starting_postmaster(pid=task.result)
+                    self.demote('immediate-nolock')
+                    return 'lost leader lock during ' + self._async_executor.scheduled_action
+        finally:
+            self.keepalive()
+
+        if self.cluster.is_unlocked():
+            logger.info('not healthy enough for leader race')
+
+        return self._async_executor.scheduled_action + ' in progress'
 
     @staticmethod
     def sysid_valid(sysid):
@@ -837,6 +928,7 @@ class Ha(object):
 
     def post_recover(self):
         if not self.state_handler.is_running():
+            self.keepalive()
             if self.has_lock():
                 self.state_handler.set_role('demoted')
                 self.dcs.delete_leader()
@@ -858,7 +950,7 @@ class Ha(object):
         if self.has_lock():
             if not self.update_lock():
                 logger.info("Lost lock while starting up. Demoting self.")
-                self.demote('immediate')
+                self.demote('immediate-nolock')
                 return 'stopped PostgreSQL while starting up because leader key was lost'
 
             timeout = self._start_timeout or self.patroni.config['master_start_timeout']
@@ -888,8 +980,14 @@ class Ha(object):
         Must be called when async_executor is busy or in the main thread."""
         self._start_timeout = value
 
+    def keepalive(self):
+        if not self.keepalive_sent:
+            self.watchdog.keepalive()
+        self.keepalive_sent = True
+
     def _run_cycle(self):
         dcs_failed = False
+        self.keepalive_sent = False
         try:
             self.load_cluster_from_dcs()
 
@@ -921,6 +1019,10 @@ class Ha(object):
 
             # is data directory empty?
             if self.state_handler.data_directory_empty():
+                # PostgreSQL is assumed to not be running if data dir is empty.
+                # TODO: detect the datadir going away (e.g. unmounted ) while PostgreSQL is running
+                self.keepalive()
+
                 # is this instance the leader?
                 if self.has_lock():
                     self.release_leader_key_voluntarily()
@@ -938,6 +1040,8 @@ class Ha(object):
                     sys.exit(1)
 
             if not self.state_handler.is_healthy():
+                # We are not running, so it's safe to send the keepalive
+                self.keepalive()
                 if self.is_paused():
                     if self.has_lock():
                         self.dcs.delete_leader()
@@ -977,12 +1081,38 @@ class Ha(object):
         finally:
             if not dcs_failed:
                 self.touch_member()
+            if not self.keepalive_sent:
+                logger.error("End of HA loop reached without sending keepalive")
 
     def run_cycle(self):
         with self._async_executor:
             info = self._run_cycle()
             return (self.is_paused() and 'PAUSE: ' or '') + info
 
+    def start(self):
+        self.watchdog.activate()
+
+    def shutdown(self):
+        if self.is_paused():
+            logger.info('Leader key is not deleted and Postgresql is not stopped due paused state')
+            self.watchdog.disable()
+        else:
+            # FIXME: If stop doesn't reach safepoint quickly enough keepalive is triggered. If shutdown checkpoint
+            # takes longer than ttl, then leader key is lost and replication might not have sent out all xlog.
+            # This might not be the desired behavior of users, as a graceful shutdown of the host can mean lost data.
+            # We probably need to something smarter here.
+            with self._background_keepalive_context(wait_for_safepoint=self.state_handler.is_leader):
+                self.while_not_sync_standby(lambda: self.state_handler.stop(checkpoint=False))
+            if not self.state_handler.is_running():
+                self.dcs.delete_leader()
+                self.watchdog.disable()
+            else:
+                # XXX: what about when Patroni is started as the wrong user that has access to the watchdog device
+                # but cannot shut down PostgreSQL. Root would be the obvious example. Would be nice to not kill the
+                # system due to a bad config.
+                logger.error("PostgreSQL shutdown failed, leader key not removed." +
+                             (" Leaving watchdog running." if self.watchdog.is_running else ""))
+
     def watch(self, timeout):
         cluster = self.cluster
         # watch on leader key changes if the postgres is running and leader is known and current node is not lock owner

patroni/postgresql.py

@@ -1,9 +1,12 @@
 import logging
+import errno
 import os
 import psycopg2
+import psutil
 import re
 import shlex
 import shutil
+import signal
 import subprocess
 import tempfile
 import time
@@ -13,9 +16,9 @@ from contextlib import contextmanager
 from patroni import call_self
 from patroni.callback_executor import CallbackExecutor
 from patroni.exceptions import PostgresConnectionException, PostgresException
-from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop
+from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop, null_context
 from six import string_types
-from threading import current_thread, Lock
+from threading import current_thread, Lock, Event
 
 logger = logging.getLogger(__name__)
 
@@ -30,6 +33,12 @@ STATE_REJECT = 'rejecting connections'
 STATE_NO_RESPONSE = 'not responding'
 STATE_UNKNOWN = 'unknown'
 
+STOP_SIGNALS = {
+    'smart': signal.SIGTERM,
+    'fast': signal.SIGINT,
+    'immediate': signal.SIGQUIT,
+}
+STOP_POLLING_INTERVAL = 1
 REWIND_STATUS = type('Enum', (), {'INITIAL': 0, 'CHECK': 1, 'NEED': 2, 'NOT_NEED': 3, 'SUCCESS': 4, 'FAILED': 5})
 
 
@@ -137,6 +146,12 @@ class Postgresql(object):
 
         self._state_entry_timestamp = None
 
+        # This event is set to true when no backends are running. Could be set in parallel by
+        # multiple processes, like when demote is racing with async restart. Needs to be cleared
+        # before invoking stop if wait for this event is desired.
+        self.stop_safepoint_reached = Event()
+        self.stop_safepoint_reached.set()
+
         if self.is_running():
             self.set_state('running')
             self.set_role('master' if self.is_leader() else 'replica')
@@ -219,14 +234,6 @@ class Postgresql(object):
         :returns: `!True` when return_code == 0, otherwise `!False`"""
 
         pg_ctl = [self._pgcommand('pg_ctl'), cmd]
-        if cmd == 'stop':
-            pg_ctl += ['-w']
-            timeout = self.config.get('pg_ctl_timeout')
-            if timeout:
-                try:
-                    pg_ctl += ['-t', str(int(timeout))]
-                except Exception:
-                    logger.error('Bad value of pg_ctl_timeout: %s', timeout)
         return subprocess.call(pg_ctl + ['-D', self._data_dir] + list(args), **kwargs) == 0
 
     def pg_isready(self):
@@ -603,8 +610,9 @@ class Postgresql(object):
 
     def is_running(self):
         if not (self._version_file_exists() and os.path.isfile(self._postmaster_pid)):
+            # XXX: This is dangerous in case somebody deletes the data directory while PostgreSQL is still running.
             return False
-        return self.is_pid_running(self.read_pid_file().get('pid', 0))
+        return self.is_pid_running(self.get_pid())
 
     def read_pid_file(self):
         """Reads and parses postmaster.pid from the data directory
@@ -618,10 +626,21 @@ class Postgresql(object):
         except IOError:
             return {}
 
+    def get_pid(self):
+        """Fetches pid value from postmaster.pid using read_pid_file
+
+        :returns pid if successful, 0 if pid file is not present"""
+        # TODO: figure out what to do on permission errors
+        pid = self.read_pid_file().get('pid', 0)
+        try:
+            return int(pid)
+        except ValueError:
+            logger.warning("Garbage pid in postmaster.pid: {0!r}".format(pid))
+            return 0
+
     @staticmethod
     def is_pid_running(pid):
         try:
-            pid = int(pid)
             if pid < 0:
                 pid = -pid
             return pid > 0 and pid != os.getpid() and pid != os.getppid() and (os.kill(pid, 0) or True)
@@ -697,7 +716,7 @@ class Postgresql(object):
         logger.warning("Timed out waiting for PostgreSQL to start")
         return False
 
-    def start(self, timeout=None, block_callbacks=False):
+    def start(self, timeout=None, block_callbacks=False, task=None):
         """Start PostgreSQL
 
         Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion
@@ -741,12 +760,21 @@ class Postgresql(object):
         # of init process to take care about postmaster.
         # In order to make everything portable we can't use fork&exec approach here, so  we will call
         # ourselves and pass list of arguments which must be used to start postgres.
-        proc = call_self(['pg_ctl_start', self._pgcommand('postgres'), '-D', self._data_dir] + options, close_fds=True,
+        with task or null_context():
-                         preexec_fn=os.setsid, stdout=subprocess.PIPE,
+            if task and task.is_cancelled:
-                         env={p: os.environ[p] for p in ('PATH', 'LC_ALL', 'LANG') if p in os.environ})
+                logger.info("PostgreSQL start cancelled.")
-        pid = int(proc.stdout.readline().strip())
+                return False
-        proc.wait()
+
-        logger.info('postmaster pid=%s', pid)
+            start_initiated = time.time()
+            proc = call_self(['pg_ctl_start', self._pgcommand('postgres'), '-D', self._data_dir] + options,
+                             close_fds=True, preexec_fn=os.setsid, stdout=subprocess.PIPE,
+                             env={p: os.environ[p] for p in ('PATH', 'LC_ALL', 'LANG') if p in os.environ})
+            pid = int(proc.stdout.readline().strip())
+            proc.wait()
+            logger.info('postmaster pid=%s', pid)
+
+            if task:
+                task.complete(pid)
 
         start_timeout = timeout
         if not start_timeout:
@@ -785,10 +813,23 @@ class Postgresql(object):
             return 'not accessible or not healty'
 
     def stop(self, mode='fast', block_callbacks=False, checkpoint=True):
-        if not self.is_running():
+        success, pg_signaled = self._do_stop(mode, block_callbacks, checkpoint)
+        if success:
+            self.stop_safepoint_reached.set()  # In case we exited early. Setting twice is not a problem.
+            # block_callbacks is used during restart to avoid
+            # running start/stop callbacks in addition to restart ones
             if not block_callbacks:
                 self.set_state('stopped')
-            return True
+                if pg_signaled:
+                    self.call_nowait(ACTION_ON_STOP)
+        else:
+            logger.warning('pg_ctl stop failed')
+            self.set_state('stop failed')
+        return success
+
+    def _do_stop(self, mode, block_callbacks, checkpoint):
+        if not self.is_running():
+            return True, False
 
         if checkpoint and not self.is_starting():
             self.checkpoint()
@@ -796,16 +837,87 @@ class Postgresql(object):
         if not block_callbacks:
             self.set_state('stopping')
 
-        ret = self.pg_ctl('stop', '-m', mode)
+        # Send signal to postmaster to stop
-        # block_callbacks is used during restart to avoid
+        pid, result = self._signal_postmaster_stop(mode)
-        # running start/stop callbacks in addition to restart ones
+        if result is not None:
-        if not ret:
+            return result, True
-            logger.warning('pg_ctl stop failed')
+
-            self.set_state('stop failed')
+        # We can skip safepoint detection if nobody is waiting for it.
-        elif not block_callbacks:
+        if not self.stop_safepoint_reached.is_set():
-            self.set_state('stopped')
+            # Wait for our connection to terminate so we can be sure that no new connections are being initiated
-            self.call_nowait(ACTION_ON_STOP)
+            self._wait_for_connection_close(pid)
-        return ret
+            self._wait_for_user_backends_to_close(pid)
+            self.stop_safepoint_reached.set()
+
+        self._wait_for_postmaster_stop(pid)
+
+        return True, True
+
+    def _wait_for_postmaster_stop(self, pid):
+        # This wait loop differs subtly from pg_ctl as we check for both the pid file going
+        # away and if the pid is running. This seems safer.
+        while pid == self.get_pid() and self.is_pid_running(pid):
+            time.sleep(STOP_POLLING_INTERVAL)
+
+    def _signal_postmaster_stop(self, mode):
+        pid = self.get_pid()
+        if pid == 0:
+            return None, True
+        elif pid < 0:
+            logger.warning("Cannot stop server; single-user server is running (PID: {0})".format(-pid))
+            return None, False
+        try:
+            os.kill(pid, STOP_SIGNALS[mode])
+        except OSError as e:
+            if e.errno == errno.ESRCH:
+                return None, True
+            else:
+                logger.warning("Could not send stop signal to PostgreSQL (error: {0})".format(e.errno))
+                return None, False
+        return pid, None
+
+    def terminate_starting_postmaster(self, pid):
+        """Terminates a postmaster that has not yet opened ports or possibly even written a pid file. Blocks
+        until the process goes away."""
+        try:
+            os.kill(pid, STOP_SIGNALS['immediate'])
+        except OSError as e:
+            if e.errno == errno.ESRCH:
+                return
+            logger.warning("Could not send stop signal to PostgreSQL (error: {0})".format(e.errno))
+
+        while self.is_pid_running(pid):
+            time.sleep(STOP_POLLING_INTERVAL)
+
+    def _wait_for_connection_close(self, pid):
+        try:
+            with self.connection().cursor() as cur:
+                while True:  # Need a timeout here?
+                    if pid == self.get_pid() and self.is_pid_running(pid):
+                        cur.execute("SELECT 1")
+                        time.sleep(STOP_POLLING_INTERVAL)
+                        continue
+                    else:
+                        break
+        except psycopg2.Error:
+            pass
+
+    @staticmethod
+    def _wait_for_user_backends_to_close(postmaster_pid):
+        # These regexps are cross checked against versions PostgreSQL 9.1 .. 9.6
+        aux_proc_re = re.compile("(?:postgres:)( .*:)? (?:""(?:startup|logger|checkpointer|writer|wal writer|"
+                                 "autovacuum launcher|autovacuum worker|stats collector|wal receiver|archiver|"
+                                 "wal sender) process|bgworker: )")
+
+        try:
+            postmaster = psutil.Process(postmaster_pid)
+            user_backends = [p for p in postmaster.children() if not aux_proc_re.match(p.cmdline()[0])]
+            logger.debug("Waiting for user backends {0} to close".format(
+                ",".join(p.cmdline()[0] for p in user_backends)))
+            psutil.wait_procs(user_backends)
+            logger.debug("Backends closed")
+        except psutil.NoSuchProcess:
+            return
 
     def reload(self):
         ret = self.pg_ctl('reload')
@@ -865,7 +977,7 @@ class Postgresql(object):
 
         return self.state == 'running'
 
-    def restart(self, timeout=None):
+    def restart(self, timeout=None, task=None):
         """Restarts PostgreSQL.
 
         When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or
@@ -875,7 +987,7 @@ class Postgresql(object):
         """
         self.set_state('restarting')
         self.__cb_pending = ACTION_ON_RESTART
-        ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True)
+        ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True, task=task)
         if not ret and not self.is_starting():
             self.set_state('restart failed ({0})'.format(self.state))
         return ret
@@ -1127,6 +1239,50 @@ class Postgresql(object):
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
         return True
 
+    def _do_rewind(self, leader):
+        logger.info("rewind flag is set")
+
+        if self.is_running() and not self.stop(checkpoint=False):
+            logger.warning('Can not run pg_rewind because postgres is still running')
+            return False
+
+        # prepare pg_rewind connection
+        r = leader.conn_kwargs(self._superuser)
+
+        # first make sure that we are really trying to rewind
+        # from the master and run a checkpoint on a t in order to
+        # make it store the new timeline ([email protected])
+        leader_status = self.checkpoint(r)
+        if leader_status:
+            logger.warning('Can not use %s for rewind: %s', leader.name, leader_status)
+            return False
+
+        # at present, pg_rewind only runs when the cluster is shut down cleanly
+        # and not shutdown in recovery. We have to remove the recovery.conf if present
+        # and start/shutdown in a single user mode to emulate this.
+        # XXX: if recovery.conf is linked, it will be written anew as a normal file.
+        if os.path.isfile(self._recovery_conf) or os.path.islink(self._recovery_conf):
+            os.unlink(self._recovery_conf)
+
+        # Archived segments might be useful to pg_rewind,
+        # clean the flags that tell we should remove them.
+        self.cleanup_archive_status()
+
+        # Start in a single user mode and stop to produce a clean shutdown
+        opts = self.read_postmaster_opts()
+        opts.update({'archive_mode': 'on', 'archive_command': 'false'})
+        self.single_user_mode(options=opts)
+
+        try:
+            if not self.rewind(r):
+                logger.error('unable to rewind the former master')
+                if self.config.get('remove_data_directory_on_rewind_failure', False):
+                    self.remove_data_directory()
+                    return False
+            return True
+        finally:
+            self._need_rewind = False
+
     def save_configuration_files(self):
         """
             copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files

patroni/utils.py

@@ -1,3 +1,4 @@
+import contextlib
 import random
 import time
 import re
@@ -280,3 +281,8 @@ def polling_loop(timeout, interval=1):
         yield iteration
         iteration += 1
         time.sleep(interval)
+
+
+@contextlib.contextmanager
+def null_context():
+    yield

patroni/watchdog/__init__.py

@@ -0,0 +1,2 @@
+from patroni.watchdog.base import WatchdogError, Watchdog
+__all__ = ['WatchdogError', 'Watchdog']

patroni/watchdog/base.py

@@ -0,0 +1,207 @@
+import abc
+import logging
+import platform
+import six
+import sys
+
+from patroni.exceptions import WatchdogError
+
+__all__ = ['WatchdogError', 'Watchdog']
+
+logger = logging.getLogger(__name__)
+
+MODE_REQUIRED = 'required'    # Will not run if a watchdog is not available
+MODE_AUTOMATIC = 'automatic'  # Will use a watchdog if one is available
+MODE_OFF = 'off'              # Will not try to use a watchdog
+
+
+def parse_mode(mode):
+    if mode is False:
+        return MODE_OFF
+    mode = mode.lower()
+    if mode in ['require', 'required']:
+        return MODE_REQUIRED
+    elif mode in ['auto', 'automatic']:
+        return MODE_AUTOMATIC
+    else:
+        if mode not in ['off', 'disable', 'disabled']:
+            logger.warning("Watchdog mode {0} not recognized, disabling watchdog".format(mode))
+        return MODE_OFF
+
+
+class Watchdog(object):
+    """Facade to dynamically manage watchdog implementations and handle config changes."""
+    def __init__(self, config):
+        self.ttl = config['ttl']
+        self.loop_wait = config['loop_wait']
+        self.mode = parse_mode(config['watchdog'].get('mode', 'automatic'))
+        self.driver = config['watchdog'].get('driver')
+        self.config = config
+
+        if self.mode == MODE_OFF:
+            self.impl = NullWatchdog()
+        else:
+            self.impl = self._get_impl()
+            if self.mode == MODE_REQUIRED and isinstance(self.impl, NullWatchdog):
+                logger.error("Configuration requires a watchdog, but watchdog is not supported on this platform.")
+                sys.exit(1)
+
+    def activate(self):
+        """Activates the watchdog device with suitable timeouts. While watchdog is active keepalive needs
+        to be called every time loop_wait expires.
+        """
+        desired_timeout = int(self.ttl // 2)
+        slack = desired_timeout - self.loop_wait
+        if slack < 0:
+            logger.warning('Watchdog not supported because leader TTL {0} is less than 2x loop_wait {1}'
+                           .format(self.ttl, self.loop_wait))
+            self.impl = NullWatchdog()
+
+        try:
+            self.impl.open()
+        except WatchdogError as e:
+            logger.warning("Could not activate %s: %s", self.impl.describe(), e)
+            self.impl = NullWatchdog()
+
+        if self.impl.is_running and not self.impl.can_be_disabled:
+            logger.warning("Watchdog implementation can't be disabled."
+                           " Watchdog will trigger after Patroni is shut down.")
+
+        if self.impl.has_set_timeout():
+            self.impl.set_timeout(desired_timeout)
+
+        # Safety checks for watchdog implementations that don't support configurable timeouts
+        actual_timeout = self.impl.get_timeout()
+        if self.impl.is_running and actual_timeout < self.loop_wait:
+            logger.error('loop_wait of {0} seconds is too long for watchdog {1} second timeout'
+                         .format(self.loop_wait, actual_timeout))
+            if self.impl.can_be_disabled:
+                logger.info('Disabling watchdog due to unsafe timeout.')
+                self.impl.close()
+                self.impl = NullWatchdog()
+
+        if not self.impl.is_running or actual_timeout > desired_timeout:
+            if self.mode == MODE_REQUIRED:
+                logger.error("Configuration requires watchdog, but a safe watchdog timeout {0} could"
+                             " not be configured. Watchdog timeout is {1}.".format(desired_timeout, actual_timeout))
+                sys.exit(1)
+            else:
+                if not isinstance(self.impl, NullWatchdog):
+                    logger.warning("Watchdog timeout {0} seconds does not ensure safe termination within {1} seconds"
+                                   .format(actual_timeout, desired_timeout))
+
+        if self.is_running:
+            logger.info("{0} activated with {1} second timeout, timing slack {2} seconds"
+                        .format(self.impl.describe(), actual_timeout, slack))
+        else:
+            if self.mode == MODE_REQUIRED:
+                logger.error("Configuration requires watchdog, but watchdog could not be activated")
+                sys.exit(1)
+
+    def disable(self):
+        try:
+            if self.impl.is_running and not self.impl.can_be_disabled:
+                # Give sysadmin some extra time to clean stuff up.
+                self.impl.keepalive()
+                logger.warning("Watchdog implementation can't be disabled. System will reboot after "
+                               "{0} seconds when watchdog times out.".format(self.impl.get_timeout()))
+            self.impl.close()
+        except WatchdogError as e:
+            logger.error("Error while disabling watchdog: %s", e)
+
+    def keepalive(self):
+        try:
+            self.impl.keepalive()
+        except WatchdogError as e:
+            logger.error("Error while sending keepalive: %s", e)
+
+    def _get_impl(self):
+        if self.mode not in [MODE_AUTOMATIC, MODE_REQUIRED]:
+            return NullWatchdog()
+
+        if self.driver == 'testing':
+            from patroni.watchdog.linux import TestingWatchdogDevice
+            return TestingWatchdogDevice.from_config(self.config['watchdog'])
+        elif platform.system() == 'Linux':
+            from patroni.watchdog.linux import LinuxWatchdogDevice
+            return LinuxWatchdogDevice.from_config(self.config['watchdog'])
+        else:
+            return NullWatchdog()
+
+    @property
+    def is_running(self):
+        return self.impl.is_running
+
+
+@six.add_metaclass(abc.ABCMeta)
+class WatchdogBase(object):
+    """A watchdog object when opened requires periodic calls to keepalive.
+    When keepalive is not called within a timeout the system will be terminated."""
+
+    @property
+    def is_running(self):
+        """Returns True when watchdog is activated and capable of performing it's task."""
+        return False
+
+    @property
+    def can_be_disabled(self):
+        """Returns True when watchdog will be disabled by calling close(). Some watchdog devices
+        will keep running no matter what once activated. May raise WatchdogError if called without
+        calling open() first."""
+        return True
+
+    @abc.abstractmethod
+    def open(self):
+        """Open watchdog device.
+
+        When watchdog is opened keepalive must be called. Returns nothing on success
+        or raises WatchdogError if the device could not be opened."""
+
+    @abc.abstractmethod
+    def close(self):
+        """Gracefully close watchdog device."""
+
+    @abc.abstractmethod
+    def keepalive(self):
+        """Resets the watchdog timer.
+
+        Watchdog must be open when keepalive is called."""
+
+    @abc.abstractmethod
+    def get_timeout(self):
+        """Returns the current keepalive timeout in effect."""
+
+    @staticmethod
+    def has_set_timeout():
+        """Returns True if setting a timeout is supported."""
+        return False
+
+    def set_timeout(self, timeout):
+        """Set the watchdog timer timeout.
+
+        :param timeout: watchdog timeout in seconds"""
+        raise WatchdogError("Setting timeout is not supported on {0}".format(self.describe()))
+
+    def describe(self):
+        """Human readable name for this device"""
+        return self.__class__.__name__
+
+    @classmethod
+    def from_config(cls, config):
+        return cls()
+
+
+class NullWatchdog(WatchdogBase):
+    """Null implementation when watchdog is not supported."""
+    def open(self):
+        return
+
+    def close(self):
+        return
+
+    def keepalive(self):
+        return
+
+    def get_timeout(self):
+        # A big enough number to not matter
+        return 1000000000

patroni/watchdog/linux.py

@@ -0,0 +1,220 @@
+import collections
+import ctypes
+import fcntl
+import os
+import platform
+from patroni.watchdog.base import WatchdogBase, WatchdogError
+
+# Pythonification of linux/ioctl.h
+IOC_NONE = 0
+IOC_WRITE = 1
+IOC_READ = 2
+
+IOC_NRBITS = 8
+IOC_TYPEBITS = 8
+IOC_SIZEBITS = 14
+IOC_DIRBITS = 2
+
+# Non-generic platform special cases
+machine = platform.machine()
+if machine in ['mips', 'sparc', 'powerpc', 'ppc64']:
+    IOC_SIZEBITS = 13
+    IOC_DIRBITS = 3
+    IOC_NONE, IOC_WRITE, IOC_READ = 1, 2, 4
+elif machine == 'parisc':
+    IOC_WRITE, IOC_READ = 2, 1
+
+IOC_NRSHIFT = 0
+IOC_TYPESHIFT = IOC_NRSHIFT + IOC_NRBITS
+IOC_SIZESHIFT = IOC_TYPESHIFT + IOC_TYPEBITS
+IOC_DIRSHIFT = IOC_SIZESHIFT + IOC_SIZEBITS
+
+
+def IOW(type_, nr, size):
+    return IOC(IOC_WRITE, type_, nr, size)
+
+
+def IOR(type_, nr, size):
+    return IOC(IOC_READ, type_, nr, size)
+
+
+def IOWR(type_, nr, size):
+    return IOC(IOC_READ | IOC_WRITE, type_, nr, size)
+
+
+def IOC(dir_, type_, nr, size):
+    return (dir_ << IOC_DIRSHIFT) \
+        | (ord(type_) << IOC_TYPESHIFT) \
+        | (nr << IOC_NRSHIFT) \
+        | (size << IOC_SIZESHIFT)
+
+
+# Pythonification of linux/watchdog.h
+
+WATCHDOG_IOCTL_BASE = 'W'
+
+
+class watchdog_info(ctypes.Structure):
+    _fields_ = [
+        ('options', ctypes.c_uint32),           # Options the card/driver supports
+        ('firmware_version', ctypes.c_uint32),  # Firmware version of the card
+        ('identity', ctypes.c_uint8 * 32),      # Identity of the board
+    ]
+
+
+struct_watchdog_info_size = ctypes.sizeof(watchdog_info)
+int_size = ctypes.sizeof(ctypes.c_int)
+
+WDIOC_GETSUPPORT = IOR(WATCHDOG_IOCTL_BASE, 0, struct_watchdog_info_size)
+WDIOC_GETSTATUS = IOR(WATCHDOG_IOCTL_BASE, 1, int_size)
+WDIOC_GETBOOTSTATUS = IOR(WATCHDOG_IOCTL_BASE, 2, int_size)
+WDIOC_GETTEMP = IOR(WATCHDOG_IOCTL_BASE, 3, int_size)
+WDIOC_SETOPTIONS = IOR(WATCHDOG_IOCTL_BASE, 4, int_size)
+WDIOC_KEEPALIVE = IOR(WATCHDOG_IOCTL_BASE, 5, int_size)
+WDIOC_SETTIMEOUT = IOWR(WATCHDOG_IOCTL_BASE, 6, int_size)
+WDIOC_GETTIMEOUT = IOR(WATCHDOG_IOCTL_BASE, 7, int_size)
+WDIOC_SETPRETIMEOUT = IOWR(WATCHDOG_IOCTL_BASE, 8, int_size)
+WDIOC_GETPRETIMEOUT = IOR(WATCHDOG_IOCTL_BASE, 9, int_size)
+WDIOC_GETTIMELEFT = IOR(WATCHDOG_IOCTL_BASE, 10, int_size)
+
+
+WDIOF_UNKNOWN = -1  # Unknown flag error
+WDIOS_UNKNOWN = -1  # Unknown status error
+
+WDIOF = {
+    "OVERHEAT": 0x0001,       # Reset due to CPU overheat
+    "FANFAULT": 0x0002,       # Fan failed
+    "EXTERN1": 0x0004,        # External relay 1
+    "EXTERN2": 0x0008,        # External relay 2
+    "POWERUNDER": 0x0010,     # Power bad/power fault
+    "CARDRESET": 0x0020,      # Card previously reset the CPU
+    "POWEROVER": 0x0040,      # Power over voltage
+    "SETTIMEOUT": 0x0080,     # Set timeout (in seconds)
+    "MAGICCLOSE": 0x0100,     # Supports magic close char
+    "PRETIMEOUT": 0x0200,     # Pretimeout (in seconds), get/set
+    "ALARMONLY": 0x0400,      # Watchdog triggers a management or other external alarm not a reboot
+    "KEEPALIVEPING": 0x8000,  # Keep alive ping reply
+}
+
+WDIOS = {
+    "DISABLECARD": 0x0001,    # Turn off the watchdog timer
+    "ENABLECARD": 0x0002,     # Turn on the watchdog timer
+    "TEMPPANIC": 0x0004,      # Kernel panic on temperature trip
+}
+
+# Implementation
+
+
+class WatchdogInfo(collections.namedtuple('WatchdogInfo', 'options,version,identity')):
+    """Watchdog descriptor from the kernel"""
+    def __getattr__(self, name):
+        """Convenience has_XYZ attributes for checking WDIOF bits in options"""
+        if name.startswith('has_') and name[4:] in WDIOF:
+            return bool(self.options & WDIOF[name[4:]])
+
+        raise AttributeError("WatchdogInfo instance has no attribute '{0}'".format(name))
+
+
+class LinuxWatchdogDevice(WatchdogBase):
+    DEFAULT_DEVICE = '/dev/watchdog'
+
+    def __init__(self, device):
+        self.device = device
+        self._support_cache = None
+        self._fd = None
+
+    @classmethod
+    def from_config(cls, config):
+        device = config.get('device', cls.DEFAULT_DEVICE)
+        return cls(device)
+
+    @property
+    def is_running(self):
+        return self._fd is not None
+
+    def open(self):
+        try:
+            self._fd = os.open(self.device, os.O_WRONLY)
+        except OSError as e:
+            raise WatchdogError("Can't open watchdog device: {0}".format(e))
+
+    def close(self):
+        if self.is_running:
+            try:
+                os.write(self._fd, b'V')
+                os.close(self._fd)
+                self._fd = None
+            except OSError as e:
+                return WatchdogError("Error while closing {0}: {1}".format(self.describe(), e))
+
+    @property
+    def can_be_disabled(self):
+        return self.get_support().has_MAGICCLOSE
+
+    def _ioctl(self, func, arg, mutate_arg=False):
+        if self._fd is None:
+            raise WatchdogError("Watchdog device is closed")
+
+        result = fcntl.ioctl(self._fd, func, arg, mutate_arg)
+        if result < 0:
+            raise IOError(result)
+
+    def get_support(self):
+        if self._support_cache is None:
+            info = watchdog_info()
+            self._ioctl(WDIOC_GETSUPPORT, info, True)
+            self._support_cache = WatchdogInfo(info.options,
+                                               info.firmware_version,
+                                               str(bytearray(info.identity)).rstrip('\x00'))
+        return self._support_cache
+
+    def describe(self):
+        dev_str = " at {0}".format(self.device) if self.device != self.DEFAULT_DEVICE else ""
+        ver_str = ""
+        identity = "Linux watchdog device"
+        if self._fd:
+            try:
+                _, version, identity = self.get_support()
+                ver_str = " (firmware {0})".format(version) if version else ""
+            except WatchdogError:
+                pass
+
+        return identity + ver_str + dev_str
+
+    def keepalive(self):
+        try:
+            os.write(self._fd, b'1')
+        except OSError as e:
+            raise WatchdogError("Could not send watchdog keepalive: {0}".format(e))
+
+    def has_set_timeout(self):
+        """Returns True if setting a timeout is supported."""
+        return self.get_support().has_SETTIMEOUT
+
+    def set_timeout(self, timeout):
+        timeout = int(timeout)
+        if not 0 < timeout < 0xFFFF:
+            raise WatchdogError("Invalid timeout {0}. Supported values are between 1 and 65535".format(timeout))
+        self._ioctl(WDIOC_SETTIMEOUT, ctypes.c_int(timeout))
+
+    def get_timeout(self):
+        timeout = ctypes.c_int()
+        self._ioctl(WDIOC_GETTIMEOUT, timeout, True)
+        return timeout.value
+
+
+class TestingWatchdogDevice(LinuxWatchdogDevice):
+    """Converts timeout ioctls to regular writes that can be intercepted from a named pipe."""
+    timeout = 60
+
+    def get_support(self):
+        return WatchdogInfo(WDIOF['MAGICCLOSE'] | WDIOF['SETTIMEOUT'], 0, "Watchdog test harness")
+
+    def set_timeout(self, timeout):
+        buf = "Ctimeout={0}\n".format(timeout).encode('utf8')
+        while len(buf):
+            buf = buf[os.write(self._fd, buf):]
+        self.timeout = timeout
+
+    def get_timeout(self):
+        return self.timeout

postgres0.yml

@@ -76,6 +76,11 @@ postgresql:
       password: zalando
   parameters:
     unix_socket_directories: '.'
+
+#watchdog:
+#  mode: automatic # Allowed values: off, automatic, required
+#  device: /dev/watchdog
+
 tags:
     nofailover: false
     noloadbalance: false

requirements.txt

@@ -11,4 +11,5 @@ click>=4.1
 prettytable>=0.7
 tzlocal
 python-dateutil
+psutil
 cdiff

tests/test_ha.py

@@ -10,6 +10,7 @@ from patroni.dcs.etcd import Client
 from patroni.exceptions import DCSError, PostgresException
 from patroni.ha import Ha, _MemberStatus
 from patroni.postgresql import Postgresql
+from patroni.watchdog import Watchdog
 from patroni.utils import tzutc
 from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get
 from test_postgresql import psycopg2_connect
@@ -84,6 +85,8 @@ postgresql:
   pg_rewind:
     username: postgres
     password: postgres
+watchdog:
+  mode: off
 zookeeper:
   exhibitor:
     hosts: [localhost]
@@ -101,6 +104,7 @@ zookeeper:
         self.nosync = False
         self.scheduled_restart = {'schedule': future_restart_time,
                                   'postmaster_start_time': str(postmaster_start_time)}
+        self.watchdog = Watchdog(self.config)
 
 
 def run_async(self, func, args=()):
@@ -334,7 +338,7 @@ class TestHa(unittest.TestCase):
         with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)):
             self.ha.restart({}, run_async=True)
             self.assertTrue(self.ha.restart_scheduled())
-            self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race')
+            self.assertEquals(self.ha.run_cycle(), 'restart in progress')
 
             self.ha.cluster = get_cluster_initialized_with_leader()
             self.assertEquals(self.ha.run_cycle(), 'restart in progress')
@@ -343,7 +347,10 @@ class TestHa(unittest.TestCase):
             self.assertEquals(self.ha.run_cycle(), 'updated leader lock during restart')
 
             self.ha.update_lock = false
-            self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart')
+            self.p.set_role('master')
+            with patch('patroni.postgresql.Postgresql.stop') as stop_mock:
+                self.assertEquals(self.ha.run_cycle(), 'lost leader lock during restart')
+                stop_mock.assert_called()
 
     @patch('requests.get', requests_get)
     def test_manual_failover_from_leader(self):

tests/test_watchdog.py

@@ -0,0 +1,90 @@
+import unittest
+from mock import patch
+import platform
+import ctypes
+
+from patroni.watchdog import Watchdog
+import patroni.watchdog.linux as linuxwd
+
+import sys
+
+class MockDevice(object):
+    def __init__(self, fd, filename, flag):
+        self.fd = fd
+        self.filename = filename
+        self.flag = flag
+        self.timeout = 60
+        self.open = True
+        self.writes = []
+
+
+mock_devices = [None]
+
+def mock_open(filename, flag):
+    fd = len(mock_devices)
+    mock_devices.append(MockDevice(fd, filename, flag))
+    return fd
+
+def mock_ioctl(fd, op, arg=None, mutate_flag=False):
+    assert 0 < fd < len(mock_devices)
+    dev = mock_devices[fd]
+    sys.stderr.write("Ioctl %d %d %r\n" %( fd, op, arg))
+    if op == linuxwd.WDIOC_GETSUPPORT:
+        sys.stderr.write("Get support\n")
+        assert(mutate_flag == True)
+        arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING', 'MAGICCLOSE']))
+        arg.identity = (ctypes.c_ubyte*32)(*map(ord, 'Mock Watchdog'))
+    elif op == linuxwd.WDIOC_GETTIMEOUT:
+        arg.value = dev.timeout
+    elif op == linuxwd.WDIOC_SETTIMEOUT:
+        sys.stderr.write("Set timeout called with %s\n" % arg.value)
+        assert 0 < arg.value < 65535
+        dev.timeout  = arg.value
+    else:
+        raise Exception("Unknown op %d", op)
+    return 0
+
+def mock_write(fd, string):
+    assert 0 < fd < len(mock_devices)
+    assert len(string) == 1
+    assert mock_devices[fd].open
+    mock_devices[fd].writes.append(string)
+
+def mock_close(fd):
+    assert 0 < fd < len(mock_devices)
+    assert mock_devices[fd].open
+    mock_devices[fd].open = False
+
+@patch('os.open', mock_open)
+@patch('os.write', mock_write)
+@patch('os.close', mock_close)
+@patch('fcntl.ioctl', mock_ioctl)
+class TestWatchdog(unittest.TestCase):
+    def setUp(self):
+        mock_devices[:] = [None]
+
+    def test_basic_operation(self):
+        if platform.system() != 'Linux':
+            return
+
+        watchdog = Watchdog({'ttl': 30, 'loop_wait': 10, 'watchdog': {'mode': 'required'}})
+
+        watchdog.activate()
+        self.assertEquals(len(mock_devices), 2)
+        device = mock_devices[-1]
+        self.assertTrue(device.open)
+
+        self.assertEquals(device.timeout, 15)
+
+        watchdog.keepalive()
+        self.assertEquals(len(device.writes), 1)
+
+        watchdog.disable()
+        self.assertFalse(device.open)
+        self.assertEquals(device.writes[-1], b'V')
+
+    def test_invalid_timings(self):
+        watchdog = Watchdog({'ttl': 30, 'loop_wait': 20, 'watchdog': {'mode': 'automatic'}})
+        watchdog.activate()
+        self.assertEquals(len(mock_devices), 1)
+        self.assertFalse(watchdog.is_running)