replication

// The streaming replication of a PostgreSQL cluster, by SSH and psql on its server, on either side. On a
// primary, one link per standby connected: its state, and how far its replay lags behind, in WAL and in time.
// On a standby, one link, "upstream": the state of its WAL receiver, and the WAL received but not replayed
// yet. The role of the server, and the number of links, so that a standby that is gone says so.
// psql connects over the local socket as the PostgreSQL role named like the SSH account (peer authentication,
// or a ~/.pgpass of that account): no password leaves banshee, none is written on a command line.
need('protocol:ssh');
$port = variable('postgresql:port', 'integer', array(
    'label'   => 'Port',
    'help'    => 'The port of the PostgreSQL cluster, which names its local socket: 5432, or 5433 for a second cluster on the same server.',
    'default' => 5432,
));
$database = variable('postgresql:database', 'text', array(
    'label'   => 'Database',
    'help'    => 'The database psql connects to: any one the role may connect to.',
    'default' => 'postgres',
));

per('link', 'Replication links');
output('role', 'text', array('label' => 'Role'));
output('links', 'number', array('label' => 'Replication links'));
output('state', 'text', array('label' => 'State', 'per' => 'link'));
output('sync', 'text', array('label' => 'Synchronous state', 'per' => 'link'));
output('lag-size', 'number', array('label' => 'WAL not replayed', 'unit' => 'MiB', 'per' => 'link'));
output('lag', 'number', array('label' => 'Replay lag', 'unit' => 's', 'per' => 'link'));

watch.below('no-link', 'No replication link', 'links', array('critical' => 1));
watch.expect('not-streaming', 'Replication not streaming', 'state', array('critical' => array('streaming'), 'for' => '3 runs'));
watch.above('lag-size', 'Replication behind', 'lag-size', array('warning' => 64, 'critical' => 1024));
watch.above('lag', 'Replication late', 'lag', array('warning' => 60, 'critical' => 300));
watch.expect('role', 'Role changed', 'role', array('enabled' => false));
watch.expect('sync', 'Synchronous state changed', 'sync', array('enabled' => false));
watch.missing('missing', 'Standby missing', 'state', array('enabled' => false));

// One query through psql, which writes one line of JSON. LC_ALL=C keeps its messages in English. A cluster
// whose socket is absent, or that refuses the connection, is not running: critical.
function query($port, $database, $sql) {
    $read = ssh.run('LC_ALL=C psql -X -A -t -q -v ON_ERROR_STOP=1', arguments=array('-p', '' . $port, '-d', $database, '-c', $sql));
    if ($read['exit'] == 2 && list.count(text.match($read['errors'], '(?i)connection refused|no such file or directory')) > 0) {
        fail('critical', 'PostgreSQL is not running', $read['errors']);
    }
    if ($read['exit'] != 0) {
        fail('misconfigured', 'psql failed', $read['errors']);
    }
    return json.read(text.trim($read['output']));
}

// standbys: one row of pg_stat_replication per standby connected to this server (on a standby, the cascading
// ones), its lag in bytes from the WAL written here to the WAL it replayed, and replay_lag, null when it is
// idle and caught up. receiver: the status of pg_stat_wal_receiver, null when no receiver runs. backlog: the
// WAL a standby received and did not replay yet. Reading them needs pg_monitor.
$found = query($port, $database, "SELECT json_build_object(
    'standby', pg_is_in_recovery(),
    'standbys', (SELECT COALESCE(json_agg(json_build_object(
        'name', application_name, 'address', host(client_addr), 'pid', pid, 'state', state, 'sync', sync_state,
        'bytes', pg_wal_lsn_diff(CASE WHEN pg_is_in_recovery() THEN pg_last_wal_receive_lsn() ELSE pg_current_wal_lsn() END, replay_lsn),
        'seconds', EXTRACT(EPOCH FROM replay_lag))), '[]'::json) FROM pg_stat_replication),
    'receiver', (SELECT status FROM pg_stat_wal_receiver),
    'backlog', COALESCE(pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn()), 0))");

// A standby is named by its application_name, unless it keeps the default one (walreceiver), then by its
// address; two of one name keep their own, the second carrying its process id.
$links = 0;
$seen = array();
foreach ($found['standbys'] as $standby) {
    $name = $standby['name'];
    if ($name == '' || $name == 'walreceiver') {
        $name = $standby['address'];
    }
    if ($name == '') {
        $name = 'local';
    }
    if (list.has($seen, $name) || $name == 'upstream') {
        $name = $name . ' #' . $standby['pid'];
    }
    $seen[$name] = true;
    $links = $links + 1;
    value('state', $standby['state'], $name);
    value('sync', $standby['sync'], $name);
    if ($standby['bytes'] != '') {
        value('lag-size', number.round($standby['bytes'] / 1048576, 1), $name);
    }
    if ($standby['seconds'] != '') {
        value('lag', number.round($standby['seconds'], 1), $name);
    }
}

$role = 'primary';
if ($found['standby']) {
    $role = 'standby';
    $receiver = $found['receiver'];
    if ($receiver == '') {
        $receiver = 'stopped';
    }
    $links = $links + 1;
    value('state', $receiver, 'upstream');
    value('lag-size', number.round($found['backlog'] / 1048576, 1), 'upstream');
}
value('role', $role);
value('links', $links);