git.8kb.co.uk Git - slony-i/pgbouncer_follower/blob - pgbouncer_follower.pl

   1 #!/usr/bin/perl
   2
   3 # Script:   pgbouncer_follower.pl
   4 # Copyright:    22/04/2012: v1.0.1 Glyn Astill <glyn@8kb.co.uk>
   5 # Requires: Perl 5.10.1+, PostgreSQL 9.0+ Slony-I 2.0+ OR Streaming Replication
   6 #
   7 # This script is a command-line utility to monitor Slony-I clusters
   8 # and reconfigure pgbouncer to follow replication sets.
   9 #
  10 # This script is free software: you can redistribute it and/or modify
  11 # it under the terms of the GNU General Public License as published by
  12 # the Free Software Foundation, either version 3 of the License, or
  13 # (at your option) any later version.
  14 #
  15 # This script is distributed in the hope that it will be useful,
  16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18 # GNU General Public License for more details.
  19 #
  20 # You should have received a copy of the GNU General Public License
  21 # along with this script.  If not, see <http://www.gnu.org/licenses/>.
  22
  23 use strict;
  24 use warnings;
  25 use experimental 'smartmatch';
  26 use DBI;
  27 use v5.10.1;
  28 use Getopt::Long qw/GetOptions/;
  29 use Digest::MD5 qw/md5 md5_hex md5_base64/;
  30 use Sys::Hostname;
  31 use IO::Socket;
  32 use Time::HiRes qw/usleep/;
  33 use sigtrap 'handler' => \&cleanExit, 'HUP', 'INT','ABRT','QUIT','TERM';
  34 Getopt::Long::Configure qw/no_ignore_case/;
  35
  36 use vars qw{%opt};
  37
  38 use constant false => 0;
  39 use constant true  => 1;
  40
  41 my $g_usage = 'Pass configuration file: pool_follower.pl -f <configuration_path> [-D]  ';
  42 my $g_method = "slony";
  43 my $g_debug = false;
  44 my $g_pidfile = "/tmp/pgbouncer_follower_%mode.pid";
  45 my $g_logfile = "/tmp/pgbouncer_follower_%mode.log";
  46 my $g_poll_interval = 1000;
  47 my $g_user = "slony";
  48 my $g_pass;
  49 my $g_clname = "replication";
  50 my $g_clsets = "1";
  51 my @g_conninfos;
  52 my @g_cluster;                  # no_id, no_comment, no_prov, orig_sets, conninfo, dbname, host, port
  53 my $g_status_file = "/tmp/pgbouncer_follower_%mode.status";
  54 my $g_conf_template = "/etc/pgbouncer/pgbouncer_%mode.template";
  55 my $g_conf_target = "/etc/pgbouncer/pgbouncer_%mode.ini";
  56 my $g_reload_command = "/etc/init.d/pgbouncer_%mode reload";
  57 my $g_mode = 'rw';
  58 my $g_all_databases=false;
  59 my $g_auth_user='';
  60 my ($year, $month, $day, $hour, $min, $sec);
  61 my $change_time;
  62 my $g_host = hostname;
  63 my ($g_addr)=inet_ntoa((gethostbyname(hostname))[4]);
  64 my $g_origins_only = false;
  65 my $g_best_config = false;
  66 my $g_max_lag = 0;
  67
  68 die $g_usage unless GetOptions(\%opt, 'config_file|f=s', 'daemon|D',) and keys %opt and ! @ARGV;
  69
  70 unless (getConfig($opt{config_file})){
  71     print ("There was a problem reading the configuration.\n");
  72 }
  73
  74 if ($g_debug) {
  75     printLogLn($g_logfile, "DEBUG: Logging to my '$g_logfile'");
  76     if ($g_method eq 'slony') {
  77         printLogLn($g_logfile, "\t Watching sets $g_clsets in Slony-I cluster '$g_clname' polling every ${g_poll_interval}ms");
  78         printLogLn($g_logfile, "\t Following " . ($g_all_databases ? "all databases" : "replicated database only") . " on an '$g_mode' node for the above replicated sets");
  79     }
  80     elsif ($g_method eq 'wal')  {
  81         printLogLn($g_logfile, "\t Watching streaming replication lag polling every ${g_poll_interval}ms");
  82     }
  83     else {
  84         printLogLn($g_logfile, "\t ERROR: Unknown replication method: '$g_method'");
  85         exit(1);
  86     }
  87     printLogLn($g_logfile, "\t Template config '$g_conf_template' Target config '$g_conf_target'");
  88     printLogLn($g_logfile, "\t Reload command is '$g_reload_command'");
  89     printLogLn($g_logfile, "\t Status stored in '$g_status_file'");
  90     printLogLn($g_logfile, "\t Using local address for '$g_host' as '$g_addr'");
  91     if (($g_max_lag > 0) && ($g_mode = 'ro')) {
  92         printLogLn($g_logfile, "\t Max lag for read only targets will be $g_max_lag seconds");
  93     }
  94     #printLogLn($g_logfile, "\t '$g_user' as '$g_pass'");
  95 }
  96
  97 if (defined($opt{daemon})) {
  98     printLogLn($g_logfile, "pgbouncer_follower starting up");
  99     if (writePID($g_pidfile)) {
 100         while (true) {
 101             doAll();
 102             if ($g_debug) {
 103                 printLogLn($g_logfile, "DEBUG: Sleeping for ${g_poll_interval}ms");
 104             }
 105             usleep($g_poll_interval * 1000);
 106         }
 107     }
 108 }
 109 else {
 110     doAll();
 111 }
 112
 113 cleanExit(0);
 114
 115 sub cleanExit {
 116     if (defined($opt{daemon})) {
 117         printLogLn($g_logfile, "pgbouncer_follower shutting down");
 118         removePID($g_pidfile);
 119     }
 120     exit(0);
 121 }
 122
 123 sub doAll {
 124     my $node_to;
 125     my $conninfo_read = 0;
 126
 127     foreach my $conninfo (@g_conninfos) {
 128         $conninfo_read++;
 129         eval {
 130             if ($g_method eq 'slony') {
 131                 @g_cluster = loadCluster($g_clname, $conninfo, $g_user, $g_pass, $g_addr, $g_clsets);
 132             }
 133             elsif ($g_method eq 'wal') {
 134                 @g_cluster = loadBinCluster($g_user, $g_pass);
 135             }
 136             if ($g_debug) {
 137                 printLogLn($g_logfile, "DEBUG: ($conninfo_read) " . (($g_method eq 'slony')?'Slony':'Streaming replication') . " cluster with " . scalar(@g_cluster) . " nodes " . (($g_method eq 'slony')?"read from conninfo: $conninfo":"provided via config conninfos"));
 138                 foreach (@g_cluster) {
 139                     printLogLn($g_logfile, "DEBUG: Node #" . @$_[0] . " DETAIL: " . @$_[1] . " " . @$_[2] . " " . (@$_[3] // "<NONE>") . " " . @$_[4] . " " . (@$_[5] // "<NONE>") . " " . @$_[6] . " " . @$_[7] . " " . @$_[8] . " " . @$_[9] . " " . @$_[10] . " "  . @$_[11]);
 140                 }
 141             }
 142         };
 143         if ($@) {
 144             printLogLn($g_logfile, "ERROR: Failed using conninfo: $conninfo DETAIL: $@");
 145         }
 146         elsif($g_best_config) {
 147             if ($g_debug) {
 148                 if ($g_method eq 'slony') {
 149                     printLogLn($g_logfile, "DEBUG: Found current origin to read config from");
 150                 }
 151                 elsif ($g_method eq 'wal') {
 152                     printLogLn($g_logfile, "DEBUG: Read config from all contactable nodes");
 153                 }
 154             }
 155             last;
 156         }
 157     }
 158     unless (checkCluster($g_status_file)) {
 159         if ($g_debug) {
 160              printLogLn ($g_logfile, "DEBUG: Cluster status unchanged");
 161         }
 162     }
 163     else {
 164         printLogLn ($g_logfile, "Cluster status changed");
 165         $node_to = generateConfig($g_conf_template, $g_conf_target, $g_mode, $g_all_databases, $g_clsets);
 166         if (reloadConfig($g_reload_command)) {
 167             printLogLn ($g_logfile, "Pool repointed to node #$node_to");
 168         }
 169     }
 170 }
 171
 172 sub reloadConfig {
 173     my $reload_command = shift;
 174     my $success = true;
 175     if(length($reload_command // '')) {
 176         printLogLn($g_logfile, "Running '$reload_command'");
 177         eval {
 178             open(RELOAD, "-|", $reload_command . " 2>&1");
 179             while (<RELOAD>) {
 180                 printLogLn($g_logfile, $_);
 181             }
 182             close(RELOAD);
 183             printLogLn($g_logfile, "Reload command has been run.");
 184         };
 185         if ($@) {
 186             printLogLn($g_logfile, "ERROR: Failed to run reload command DETAIL: $@");
 187             $success = false;
 188         }
 189     }
 190     return $success;
 191 }
 192
 193 sub generateConfig {
 194     my $template = shift;
 195     my $target = shift;
 196     my $mode = shift;
 197     my $all_databases = shift;
 198     my $clsets = shift;
 199
 200     my $success = false;
 201     my @sets_to_follow;
 202     my @sets_origin;
 203     my @sets_subscribed;
 204     my $target_node_id;
 205     my $target_db;
 206     my $target_host;
 207     my $target_sets;
 208     my $target_port = 5432;
 209     my $target_is_origin;
 210     my $target_auth = "";
 211
 212     if ($g_debug) {
 213         printLogLn($g_logfile, "DEBUG: All databases = " . ($all_databases ? 'true' : 'false'));
 214     }
 215
 216     if ($g_auth_user ne "") {
 217         $target_auth = " auth_user=" . $g_auth_user;
 218     }
 219
 220     if (open(INFILE, "<", $template)) {
 221         if (open(OUTFILE, ">", $target)) {
 222             print OUTFILE "# Configuration file autogenerated at " . getRuntime() . " from $template\n";
 223             foreach (<INFILE>) {
 224                if (m/\[databases]/) {
 225
 226                     # Try and choose a node; we always assign the origin initially regardless of rw/ro status
 227                     # when in ro mode and if we then  find a suitable subscriber we'll reassign to it.
 228                     foreach my $node (@g_cluster) {
 229
 230
 231                         # If the node is lagging anyway skip it
 232                         if (($g_mode eq 'ro') && ($g_max_lag > 0) && ($node->[11])) {
 233                             printLogLn ($g_logfile, "Lag on node $node->[0] exceeds $g_max_lag seconds");
 234                             next;
 235                         }
 236
 237                         if ($clsets ne 'all') {
 238                             @sets_to_follow = split(',', $clsets);
 239                             if (defined($node->[3])) {
 240                                 @sets_origin =  split(',', $node->[3]);
 241                             }
 242                             else {
 243                                 undef @sets_origin;
 244                             }
 245                             if (defined($node->[5])) {
 246                                 @sets_subscribed =  split(',', $node->[5]);
 247                             }
 248                             else {
 249                                 undef @sets_subscribed;
 250                             }
 251                         }
 252
 253                         if (($clsets eq 'all' && defined($node->[3])) || (@sets_to_follow && @sets_origin && checkProvidesAllSets(\@sets_to_follow, \@sets_origin))) {
 254                             if (defined($node->[8])) {
 255                                 $target_db = $node->[7];
 256                                 $target_host = $node->[8];
 257                                 $target_node_id = $node->[0];
 258                                 $target_sets = $node->[3];
 259                                 $target_is_origin = true;
 260                             }
 261                             if (defined($node->[9])) {
 262                                 $target_port = $node->[9];
 263                             }
 264                             if ($mode eq "rw") {
 265                                 last;
 266                             }
 267                         }
 268                         elsif (($mode eq "ro") && (($clsets eq 'all') || (@sets_to_follow && @sets_subscribed && checkProvidesAllSets(\@sets_to_follow, \@sets_subscribed)))) {
 269                             if (defined($node->[8])) {
 270                                 $target_db = $node->[7];
 271                                 $target_host = $node->[8];
 272                                 $target_node_id = $node->[0];
 273                                 $target_sets = ($node->[5] // $node->[3]);
 274                                 $target_is_origin = false;
 275                             }
 276                             if (defined($node->[9])) {
 277                                 $target_port = $node->[9];
 278                             }
 279                             last;
 280                         }
 281                     }
 282                     if (defined($target_host)) {
 283                         $_ = "# Configuration for " . ($target_is_origin ? "origin" : "subscriber") . " of sets $target_sets node #$target_node_id $target_host:$target_port\n" . $_;
 284                         if ($g_debug) {
 285                             printLogLn ($g_logfile, "DEBUG: Configuration for " . ($target_is_origin ? "origin" : "subscriber") . " of sets $target_sets node #$target_node_id $target_host:$target_port");
 286                         }
 287                         if ($all_databases || $target_db eq '*') {
 288                             $_ =~ s/(\[databases\])/$1\n\* = host=$target_host port=$target_port$target_auth/;
 289                         }
 290                         else {
 291                             $_ =~ s/(\[databases\])/$1\n$target_db = host=$target_host port=$target_port dbname=$target_db$target_auth/;
 292                         }
 293                     }
 294                     else {
 295                             $_ = "# Could not find any node providing sets $g_clsets in mode $mode\n";
 296                             printLogLn ($g_logfile, "DEBUG: Could not find any node providing sets $g_clsets in mode $mode");
 297                     }
 298
 299                }
 300                print OUTFILE $_;
 301             }
 302             close (OUTFILE);
 303         }
 304         else {
 305             print ("ERROR: Can't open file $target\n");
 306         }
 307         close(INFILE);
 308     }
 309     else {
 310         print ("ERROR: Can't open file $template\n");
 311     }
 312     return $target_node_id;
 313 }
 314
 315 sub checkCluster {
 316     my $infile = shift;
 317     my $changed = false;
 318     my $current_state = md5_hex('INIT');
 319     my $previous_state;
 320     foreach (@g_cluster) {
 321         if (!$g_origins_only || defined($_->[3])) {
 322             $current_state = md5_hex(($current_state // "") . $_->[0] . $_->[2] . (defined($_->[3]) ? 't' : 'f') . $_->[6] . $_->[11]);
 323             if ($g_debug) {
 324                 printLogLn($g_logfile, "DEBUG: Node " . $_->[0] . " detail = " . $_->[2] . (defined($_->[3]) ? 't' : 'f') . $_->[6] . $_->[11]);
 325             }
 326         }
 327     }
 328
 329     if (-f $infile) {
 330         if (open(CLUSTERFILE, "<", $infile)) {
 331             $previous_state = <CLUSTERFILE>;
 332             close(CLUSTERFILE);
 333         }
 334         else {
 335             printLogLn ($g_logfile, "ERROR: Can't open file $infile for reading");
 336         }
 337     }
 338
 339     unless (-f $infile && ($current_state eq $previous_state)) {
 340         if ($g_debug) {
 341                 printLogLn($g_logfile, "DEBUG: Writing to status file");
 342         }
 343         if (open(CLUSTERFILE, ">", $infile)) {
 344             print CLUSTERFILE $current_state;
 345             close(CLUSTERFILE);
 346         }
 347         else {
 348             printLogLn ($g_logfile, "ERROR: Can't open file $infile for writing");
 349         }
 350     }
 351
 352     if ((($previous_state // "") ne "") && ($current_state ne $previous_state)){
 353         $changed = true;
 354     }
 355
 356     return $changed
 357 }
 358
 359 sub loadCluster {
 360     my $clname = shift;
 361     my $conninfo = shift;
 362     my $dbuser = shift;
 363     my $dbpass = shift;
 364     my $addr = shift;
 365     my $clsets = shift;
 366     my $param_on = 1;
 367
 368     my $dsn;
 369     my $dbh;
 370     my $sth;
 371     my $query;
 372     my $version;
 373     my $qw_clname;
 374     my @cluster;
 375
 376     $g_best_config = false;
 377     $dsn = "DBI:Pg:$conninfo};";
 378
 379     eval {
 380         $dbh = DBI->connect($dsn, $dbuser, $dbpass, {RaiseError => 1});
 381         $qw_clname = $dbh->quote_identifier("_" . $clname);
 382
 383         $query = "SELECT $qw_clname.getModuleVersion()";
 384         $sth = $dbh->prepare($query);
 385         $sth->execute();
 386         ($version) = $sth->fetchrow;
 387         $sth->finish;
 388
 389         $query = "WITH x AS (
 390                 SELECT a.no_id,
 391                     a.no_comment,
 392                     COALESCE(b.sub_provider, 0) AS no_prov,
 393                     NULLIF(array_to_string(array(SELECT set_id FROM $qw_clname.sl_set WHERE set_origin = a.no_id" .
 394                     ($clsets ne "all" ? " AND set_id IN (" . substr('?, ' x scalar(split(',', $clsets)), 0, -2) . ")" : "")
 395                     . " ORDER BY set_id), ','), '') AS origin_sets,
 396                     CASE " . ((substr($version,0,3) >= 2.2) ? "WHEN a.no_failed THEN 'FAILED' " : "") . "WHEN a.no_active THEN 'ACTIVE' ELSE 'INACTIVE' END AS no_status,
 397                     string_agg(CASE WHEN b.sub_receiver = a.no_id AND b.sub_forward AND b.sub_active" .
 398                     ($clsets ne "all" ? " AND b.sub_set IN (" . substr('?, ' x scalar(split(',', $clsets)), 0, -2) . ")" : "")
 399                     . " THEN b.sub_set::text END, ',' ORDER BY b.sub_set) AS prov_sets,
 400                     COALESCE(c.pa_conninfo,(SELECT pa_conninfo FROM $qw_clname.sl_path WHERE pa_server = $qw_clname.getlocalnodeid(?) LIMIT 1)) AS no_conninfo
 401                 FROM $qw_clname.sl_node a
 402                 LEFT JOIN $qw_clname.sl_subscribe b ON a.no_id = b.sub_receiver AND b.sub_set <> 999
 403                 LEFT JOIN $qw_clname.sl_path c ON c.pa_server = a.no_id AND c.pa_client = $qw_clname.getlocalnodeid(?)
 404                 LEFT JOIN $qw_clname.sl_set d ON d.set_origin = a.no_id
 405                 GROUP BY b.sub_provider, a.no_id, a.no_comment, c.pa_conninfo, a.no_active
 406                 ORDER BY (COALESCE(b.sub_provider, 0) = 0) DESC, a.no_id ASC
 407                 ), z AS (
 408                 SELECT x.*,
 409                     CASE WHEN x.no_conninfo ilike '%dbname=%' THEN(regexp_matches(x.no_conninfo, E'dbname=(.+?)\\\\M', 'ig'))[1] END AS database,
 410                     CASE WHEN x.no_conninfo ilike '%host=%' THEN(regexp_matches(x.no_conninfo, E'host=(.+?)(?=\\\\s|\$)', 'ig'))[1] END AS host,
 411                     CASE WHEN x.no_conninfo ilike '%port=%' THEN(regexp_matches(x.no_conninfo, E'port=(.+?)\\\\M', 'ig'))[1] ELSE '5432' END AS port,
 412                     (no_id = $qw_clname.getlocalnodeid(?)) AS this_node,
 413                     COALESCE((? BETWEEN 1 AND extract(epoch from s.st_lag_time)),false) AS lag_exceeded
 414                 FROM x
 415                 LEFT JOIN $qw_clname.sl_status s ON s.st_received = x.no_id
 416                 WHERE x.no_conninfo != '<event pending>'
 417                 )
 418                 SELECT * FROM z
 419                 ORDER BY origin_sets, @(CASE WHEN (host ~ '^[0-9]{1,3}(.[0-9]{1,3}){3}\$') THEN host::inet ELSE '255.255.255.255'::inet END - ?::inet) ASC";
 420
 421         if ($g_debug) {
 422 #            printLogLn($g_logfile, "DEBUG: " . $query);
 423         }
 424
 425         $sth = $dbh->prepare($query);
 426
 427         if ($clsets ne "all") {
 428             for (0..1) {
 429                 foreach my $param (split(",", $clsets)) {
 430                     $sth->bind_param($param_on, $param);
 431                     $param_on++;
 432                 }
 433             }
 434         }
 435         # This param is taken 3 times
 436         for (0..2) {
 437             $sth->bind_param($param_on, "_" . $clname);
 438             $param_on++;
 439         }
 440         $sth->bind_param($param_on, $g_max_lag);
 441         $param_on++;
 442         $sth->bind_param($param_on, (isInet($addr) ? $addr : '255.255.255.255'));
 443         $sth->execute();
 444
 445         while (my @node = $sth->fetchrow) {
 446             # If some origin sets exist for this node row (we can assume they're the sets we're following since they're filtered in the query)
 447             # and the row is flagged as this_node then we have found the best node to read the configuration from.
 448             if (defined($node[3]) && $node[10]) {
 449                 $g_best_config = true;
 450             }
 451             push(@cluster,  \@node);
 452         }
 453
 454         $sth->finish;
 455         $dbh->disconnect();
 456     };
 457     if ($@) {
 458         printLogLn($g_logfile, "ERROR: Failed to execute query against Postgres server: $@");
 459     }
 460
 461     return @cluster;
 462 }
 463
 464 sub loadBinCluster {
 465     my $dbuser = shift;
 466     my $dbpass = shift;
 467
 468     my $dsn;
 469     my $dbh;
 470     my $sth;
 471     my $query;
 472     my $recovery;
 473     my $xlog_location;
 474     my $apply_lag;
 475     my $primaries = 0;
 476
 477     my @parts;
 478     my $timeline;
 479     my $location;
 480     my $primary_location;
 481
 482     my $hostname;
 483     my $port;
 484     my $database;
 485     my @tmp_cluster;
 486     my @cluster;
 487     my $node_id = 1;
 488     $g_best_config = true;
 489
 490     foreach my $conninfo (@g_conninfos) {
 491         $dsn = "DBI:Pg:$conninfo};";
 492
 493         eval {
 494             $dbh = DBI->connect($dsn, $dbuser, $dbpass, {RaiseError => 1});
 495             # Check to see if the server is a secondary, and also pull the current xlog
 496             # location and the apply lag using pg_last_xact_replay_timestamp to get the
 497             # last commit timestamp from the primary applied on the secondary.
 498             # We will need to compare the current receive location to the primary xlog
 499             # location, if they differ we can then use the apply_lag value; we'll have
 500             # to post-process this.
 501             # In 9.6+ we might want to pull the system identifier from pg_controldata view too
 502             $query = "SELECT pg_is_in_recovery(),
 503                           CASE
 504                               WHEN pg_is_in_recovery() THEN pg_last_xlog_receive_location()
 505                               ELSE pg_current_xlog_location()
 506                           END,
 507                           COALESCE(extract(epoch from current_timestamp-pg_last_xact_replay_timestamp()),0)";
 508             $sth = $dbh->prepare($query);
 509             $sth->execute();
 510             ($recovery, $xlog_location, $apply_lag) = $sth->fetchrow;
 511             $sth->finish;
 512
 513             ($port) = $conninfo =~ m/port=(.+?)(\s|$)/g;
 514             ($hostname) = $conninfo =~ m/host=(.+?)(\s|$)/g;
 515             ($database) = $conninfo =~ m/dbname=(.+?)(\s|$)/g;
 516             @parts = split('/', $xlog_location, 2);
 517             $timeline = qtrim(trim($parts[0]));
 518             $location = hex(qtrim(trim($parts[1])));
 519
 520             if ($g_debug) {
 521                 printLogLn($g_logfile, "DEBUG: Server: $hostname:$port " . ($recovery ? 'secondary' : 'primary') . " at $xlog_location ($timeline/$location)");
 522             }
 523
 524             # For WAL replication we assume if we can contact a server then it is active,
 525             # which isn't strictly true, but nodes that have fallen behind can be excluded
 526             # via the max_ro_lag setting.  We also substitute timeline+1 for the slony
 527             # replication set.
 528             if (!$recovery) {
 529                 $primaries++;
 530                 $primary_location = $xlog_location;
 531                 my @node=(1,'Primary',0,$timeline,"ACTIVE",($timeline+1),$conninfo,$database,$hostname,$port,1,$xlog_location,$apply_lag);
 532                 push(@tmp_cluster,  \@node);
 533             }
 534             else {
 535                 $node_id++;
 536                 my @node=($node_id,"Secondary".($node_id-1),1,undef,"ACTIVE",($timeline+1),$conninfo,$database,$hostname,$port,$node_id,$xlog_location,$apply_lag);
 537                 push(@tmp_cluster,  \@node);
 538             }
 539         };
 540         if ($@) {
 541             printLogLn($g_logfile, "ERROR: Could not connect to server with conninfo: $conninfo DETAIL: $@");
 542         }
 543     }
 544
 545     # Error if more than one primary discovered.
 546     if ($primaries != 1) {
 547         printLogLn($g_logfile, "ERROR: Invalid quantity of primaries: $primaries");
 548         die "no primaries found";
 549     }
 550     # Do the post processing we mentioned above, once we know xlog locations differ
 551     # then we can say the apply lag is correct, else it's just there has been no
 552     # activity on the master. In the case the xlog locations differ but apply_lag
 553     # is 0 then it means no WAL has been applied since srver start; we don't know
 554     # a time for lag in this case.
 555     else {
 556         foreach (@tmp_cluster) {
 557             $apply_lag = false;
 558             if ($g_max_lag > 0) {
 559                 if ((@$_[11] ne $primary_location) && (@$_[12] > $g_max_lag || @$_[12] == 0)) {
 560                    $apply_lag  = true;
 561                 }
 562             }
 563             my @node=(@$_[0],@$_[1],@$_[2],@$_[3],@$_[4],@$_[5],@$_[6],@$_[7],@$_[8],@$_[9],@$_[10],$apply_lag);
 564             push(@cluster,  \@node);
 565         }
 566     }
 567
 568     return @cluster;
 569 }
 570
 571 sub getConfig {
 572     my @fields;
 573     my $success = false;
 574     my $infile = shift;
 575     my $value;
 576
 577     if (open(CFGFILE, "<", $infile)) {
 578         foreach (<CFGFILE>) {
 579             chomp $_;
 580             for ($_) {
 581                 s/\r//;
 582                 #s/\#.*//;
 583                 s/#(?=(?:(?:[^']|[^"]*+'){2})*+[^']|[^"]*+\z).*//;
 584             }
 585             if (length(trim($_))) {
 586                 @fields = split('=', $_, 2);
 587                 $value = qtrim(trim($fields[1]));
 588                 given(lc($fields[0])) {
 589                     when(/\breplication_method\b/i) {
 590                         $g_method = $value;
 591                     }
 592                     when(/\bdebug\b/i) {
 593                         $g_debug = checkBoolean($value);
 594                     }
 595                     when(/\bpid_file\b/i) {
 596                         $g_pidfile = $value;
 597                     }
 598                     when(/\blog_file\b/i) {
 599                         $g_logfile = $value;
 600                     }
 601                     when(/\bslony_user\b/i) { # Depreciated
 602                         $g_user = $value;
 603                     }
 604                     when(/\bslony_pass\b/i) { # Depreciated
 605                         $g_pass = $value;
 606                     }
 607                     when(/\breplication_user\b/i) {
 608                         $g_user = $value;
 609                     }
 610                     when(/\breplication_pass\b/i) {
 611                         $g_pass = $value;
 612                     }
 613                     when(/\bslony_cluster_name\b/i) {
 614                         $g_clname = $value;
 615                     }
 616                     when(/\bslony_sets_to_follow\b/i) {
 617                         $g_clsets = $value;
 618                     }
 619                     when(/\bserver_conninfo\b/i) {
 620                         push(@g_conninfos, $value);
 621                     }
 622                     when(/\bfollower_poll_interval\b/i) {
 623                         $g_poll_interval = checkInteger($value);
 624                     }
 625                     when(/\bstatus_file\b/i) {
 626                         $g_status_file = $value;
 627                     }
 628                     when(/\bpool_conf_template\b/i) {
 629                         $g_conf_template = $value;
 630                     }
 631                     when(/\bpool_conf_target\b/i) {
 632                         $g_conf_target = $value;
 633                     }
 634                     when(/\bpool_reload_command\b/i) {
 635                         $g_reload_command = $value;
 636                     }
 637                     when(/\bpool_mode\b/i) {
 638                         $g_mode = lc($value);
 639                     }
 640                     when(/\bpool_all_databases\b/i) {
 641                         $g_all_databases = checkBoolean($value);
 642                     }
 643                     when(/\bauth_user\b/i) {
 644                         $g_auth_user = $value;
 645                     }
 646                     when(/\bonly_follow_origins\b/i) {
 647                         $g_origins_only = checkBoolean($value);
 648                     }
 649                     when(/\bmax_ro_lag\b/i) {
 650                         $g_max_lag = checkInteger($value);
 651                     }
 652                 }
 653             }
 654         }
 655         close (CFGFILE);
 656         if (defined($g_user) && (scalar(@g_conninfos) > 0)) {
 657            $success = true;
 658         }
 659         # Replace %mode and %clname here for actual value
 660         for ($g_pidfile, $g_logfile, $g_status_file, $g_conf_template, $g_conf_target, $g_reload_command) {
 661             s/\%mode/$g_mode/g;
 662             s/\%clname/$g_clname/g;
 663         }
 664
 665
 666     }
 667     else {
 668         printLogLn($g_logfile, "ERROR: Could not read configuration from '$infile'");
 669     }
 670     return $success;
 671 }
 672
 673 sub writePID {
 674     my $pidfile = shift;
 675     my $success = true;
 676
 677     eval {
 678         open (PIDFILE, ">", $pidfile);
 679         print PIDFILE $$;
 680         close (PIDFILE);
 681         if ($g_debug) {
 682             printLogLn($g_logfile, "DEBUG: Created PID file '$pidfile' for process $$");
 683         }
 684     };
 685     if ($@) {
 686         printLogLn($g_logfile, "ERROR: unable to write pidfile at '$pidfile' DETAIL $!");
 687         $success = false;
 688     }
 689     return $success;
 690 }
 691
 692 sub removePID {
 693     my $pidfile = shift;
 694     my $success = true;
 695
 696     eval {
 697         if (-f $pidfile) {
 698             unlink $pidfile;
 699             if ($g_debug) {
 700                 printLogLn($g_logfile, "DEBUG: Removed PID file '$pidfile'");
 701             }
 702         }
 703         elsif ($g_debug){
 704             printLogLn($g_logfile, "DEBUG: PID file '$pidfile' never existed to be removed");
 705         }
 706     };
 707     if ($@) {
 708         printLogLn($g_logfile, "ERROR: unable to remove pidfile at '$pidfile' DETAIL $!");
 709         $success = false;
 710     }
 711     return $success
 712 }
 713
 714 sub checkBoolean {
 715     my $text = shift;
 716     my $value = undef;
 717     if ( grep /^$text$/i, ("y","yes","t","true","on") ) {
 718         $value = true;
 719     }
 720     elsif ( grep /^$text$/i, ("n","no","f","false","off") ) {
 721         $value = false;
 722     }
 723     return $value;
 724 }
 725
 726 sub checkInteger {
 727     my $integer = shift;
 728     my $value = undef;
 729
 730     if (($integer * 1) eq $integer) {
 731         $value = int($integer);
 732     }
 733     return $value;
 734 }
 735
 736 sub checkProvidesAllSets {
 737     my ($originSets, $providerSets) = @_;
 738     my %test_hash;
 739
 740     undef @test_hash{@$originSets};       # add a hash key for each element of @$originSets
 741     delete @test_hash{@$providerSets};    # remove all keys for elements of @$providerSets
 742
 743     return !%test_hash;              # return false if any keys are left in the hash
 744 }
 745
 746 sub isInet {
 747     my $address = shift;
 748     my $success = true;
 749
 750     my(@octets) = $address =~ /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/;
 751     if (@octets == 4) {
 752         foreach (@octets) {
 753             unless ($_ <= 255) {
 754                 $success = false;
 755             }
 756         }
 757     }
 758     else {
 759         $success = false;
 760     }
 761
 762     return $success;
 763 }
 764
 765 sub qtrim {
 766     my $string = shift;
 767     $string =~ s/^('|")+//;
 768     $string =~ s/('|")+$//;
 769     return $string;
 770 }
 771
 772 sub trim {
 773     my $string = shift;
 774     $string =~ s/^\s+//;
 775     $string =~ s/\s+$//;
 776     return $string;
 777 }
 778
 779 sub getRuntime {
 780     my ($year, $month, $day, $hour, $min, $sec) = (localtime(time))[5,4,3,2,1,0];
 781     my $time = sprintf ("%02d:%02d:%02d on %02d/%02d/%04d", $hour, $min, $sec, $day, $month+1, $year+1900);
 782     return $time;
 783 }
 784
 785 sub printLog {
 786     my $logfile = shift;
 787     my $message = shift;
 788
 789     print $message;
 790
 791     if (open(LOGFILE, ">>", $logfile)) {
 792         print LOGFILE getRuntime() . " " . $message;
 793         close (LOGFILE);
 794     }
 795     else {
 796         printLn("ERROR: Unable to write to logfile $logfile");
 797     }
 798 }
 799
 800 sub printLogLn {
 801     printLog ($_[0], $_[1] . $/);
 802 }
 803
 804 sub printLn {
 805     print ((@_ ? join($/, @_) : $_), $/);
 806 }