bucardo_ctl

#!/usr/bin/perl -- -*-cperl-*-

## Script to control Bucardo
##
## Copyright 2006-2009 Greg Sabino Mullane <greg@endpoint.com>

package bucardo_ctl;
use strict;
use warnings;
use 5.008003;
use DBI;
use IO::Handle;
use Getopt::Long;
use Time::HiRes 'sleep';
use POSIX qw/ceil setsid/;
use Data::Dumper;

return 1 if $ENV{BUCARDO_CTL_TEST};

our $VERSION = '3.2.7';

*STDOUT->autoflush(1);
*STDERR->autoflush(1);

use vars qw/$dbh $SQL $sth %sth $count $info/;
my $DEBUG            = 0;
my $DATEFORMAT       = q{Mon DD, YYYY HH24:MI:SS};
my $DEFAULT_DAYSBACK = 3;
my $WAITSLEEP        = 1;
my $PROGRESS         = 1;

help() unless @ARGV;

## Default arguments - most are for the bc constructor
my $bcargs = {
			  ctlquiet     => 0,
			  ctlverbose   => 0,
			  dbname       => 'bucardo',
			  dbuser       => 'bucardo',
			  dbpass       => '',
			  verbose      => 1, ## Highly recommended to leave this on
			  sendmail     => 0,
			  extraname    => '',
			  debugfilesep => 0,
			  debugdir     => '.',
			  debugname    => '',
			  debugstderr  => 1,
			  debugstdout  => 0,
			  debugsyslog  => 1,
			  debugfile    => 1,
			  cleandebugs  => 0,
		  };

## Values are first read from a .bucardorc, either in the current dir, or the home dir.
## These will be overwritten by command-line args.
my $file;
if (-e '.bucardorc') {
	$file = '.bucardorc';
}
elsif (-e "$ENV{HOME}/.bucardorc") {
	$file = "$ENV{HOME}/.bucardorc";
}
if (defined $file) {
	open my $rc, '<', $file or die qq{Could not open "$file": $!\n};
	while (<$rc>) {
		next if /^\s*#/;
		next unless /^\s*(\w+)\s*=\s*(.+?)\s*$/o;
		my ($name,$value) = ($1,$2); ## no critic (ProhibitCaptureWithoutTest)
		$bcargs->{$name} = $value;
	}
	close $rc or die;
}

GetOptions ## no critic (ProhibitCallsToUndeclaredSubs)
	($bcargs,
	 'ctlverbose+',
	 'ctlquiet+',
	 'notimer',
	 'help',
	 'daysback=i',
	 'sort=i',
	 'showdays',
	 'compress',
	 'retry=i',
	 'retrysleep=i',

	 ## These are sent to the constructor:
	 'dbport=i',
	 'dbhost=s',
	 'dbname=s',
	 'dbuser=s',
	 'dbpass=s',
	 'verbose=i',
	 'sendmail=i',
	 'extraname=s',
	 'debugfilesep',
	 'debugname=s',
	 'debugstderr=i',
	 'debugstdout=i',
	 'debugsyslog=i',
	 'debugdir=s',
	 'debugfile=i',
	 'cleandebugs=i',
) or die "\n";

my $QUIET      = delete $bcargs->{ctlquiet};
help() if $bcargs->{help};

my $VERBOSE    = delete $bcargs->{ctlverbose};
my $DAYSBACK   = delete $bcargs->{daysback} || $DEFAULT_DAYSBACK;
my $COMPRESS   = delete $bcargs->{compress};
my $SHOWDAYS   = delete $bcargs->{showdays} || 1;
my $RETRY      = delete $bcargs->{retry} || 0;
my $RETRYSLEEP = delete $bcargs->{retrysleep} || 0;
my $NOTIMER    = delete $bcargs->{notimer} || 0;

my $DBCONN = "User: $bcargs->{dbuser}  Database: $bcargs->{dbname}";
$bcargs->{dbhost} and length $bcargs->{dbhost} and $DBCONN .= "  Host: $bcargs->{dbhost}";
$bcargs->{dbport} and length $bcargs->{dbport} and $DBCONN .= "  Port: $bcargs->{dbport}";

## Anything left over is the verb and noun(s)

my $verb = shift || '';
help() unless $verb;
$verb = lc $verb;
my @nouns = @ARGV;
my $nouns = join ' ' => @nouns;

## Grab current information from the bucardo_config file
my $DSN = "dbi:Pg:dbname=$bcargs->{dbname}";
$bcargs->{dbhost} and length $bcargs->{dbhost} and $DSN .= ";host=$bcargs->{dbhost}";
$bcargs->{dbport} and length $bcargs->{dbport} and $DSN .= ";port=$bcargs->{dbport}";
$dbh = DBI->connect($DSN, $bcargs->{dbuser}, $bcargs->{dbpass}, {AutoCommit=>0,RaiseError=>1,PrintError=>0});
$dbh->do("SET search_path = bucardo");
$dbh->commit();

my $REASONFILE = get_config('reason_file') or die "Invalid reason_file!\n";
my $PIDDIR     = get_config('piddir') or die "Invalid piddir!\n";
my $pidfile    = get_config('pidfile') or die "Invalid pidfile!\n";
my $stopfile   = get_config('stopfile') or die "Invalid stopfile!\n";

my $PIDFILE        = "$PIDDIR/$pidfile";
my $REASONFILE_LOG = "$REASONFILE.log";
my $STOPFILE       = "$PIDDIR/$stopfile";

## Handle all the verbs
status_all()    if $verb eq 'status' and ! @nouns;
status_detail() if $verb eq 'status';
start()         if $verb eq 'start';
stop()          if $verb eq 'stop';
reload_config() if $verb eq 'reload_config';
ping()          if $verb eq 'ping';
add_item()      if $verb eq 'add';
alter_item()    if $verb eq 'alter';
remove_item()   if $verb eq 'remove';
upgrade()       if $verb eq 'upgrade';
message()       if $verb eq 'message';
list()          if $verb eq 'list';
config()        if $verb eq 'set' or $verb eq 'show';


## For the rest, we expect a list of sync with an optional decimal "timeout"
my $adverb;
my $syncs = get_syncs();
my @syncs;
SYNCMATCH: for my $sync (@nouns) {
	if ($sync =~ /^\d+$/) {
		$adverb = $sync;
		next SYNCMATCH;
	}
    if ($sync =~ /%/) {
        my $tmp = $dbh->selectall_arrayref(qq{SELECT name FROM sync WHERE name LIKE '$sync'});
        if ($#{ $tmp } >= 0) {
            push @syncs, (map { $_->[0] } @$tmp);
            next SYNCMATCH;
        }
    }
	if (! exists $syncs->{$sync}) {
		die qq{Sync "$sync" does not appear to exist\n};
	}
	push @syncs, $sync;
}
vate_sync() if $verb eq 'activate' or $verb eq 'deactivate';
kick() if $verb eq 'kick';

help();

sub help {
	warn qq{Usage: $0 args
  start <reason>  ** force existing sync(s) to quit, then start

  stop <reason>   ** tell everyone to stop permanently, even the MCP

  list <type>     ** view short info about dbs, herds, goats, or syncs

  add <item_type> <item_name>
                  ** add a new database, table, sync, herd, or database group
    add database <dbname> name=internal_name port=xxx host=xxx user=xxx pass=xxx service=xxx conn=xxx sourcelimit=xxx targetlimit=xxx
    add table [schema].table db=internal_db_name ping=bool standard_conflict=xxx makedelta=bool
    add sync syncname source=xxx targetdb=xxx targetgroup=xxx type=xxx makedelta=xxx 
    add herd name
    add dbgroup name db1 db2 db3 ...

  kick <syncname(s)> [timeout]
                  ** kick off one or more syncs, optionally wait for result (0 = wait until done)

  ping [timeout]  ** ping the MCP for a response

  status [--sort=col#] [--daysback=#]
                  ** list information about all syncs

  status syncname[s] [--daysback=#]
                  ** list detailed information about one or more syncs

  reload_config   ** force a running Bucardo to reload the bucardo_config table
For more details, try 'man bucardo_ctl'
} unless $QUIET; ## Mostly for the test suite
	exit 0;
}


sub start {

	## Attempt to start Bucardo

	## Refuse to go on if we get a ping response within 5 seconds
	$QUIET or print "Checking for existing processes\n";

	my $oldpid = ping({quiet => 1, timeout => 5, noexit => 1});

	if ($oldpid) {
		$QUIET or print "Cannot start, process $oldpid is already running\n";
		exit 1;
	}

	## Create a new Bucardo instance and connect to its database
	require Bucardo;
	my $bc = Bucardo->new($bcargs);

	my $pm_version = $bc->{version} || 'unknown';
	if ($VERSION ne $pm_version) {
		die "Version mismatch: bucardo_ctl is $VERSION, but Bucardo.pm is $pm_version\n";
	}

	set_reason();

	stop_bucardo();

	sleep 2; ## Give everyone a chance to notice it
	## TODO: be smarter about this by scanning PIDDIR

	if (-e $STOPFILE) {
		print "Removing $STOPFILE\n" unless $QUIET;
		unlink $STOPFILE;
	}
	if (-e $PIDFILE) {
		print "Removing $STOPFILE\n" unless $QUIET;
		unlink $PIDFILE;
	}

	$QUIET or print qq{Starting Bucardo\n};

	$dbh->disconnect();
	if (fork) {
	}
	else {
		close STDERR;
		close STDOUT;
		setsid() or die;
		$bc->start_mcp();
	}
	exit 0;

}


sub stop {

	## Attempt to stop Bucardo

	set_reason();

	print "Creating $STOPFILE ... " unless $QUIET;
	stop_bucardo();
	print "Done\n" unless $QUIET;

	exit 0;
}


sub reload_config {

	## Reload configuration settings from the DB, restart all controllers and kids
	for (@nouns) {
		if (/^\d+$/) {
			$adverb = $1;
			last;
		}
	}

	$QUIET or print qq{Forcing Bucardo to reload the bucardo_config table};

	my $done = 'bucardo_reload_config_finished';
	$dbh->do('NOTIFY bucardo_reload_config');
	if (defined $adverb) {
		print '...';
		$dbh->do("LISTEN $done");
	}
	$dbh->commit();

	if (!defined $adverb) {
		print "\n";
		exit 0;
	}
	sleep 0.1;
  WAITIN: {
		while (my $notify = $dbh->func('pg_notifies')) {
			my ($name, $pid) = @$notify;
			last WAITIN if $name eq $done;
		}
		$dbh->commit();
		sleep($WAITSLEEP);
		redo;
	}
	print "DONE!\n";

	exit 0;

} ## end of reload_config


sub ping {

	## See if the MCP is alive and responds to pings
	## Default is to wait 15 seconds

	my $arg = shift || {};

	my $timeout = $arg->{timeout} || $ARGV[0] || 15;
	my $quiet = $arg->{quiet} || $ARGV[1] || 0;

	if (defined $nouns[0] and $nouns[0] =~ /^\d+$/) {
		$timeout = $nouns[0];
	}

	$VERBOSE and print "Pinging MCP, timeout = $timeout\n";
	$dbh->do('LISTEN bucardo_mcp_pong');
	$dbh->do('NOTIFY bucardo_mcp_ping');
	$dbh->commit();
	my $starttime = time;
	sleep 0.1;

  P:{
		my $notify = $dbh->func('pg_notifies');
		if (defined $notify) {
			my ($name, $pid) = @$notify;
			$quiet or print "OK: Got response from PID $pid\n";
			return $pid if $arg->{noexit};
			exit 0;
		}
		$dbh->rollback();
		sleep 0.5;
		my $totaltime = time - $starttime;
		if ($timeout and $totaltime >= $timeout) {
			$quiet or print "CRITICAL: Timed out ($totaltime s), no ping response from MCP\n";
			return 0 if $arg->{noexit};
			exit 1;
		}
		redo;
	}

	return;

} ## end of ping

sub kick {

	@nouns or die qq{kick requires at least one sync name\n};

	my ($exitstatus, $retries, $do_retry) = (0,0,0);

  RETRY: {
		$dbh->rollback();
		$exitstatus = 0;
	  SYNC: for my $sync (@syncs) {
			$dbh->do(qq{NOTIFY "bucardo_kick_sync_$sync"});
			my $done = "bucardo_syncdone_$sync";
			my $killed = "bucardo_synckill_$sync";
			if (! defined $adverb) {
				$dbh->commit();
				$QUIET or print qq{Kicked sync $sync\n};
				next;
			}

			$QUIET or print qq{Kick $sync: };
			$dbh->do(qq{LISTEN "$done"});
			$dbh->do(qq{LISTEN "$killed"});
			my $s = $syncs->{$sync};
			if ($s->{targetgroup}) {
				for (@{$s->{targets}}) {
					$dbh->do(qq{LISTEN "bucardo_syncdone_${sync}_$_"});
					$dbh->do(qq{LISTEN "bucardo_synckill_${sync}_$_"});
				}
			}
			else {
				$dbh->do(qq{LISTEN "bucardo_syncdone_${sync}_$s->{targetdb}"});
				$dbh->do(qq{LISTEN "bucardo_synckill_${sync}_$s->{targetdb}"});
			}
			$dbh->commit();

			my $time = time;
			sleep 0.1;

			my $timeout = (defined $adverb and $adverb > 0) ? $adverb : 0;

			my $printstring = $NOTIMER ? '' : '[0 s] ';
			print $printstring unless $QUIET or $NOTIMER;
			my $oldtime = 0;
			local $SIG{ALRM} = sub { die 'Timed out' };
			$do_retry = 0;
			eval {
				if ($timeout) {
					alarm $timeout;
				}
			  WAITIN: {
					my $lastwait = '';
					if ($PROGRESS and time - $time != $oldtime) {
						$oldtime = time - $time;
						if (!$QUIET and !$NOTIMER) {
							print "\b" x length($printstring);
							$printstring =~ s/\d+/$oldtime/;
							print $printstring;
						}
					}
				  W: while (my $notify = $dbh->func('pg_notifies')) {
						my ($name, $pid) = @$notify;
						if ($name eq $done) {
							$lastwait = 'DONE!';
						}
						elsif ($name eq $killed) {
							$lastwait = 'KILLED!';
							$exitstatus = 2;
						}
						elsif ($name =~ /^bucardo_syncdone_${sync}_(.+)$/) {
							my $new = sprintf "$1(%ds) ", ceil(time-$time);
							print $new unless $QUIET;
							$printstring .= $new;
						}
						elsif ($name =~ /^bucardo_synckill_${sync}_(.+)$/) {
							my $new = sprintf "$1 KILLED (%ds) ", ceil(time-$time);
							print $new unless $QUIET;
							$printstring .= $new;
							$exitstatus = 2;
							$lastwait = " ";
						}
					}
					$dbh->rollback();
					if ($lastwait) {
						print $lastwait unless $QUIET;
						if ($lastwait ne 'DONE!' and $RETRY and ++$retries <= $RETRY) {
							print "Retry #$retries\n";
							$do_retry = 1;
							die "Forcing eval to exit for retry attempt\n";
						}
						last WAITIN;
					}
					sleep($WAITSLEEP);
					redo WAITIN;
				}
				alarm 0 if $timeout;
			};
			alarm 0 if $timeout;
			if ($do_retry) {
				$do_retry = 0;
				redo RETRY;
			}
			if ($@) {
				if ($@ =~ /Timed out/o) {
					$exitstatus = 1;
					warn "Timed out!\n";
				}
				else {
					$exitstatus = 2;
					warn "Error: $@\n";
				}
				next SYNC;
			}

			next SYNC if $QUIET;

			if ($PROGRESS) {
				print "\n";
			}
			else {
				printf "(%ds)\n", ceil(time - $time);
			}
		} ## end each sync

	} ## end RETRY

	exit $exitstatus;

} ## end of kick


sub status_all {

	## Show status of all syncs in the database

	print "Days back: $DAYSBACK  $DBCONN  ";

	## See if the MCP is running and what its PID is
	if (! -e $PIDFILE) {
		print "\nBucardo may not be running. No file found at $PIDFILE";
	}
	else {
		my $fh;
		if (!open $fh, '<', $PIDFILE) {
			print "\nERROR: Could not open $PIDFILE: $!";
		}
		else {
			my $pid = <$fh>;
			chomp $pid;
			close $fh or warn qq{Could not close $PIDFILE: $!\n};
			if ($pid =~ /^\d+$/) {
				print "PID of Bucardo MCP: $pid";
			}
			else {
				print "\nERROR: $PIDFILE contained: $pid";
			}
		}
	}
	print "\n";

	my $orderby = $bcargs->{sort} || '1';
	if ($orderby !~ /^\+?\-?\d$/) {
		die "Invalid sort option, must be +- 1 through 9\n";
	}

	our $syncs;
	my $max;
	($syncs,$max) = get_detailed_syncs();

	printf qq{%-*s %-*s %-*s %-*s %-*s %-*s %-*s %-*s %-*s\n},
		$max->{name},'Name',
		$max->{type}, 'Type',
		$max->{stat}, 'State',
		$max->{pid}, 'PID',
		$max->{good}, 'Last_good',
		$max->{total}, 'Time',
		$max->{iud}, 'I/U/D',
		$max->{bad}, 'Last_bad',
		$max->{totalbad}, 'Time';

	printf qq{%s+%s+%s+%s+%s+%s+%s+%s+%s\n},
		'=' x $max->{name},
		'=' x $max->{type},
		'=' x $max->{stat},
		'=' x $max->{pid},
		'=' x $max->{good},
		'=' x $max->{total},
		'=' x $max->{iud},
		'=' x $max->{bad},
		'=' x $max->{totalbad};

	## If fancy sorting desired, call the list ourself to sort
	sub sortme {
		my $sortcol = $bcargs->{sort} || 1;

		+1 == $sortcol and return $a cmp $b;
		-1 == $sortcol and return $b cmp $a;

		my ($uno,$dos) = ($syncs->{$a}, $syncs->{$b});

		## Synctype
		+2 == $sortcol and return ($uno->{synctype} cmp $dos->{synctype} or $a cmp $b);
		-2 == $sortcol and return ($dos->{synctype} cmp $uno->{synctype} or $a cmp $b);

		## Status
		+3 == $sortcol and return ($uno->{state} cmp $dos->{state} or $a cmp $b);
		-3 == $sortcol and return ($dos->{state} cmp $uno->{state} or $a cmp $b);

		## PID
		+4 == $sortcol and return $uno->{PID} <=> $dos->{PID};
		-4 == $sortcol and return $dos->{PID} <=> $uno->{PID};

		## Last good
		+5 == $sortcol and return ($uno->{lastgoodsecs} <=> $dos->{lastgoodsecs} or $a cmp $b);
		-5 == $sortcol and return ($dos->{lastgoodsecs} <=> $uno->{lastgoodsecs} or $a cmp $b);

		## Good time
		+6 == $sortcol and return ($uno->{lastgoodtime} <=> $dos->{lastgoodtime} or $a cmp $b);
		-6 == $sortcol and return ($dos->{lastgoodtime} <=> $uno->{lastgoodtime} or $a cmp $b);

		if ($sortcol == 7 or $sortcol == -7) {
			my ($total1,$total2) = (0,0);
			while ($uno->{iud} =~ /(\d+)/go) {
				$total1 += $1;
			}
			while ($dos->{iud} =~ /(\d+)/go) {
				$total2 += $1;
			}

			7 == $sortcol and return ($total1 <=> $total2 or $a cmp $b);

			return ($total2 <=> $total1 or $a cmp $b);
		}

		## Last bad
		+8 == $sortcol and return ($uno->{lastbadsecs} <=> $dos->{lastbadsecs} or $a cmp $b);
		-8 == $sortcol and return ($dos->{lastbadsecs} <=> $uno->{lastbadsecs} or $a cmp $b);

		## Bad time
		+9 == $sortcol and return ($uno->{lastbadtime} <=> $dos->{lastbadtime} or $a cmp $b);
		-9 == $sortcol and return ($dos->{lastbadtime} <=> $uno->{lastbadtime} or $a cmp $b);


		return $a cmp $b;

	}

	for my $sync (sort sortme keys %$syncs) {
		my $s = $syncs->{$sync};
		my $X = '|';
		printf qq{%-*s$X%-*s$X%-*s$X%-*s$X%-*s$X%-*s$X%-*s$X%-*s$X%-*s\n},
			$max->{name},$sync,
			$max->{type},$s->{typetext},
			$max->{stat}, $s->{state},
			$max->{pid}, $s->{PID} || '',
			$max->{good}, $s->{lastgood},
			$max->{total}, $s->{timegood},
			$max->{iud}, $s->{iud},
			$max->{bad}, $s->{lastbad},
			$max->{totalbad}, $s->{timebad};
	}


	exit 0;

} ## end of status_all


sub status_detail {

	## Show detailed information about one or more syncs

	## Verify that all named syncs exist
	my ($syncs,$max) = get_detailed_syncs({syncs => \@nouns});

	my @syncs;
	for my $sync (sort @nouns) {
		$DEBUG and warn "Verify sync: $sync\n";
		if (!exists $syncs->{$sync}) {
            ## If a number, skip for ease of "kick name #" toggling
            $sync !~ /^\d+$/ and die "No such sync: $sync\n";
		}
		else {
			push @syncs => $sync;
		}
	}

	print "Days back: $DAYSBACK  $DBCONN\n";

	## Present each ordered by name
	for my $sync (sort { lc $a cmp lc $b } @syncs) {
		print '=' x 70; print "\n";
		my $s = $syncs->{$sync};

		## Undefined should be written as 'none'
		for (qw/checktime/) {
			$s->{$_} ||= 'none';
		}
		## Not true should be empty
		for (qw/PID PIDFILE CREATED/) {
			$s->{$_} ||= '';
		}

		my $morepid = '';
		if ($s->{stayalive} or $s->{kidsalive}) {
			$morepid = " (PID = $s->{PID})";
		}

		## Should be 'yes' or 'no'
		for (qw/makedelta analyze_after_copy stayalive kidsalive ping rebuild_index do_listen usecustomselect/) {
			$s->{$_} = (defined $s->{$_} and $s->{$_}) ? 'yes' : 'no ';
		}

		my $target;
		if ($s->{targetdb}) {
			$target = "Target database:   $s->{targetdb}";
		}
		else {
			$target = "Target group:      $s->{targetgroup} (";
			$target .= join ',' => @{$s->{targets}};
			$target .= ')';
		}

		$s->{iud} =~ s#/# / #g;
		$s->{state} =~ s/>/ > /;

		my $moregood = '';
		if ($s->{lastgood} ne 'unknown') {
			$moregood .= " (time to run: $s->{timegood})";
			$moregood .= "\nLast good time:    $s->{latest_good}{last_ended_date}";
			$moregood .= "  Target: $s->{latest_good}{targetdb}";
		}

		my $morebad = '';
		if ($s->{lastbad} ne 'unknown') {
			$morebad = " (time to run: $s->{timebad})";
			$morebad .= "\nLast bad time:     $s->{latest_bad}{last_aborted_date}";
			$morebad .= "  Target: $s->{latest_bad}{targetdb}";
			(my $why = $s->{latest_bad}{whydie}) =~ s/\s+$//;
			$why =~ s/^/                  /mg;

			$why =~ s/^\s+//;
			$morebad .= "\nLatest bad reason: $why";
		}

		print qq{Sync name:         $sync
Current state:     $s->{state}$morepid
Type:              $s->{synctype}
Source database:   $s->{source}
$target
Last good:         $s->{lastgood}$moregood
Ins/Upd/Del:       $s->{iud}
Last bad:          $s->{lastbad}$morebad
PID file:          $s->{PIDFILE}
PID file created:  $s->{CREATED}
Status:            $s->{status}
Limitdbs:          $s->{limitdbs}
Priority:          $s->{priority}
Checktime:         $s->{checktime}
Overdue time:      $s->{overdue}
Expired time:      $s->{expired}
Stayalive:         $s->{stayalive}          Kidsalive: $s->{kidsalive}
Rebuild index:     $s->{rebuild_index}          Do_listen: $s->{do_listen}
Ping:              $s->{ping}          Makedelta: $s->{makedelta}
};
		if ($s->{synctype} eq 'fullcopy'){
			print qq{Custom select:     $s->{usecustomselect} $s->{copyextra}\n};
			print qq{Post-copy analyze: $s->{analyze_after_copy}\n};
			print qq{Delete method:     $s->{deletemethod}\n};
		}

	}
	exit 0;

} ## end of status_detail

sub get_config {

	my $name = shift;

	$SQL = "SELECT value FROM bucardo_config WHERE lower(setting) = ?";
	$sth = $dbh->prepare_cached($SQL);
	$count = $sth->execute(lc $name);
	if ($count < 1) {
		$sth->finish();
		die "Invalid bucardo_config setting: $name\n";
	}
	return $sth->fetchall_arrayref()->[0][0];
}


sub set_reason {

	if (! length $nouns) {
		warn qq{Please provide a reason. For example:\n$0 $verb "Adding new table -Greg"\n};
		exit 1;
	}

	open my $fh, '>', $REASONFILE or die qq{Could not open "$REASONFILE": $!\n};
	print $fh (scalar localtime) . " | $nouns\n";
	close $fh or warn qq{Could not close $REASONFILE: $!\n};
	open $fh, '>>', $REASONFILE_LOG or die qq{Could not open "$REASONFILE_LOG": $!\n};
	print $fh (scalar localtime) . " | $verb | $nouns\n";
	close $fh or warn qq{Could not close $REASONFILE_LOG: $!\n};
	return 1;
}


sub stop_bucardo {
	open my $stop, '>', $STOPFILE or die qq{Could not create "$STOPFILE": $!\n};
	print {$stop} "Stopped by $0 on " . (scalar localtime) . "\n";
	close $stop or warn qq{Could not close "$STOPFILE": $!\n};
	return;
}


sub get_detailed_syncs {

	my $arg = shift || {};

	my $syncs = get_syncs();

	## Now see exactly what's going on with it at this moment (according to q)

	## First, we need to hard-code our time backwards

	$SQL = "SELECT date (now() - '$DAYSBACK days'::interval)";
	my $oldtime = $dbh->selectall_arrayref($SQL)->[0][0];
	$dbh->do("SET constraint_exclusion = true");

	## We want information about the last time it ran successfully
	$SQL = qq{
SELECT *, 
TO_CHAR(ended,'$DATEFORMAT') AS last_ended_date,
TO_CHAR(aborted,'$DATEFORMAT') AS last_aborted_date,
round(extract(epoch FROM ended-started)) AS total_time_ended,
round(extract(epoch FROM aborted-started)) AS total_time_aborted,
round(extract(epoch FROM now()-ended)) AS last_ended_secs,
round(extract(epoch FROM now()-aborted)) AS last_aborted_secs
FROM
(SELECT * FROM q WHERE sync = \$1 AND cdate >= '$oldtime'
UNION ALL
SELECT * FROM freezer.master_q WHERE sync = \$1 AND cdate >= '$oldtime') AS foo
WHERE ended is NOT NULL AND aborted IS NULL
ORDER BY ended DESC LIMIT 1;
};
	$sth{latest_good} = $dbh->prepare($SQL);

	## If the last one was an abort, find the latest good run:
	$SQL =~ s/IS NULL/IS NOT NULL/;
	$sth{latest_bad} = $dbh->prepare($SQL);

	## Is it running right now?
	$SQL = qq{
SELECT *,
round(extract(epoch FROM now()-cdate)) AS q1,
round(extract(epoch FROM now()-started)) AS q2
FROM q
WHERE sync = \$1
AND ended IS NULL
AND aborted IS NULL
ORDER BY started DESC LIMIT 1
};
	$sth{active_now} = $dbh->prepare($SQL);

	my %max = (
			   name     => 4,
			   type     => 5,
			   stat     => 5,
			   pid      => 3,
			   q1       => 6,
			   q2       => 6,
			   iud      => 5,
			   good     => 9,
			   total    => 5,
			   bad      => 8,
			   totalbad => 4,
			   );
	for my $sync (sort keys %$syncs) {
		if ($arg->{syncs}) {
			next unless grep { $_ eq $sync } @{$arg->{syncs}}; ## no critic (ProhibitBooleanGrep)
		}
		$DEBUG and warn "Reading sync $sync...\n";

		my $s = $syncs->{$sync};

		## Normal types are too long
		$s->{typetext} = ' ' . uc(substr $s->{synctype},0,1);

		## Set some basic lengths
		$max{name} = length($sync) if length($sync) > $max{name};
		$s->{PID} = 0 if ! defined $s->{PID};
		$max{pid} = length($s->{PID}) if length($s->{PID}) > $max{pid};

		## Grab information from the q tables
		$sth{latest_good}->execute($sync);
		$s->{latest_good}  = $sth{latest_good}->fetchall_arrayref({})->[0];

		$sth{latest_bad}->execute($sync);
		$s->{latest_bad} = $sth{latest_bad}->fetchall_arrayref({})->[0];

		$sth{active_now}->execute($sync);
		$s->{active_now} = $sth{active_now}->fetchall_arrayref({});

		## Pretty up the times
		$s->{lastgoodsecs} = $s->{lastgoodtime} = 0;
		$s->{lastbadsecs} = $s->{lastbadtime} = 0;
		$s->{lastgood} = $s->{lastbad} = 'unknown';
		$s->{timegood} = $s->{timebad} = '';
		$s->{iud} = '';
		$s->{is_overdue} = $s->{is_expired} = '?';
		if (defined $s->{latest_good}) {
			my $g = $s->{latest_good};
			$s->{iud} = "$g->{inserts}/$g->{updates}/$g->{deletes}";
			$max{iud} = length($s->{iud}) if length($s->{iud}) > $max{iud};

			$s->{timegood} = pretty_time($g->{total_time_ended});
			$max{total} = length($s->{timegood}) if length($s->{timegood}) > $max{total};
			$s->{lastgoodtime} = $g->{total_time_ended};

			$s->{lastgood} = pretty_time($g->{last_ended_secs});
			$max{good} = length($s->{lastgood}) if length($s->{lastgood}) > $max{good};
			$s->{lastgoodsecs} = $g->{last_ended_secs};

			$s->{is_overdue} = ($g->{last_ended_secs} > $s->{overdue_secs}) ? 'yes' : 'no';
			$s->{is_expired} = ($g->{last_ended_secs} > $s->{expired_secs}) ? 'yes' : 'no';

		}
		$s->{is_expired} = 'no' if $s->{expired_secs} < 1;
		$s->{is_overdue} = 'no' if $s->{overdue_secs} < 1;

		if (defined $s->{latest_bad}) {
			$s->{timebad} = pretty_time($s->{latest_bad}{total_time_aborted});
			$max{totalbad} = length($s->{timebad}) if length($s->{timebad}) > $max{totalbad};
			$s->{lastbadtime} = $s->{latest_bad}{total_time_aborted};

			$s->{lastbad} = pretty_time($s->{latest_bad}{last_aborted_secs});
			$max{bad} = length($s->{lastbad}) if length($s->{lastbad}) > $max{bad};
			$s->{lastbadsecs} = $s->{latest_bad}{last_aborted_secs};
		}

		## Check for current activity
		$s->{timeq1} = $s->{timeq2} = '';
		for my $row (@{$s->{active_now}}) {
			$s->{timeq} = $s->{timeq1} = pretty_time($row->{q1});
			if (defined $row->{q2}) {
				$s->{timeq} = $s->{timeq2} = pretty_time($row->{q2});
			}
			my $len = length('WAIT:') + length($s->{timeq});
			if (length $s->{timeq2}) {
				$len += (1 + length($s->{targetdb}));
			}
			$max{stat} = $len if $len > $max{stat};
			last;
		}

		$s->{state} = length($s->{timeq2}) ? "RUN:$s->{timeq}>$s->{targetdb}" :
			length($s->{timeq1}) ? "WAIT:$s->{timeq}" :
			($s->{status} eq 'active') ? 'idle' : 'off';

		## Is anything overdue or expired?
		if ($s->{is_expired} eq 'yes') {
			$s->{typetext} .= ' E!'
		}
		elsif ($s->{is_overdue} eq 'yes') {
			$s->{typetext} .= ' O!';
		}
		elsif ($s->{is_expired} ne 'no') {
			$s->{typetext} .= ' ?';
		}

	}

	return $syncs, \%max;

} ## end of get_detailed_syncs


sub list {

	my $msg = "Need to specify what to list: db, goat, herd, or sync\n";
	if (!@nouns) {
		die $msg;
	}
	my $thing = shift @nouns;

	list_dbs()   if $thing =~ /db/i   or $thing eq 'd';
	list_goats() if $thing =~ /goat/i or $thing eq 'g';
	list_herds() if $thing =~ /herd/i or $thing eq 'h';
	list_syncs() if $thing =~ /sync/i or $thing eq 's';

	die $msg;

	exit;

} ## end of list


sub list_dbs {

	## Show information about all or some subset of the 'db' table

	## If any nouns left, treat as wildcards
	my $WHERE = '';
	for my $term (@nouns) {
		$WHERE .= sprintf qq{%s lower(name) ~ %s},
			$WHERE ? ' OR' : 'WHERE',
			$dbh->quote(lc $term);
	}

	$SQL = "SELECT * FROM db $WHERE ORDER BY name";
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute();
	if ($count < 1) {
		$sth->finish();
		printf "There are no%s entries in the 'db' table.\n",
			$WHERE ? ' matching' : '';
		exit;
	}

	$info = $sth->fetchall_arrayref({});
	for my $row (@$info) {
		print "Database: $row->{name}   Status: $row->{status}\n";
		if ($row->{sourcelimit} != 0 or $row->{targetlimit} != 0) {
			print "Limits: source:$row->{sourcelimit} / target:$row->{targetlimit}\n";
		}
		print "Conn: psql -h $row->{dbhost} -p $row->{dbport} -U $row->{dbuser} -d $row->{dbname}\n";
		print "\n";
	}

	exit;

} ## end of list_dbs


sub list_goats {

	## Show information about all or some subset of the 'goat' table
	## Keep to three lines or less if possible

	## If any nouns left, treat as wildcards
	my $WHERE = '';
	for my $term (@nouns) {
		$WHERE .= sprintf qq{%s lower(tablename) ~ %s},
			$WHERE ? ' OR' : 'WHERE',
				$dbh->quote(lc $term);
	}

	$SQL = "SELECT * FROM goat $WHERE ORDER BY schemaname, tablename";
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute();
	if ($count < 1) {
		$sth->finish();
		printf "There are no%s entries in the 'goat' table.\n",
			$WHERE ? ' matching' : '';
		exit;
	}

	$info = $sth->fetchall_arrayref({});
	for my $row (@$info) {
		printf "Table $row->{id}: $row->{schemaname}.$row->{tablename}%s\n",
			$row->{ghost} ? ' GHOST TABLE' : '';
		if ($row->{pkey}) {
			print "  PK: $row->{pkey} ($row->{pkeytype})\n";
		}
		if ($row->{customselect}) {
			print "  Custom select: $row->{customselect}\n";
		}
		printf "  DB: %s | Ping: %s  | Makedelta: %s | Has delta: %s\n",
			$row->{db},
			$row->{ping} ? 'Yes' : 'No',
			$row->{makedelta} ? 'Yes' : 'No',
			$row->{hasdelta} ? 'Yes' : 'No';
	}

	exit;

} ## end of list_goats


sub list_herds {

	## Show information about all or some subset of the 'herd' table

	$SQL = 'SELECT herd, schemaname, tablename FROM herdmap h JOIN goat g ON (g.id = h.goat)';
	$sth = $dbh->prepare($SQL);
	$sth->execute();
	my $map = $sth->fetchall_arrayref({});
	my $herdmap;
	for my $row (@$map) {
		push @{$herdmap->{$row->{herd}}}, "$row->{schemaname}.$row->{tablename}";
	}

	## If any nouns left, treat as wildcards
	my $WHERE = '';
	for my $term (@nouns) {
		$WHERE .= sprintf qq{%s lower(name) ~ %s},
			$WHERE ? ' OR' : 'WHERE',
			$dbh->quote(lc $term);
	}

	$SQL = "SELECT * FROM herd $WHERE ORDER BY name";
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute();
	if ($count < 1) {
		$sth->finish();
		printf "There are no%s entries in the 'herd' table.\n",
			$WHERE ? ' matching' : '';
		exit;
	}

	$info = $sth->fetchall_arrayref({});
	my ($namelen,$goatlen) = (5,1);
	for my $row (@$info) {
		my $num = @{$herdmap->{$row->{name}}};
		$namelen = length $row->{name} if length $row->{name} > $namelen;
		$goatlen = length $num if length $num > $goatlen;
	}
	for my $row (@$info) {
		printf "Herd: %-*s   Goats: %*d\n",
			$namelen, $row->{name},
			$goatlen, scalar @{$herdmap->{$row->{name}}};
	}

	## If requesting (or matching) a single herd, show the goats in detail
	if (@nouns and $count == 1) {
		for my $name (@{$herdmap->{$info->[0]{name}}}) {
			print "  $name\n";
		}
	}

	exit;

} ## end of list_herds


sub list_syncs {

	## Show information about all or some subset of the 'sync' table

	## If any nouns left, treat as wildcards
	my $WHERE = '';
	for my $term (@nouns) {
		$WHERE .= sprintf qq{%s lower(name) ~ %s},
			$WHERE ? ' OR' : 'WHERE',
			$dbh->quote(lc $term);
	}

	$SQL = "SELECT * FROM sync $WHERE ORDER BY name";
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute();
	if ($count < 1) {
		$sth->finish();
		printf "There are no%s entries in the 'sync' table.\n",
			$WHERE ? ' matching' : '';
		exit;
	}

	$info = $sth->fetchall_arrayref({});
	my ($namelen,$goatlen) = (5,1);
	for my $row (@$info) {
		$namelen = length $row->{name} if length $row->{name} > $namelen;
	}
	for my $row (@$info) {
		printf "Sync: %-*s %s %s=>%s %s\n",
			$namelen,
			$row->{name},
			$row->{synctype},
			$row->{source},
			$row->{targetdb} ? $row->{targetdb} : $row->{targetgroup},
			$row->{status} eq 'active' ? '' : "$row->{status}";
	}

	exit;

} ## end of list_syncs


sub pretty_time {

	my $secs = shift;
	return '?' if ! defined $secs or $secs !~ /^\-?\d+$/o or $secs < 0;
	my ($D,$H,$M,$S) = (0,0,0,0);

	if ($bcargs->{showdays}) {
		if ($secs > 60*60*24) {
			$D = int $secs/(60*60*24);
			$secs -= $D*60*60*24;
		}
	}
	if ($secs > 60*60) {
		$H = int $secs/(60*60);
		$secs -= $H*60*60;
	}
	if ($secs > 60) {
		$M = int $secs/60;
		$secs -= $M*60;
	}
	$secs = int $secs;
	my $answer = sprintf "%s%s%s${secs}s",$D ? "${D}d " : '',$H ? "${H}h " : '',$M ? "${M}m " : '';

	## Detailed listings get compressed
	if ((defined $COMPRESS and $COMPRESS) or (!defined $COMPRESS and !@nouns)) {
		$answer =~ s/ //g;
	}
	return $answer;
}


sub get_syncs {

	my %dbgroup;
	$SQL = "SELECT dbgroup, db FROM dbmap ORDER BY priority, db";
	for my $row (@{$dbh->selectall_arrayref($SQL)}) {
		push @{$dbgroup{$row->[0]}}, $row->[1];
	}

	$SQL = qq{
        SELECT *,
            COALESCE(EXTRACT(epoch FROM checktime),0) AS checksecs,
            now()-overdue AS overdue_time,
            now()-expired AS expired_time,
            extract(epoch FROM overdue) AS overdue_secs,
            extract(epoch FROM expired) AS expired_secs
        FROM     bucardo.sync
        ORDER BY priority DESC, name DESC
    };
	$sth = $dbh->prepare($SQL);
	$sth->execute();
	my $sync = $sth->fetchall_hashref("name");

	## Expand any targetgroups in use
	for (keys %$sync) {
		my $s = $sync->{$_};
		if (defined $s->{targetgroup}) {
			$s->{targets} = $dbgroup{$s->{targetgroup}};
		}
	}
	## Check what exists in the pid directory
	opendir my $sdir, $PIDDIR or die qq{Could not opendir "$PIDDIR": $!\n};
	my $pidfile;
	while (defined ($pidfile = readdir($sdir))) {
		next if $pidfile =~ /^\.\.?$/
			or "$PIDDIR/$pidfile" eq $STOPFILE
			or "$PIDDIR/$pidfile" eq $PIDFILE;
		if ($pidfile !~ /^bucardo_sync_(.+)\.pid$/) {
			warn "Skipping unknown file in pid directory: $pidfile\n";
			next;
		}
		my $syncname = $1; ## no critic (ProhibitCaptureWithoutTest)

		## Is this a valid syncname?
		if (! exists $sync->{$syncname}) {
			warn qq{Invalid pid file found: $PIDDIR/$pidfile - removing it\n};
			unlink "$PIDDIR/$pidfile";
			next;
		}

		my $cdate = localtime ($^T - (-C "$PIDDIR/$pidfile")*24*60*60);
		$sync->{$syncname}{PIDFILE} = "$PIDDIR/$pidfile";
		$sync->{$syncname}{CREATED} = $cdate;

		## Does it contain a pid?
		open my $fh, '<', "$PIDDIR/$pidfile" or die qq{Could not open "$PIDDIR/$pidfile": $!\n};
		my $pid = <$fh>;
		chomp $pid;
		if (! defined $pid) { $pid = ''; }
		close $fh or warn qq{Could not close $PIDDIR/$pidfile: $!\n};
		if ($pid !~ /^\d+$/) {
			$sync->{$syncname}{NOPID} = 1;
		}
		else {
			$sync->{$syncname}{PID} = $pid;
			$sync->{$syncname}{PIDPING} = kill 0, $pid;
		}
	}
	return $sync;

} ## end of get_syncs


sub vate_sync {

	my $name = lc $verb;
	my $ucname = ucfirst $name;
	@nouns or die qq{${name}_sync requires at least one sync name\n};

	my $wait = (defined $adverb and $adverb eq '0') ? 1 : 0;
	for my $sync (@syncs) {
		$QUIET or print qq{$ucname sync $sync};
		my $done = "bucardo_${name}d_sync_$sync";
		$dbh->do(qq{NOTIFY "bucardo_${name}_sync_$sync"});
		if ($wait) {
			print "...";
			$dbh->do(qq{LISTEN "$done"});
		}
		$dbh->commit();
		if (!$wait) {
			print "\n";
			next;
		}
		sleep 0.1;
	  WAITIN: {
			while (my $notify = $dbh->func('pg_notifies')) {
				my ($name, $pid) = @$notify;
				last WAITIN if $name eq $done;
			}
			$dbh->commit();
			sleep($WAITSLEEP);
			redo;
		}
		print "OK\n";
	} ## end each sync
	exit 0;
}

sub add_item {

	my $self = shift;

	if (!@nouns) {
		warn "Usage: add <item_type> <item_name>\n";
		exit 1;
	}

	## First word must be a type we know about
	my $type = shift @nouns;
	$type = lc $type;

	if ($type eq 'db' or $type eq 'database') {
		add_database($type);
	}
	elsif ($type eq 'table' or $type eq 'goat') {
		add_table($type);
	}
	elsif ($type eq 'sync') {
		add_sync($type);
	}
	elsif ($type eq 'herd') {
		add_herd($type);
	}
	elsif ($type eq 'dbgroup') {
		add_dbgroup($type);
	}
	else {
		warn "Cannot add: unknown type\n";
	}
	exit 1;

} ## end of add_item

sub alter_item {

	my $self = shift;

	if (!@nouns) {
		warn "Usage: alter <item_type> <item_name>\n";
		exit 1;
	}

	## First word must be a type we know about
	my $type = shift @nouns;
	$type = lc $type;

	if ($type eq 'sync') {
		alter_sync($type);
	}
	else {
		warn "Cannot alter: unknown type\n";
	}
	exit 1;

} ## end of alter_item

sub remove_item {

	my $self = shift;

	if (!@nouns) {
		warn "Usage: remove <item_type> <item_name>\n";
		exit 1;
	}

	## First word must be a type we know about
	my $type = shift @nouns;
	$type = lc $type;

	if ($type eq 'db' or $type eq 'database') {
		remove_database($type);
	}
	elsif ($type eq 'table' or $type eq 'goat') {
		remove_table($type);
	}
	elsif ($type eq 'sync') {
		remove_sync($type);
	}
	elsif ($type eq 'herd') {
		remove_herd($type);
	}
	else {
		warn "Cannot add: unknown type\n";
	}
	exit 1;

} ## end of remove_item

sub clog {

	my $msg = shift;
	chomp $msg;

	warn "$msg\n";

} ## end of clog

sub schema_exists {

	my ($schema) = @_;
	my $SQL = 'SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = ?';
	my $sth = $dbh->prepare_cached($SQL);
	my $count = $sth->execute($schema);
	$sth->finish();
	return $count < 1 ? 0 : 1;

} ## end of schema_exists


sub relation_exists {

	## Checks if a relation exists. Returns the oid or 0
	my ($schema,$name) = @_;
	my $SQL = 'SELECT c.oid FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n '.
		'WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ?';
	my $sth = $dbh->prepare_cached($SQL);
	my $count = $sth->execute($schema,$name);
	if ($count == 1) {
		return $sth->fetchall_arrayref()->[0][0];
	}
	$sth->finish();
	return 0;

} ## end of relation_exists

sub constraint_exists {

	my ($schema,$table,$constraint) = @_;
	my $SQL = 'SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, pg_catalog.pg_constraint o '.
		'WHERE n.oid=c.relnamespace AND c.oid=o.conrelid AND n.nspname = ? AND c.relname = ? AND o.conname = ?';
	my $sth = $dbh->prepare_cached($SQL);
	my $count = $sth->execute($schema,$table,$constraint);
	$sth->finish();
	return $count < 1 ? 0 : 1;

} ## end of constraint_exists

sub column_exists {

	my ($schema,$table,$column) = @_;
	my $SQL = 'SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, '.
		'pg_catalog.pg_attribute a WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ? '.
		'AND a.attname = ? AND a.attrelid = c.oid';
	my $sth = $dbh->prepare_cached($SQL);
	my $count = $sth->execute($schema,$table,$column);
	$sth->finish();
	return $count < 1 ? 0 : 1;

} ## end of column_exists

sub trigger_exists {

	my $name = shift;
	my $SQL = 'SELECT 1 FROM pg_catalog.pg_trigger WHERE tgname = ?';
	my $sth = $dbh->prepare_cached($SQL);
	my $count = $sth->execute($name);
	$sth->finish();
	return $count < 1 ? 0 : 1;

} ## end of trigger_exists

sub column_default {

	my ($schema,$table,$column) = @_;
	my $SQL = 'SELECT pg_get_expr(adbin,adrelid) FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, '.
		'pg_catalog.pg_attribute a, pg_attrdef d '.
		'WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ? '.
		'AND a.attname = ? AND a.attrelid = c.oid AND d.adnum = a.attnum AND d.adrelid = a.attrelid';
	my $sth = $dbh->prepare_cached($SQL);
	my $count = $sth->execute($schema,$table,$column);
	if ($count eq '0E0') {
		$sth->finish();
		return '';
	}
	return $sth->fetchall_arrayref()->[0][0];

} ## end of column_default

sub column_value {

	my ($schema,$table,$column,$where) = @_;
	my $SQL = "SELECT $column FROM $schema.$table WHERE $where";
	return $dbh->selectall_arrayref($SQL)->[0][0];

} ## end of column_value

sub table_definition {

	## Pull the complete table definition from the bucardo.schema file
	## Returns an arrayref of sequences, and the textual table def

	my $name = shift;

	my $def = '';

	my $file = 'bucardo.schema';
	open my $fh, '<', $file or die qq{Could not open "$file": $!\n};
	my @seq;
	while (<$fh>) {
		if (!$def) {
			if (/^CREATE TABLE $name/) {
				$def .= $_;
			}
		}
		else {
			$def .= $_;
			last if /^\);/;
		}
	}
	close $fh or die qq{Could not close "$file": $!\n};
	while ($def =~ /nextval\('(.+?)'/g) {
		push @seq => $1;
	}

	return \@seq, $def;

} ## end of table_definition


sub upgrade_and_log {

	my $action = shift;
	my $short = shift || $action;

	eval {
		$dbh->do($action);
	};
	if ($@) {
		my $line = (caller)[2];
		die "From line $line, action $action\n$@\n";
	}

	$SQL = "INSERT INTO upgrade_log(action,version,summary) VALUES (?,?,?)";
	$sth = $dbh->prepare($SQL);
	$sth->execute($action,$VERSION,$short);

} ## end of upgrade_and_log


sub function_exists {

	my ($schema,$name,$args) = @_;
	$name = lc $name;
	$SQL = "SELECT md5(prosrc) FROM pg_proc p, pg_language l ".
		"WHERE p.prolang = l.oid AND proname = ? AND oidvectortypes(proargtypes) = ?";
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute($name,$args);
	if ($count eq '0E0') {
		$sth->finish();
		return '';
	}
	return $sth->fetchall_arrayref()->[0][0];

} ## end of function_exists


sub upgrade {

	## Make upgrades to an existing schema to match the current version
	## This is generally not reversible, so we throw plenty of warnings

	my $self = shift;

	my $schema = 'bucardo';

	$dbh->do("SET escape_string_warning = 'OFF'");

	my $changes = 0;

	## Make sure the upgrade_log table is in place

	if (!relation_exists($schema, 'upgrade_log')) {
		my ($seqlist, $tabledef) = table_definition('upgrade_log');
		upgrade_and_log($tabledef,"CREATE TABLE upgrade_log");
		$dbh->commit();
	}

	my @new_sequences = (
		'audit_pid_id_seq',
	);

	my @old_sequences = (
		'dbgroup_id_seq',
	);

	my @new_tables = (
		['bucardo', 'bucardo_log_message'],
		['bucardo', 'bucardo_rate'],
	);

	my @new_columns = (
		['bucardo', 'audit_pid', 'id',              q{INTEGER NOT NULL DEFAULT nextval('audit_pid_id_seq')}],
		['bucardo', 'audit_pid', 'parentid',        q{INTEGER     NULL}],
		['bucardo', 'audit_pid', 'familyid',        q{INTEGER     NULL}],
		['bucardo', 'audit_pid', 'source',          q{TEXT        NULL}],
		['bucardo', 'audit_pid', 'target',          q{TEXT        NULL}],
		['bucardo', 'audit_pid', 'master_backend',  q{INT     NOT NULL DEFAULT pg_backend_pid()}],
		['bucardo', 'audit_pid', 'source_backend',  q{INT         NULL}],
		['bucardo', 'audit_pid', 'target_backend',  q{INT         NULL}],
		['bucardo', 'db',        'dbservice',       q{TEXT        NULL}],
		['bucardo', 'goat',      'qpkey',           q{TEXT        NULL}],
		['bucardo', 'sync',      'strict_checking', q{BOOLEAN NOT NULL DEFAULT 'true'}],
		['bucardo', 'sync',      'track_rates',     q{BOOLEAN NOT NULL DEFAULT 'false'}],

		['freezer', 'old_audit_pid', 'id',              q{INTEGER}],
		['freezer', 'old_audit_pid', 'parentid',        q{INTEGER}],
		['freezer', 'old_audit_pid', 'familyid',        q{INTEGER}],
		['freezer', 'old_audit_pid', 'source',          q{TEXT}   ],
		['freezer', 'old_audit_pid', 'target',          q{TEXT}   ],
		['freezer', 'old_audit_pid', 'master_backend',  q{INT}    ],
		['freezer', 'old_audit_pid', 'source_backend',  q{INT}    ],
		['freezer', 'old_audit_pid', 'target_backend',  q{INT}    ],
	);

	my @altered_columns = (
		['bucardo', 'goat', 'schemaname', 'NO DEFAULT'],
	);

	my @old_columns = (
		['bucardo', 'sync', 'disable_rules'],
		['bucardo', 'sync', 'disable_triggers'],
	);

	my @row_values = (

		['bucardo_config','about',q{setting = 'log_showtime'}, 1,
		 'Show timestamp in the log output?  0=off  1=seconds since epoch  2=scalar gmtime  3=scalar localtime'],
	);

	my @old_constraints = (
		['bucardo', 'goat', 'goat_pkeytype_check'],
		['bucardo', 'sync', 'sync_replica_allornone'],
		['bucardo', 'sync', 'sync_disable_triggers_method'],
		['bucardo', 'sync', 'sync_disable_rules_method'],
	);

	my @new_constraints = (
	);

	my @old_functions = (
		['create_child_q', 'text'],
	);

	my @old_indexes = (
        ['bucardo', 'sync', 'sync_source_targetdb'],
        ['bucardo', 'sync', 'sync_source_targetgroup'],
	);

	my @new_indexes = (
        ['bucardo', 'sync', 'sync_source_targetdb_type', 'UNIQUE source,targetdb,synctype'],
        ['bucardo', 'sync', 'sync_source_targetgroup_type', 'UNIQUE source,targetgroup,synctype'],
	);

	my @drop_all_rules = (
		['freezer','master_q'],
	);

	## Drop all existing rules from this table:
	for my $row (@drop_all_rules) {
		my ($schema,$table) = @$row;
		print "Checking $schema and $table\n";
		my $oid = relation_exists($schema,$table);
		if (!$oid) {
			warn "Could not find table $schema.$table to check!\n";
			next;
		}
		$SQL = "SELECT rulename FROM pg_rewrite WHERE ev_class = ? ORDER BY rulename";
		$sth = $dbh->prepare($SQL);
		$count = $sth->execute($oid);
		if ($count < 1) {
			$sth->finish();
			next;
		}
		for my $rule (map { $_->[0] } @{$sth->fetchall_arrayref()}) {
			upgrade_and_log(qq{DROP RULE "$rule" ON $schema.$table});
			clog "Dropped rule $rule on table $schema.$table";
			$changes++;
		}
	}

	for my $con (@old_constraints) {
		my ($schema, $table, $constraint) = @$con;
		if (!constraint_exists($schema, $table, $constraint)) {
			clog "  Constraint $constraint does not exist";
		}
		else {
			upgrade_and_log(qq{ALTER TABLE $schema.$table DROP CONSTRAINT "$constraint"});
			clog "Dropped constraint $constraint ON $schema.$table";
			$changes++;
		}
	}

	for my $sequence (@new_sequences) {
		if (relation_exists($schema, $sequence)) {
			clog "  Sequence already exists: $sequence";
		}
		else {
			upgrade_and_log("CREATE SEQUENCE $schema.$sequence");
			clog "Created sequence: $sequence";
			$changes++;
		}
	}

	for my $sequence (@old_sequences) {
		if (!relation_exists($schema, $sequence)) {
			clog "  Sequence does not exist: $sequence";
		}
		else {

			upgrade_and_log("DROP SEQUENCE $schema.$sequence");
			clog "Dropped sequence: $sequence";
			$changes++;
		}
	}

	for my $row (@new_tables) {
		my ($schema,$table) = @$row;
		if (relation_exists($schema, $table)) {
			clog "  Table already exists: $table\n";
		}
		else {
			my ($seqlist, $tabledef) = table_definition($table);
			for my $sequence (@$seqlist) {
				if (!relation_exists($schema, $sequence)) {
					upgrade_and_log("CREATE SEQUENCE $schema.$sequence");
					clog "Created sequence: $sequence";
				}
			}
			upgrade_and_log($tabledef, "CREATE TABLE $table");
			clog "Created table: $table\n";
			$changes++;
		}
	}

	for my $row (@new_columns) {
		my ($schema,$table,$column,$def) = @$row;
		if (column_exists($schema, $table, $column)) {
			clog "  Column already exists: $schema.$table.$column";
		}
		else {
			$def =~ s/\s+/ /g;
			$dbh->do("ALTER TABLE $schema.$table ADD COLUMN $column $def");
			clog "Created column: $schema.$table.$column $def";
			$changes++;
		}
	}

	for my $row (@altered_columns) {
		my ($schema,$table,$column,$change) = @$row;
		next if ! column_exists($schema, $table, $column);
		if ($change eq 'NO DEFAULT') {
			my $def = column_default($schema, $table, $column);
			if (!$def) {
				clog "  Column $schema.$table.$column already has no DEFAULT";
			}
			else {
				upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column DROP DEFAULT");
				clog "Removed DEFAULT ($def) from $schema.$table.$column";
				$changes++;
			}
		}
		elsif ($change =~ /^RENAME\s+(\w+)/) {
			my $newname = $1;
			upgrade_and_log("ALTER TABLE $schema.$table RENAME COLUMN $column TO $newname");
			clog("Renamed $schema.$table.$column to $newname");
			$changes++;
		}
		elsif ($change =~ /^DEFAULT\s+(.+)/) {
			my $newname = $1;
			my $oldname = column_default($schema, $table, $column);
			if ($newname eq $oldname) {
				clog "  Column $schema.$table.$column already has a DEFAULT of $newname";
			}
			else {
				upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column SET DEFAULT $newname");
				clog("Changed DEFAULT on $schema.$table.$column to $newname");
				$changes++;
			}
		}
		else {
			die qq{Do not know how to handle altered column spec of "$change"};
		}
	}

	for my $row (@old_columns) {
		my ($schema,$table,$column) = @$row;
		if (!column_exists($schema, $table, $column)) {
			clog "  Column already removed: $schema.$table.$column";
		}
		else {
			$dbh->do("ALTER TABLE $schema.$table DROP COLUMN $column");
			clog "Dropped column: $schema.$table.$column";
			$changes++;
		}
	}

	for my $row (@new_constraints) {
		my ($schema,$table,$name,$def) = @$row;
		if (constraint_exists($schema, $table, $name)) {
			clog "  Contraint $name on $schema.$table already exists";
		}
		else {
			$dbh->do("ALTER TABLE $schema.$table ADD CONSTRAINT $name $def");
			clog "Added constraint $name to $schema.$table";
			$changes++;
		}
	}

	for my $row (@old_indexes) {
		my ($schema,$table,$name) = @$row;
		if (!relation_exists($schema, $name)) {
			clog "  Index already removed: $schema.$table.$name";
		}
		else {
			$dbh->do("DROP INDEX $name");
			clog "Dropped index $name";
			$changes++;
		}
	}

	for my $row (@new_indexes) {
		my ($schema,$table,$name,$def) = @$row;
		if (relation_exists($schema, $name)) {
			clog "  Index already exists: $schema.$table.$name";
		}
		else {
			my $unique = $def =~ s/^UNIQUE\s*// ? 1 : 0;
			my $COM = sprintf 'CREATE %sINDEX %s ON %s.%s (%s)',
				$unique ? 'UNIQUE ' : '',
                $name,
				$schema,
                $table,
                $def;
			$dbh->do($COM);
			clog "Created index $name on $name";
			$changes++;
		}
	}


	## Check for any new config items
	$SQL = "SELECT value FROM bucardo_config WHERE lower(setting) = ?";
	my $cfgsth = $dbh->prepare($SQL);
	$SQL = 'INSERT INTO bucardo_config(setting,value,about) VALUES (?,?,?)';
	my $newcfg = $dbh->prepare($SQL);
	my $file = 'bucardo.schema';
	open my $fh, '<', $file or die qq{Could not open "$file": $!\n};

	my %config;
	my $inside = 0;
	while (<$fh>) {
		if (!$inside) {
			if (/^WITH DELIMITER/) {
				$inside = 1;
			}
			next;
		}
		if (/^\\/) {
			$inside = 0;
			next;
		}
		## Scoop
		my ($setting,$value,$about) = split /\|/ => $_;
		$config{$setting} = [$value,$about];
		$count = $cfgsth->execute($setting);
		$cfgsth->finish();
		if ($count eq '0E0') {
			clog "Added new bucardo_config setting: $setting";
			$changes++;
			$newcfg->execute($setting,$value,$about);
		}
	}
	close $fh or die qq{Could not close file "$file": $!\n};

	for my $row (@row_values) {
		my ($table,$column,$where,$force,$value) = @$row;
		my $val = column_value($schema,$table,$column,$where);
		if (!defined $val) {
			die "Failed to find $table.$column where $where!\n";
		}
		if ($val eq $value) {
			clog "  Value of $table.$column where $where is already correct";
		}
		else {
			$SQL = sprintf "UPDATE $schema.$table SET $column=%s WHERE $where",
				$dbh->quote($value);
			upgrade_and_log($SQL);
			clog "New value set for $schema.$table.$column WHERE $where";
			$changes++;
		}
	}

	for my $row (@old_functions) {
		my ($name, $args) = @$row;
		if (function_exists('bucardo', $name, $args)) {
			clog "Dropping deprecated function $name($args)";
			upgrade_and_log(qq{DROP FUNCTION bucardo."$name"($args)});
			$changes++;
		}
		else {
			clog "  Deprecated function $name($args) does not exist";
		}
	}

	## Now we check all the functions, triggers, and indexes within bucardo.schema
	$file = 'bucardo.schema';
	open $fh, '<', $file or die qq{Could not open "$file": $!\n};
	my (@flist, @tlist, @ilist);
	my ($fname,$args,$fbody) = ('','','');
	my ($tname,$tbody) = ('','');
	while (<$fh>) {
		if ($fbody) {
			if (/^(\$bc\$;)/) {
				$fbody .= $1;
				push @flist, [$fname, $args, $fbody];
				$fbody = $fname = $args = '';
			}
			else {
				$fbody .= $_;
			}
			next;
		}
		if ($tbody) {
			$tbody .= $_;
			if (/;/) {
				push @tlist, [$tname, $tbody];
				$tbody = $tname = '';
			}
			next;
		}
		if (/^CREATE (?:OR REPLACE )?FUNCTION\s+(\S+)/) {
			$fname = $1;
			$fbody .= $_;
			$fname =~ s/bucardo\.//;
			$fname =~ s/\((.*)\)// or die "No args found for function: $_\n";
			$args = $1;
			$args =~ s/,(\S)/, $1/g;
			next;
		}
		if (/^CREATE TRIGGER (\S+)/) {
			$tname = $1;
			$tbody .= $_;
			next;
		}
		if (/^CREATE INDEX (\S+)/) {
			push @ilist, [$1, $_];
			next;
		}
	}
	close $fh or die qq{Could not close file "$file": $!\n};

	$SQL = "SELECT md5(?)";
	my $md5sth = $dbh->prepare($SQL);
	for my $row (@flist) {
		my ($name,$args,$body) = @$row;
		my $oldbody = function_exists($schema,$name,$args);
		if (!$oldbody) {
			$body =~ s/FUNCTION /FUNCTION $schema./;
			upgrade_and_log($body,"CREATE FUNCTION $schema.$name($args)");
			clog "Added function $schema.$name($args)";
			$changes++;
			next;
		}
		my $realbody = $body;
		$realbody =~ s/.*?\$bc\$(.+)\$bc\$;/$1/sm;
		$md5sth->execute($realbody);
		my $newbody = $md5sth->fetchall_arrayref()->[0][0];
		if ($oldbody eq $newbody) {
			clog "  Function unchanged: $schema.$name($args)";
			next;
		}
		$body =~ s/^CREATE FUNCTION/CREATE OR REPLACE FUNCTION/;
		$body =~ s/FUNCTION /FUNCTION $schema./;
		(my $short = $body) =~ s/^(.+?)\n.*/$1/s;
		$dbh->do("SAVEPOINT bucardo_upgrade");
		eval { upgrade_and_log($body,$short); };
		if ($@) {
			$dbh->do("ROLLBACK TO bucardo_upgrade");
			(my $dropbody = $short) =~ s/CREATE OR REPLACE/DROP/;
			$dropbody .= ' CASCADE';
			upgrade_and_log($dropbody);
			upgrade_and_log($body,$short);
		}
		else {
			$dbh->do("RELEASE bucardo_upgrade");
		}
		clog "Updated function: $schema.$name($args)";
		$changes++;
	}

	## Check for any added triggers
	for my $row (@tlist) {
		my ($name,$body) = @$row;
		if (trigger_exists($name)) {
			clog "  Trigger $name already exists";
		}
		else {
			upgrade_and_log($body);
			clog "Created trigger $name";
			$changes++;
		}
	}

	## Check for any added indexes
	for my $row (@ilist) {
		my ($name,$body) = @$row;
		if (relation_exists('bucardo',$name)) {
			clog "  Index $name already exists";
		}
		else {
			warn "DOES NOT EXIST? $name";
			upgrade_and_log($body);
			clog "Created index $name";
			$changes++;
		}
	}

	## The freezer.q_staging table is no longer needed, but we must empty it before dropping
	if (relation_exists('freezer','q_staging')) {
		upgrade_and_log("INSERT INTO freezer.master_q SELECT * FROM freezer.q_staging");
		upgrade_and_log("DROP TABLE freezer.q_staging");
		clog 'Dropped deprecated table freezer.q_staging';
		$changes++;
	}

	if ($changes) {
		printf "Okay to commit $changes %s? ", $changes==1 ? 'change' : 'changes';
		exit if <STDIN> !~ /Y/i;
		$dbh->commit();
		print "Changes have been commited\n";
	}

	exit 1;

} ## end of upgrade


sub add_database {

	my $type = shift;

	## Usage: add database <dbname> name=internal_name port=xxx host=xxx user=xxx pass=xxx
	## service=xxx conn=xxx sourcelimit=xxx targetlimit=xxx

	## The first word is the name of the database we connect to (column dbname)
	my $item_name = shift @nouns || '';
	if (!length $item_name) {
		warn "Usage: add $type <name>";
		exit 1;
	}

	## The rest is the options. If not there, we fill in the defaults
	my $info = join ' ' => @nouns;
	my $delim = $info =~ /\|/ ? "\\s*\\|\\s*" : '\\s+';
	## Command line name, default value, database column name
	my %arg = (
			   name        => [$item_name, ''],
			   user        => ['bucardo',  'dbuser'],
			   port        => ['',         'dbport'],
			   host        => ['',         'dbhost'],
			   pass        => ['',         'dbpass'],
			   status      => ['',         ''],
			   conn        => ['',         'dbconn'],
			   service     => ['',         'dbservice'],
			   sourcelimit => ['',         ''],
			   targetlimit => ['',         ''],
			   );
	for my $arg (split /$delim/ => $info) {
		if ($arg !~ /(.+)=(.+)/o) {
			die qq{Invalid argument\n};
		}
		my ($name,$value) = ($1,$2);
		if (!exists $arg{$name}) {
			die "Unknown option '$name'\n";
		}
		$arg{$name}[0] = $value;
	}
	$arg{dbname} = [$item_name,''];

	## Lots of checks go here...

	my ($cols,$phs) = ('','');
	my @vals;
	for my $col (sort keys %arg) {
		next if ! length $arg{$col}[0];
		my $name = $arg{$col}[1] || $col;
		$cols .= "$name,";
		$phs .= '?,';
		push @vals, $arg{$col}[0];
	}
	$cols =~ s/,$//;
	$phs =~ s/,$//;

	## Add this to the database
	$SQL = "INSERT INTO bucardo.db ($cols) VALUES ($phs)";
	$DEBUG and warn "SQL: $SQL\n";
	$DEBUG and warn Dumper \@vals;
	$sth = $dbh->prepare($SQL);
	eval {
		$count = $sth->execute(@vals);
	};
	if ($@) {
		die "Fail: $@\n";
	}
	## Prompt for final go-ahead

	$dbh->commit();

	warn "Database added: $arg{name}[0]\n";
	exit 0;

} ## end of add_database


sub add_table {

	my $type = shift;

	## Usage: add table [schema].table db=internal_db_name ping=bool standard_conflict=xxx makedelta=bool

	## TODO: Make this smarter
	my $DEFAULT_SCHEMA = 'public';

	## Any single words are table names.
	my (@newnouns,@tables);
	for (@nouns) {
		if (/=/) {
			push @newnouns => $_;
		}
		else {
			if (/^(\w*?)?\.?(\w+)$/) {
				push @tables => length $1 ? [$1,$2] : [$DEFAULT_SCHEMA,$2];
			}
			else {
				warn "Invalid table name: $_\n";
				exit 1;
			}
		}
	}
	@nouns = @newnouns;

	if (! @tables) {
		warn "Usage: add $type <name>";
		exit 1;
	}

	## The rest is the options. If not there, we fill in the defaults
	my $info = join ' ' => @nouns;
	my $delim = $info =~ /\|/ ? "\\s*\\|\\s*" : '\\s+';
	## Command line name, default value, database column name
	my %arg = (
			   db                => ['',      ''],
			   ping              => ['',      ''],
			   standard_conflict => ['',      ''],
			   herd              => ['',      ''],
			   makedelta         => ['',      ''],
			   );
	for my $arg (split /$delim/ => $info) {
		if ($arg !~ /(.+)=(.+)/o) {
			die qq{Invalid argument: ($arg)\n};
		}
		my ($name,$value) = ($1,$2);
		if (!exists $arg{$name}) {
			die "Unknown option '$name'\n";
		}
		$arg{$name}[0] = $value;
	}

	if (! length $arg{db}[0]) {
		die "Please specify a database with db=<name>\n";
	}

	## Lots of checks go here...

	my $herd;
	if ($arg{herd}->[0]) {
		$herd = $arg{herd}->[0];
		delete $arg{herd};
		## Does this herd exist?
		$SQL = 'SELECT count(*) FROM herd WHERE name = ?';
		$sth = $dbh->prepare($SQL);
		$count = $sth->execute($herd);
		$sth->finish();
		if ($count != 1) {
			die "Cannot add to non-existent herd: $herd\n";
		}
	}

	my ($cols,$phs) = ('','');
	my @vals;
	for my $col (sort keys %arg) {
		next if ! length $arg{$col}[0];
		my $name = $arg{$col}[1] || $col;
		$cols .= "$name,";
		$phs .= '?,';
		push @vals, $arg{$col}[0] if $arg{$col}[0];
	}
	$cols =~ s/,$//;
	$phs =~ s/,$//;

	## Add these to the database
	$SQL = "INSERT INTO bucardo.goat ($cols,schemaname,tablename) VALUES ($phs,?,?) RETURNING id";
	if ($DEBUG) {
		warn "SQL: $SQL\n";
		warn Dumper \@tables;
	}
	$sth = $dbh->prepare($SQL);
	my @id;
	for (@tables) {
		my ($schema,$table) = @$_;
		eval {
			if ($DEBUG >= 2) {
				warn "SQL: $SQL\n";
				my $args = join ',' => @vals;
				my $ss = defined $schema ? $schema : '<undef>';
				warn "Args: $args, $ss, $table\n";
			}
			$count = $sth->execute(@vals,$schema,$table);
			push @id => $sth->fetch()->[0];
		};
		if ($@) {
			die "Table add failed: $@\n";
		}
		$DEBUG and warn "Insert count: $count\n";
	}

	## Add them to the herd if it was specificed
	if (defined $herd) {
		$SQL = 'INSERT INTO herdmap (goat,herd,priority) VALUES (?,?,?)';
		$sth = $dbh->prepare($SQL);
		for (@id) {
			$sth->execute($_,$herd,0);
		}
	}

	## Prompt for final go-ahead

	$dbh->commit();

	my $tablelist = sprintf "%s added: ", @tables > 1 ? 'Tables' : 'Table';
	for (@tables) {
		$tablelist .= sprintf "%s%s, ", $_->[0] ? "$_->[0]." : '', $_->[1];
	}
	$tablelist =~ s/, $//;
	warn "$tablelist\n";
	exit 0;

} ## end of add_table


sub add_sync {

	my $type = shift;

	## Usage: add sync syncname options

	## The first word is the sync name
	my $item_name = shift @nouns || '';
	if (!length $item_name) {
		warn "Usage: add $type <name>";
		exit 1;
	}

	## The rest is the options. If not there, we fill in the defaults
	my $info = join ' ' => @nouns;
	my $delim = $info =~ /\|/ ? "\\s*\\|\\s*" : '\\s+';
	## Command line name, default value, database column name
	my %arg = (
			   source      => ['', ''],
			   targetdb    => ['', ''],
			   targetgroup => ['', ''],
			   type    => ['', 'synctype'],
			   makedelta   => ['', ''],
			   );
	for my $arg (split /$delim/ => $info) {
		if ($arg !~ /(.+)=(.+)/o) {
			die qq{Invalid argument\n};
		}
		my ($name,$value) = ($1,$2);
		if (!exists $arg{$name}) {
			die "Unknown option '$name'\n";
		}
		$arg{$name}[0] = $value;
	}
	$arg{name} = [$item_name];

	## Lots of checks go here...

	my ($cols,$phs) = ('','');
	my @vals;
	for my $col (sort keys %arg) {
		next if ! length $arg{$col}[0];
		my $name = $arg{$col}[1] || $col;
		$cols .= "$name,";
		$phs .= '?,';
		push @vals, $arg{$col}[0];
	}
	$cols =~ s/,$//;
	$phs =~ s/,$//;

	if ($arg{type}[0] eq 'fullcopy') {
		$cols .= ',stayalive,kidsalive,ping';
		$phs .= q{,'f','f','f'};
	}

	## Add this to the database
	$SQL = "INSERT INTO bucardo.sync ($cols) VALUES ($phs)";
	if ($DEBUG) {
		warn "SQL: $SQL\n";
		warn Dumper \@vals;
	}
	$sth = $dbh->prepare($SQL);
	eval {
		$count = $sth->execute(@vals);
	};
	if ($@) {
		die "Failed to add sync: $@\n";
	}

	$dbh->commit();

	warn "Sync added: $item_name\n";
	exit 0;

} ## end of add_sync


sub alter_sync {

	## Modify an existing sync
	## Usage: alter sync syncname options
	## targetgroup=foo - add/change the targetgroup
	## targetdb=foo    - add/change the targetdb
	## If a target already exists of other type, it is set to null

	my $type = shift;

	## The first word is the sync name
	my $item_name = shift @nouns || '';
	if (!length $item_name) {
		warn "Usage: alter $type <name>";
		exit 1;
	}

	## The rest is the options.
	my $info = join ' ' => @nouns;
	my $delim = $info =~ /\|/ ? "\\s*\\|\\s*" : '\\s+';
	## Command line name, value
	my %arg = (
			   targetdb    => '',
			   targetgroup => '',
			   );
	for my $arg (split /$delim/ => $info) {
		if ($arg !~ /(.+)=(.+)/o) {
			die qq{Invalid argument\n};
		}
		my ($name,$value) = ($1,$2);
		if (!exists $arg{$name}) {
			die "Unknown option '$name'\n";
		}
		$arg{$name} = $value;
	}
	$arg{name} = $item_name;

	## Make sure this sync exists and gather current information from it
	$SQL = 'SELECT * FROM bucardo.sync WHERE name = ?';
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute($item_name);
	if ($count != 1) {
		warn "No such sync: $item_name\n";
		exit 1;
	}
	my $sync = $sth->fetchall_arrayref({})->[0];

	my %set;

	## Do we have a targetgroup?
	if (length $arg{targetgroup}) {
		my $group = $arg{targetgroup};
		if (defined $sync->{targetgroup} and $group eq $sync->{targetgroup}) {
			warn qq{targetgroup "$group" already exists: no change made\n};
		}
		else {
			$set{'targetgroup'} = $group;
			if (defined $sync->{targetdb}) {
				warn qq{Removing targetdb "$sync->{targetdb}"\n};
				$set{'targetdb'} = undef;
			}
		}
	}
	elsif (length $arg{targetdb}) {
		my $db = $arg{targetdb};
		if (defined $sync->{targetdb} and $db eq $sync->{targetdb}) {
			warn qq{targetdb "$db" already exists: no change made\n};
		}
		else {
			$set{'targetdb'} = $db;
			if (defined $sync->{targetgroup}) {
				warn qq{Removing targetgroup "$sync->{targetgroup}"\n};
				$set{'targetgroup'} = undef;
			}
		}
	}

	if (! keys %set) {
		warn "No change made\n";
		exit 0;
	}

	$SQL = 'UPDATE sync SET ';
	my @args;
	for my $c (sort keys %set) {
		$SQL .= "$c = ?, ";
		push @args, $set{$c};
	}
	$SQL =~ s/, $//;
	$SQL .= ' WHERE name = ?';

	$sth = $dbh->prepare($SQL);
	eval {
		$count = $sth->execute(@args,$item_name);
	};

	## Prompt for final go-ahead

	$dbh->commit();

	warn "Sync updated: $item_name ($SQL)\n";

	# warn about not forgetting to reload the sync if active?

	exit 0;

} ## end of alter_sync


sub remove_sync {

	my $type = shift;

	## Usage: remove sync syncname

	## The first word is the sync name
	my $item_name = shift @nouns || '';
	if (!length $item_name) {
		warn "Usage: remove $type <name>";
		exit 1;
	}

	## Does this sync exist?
	$SQL = 'SELECT * FROM bucardo.sync WHERE name = ?';
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute($item_name);
	if ($count != 1) {
		warn "No such sync: $item_name\n"; ## XXX Suggest alternatives?
		exit 1;
	}
	## Is it active?
	## Is it in use?

	## Verify removal
	## Remove vs inactivate
	$SQL = 'DELETE FROM bucardo.sync WHERE name = ?';
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute($item_name);
	if ($count != 1) {
		warn "Could not delete from sync table!\n";
		$dbh->rollback();
		exit 1;
	}
	$dbh->commit();
	warn "Sync removed: $item_name\n";
	exit 0;

} ## end of remove_sync


sub add_herd {

	my $type = shift;

	## Usage: add herd name (goat1, goat2, ...)

	my $item_name = shift @nouns || '';
	if (!length $item_name) {
		warn "Usage: add $type <name> goats";
		exit 1;
	}

	## The rest is the options. If not there, we fill in the defaults
	my $info = join ' ' => @nouns;
	my $delim = $info =~ /\|/ ? "\\s*\\|\\s*" : '\\s+';
	## Command line name, default value, database column name
	my %arg = (
			   );

	my @goatlist;
	for my $arg (split /$delim/ => $info) {
		if ($arg !~ /(.+)=(.+)/o) {
			push @goatlist, $arg;
			next;
		}
		my ($name,$value) = ($1,$2);
		if (!exists $arg{$name}) {
			die "Unknown option '$name'\n";
		}
		$arg{$name}[0] = $value;
	}
	$arg{name} = [$item_name];

	## Lots of checks go here...

	my ($cols,$phs) = ('','');
	my @vals;
	for my $col (sort keys %arg) {
		next if ! length $arg{$col}[0];
		my $name = $arg{$col}[1] || $col;
		$cols .= "$name,";
		$phs .= '?,';
		push @vals, $arg{$col}[0];
	}
	$cols =~ s/,$//;
	$phs =~ s/,$//;

	## Add this to the database
	$SQL = "INSERT INTO bucardo.herd ($cols) VALUES ($phs)";
	$sth = $dbh->prepare($SQL);
	eval {
		$count = $sth->execute(@vals);
	};
	if ($@) {
		die "Failed to add herd: $@\n";
	}
	warn "Herd added: $item_name\n";
	if (@goatlist) {
		$SQL = "INSERT INTO bucardo.herdmap(herd,goat) VALUES (?,(SELECT id FROM bucardo.goat WHERE tablename=?))";
		$sth = $dbh->prepare($SQL);
		for my $goat (@goatlist) {
			eval {
				$sth->execute($item_name, $goat);
			};
			if ($@) {
				die "Failed to add goat to herdmap: $@\n";
			}
			print qq{Added goat "$goat" to new herd\n};
		}
	}
	$dbh->commit();

	exit 0;

} ## end of add_herd


sub add_dbgroup {

	my $type = shift;

	## Usage: add dbgroup name db1 db2 db3 ...

	my $name = shift @nouns || '';
	if (!length $name) {
		warn "Usage: add $type <name>";
		exit 1;
	}

	## Create the group if it does not exist
	$SQL = 'SELECT 1 FROM bucardo.dbgroup WHERE name = ?';
	$sth = $dbh->prepare($SQL);
	$count = $sth->execute($name);
	$sth->finish();
	if ($count != 1) {
		$SQL = 'INSERT INTO dbgroup(name) VALUES (?)';
		$sth = $dbh->prepare($SQL);
		$sth->execute($name);
	}

	## Add any dbs to this group, if they are not already there
	if (@nouns) {
		$SQL = 'SELECT 1 FROM bucardo.db WHERE name = ?';
		my $gotdb = $dbh->prepare($SQL);
		$SQL = 'SELECT priority FROM dbmap WHERE db=? AND dbgroup=?';
		my $getpri = $dbh->prepare($SQL);
		$SQL = 'INSERT INTO dbmap(db,dbgroup,priority) VALUES (?,?,?)';
		my $addrow = $dbh->prepare($SQL);
		$SQL = 'UPDATE dbmap SET priority = ? WHERE db=? AND dbgroup=?';
		my $changepri = $dbh->prepare($SQL);
		for my $db (@nouns) {
			my $newpri  = 1;
			if ($db =~ s/\=(\d+)//) { $newpri = $1; }
			$count = $gotdb->execute($db);
			$gotdb->finish();
			if ($count != 1) {
				warn "No such database: $db\n";
				exit 1;
			}
			$count = $getpri->execute($db,$name);
			if ($count == 1) {
				my $oldpri = $getpri->fetchall_arrayref()->[0][0];
				if ($oldpri != $newpri) {
					$changepri->execute($newpri,$db,$name);
				}
			}
			else {
				## Does not exist
				$addrow->execute($db,$name,$newpri);
			}
		}
	}
	$dbh->commit();

	warn "Group updated\n";
	exit 0;

} ## end of add_dbgroup


sub config {

	## View or change a value inside the bucardo_config table

	my $setusage = "Usage: $0 set setting=value [setting=value ...]\n";

	if (!@nouns) {
		$verb eq 'set' and die $setusage;
		die "Usage: $0 show <all|setting1> [settting2 ...]\n";
	}

	$SQL = 'SELECT * FROM bucardo.bucardo_config';
	$sth = $dbh->prepare($SQL);
	$sth->execute();
	my $config = $sth->fetchall_hashref('setting');
	if ($verb eq 'show') {
		my $all = $nouns[0] =~ /\ball\b/i ? 1 : 0;
		my $maxsize = 3;
		for my $s (keys %$config) {
			next if ! $all and ! grep { $s =~ /$_/i } @nouns;
			$maxsize = length $s if length $s > $maxsize;
		}
		for my $s (sort keys %$config) {
			next if ! $all and ! grep { $s =~ /$_/i } @nouns;
			printf "%-*s = %s\n", $maxsize, $s, $config->{$s}{value};
		}
		exit;
	}

	$SQL = 'UPDATE bucardo.bucardo_config SET value = ? WHERE setting = ?';
	$sth = $dbh->prepare($SQL);

	for my $noun (@nouns) {
		$noun =~ /(\w+)=(.+)/ or die $setusage;
		my ($set,$val) = (lc $1,$2);

		if (! exists $config->{$set}) {
			die qq{Unknown setting "$set"\n};
		}

		$sth->execute($val,$set);
		print qq{Set "$set" to "$val"\n};

	}

	$dbh->commit();

	exit 0;

} ## end of config


sub message {

	## Add a message to the Bucardo logs, via the bucardo_log_message table
	## Note: If no MCP processes are listening, the message will hang out until an MCP processes it

	if (! length($nouns)) {
		die qq{Usage: bucardo_ctl message "Some message to send to the logs"\n};
	}

	$SQL = "INSERT INTO bucardo_log_message(msg) VALUES (?)";
	$sth = $dbh->prepare($SQL);
	$sth->execute($nouns);
	$dbh->commit();
	$VERBOSE and print "Message added\n";
	exit;

} ## end of message


__END__

=head1 NAME

bucardo_ctl - utility script for controlling the Bucardo program

=head1 VERSION

This documents describes bucardo_ctl version 3.2.7

=head1 SYNOPSIS

  ./bucardo_ctl start "Starting up - Greg"

  ./bucardo_ctl stop "Bringing down for debugging - Raul E."

  ./bucardo_ctl list dbs

  ./bucardo_ctl add sync testsync source=herd1 type=pushdelta targetdb=B

  ./bucardo_ctl add database newdb name=internal_name port=5432 host=myserver user=postgres

  ./bucardo_ctl add table some_schema.some_table db=internal_db_name ping=bool standard_conflict=source

  ./bucardo_ctl add herd newherd

  ./bucardo_ctl add dbgroup name db1 db2 db3 ...

  ./bucardo_ctl alter sync syncname targetgroup=foo targetdb=foo

  ./bucardo_ctl ping

  ./bucardo_ctl status

  ./bucardo_ctl status sync1 sync2

  ./bucardo_ctl kick sync1 sync2

  ./bucardo_ctl kick sync1 0 --retry=10

  ./bucardoctl reload_config

=head1 DESCRIPTION

The bucardo_ctl script is used to control the operation of a Bucardo process. Primarily it is 
used for stopping and starting Bucardo, viewing current information, and for kicking off named 
syncs. The script needs to know how to connect to the Bucardo database: this information can 
be passed as arguments or hard-coded into the script itself. If you have multiple Bucardos 
running, it is recommended that you create aliases for each database.

=head1 COMMANDS

=over 4

=item B<start>

Usage: ./bucardo_ctl start "Reason --name"

Restarts Bucardo cleanly by first issuing the equivalent of a stop to ask any existing Bucardo 
proceses to exit, and then starting a new Bucardo MCP process. A short reason and name should 
be provided - these are logged in the reason_file file and sent in the email sent when Bucardo 
has been started up.

Before attempting to kill any old processes, a ping command with a timeout of 5 seconds is issued. 
If this returns successfully (indicating an active MCP process already running), the script will 
exit with a return value of 2.

=item B<stop>

Usage: ./bucardo_ctl stop "Reason --name"

Forces Bucardo to quit by creating a stop file which all MCP, CTL, and KID processes should 
detect and cause them to exit. Note that active syncs will not exit right away, as they 
will not look for the stop file until they have finished their current run. Typically, 
you should scan the list of processes after running this program to make sure that all Bucardo 
processes have stopped. One should also provide a reason for issuing the stop - usually 
this is a short explanation and your name. This is logged in the reason_file file and 
is also used by Bucardo when it exits and sends out mail about its death.

=item B<list>

Usage: ./bucardo_ctl list <type> <regex>

Lists summary information about databases, goats, herds, or syncs. Adding anything after the 
type will look up all matching entries.

=item B<add>

Usage:  add <item_type> <item_name>

Usage:  add database <dbname> name=internal_name port=xxx host=xxx user=xxx pass=xxx service=xxx conn=xxx sourcelimit=xxx targetlimit=xxx

Usage:  add table [schema].table db=internal_db_name ping=bool standard_conflict=xxx makedelta=bool

Usage:  add sync syncname options

Usage:  add herd name

Usage:  add dbgroup name db1 db2 db3 ...

Tells Bucardo about new objects it should know about. These commands can
replace direct manipulation of the tables in the bucardo schema for the
supported object types (you'll still need to add things like the mappings between objects on your own).

=item B<alter>

Usage:  alter sync syncname [targetgroup=foo | targetdb=foo]

Alters an existing sync, allowing modification to the target database or target database group

=item B<kick>

Usage: ./bucardo_ctl kick <syncname(s)> [timeout]

Tells one or more named syncs to fire as soon as possible. Note that this simply sends a request that 
the sync fire: it may not start right away if the same sync is already running, or if the source or 
target database has exceeded the number of allowed Bucardo connections. If the final argument is a 
number, it is treated as a timeout. If this number is zero, the bucardo_ctl command will not return 
until the sync has finished. For any other number, the sync will wait at most that number of seconds. 
If any sync has not finished before the timeout, a false value is returned. In all other cases, a 
true value is returned.

If a timeout is given, the total completion time in seconds is also displayed. If the sync is going to 
multiple targets, the time that each target takes from the start of the kick is also shown as each 
target finishes.

=item B<reload_config>

Forces Bucardo to reload the bucardo_config file, and then restart all processes to ensure that the new 
information is loaded.

=item B<show>

Usage: ./bucardo_ctl show <all|setting1> [setting2..]

Shows the current values in the bucardo_config table. Use the keyword 'all' to see all the settings, or 
specify one or more search terms.

=item B<set>

Usage: ./bucardo_ctl set setting1=value [setting2=value]

Sets one or more items inside the bucardo_config table. Setting names are case-insensitive.

=item B<ping>

Sends a ping notice to the MCP process to see if it will respond. By default, it will wait 15 seconds. A 
numeric argument will change this timeout. Using a 0 as the timeout indicates waiting forever. If a response 
was returned, the program will exit with a value of 0. If it times out, the value will be 1.

=item B<status>

Usage: ./bucardo_ctl status [syncname(s)] [--sort=#] [--daysback=#] [--showdays]

Shows the current status of all known syncs in a tabular format. If given one or more syncnames, 
shows detailed information for each one.

When showing all syncs, the columns are:

=over 8

=item 1. B<Name>

The name of the sync

=item 2. B<Type>

The type of the sync. C<F> = fullcopy, C<S> = swap, C<P> = pushdelta. In addition, if a sync is overdue, a C<O!> will 
appear, and if it is expired, a C<E!> will appear.

=item 3. B<State>

The current status of this sync. If no sync is running, C<idle> will appear. If a sync has been rquested, but has not 
started yet, C<WAIT> will appear, along with how long since the sync was requested. If a sync is 
currently running, C<RUN> will appear, followed by the amount of time the sync has been running, followed by which 
target the sync is running against. Note that syncs running to more than one database at a time will only show 
the one most recently started.

=item 4. B<PID>

The PID of the current sync's controller (CTL). Note that if this is not a persistent sync and the state is C<idle>, 
this is merely a historical record and does not represent an active process.

=item 5. B<Last_good>

How long since this sync last ran succesfully. Remember that this is affected by the --daysback parameter.

=item 6. B<Time>

The amount of time the last successful sync took to run.

=item 7. B<I/U/D>

The number of inserts. updates, and deletes performed by the last successful sync.

=item 8. B<Last_bad>

How long since this sync failed to run successfully. Strongly affected by the --daysback parameter.

=item 9. B<Time>

The amount of time the last failed sync took before it was aborted.

=back


=item B<activate> syncname [syncname2 syncname3 ...] [timeout]

Activates one or more named syncs. If given a timeout argument, it will wait until it has received 
confirmation from Bucardo that each sync has been successfully activated.

=item B<deactivate> syncname [syncname2 syncname3 ...] [timeout]

Deactivates one or more named syncs. If given a timeout argument, it will wait until it has received 
confirmation from Bucardo that the sync has been successfully deactivated.

=item B<upgrade>

Performs an upgrade of Bucardo itself, by making sure the connected Bucardo database has a copy of the 
latest schema. This should always be run after upgrading Bucardo, and will prompt you before making 
any permanent changes.

=item B<message>

Adds a message to the running Bucardo logs. This message will appear prefixed with "MESSAGE: ". If 
Bucardo is not running, the message will go to the logs the next time Bucardo is running and someone 
adds another message.

=back

=head1 OPTIONS

It is usually easier to set most of these options at the top of the script, or make an alias for them, 
as they will not change very often if at all.

=over 4

=item B<--dbport=number>

=item B<--dbhost=string>

=item B<--dbname=string>

=item B<--dbuser=string>

=item B<--dbpass=string>

The port, host, and name of the Bucardo database, the user to connect as, and the password to use.

=item B<--verbose>

Tells Bucardo (not bucardo_ctl) to run in verbose mode. Default is on. Turning this off is not recommended.

=item B<--ctlverbose>

Makes bucardo_ctl run verbosely. Default is off.

=item B<--ctlquiet>

Tells bucardo_ctl to be as quiet as possible. Default is off.

=item B<--help>

Shows a brief summary of usage for bucardo_ctl.

=back

=head2 Kick arguments

The following arguments are only used with the 'kick' command:

=over 4

=item B<--retry=#>

The number of times to retry a sync if it fails. Defaults to 0.

=item B<--retrysleep>

How long to sleep, in seconds, between each retry attempt.

=item B<--notimer>

By default, kicks with a timeout argument give a running real-time summary of time elapsed by 
using the backspace character. This may not be wanted if running a kick, for example, 
via a cronjob, so turning --notimer on will simply print the entire message without backspaces.

=back

=head2 Status arguments

The following arguments are only used with the 'status' command:

=over 4

=item B<--daysback=#>

Sets how many days backwards to search the old 'q' logs for information. Defaults to 3 days.

=item B<--showdays>

Specifies whether or not do list the time interval with days, or simply show the hours. For example, 
"3d 12h 6m 3s" vs. "48h 6m 3s"

=item B<--compress>

Specifies whether or not to compress the time interval by removing spaces. Mostly used to limit 
the width of the 'status' display.

=item B<--sort=#>

Requests sorting of the 'status' output by one of the nine columns. Use a negative number to reverse 
the sort order.

=back

=head2 Startup arguments

The following arguments are only applicable when using the "start" command:

=over 4

=item B<--sendmail>

Tells Bucardo whether or not to send mail on interesting events: startup, shutdown, and errors. Default is on.
Only applicable when using ./bucardo_ctl start.

=item B<--extraname=string>

A short string that will be appended to the version string as output by the Bucardo process names. Mostly 
useful for debugging.

=item B<--debugfilesep>

Forces creation of separate log files for each Bucardo process of the form "log.bucardo.X.Y", 
where X is the type of process (MCP, CTL, or KID), and Y is the process ID.

=item B<--debugstderr>

Sends all log messages to standard error. Off by default.

=item B<--debugstdout>

Sends all log messages to standard out. Off by default.

=item B<--debugsyslog>

Sends all log messages to the syslog daemon. On by default. The facility used is controlled by 
the row "syslog_facility" in the bucardo_config table, and defaults to "LOG_LOCAL1".

=item B<--debugfile>

If set, writes detailed debugging information to one or more files.

=item B<--debugdir=directory name>

Directory where the debug files should go.

=item B<--debugname=string>

Appends the given string to the end of the default debug file name, "log.bucardo". A dot is added 
before the name as well, so a debugname of "rootdb" would produce a log file named "log.bucardo.rootdb".

=item B<--cleandebugs>

Forces removal of all old debug files before running.

=back

=head1 FILES

In addition to command-line configurations, you can put any options inside of a file. The file F<.bucardorc> in 
the current directory will be used if found. If not found, then the file F<~/.bucardorc> will be used. The format 
of the file is option = value, one per line. Any line starting with a '#' will be skipped. Any values loaded 
from a .bucardorc file will be overwritten by command-line options.

=head1 ENVIRONMENT VARIABLES

The bucardo_ctl script uses I<$ENV{HOME}> to look for a F<.bucardorc> file.

=head1 BUGS

The 'status' command does not yet return current information, and the start time in particular should be 
taken with a grain of salt.

Bug reports and feature requests are always welcome, please visit http://bucardo.org or email bucardo-general@bucardo.org.

=head1 SEE ALSO

Bucardo

=head1 COPYRIGHT

Copyright 2006-2009 Greg Sabino Mullane <greg@endpoint.com>

This program is free to use, subject to the limitations in the LICENSE file.

=cut