percona-toolkit 2.2.7-1~dfsg1 / usr / bin / pt-agent

#!/usr/bin/env perl

# This program is part of Percona Toolkit: http://www.percona.com/software/
# See "COPYRIGHT, LICENSE, AND WARRANTY" at the end of this file for legal
# notices and disclaimers.

use strict;
use warnings FATAL => 'all';

# This tool is "fat-packed": most of its dependent modules are embedded
# in this file.  Setting %INC to this file for each module makes Perl aware
# of this so it will not try to load the module from @INC.  See the tool's
# documentation for a full list of dependencies.
BEGIN {
   $INC{$_} = __FILE__ for map { (my $pkg = "$_.pm") =~ s!::!/!g; $pkg } (qw(
      Percona::Toolkit
      Lmo::Utils
      Lmo::Meta
      Lmo::Object
      Lmo::Types
      Lmo
      Percona::WebAPI::Representation
      Percona::WebAPI::Client
      Percona::WebAPI::Exception::Request
      Percona::WebAPI::Exception::Resource
      Percona::WebAPI::Resource::Agent
      Percona::WebAPI::Resource::Config
      Percona::WebAPI::Resource::Service
      Percona::WebAPI::Resource::Task
      Percona::WebAPI::Resource::LogEntry
      VersionCheck
      DSNParser
      OptionParser
      Cxn
      Quoter
      VersionParser
      Daemon
      Transformers
      Safeguards
      Percona::Agent::Logger
   ));
}

# ###########################################################################
# Percona::Toolkit package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/Toolkit.pm
#   t/lib/Percona/Toolkit.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::Toolkit;

our $VERSION = '2.2.7';

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use Carp qw(carp cluck);
use Data::Dumper qw();

require Exporter;
our @ISA         = qw(Exporter);
our @EXPORT_OK   = qw(
   have_required_args
   Dumper
   _d
);

sub have_required_args {
   my ($args, @required_args) = @_;
   my $have_required_args = 1;
   foreach my $arg ( @required_args ) {
      if ( !defined $args->{$arg} ) {
         $have_required_args = 0;
         carp "Argument $arg is not defined";
      }
   }
   cluck unless $have_required_args;  # print backtrace
   return $have_required_args;
}

sub Dumper {
   local $Data::Dumper::Indent    = 1;
   local $Data::Dumper::Sortkeys  = 1;
   local $Data::Dumper::Quotekeys = 0;
   Data::Dumper::Dumper(@_);
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End Percona::Toolkit package
# ###########################################################################

# ###########################################################################
# Lmo::Utils package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Lmo/Utils.pm
#   t/lib/Lmo/Utils.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Lmo::Utils;

use strict;
use warnings qw( FATAL all );
require Exporter;
our (@ISA, @EXPORT, @EXPORT_OK);

BEGIN {
   @ISA = qw(Exporter);
   @EXPORT = @EXPORT_OK = qw(
      _install_coderef
      _unimport_coderefs
      _glob_for
      _stash_for
   );
}

{
   no strict 'refs';
   sub _glob_for {
      return \*{shift()}
   }

   sub _stash_for {
      return \%{ shift() . "::" };
   }
}

sub _install_coderef {
   my ($to, $code) = @_;

   return *{ _glob_for $to } = $code;
}

sub _unimport_coderefs {
   my ($target, @names) = @_;
   return unless @names;
   my $stash = _stash_for($target);
   foreach my $name (@names) {
      if ($stash->{$name} and defined(&{$stash->{$name}})) {
         delete $stash->{$name};
      }
   }
}

1;
}
# ###########################################################################
# End Lmo::Utils package
# ###########################################################################

# ###########################################################################
# Lmo::Meta package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Lmo/Meta.pm
#   t/lib/Lmo/Meta.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Lmo::Meta;
use strict;
use warnings qw( FATAL all );

my %metadata_for;

sub new {
   my $class = shift;
   return bless { @_ }, $class
}

sub metadata_for {
   my $self    = shift;
   my ($class) = @_;

   return $metadata_for{$class} ||= {};
}

sub class { shift->{class} }

sub attributes {
   my $self = shift;
   return keys %{$self->metadata_for($self->class)}
}

sub attributes_for_new {
   my $self = shift;
   my @attributes;

   my $class_metadata = $self->metadata_for($self->class);
   while ( my ($attr, $meta) = each %$class_metadata ) {
      if ( exists $meta->{init_arg} ) {
         push @attributes, $meta->{init_arg}
               if defined $meta->{init_arg};
      }
      else {
         push @attributes, $attr;
      }
   }
   return @attributes;
}

1;
}
# ###########################################################################
# End Lmo::Meta package
# ###########################################################################

# ###########################################################################
# Lmo::Object package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Lmo/Object.pm
#   t/lib/Lmo/Object.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Lmo::Object;

use strict;
use warnings qw( FATAL all );

use Carp ();
use Scalar::Util qw(blessed);

use Lmo::Meta;
use Lmo::Utils qw(_glob_for);

sub new {
   my $class = shift;
   my $args  = $class->BUILDARGS(@_);

   my $class_metadata = Lmo::Meta->metadata_for($class);

   my @args_to_delete;
   while ( my ($attr, $meta) = each %$class_metadata ) {
      next unless exists $meta->{init_arg};
      my $init_arg = $meta->{init_arg};

      if ( defined $init_arg ) {
         $args->{$attr} = delete $args->{$init_arg};
      }
      else {
         push @args_to_delete, $attr;
      }
   }

   delete $args->{$_} for @args_to_delete;

   for my $attribute ( keys %$args ) {
      if ( my $coerce = $class_metadata->{$attribute}{coerce} ) {
         $args->{$attribute} = $coerce->($args->{$attribute});
      }
      if ( my $isa_check = $class_metadata->{$attribute}{isa} ) {
         my ($check_name, $check_sub) = @$isa_check;
         $check_sub->($args->{$attribute});
      }
   }

   while ( my ($attribute, $meta) = each %$class_metadata ) {
      next unless $meta->{required};
      Carp::confess("Attribute ($attribute) is required for $class")
         if ! exists $args->{$attribute}
   }

   my $self = bless $args, $class;

   my @build_subs;
   my $linearized_isa = mro::get_linear_isa($class);

   for my $isa_class ( @$linearized_isa ) {
      unshift @build_subs, *{ _glob_for "${isa_class}::BUILD" }{CODE};
   }
   my @args = %$args;
   for my $sub (grep { defined($_) && exists &$_ } @build_subs) {
      $sub->( $self, @args);
   }
   return $self;
}

sub BUILDARGS {
   shift; # No need for the classname
   if ( @_ == 1 && ref($_[0]) ) {
      Carp::confess("Single parameters to new() must be a HASH ref, not $_[0]")
         unless ref($_[0]) eq ref({});
      return {%{$_[0]}} # We want a new reference, always
   }
   else {
      return { @_ };
   }
}

sub meta {
   my $class = shift;
   $class    = Scalar::Util::blessed($class) || $class;
   return Lmo::Meta->new(class => $class);
}

1;
}
# ###########################################################################
# End Lmo::Object package
# ###########################################################################

# ###########################################################################
# Lmo::Types package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Lmo/Types.pm
#   t/lib/Lmo/Types.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Lmo::Types;

use strict;
use warnings qw( FATAL all );

use Carp ();
use Scalar::Util qw(looks_like_number blessed);


our %TYPES = (
   Bool   => sub { !$_[0] || (defined $_[0] && looks_like_number($_[0]) && $_[0] == 1) },
   Num    => sub { defined $_[0] && looks_like_number($_[0]) },
   Int    => sub { defined $_[0] && looks_like_number($_[0]) && $_[0] == int($_[0]) },
   Str    => sub { defined $_[0] },
   Object => sub { defined $_[0] && blessed($_[0]) },
   FileHandle => sub { local $@; require IO::Handle; fileno($_[0]) && $_[0]->opened },

   map {
      my $type = /R/ ? $_ : uc $_;
      $_ . "Ref" => sub { ref $_[0] eq $type }
   } qw(Array Code Hash Regexp Glob Scalar)
);

sub check_type_constaints {
   my ($attribute, $type_check, $check_name, $val) = @_;
   ( ref($type_check) eq 'CODE'
      ? $type_check->($val)
      : (ref $val eq $type_check
         || ($val && $val eq $type_check)
         || (exists $TYPES{$type_check} && $TYPES{$type_check}->($val)))
   )
   || Carp::confess(
        qq<Attribute ($attribute) does not pass the type constraint because: >
      . qq<Validation failed for '$check_name' with value >
      . (defined $val ? Lmo::Dumper($val) : 'undef') )
}

sub _nested_constraints {
   my ($attribute, $aggregate_type, $type) = @_;

   my $inner_types;
   if ( $type =~ /\A(ArrayRef|Maybe)\[(.*)\]\z/ ) {
      $inner_types = _nested_constraints($1, $2);
   }
   else {
      $inner_types = $TYPES{$type};
   }

   if ( $aggregate_type eq 'ArrayRef' ) {
      return sub {
         my ($val) = @_;
         return unless ref($val) eq ref([]);

         if ($inner_types) {
            for my $value ( @{$val} ) {
               return unless $inner_types->($value)
            }
         }
         else {
            for my $value ( @{$val} ) {
               return unless $value && ($value eq $type
                        || (Scalar::Util::blessed($value) && $value->isa($type)));
            }
         }
         return 1;
      };
   }
   elsif ( $aggregate_type eq 'Maybe' ) {
      return sub {
         my ($value) = @_;
         return 1 if ! defined($value);
         if ($inner_types) {
            return unless $inner_types->($value)
         }
         else {
            return unless $value eq $type
                        || (Scalar::Util::blessed($value) && $value->isa($type));
         }
         return 1;
      }
   }
   else {
      Carp::confess("Nested aggregate types are only implemented for ArrayRefs and Maybe");
   }
}

1;
}
# ###########################################################################
# End Lmo::Types package
# ###########################################################################

# ###########################################################################
# Lmo package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Lmo.pm
#   t/lib/Lmo.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
BEGIN {
$INC{"Lmo.pm"} = __FILE__;
package Lmo;
our $VERSION = '0.30_Percona'; # Forked from 0.30 of Mo.


use strict;
use warnings qw( FATAL all );

use Carp ();
use Scalar::Util qw(looks_like_number blessed);

use Lmo::Meta;
use Lmo::Object;
use Lmo::Types;

use Lmo::Utils;

my %export_for;
sub import {
   warnings->import(qw(FATAL all));
   strict->import();

   my $caller     = scalar caller(); # Caller's package
   my %exports = (
      extends  => \&extends,
      has      => \&has,
      with     => \&with,
      override => \&override,
      confess  => \&Carp::confess,
   );

   $export_for{$caller} = \%exports;

   for my $keyword ( keys %exports ) {
      _install_coderef "${caller}::$keyword" => $exports{$keyword};
   }

   if ( !@{ *{ _glob_for "${caller}::ISA" }{ARRAY} || [] } ) {
      @_ = "Lmo::Object";
      goto *{ _glob_for "${caller}::extends" }{CODE};
   }
}

sub extends {
   my $caller = scalar caller();
   for my $class ( @_ ) {
      _load_module($class);
   }
   _set_package_isa($caller, @_);
   _set_inherited_metadata($caller);
}

sub _load_module {
   my ($class) = @_;
   
   (my $file = $class) =~ s{::|'}{/}g;
   $file .= '.pm';
   { local $@; eval { require "$file" } } # or warn $@;
   return;
}

sub with {
   my $package = scalar caller();
   require Role::Tiny;
   for my $role ( @_ ) {
      _load_module($role);
      _role_attribute_metadata($package, $role);
   }
   Role::Tiny->apply_roles_to_package($package, @_);
}

sub _role_attribute_metadata {
   my ($package, $role) = @_;

   my $package_meta = Lmo::Meta->metadata_for($package);
   my $role_meta    = Lmo::Meta->metadata_for($role);

   %$package_meta = (%$role_meta, %$package_meta);
}

sub has {
   my $names  = shift;
   my $caller = scalar caller();

   my $class_metadata = Lmo::Meta->metadata_for($caller);
   
   for my $attribute ( ref $names ? @$names : $names ) {
      my %args   = @_;
      my $method = ($args{is} || '') eq 'ro'
         ? sub {
            Carp::confess("Cannot assign a value to a read-only accessor at reader ${caller}::${attribute}")
               if $#_;
            return $_[0]{$attribute};
         }
         : sub {
            return $#_
                  ? $_[0]{$attribute} = $_[1]
                  : $_[0]{$attribute};
         };

      $class_metadata->{$attribute} = ();

      if ( my $type_check = $args{isa} ) {
         my $check_name = $type_check;
         
         if ( my ($aggregate_type, $inner_type) = $type_check =~ /\A(ArrayRef|Maybe)\[(.*)\]\z/ ) {
            $type_check = Lmo::Types::_nested_constraints($attribute, $aggregate_type, $inner_type);
         }
         
         my $check_sub = sub {
            my ($new_val) = @_;
            Lmo::Types::check_type_constaints($attribute, $type_check, $check_name, $new_val);
         };
         
         $class_metadata->{$attribute}{isa} = [$check_name, $check_sub];
         my $orig_method = $method;
         $method = sub {
            $check_sub->($_[1]) if $#_;
            goto &$orig_method;
         };
      }

      if ( my $builder = $args{builder} ) {
         my $original_method = $method;
         $method = sub {
               $#_
                  ? goto &$original_method
                  : ! exists $_[0]{$attribute}
                     ? $_[0]{$attribute} = $_[0]->$builder
                     : goto &$original_method
         };
      }

      if ( my $code = $args{default} ) {
         Carp::confess("${caller}::${attribute}'s default is $code, but should be a coderef")
               unless ref($code) eq 'CODE';
         my $original_method = $method;
         $method = sub {
               $#_
                  ? goto &$original_method
                  : ! exists $_[0]{$attribute}
                     ? $_[0]{$attribute} = $_[0]->$code
                     : goto &$original_method
         };
      }

      if ( my $role = $args{does} ) {
         my $original_method = $method;
         $method = sub {
            if ( $#_ ) {
               Carp::confess(qq<Attribute ($attribute) doesn't consume a '$role' role">)
                  unless Scalar::Util::blessed($_[1]) && eval { $_[1]->does($role) }
            }
            goto &$original_method
         };
      }

      if ( my $coercion = $args{coerce} ) {
         $class_metadata->{$attribute}{coerce} = $coercion;
         my $original_method = $method;
         $method = sub {
            if ( $#_ ) {
               return $original_method->($_[0], $coercion->($_[1]))
            }
            goto &$original_method;
         }
      }

      _install_coderef "${caller}::$attribute" => $method;

      if ( $args{required} ) {
         $class_metadata->{$attribute}{required} = 1;
      }

      if ($args{clearer}) {
         _install_coderef "${caller}::$args{clearer}"
            => sub { delete shift->{$attribute} }
      }

      if ($args{predicate}) {
         _install_coderef "${caller}::$args{predicate}"
            => sub { exists shift->{$attribute} }
      }

      if ($args{handles}) {
         _has_handles($caller, $attribute, \%args);
      }

      if (exists $args{init_arg}) {
         $class_metadata->{$attribute}{init_arg} = $args{init_arg};
      }
   }
}

sub _has_handles {
   my ($caller, $attribute, $args) = @_;
   my $handles = $args->{handles};

   my $ref = ref $handles;
   my $kv;
   if ( $ref eq ref [] ) {
         $kv = { map { $_,$_ } @{$handles} };
   }
   elsif ( $ref eq ref {} ) {
         $kv = $handles;
   }
   elsif ( $ref eq ref qr// ) {
         Carp::confess("Cannot delegate methods based on a Regexp without a type constraint (isa)")
            unless $args->{isa};
         my $target_class = $args->{isa};
         $kv = {
            map   { $_, $_     }
            grep  { $_ =~ $handles }
            grep  { !exists $Lmo::Object::{$_} && $target_class->can($_) }
            grep  { !$export_for{$target_class}->{$_} }
            keys %{ _stash_for $target_class }
         };
   }
   else {
         Carp::confess("handles for $ref not yet implemented");
   }

   while ( my ($method, $target) = each %{$kv} ) {
         my $name = _glob_for "${caller}::$method";
         Carp::confess("You cannot overwrite a locally defined method ($method) with a delegation")
            if defined &$name;

         my ($target, @curried_args) = ref($target) ? @$target : $target;
         *$name = sub {
            my $self        = shift;
            my $delegate_to = $self->$attribute();
            my $error = "Cannot delegate $method to $target because the value of $attribute";
            Carp::confess("$error is not defined") unless $delegate_to;
            Carp::confess("$error is not an object (got '$delegate_to')")
               unless Scalar::Util::blessed($delegate_to) || (!ref($delegate_to) && $delegate_to->can($target));
            return $delegate_to->$target(@curried_args, @_);
         }
   }
}

sub _set_package_isa {
   my ($package, @new_isa) = @_;
   my $package_isa  = \*{ _glob_for "${package}::ISA" };
   @{*$package_isa} = @new_isa;
}

sub _set_inherited_metadata {
   my $class = shift;
   my $class_metadata = Lmo::Meta->metadata_for($class);
   my $linearized_isa = mro::get_linear_isa($class);
   my %new_metadata;

   for my $isa_class (reverse @$linearized_isa) {
      my $isa_metadata = Lmo::Meta->metadata_for($isa_class);
      %new_metadata = (
         %new_metadata,
         %$isa_metadata,
      );
   }
   %$class_metadata = %new_metadata;
}

sub unimport {
   my $caller = scalar caller();
   my $target = caller;
  _unimport_coderefs($target, keys %{$export_for{$caller}});
}

sub Dumper {
   require Data::Dumper;
   local $Data::Dumper::Indent    = 0;
   local $Data::Dumper::Sortkeys  = 0;
   local $Data::Dumper::Quotekeys = 0;
   local $Data::Dumper::Terse     = 1;

   Data::Dumper::Dumper(@_)
}

BEGIN {
   if ($] >= 5.010) {
      { local $@; require mro; }
   }
   else {
      local $@;
      eval {
         require MRO::Compat;
      } or do {
         *mro::get_linear_isa = *mro::get_linear_isa_dfs = sub {
            no strict 'refs';

            my $classname = shift;

            my @lin = ($classname);
            my %stored;
            foreach my $parent (@{"$classname\::ISA"}) {
               my $plin = mro::get_linear_isa_dfs($parent);
               foreach (@$plin) {
                     next if exists $stored{$_};
                     push(@lin, $_);
                     $stored{$_} = 1;
               }
            }
            return \@lin;
         };
      }
   }
}

sub override {
   my ($methods, $code) = @_;
   my $caller          = scalar caller;

   for my $method ( ref($methods) ? @$methods : $methods ) {
      my $full_method     = "${caller}::${method}";
      *{_glob_for $full_method} = $code;
   }
}

}
1;
}
# ###########################################################################
# End Lmo package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Representation package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Representation.pm
#   t/lib/Percona/WebAPI/Representation.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Representation;

eval {
   require JSON;
};

require Exporter;
our @ISA       = qw(Exporter);
our @EXPORT_OK = qw(
   as_hashref
   as_json
   as_config
);

sub as_hashref {
   my ($resource, %args) = @_;

   my $as_hashref = { %$resource };

   if ( !defined $args{with_links} || !$args{with_links} ) {
      delete $as_hashref->{links};
   }

   return $as_hashref;
}

sub as_json {
   my ($resource, %args) = @_;

   my $json = $args{json} || JSON->new;
   $json->allow_blessed([]);
   $json->convert_blessed([]);

   my $text = $json->encode(
      ref $resource eq 'ARRAY' ? $resource : as_hashref($resource, %args)
   );
   if ( $args{json} && $text ) {  # for testing
      chomp($text);
      $text .= "\n";
   }
   return $text;
}

sub as_config {
   my $resource = shift;
   if ( !$resource->isa('Percona::WebAPI::Resource::Config') ) {
      die "Only Config resources can be represented as config.\n";
   }
   my $as_hashref = as_hashref($resource);
   my $options    = $as_hashref->{options};
   my $config     = join("\n",
      map { defined $options->{$_} ?  "$_=$options->{$_}" : "$_" }
      sort keys %$options
   ) . "\n";
   return $config;
}

1;
}
# ###########################################################################
# End Percona::WebAPI::Representation package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Client package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Client.pm
#   t/lib/Percona/WebAPI/Client.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Client;

our $VERSION = '0.01';

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

eval {
   require LWP;
   require JSON;
};

use Scalar::Util qw(blessed); 

use Lmo;
use Percona::Toolkit;
use Percona::WebAPI::Representation;
use Percona::WebAPI::Exception::Request;
use Percona::WebAPI::Exception::Resource;

Percona::WebAPI::Representation->import(qw(as_json));
Percona::Toolkit->import(qw(_d Dumper have_required_args));

has 'api_key' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'entry_link' => (
   is       => 'rw',
   isa      => 'Str',
   required => 0,
   default  => sub { return 'https://cloud-api.percona.com' },
);

has 'ua' => (
   is       => 'rw',
   isa      => 'Object',
   lazy     => 1,
   required => 0,
   builder  => '_build_ua',
);

has 'response' => (
   is       => 'rw',
   isa      => 'Object',
   required => 0,
   default  => undef,
);

sub _build_ua {
   my $self = shift;
   my $ua = LWP::UserAgent->new;
   $ua->agent("Percona::WebAPI::Client/$Percona::WebAPI::Client::VERSION");
   $ua->default_header('Content-Type', 'application/json');
   $ua->default_header('X-Percona-API-Key', $self->api_key);
   return $ua;
}

sub get {
   my ($self, %args) = @_;
   
   have_required_args(\%args, qw(
      link
   )) or die;
   my ($link) = $args{link};

   eval {
      $self->_request(
         method => 'GET',
         link   => $link,
      );
   };
   if ( my $e = $EVAL_ERROR ) {
      if (blessed($e) && $e->isa('Percona::WebAPI::Exception::Request')) {
         die $e;
      }
      else {
         die "Unknown error: $e";
      }
   }

   my $resource = eval {
      JSON::decode_json($self->response->content);
   };
   if ( $EVAL_ERROR ) {
      warn sprintf "Error decoding resource: %s: %s",
         $self->response->content,
         $EVAL_ERROR;
      return;
   }

   my $resource_objects;
   if ( my $type = $self->response->headers->{'x-percona-resource-type'} ) {
      eval {
         $type = "Percona::WebAPI::Resource::$type";
         if ( ref $resource eq 'ARRAY' ) {
            PTDEBUG && _d('Got a list of', $type, 'resources');
            $resource_objects = [];
            foreach my $attribs ( @$resource ) {
               my $obj = $type->new(%$attribs);
               push @$resource_objects, $obj;
            }
         }
         else {
            PTDEBUG && _d('Got a', $type, 'resource', Dumper($resource));
            $resource_objects = $type->new(%$resource);
         }
      };
      if ( my $e = $EVAL_ERROR ) {
         die Percona::WebAPI::Exception::Resource->new(
            type  => $type,
            link  => $link,
            data  => (ref $resource eq 'ARRAY' ? $resource : [ $resource ]),
            error => $e,
         );
      }
   }
   elsif ( exists $resource->{links} ) {
      $resource_objects = $resource->{links};
   }
   elsif ( exists $resource->{pong} ) {
      PTDEBUG && _d("Ping pong!");
   }
   else {
      warn "Did not get X-Percona-Resource-Type or links from $link\n";
   }

   return $resource_objects;
}

sub post {
   my $self = shift;
   $self->_set(
      @_,
      method => 'POST',
   );
   return $self->response->header('Location');
}

sub put {
   my $self = shift;
   $self->_set(
      @_,
      method => 'PUT',
   );
   return $self->response->header('Location');
}

sub delete {
   my ($self, %args) = @_;
   have_required_args(\%args, qw(
      link 
   )) or die;
   my ($link) = $args{link};

   eval {
      $self->_request(
         method  => 'DELETE',
         link    => $link,
         headers => { 'Content-Length' => 0 },
      ); 
   };
   if ( my $e = $EVAL_ERROR ) {
      if (blessed($e) && $e->isa('Percona::WebAPI::Exception::Request')) {
         die $e;
      }
      else {
         die "Unknown error: $e";
      }
   }

   return;
}

sub _set {
   my ($self, %args) = @_;
   have_required_args(\%args, qw(
      method
      resources
      link
   )) or die;
   my $method = $args{method};
   my $res    = $args{resources};
   my $link   = $args{link};

   my $headers = $args{headers};

   my $content = '';
   if ( ref($res) eq 'ARRAY' ) {
      PTDEBUG && _d('List of resources');
      $content = '[' . join(",\n", map { as_json($_) } @$res) . ']';
   }
   elsif ( ref($res) ) {
      PTDEBUG && _d('Resource object');
      $content = as_json($res);
   }
   elsif ( $res !~ m/\n/ && -f $res ) {
      PTDEBUG && _d('List of resources in file', $res);
      $content = '[';
      my $data = do {
         local $INPUT_RECORD_SEPARATOR = undef;
         open my $fh, '<', $res
            or die "Error opening $res: $OS_ERROR";
         <$fh>;
      };
      $data =~ s/,?\s*$/]/;
      $content .= $data;
   }
   else {
      PTDEBUG && _d('Resource text');
      $content = $res;
   }

   eval {
      $self->_request(
         method  => $method,
         link    => $link,
         content => $content,
         headers => $headers,
      );
   };
   if ( my $e = $EVAL_ERROR ) {
      if (blessed($e) && $e->isa('Percona::WebAPI::Exception::Request')) {
         die $e;
      }
      else {
         die "Unknown error: $e";
      }
   }

   return;
}

sub _request {
   my ($self, %args) = @_;

   have_required_args(\%args, qw(
      method
      link 
   )) or die;
   my $method = $args{method};
   my $link   = $args{link};
   
   my $content = $args{content};
   my $headers = $args{headers};

   my $req = HTTP::Request->new($method => $link);
   if ( $content ) {
      $req->content($content);
   }
   if ( $headers ) {
      map { $req->header($_ => $headers->{$_}) } keys %$headers;
   }
   PTDEBUG && _d('Request', $method, $link, Dumper($req));

   my $response = $self->ua->request($req);
   PTDEBUG && _d('Response', Dumper($response));

   $self->response($response);

   if ( !($response->code >= 200 && $response->code < 400) ) {
      die Percona::WebAPI::Exception::Request->new(
         method  => $method,
         url     => $link,
         content => $content,
         status  => $response->code,
         error   => "Failed to $method $link",
      );
   }

   return;
}

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Client package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Exception::Request package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Exception/Request.pm
#   t/lib/Percona/WebAPI/Exception/Request.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Exception::Request;

use Lmo;
use overload '""' => \&as_string;

has 'method' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'url' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'content' => (
   is       => 'ro',
   isa      => 'Maybe[Str]',
   required => 0,
);

has 'status' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'error' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

sub as_string {
   my $self = shift;
   chomp(my $error = $self->error);
   $error =~ s/\n/ /g;
   return sprintf "%s\nRequest: %s %s %s\nStatus: %d\n",
      $error, $self->method, $self->url, $self->content || '', $self->status;
}

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Exception::Request package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Exception::Resource package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Exception/Resource.pm
#   t/lib/Percona/WebAPI/Exception/Resource.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Exception::Resource;

use Lmo;
use overload '""' => \&as_string;
use Data::Dumper;

has 'type' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'link' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'data' => (
   is       => 'ro',
   isa      => 'ArrayRef',
   required => 1,
);

has 'error' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

sub as_string {
   my $self = shift;
   chomp(my $error = $self->error);
   local $Data::Dumper::Indent    = 1;
   local $Data::Dumper::Sortkeys  = 1;
   local $Data::Dumper::Quotekeys = 0;
   return sprintf "Invalid %s resource from %s:\n\n%s\nError: %s\n\n",
      $self->type, $self->link, Dumper($self->data), $error;
}

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Exception::Resource package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Resource::Agent package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Resource/Agent.pm
#   t/lib/Percona/WebAPI/Resource/Agent.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Resource::Agent;

use Lmo;

has 'uuid' => (
   is       => 'ro',
   isa      => 'Str',
   required => 0,
);

has 'username' => (
   is       => 'rw',
   isa      => 'Str',
   required => 0,
   default  => sub { return $ENV{USER} || $ENV{LOGNAME} },
);

has 'hostname' => (
   is       => 'rw',
   isa      => 'Str',
   required => 0,
   default  => sub {
      chomp(my $hostname = `hostname`);
      return $hostname;
   },
);

has 'alias' => (
   is       => 'rw',
   isa      => 'Str',
   required => 0,
);

has 'versions' => (
   is       => 'rw',
   isa      => 'Maybe[HashRef]',
   required => 0,
);

has 'links' => (
   is       => 'rw',
   isa      => 'Maybe[HashRef]',
   required => 0,
   default  => sub { return {} },
);

sub name {
   my ($self) = @_;
   return $self->alias || $self->hostname || $self->uuid || 'Unknown';
}

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Resource::Agent package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Resource::Config package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Resource/Config.pm
#   t/lib/Percona/WebAPI/Resource/Config.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Resource::Config;

use Lmo;

has 'ts' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'name' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'options' => (
   is       => 'ro',
   isa      => 'HashRef',
   required => 1,
);

has 'links' => (
   is       => 'rw',
   isa      => 'Maybe[HashRef]',
   required => 0,
   default  => sub { return {} },
);

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Resource::Config package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Resource::Service package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Resource/Service.pm
#   t/lib/Percona/WebAPI/Resource/Service.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Resource::Service;

use Lmo;

has 'ts' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'name' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'tasks' => (
   is       => 'ro',
   isa      => 'ArrayRef[Percona::WebAPI::Resource::Task]',
   required => 1,
);

has 'run_schedule' => (
   is       => 'ro',
   isa      => 'Str',
   required => 0,
);

has 'spool_schedule' => (
   is       => 'ro',
   isa      => 'Str',
   required => 0,
);

has 'meta' => (
   is       => 'ro',
   isa      => 'Bool',
   required => 0,
   default  => sub { return 0 },
);

has 'run_once' => (
   is       => 'ro',
   isa      => 'Bool',
   required => 0,
   default  => sub { return 0 },
);

has 'links' => (
   is       => 'rw',
   isa      => 'Maybe[HashRef]',
   required => 0,
   default  => sub { return {} },
);

sub BUILDARGS {
   my ($class, %args) = @_;
   if ( ref $args{tasks} eq 'ARRAY' ) {
      my @tasks;
      foreach my $run_hashref ( @{$args{tasks}} ) {
         my $task = Percona::WebAPI::Resource::Task->new(%$run_hashref);
         push @tasks, $task;
      }
      $args{tasks} = \@tasks;
   }
   return $class->SUPER::BUILDARGS(%args);
}

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Resource::Service package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Resource::Task package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Resource/Task.pm
#   t/lib/Percona/WebAPI/Resource/Task.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Resource::Task;

use Lmo;

has 'name' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'number' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'program' => (
   is       => 'ro',
   isa      => 'Maybe[Str]',
   required => 0,
);

has 'query' => (
   is       => 'ro',
   isa      => 'Maybe[Str]',
   required => 0,
);

has 'output' => (
   is       => 'ro',
   isa      => 'Maybe[Str]',
   required => 0,
);

sub TO_JSON { return { %{ shift() } }; }

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Resource::Task package
# ###########################################################################

# ###########################################################################
# Percona::WebAPI::Resource::LogEntry package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/WebAPI/Resource/LogEntry.pm
#   t/lib/Percona/WebAPI/Resource/LogEntry.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::WebAPI::Resource::LogEntry;

use Lmo;

has 'pid' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'service' => (
   is       => 'ro',
   isa      => 'Str',
   required => 0,
);

has 'data_ts' => (
   is       => 'ro',
   isa      => 'Int',
   required => 0,
);

has 'entry_ts' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

has 'log_level' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'message' => (
   is       => 'ro',
   isa      => 'Str',
   required => 1,
);

no Lmo;
1;
}
# ###########################################################################
# End Percona::WebAPI::Resource::LogEntry package
# ###########################################################################

# ###########################################################################
# VersionCheck package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/VersionCheck.pm
#   t/lib/VersionCheck.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package VersionCheck;


use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);

use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use Data::Dumper;
local $Data::Dumper::Indent    = 1;
local $Data::Dumper::Sortkeys  = 1;
local $Data::Dumper::Quotekeys = 0;

use Digest::MD5 qw(md5_hex);
use Sys::Hostname qw(hostname);
use File::Basename qw();
use File::Spec;
use FindBin qw();

eval {
   require Percona::Toolkit;
   require HTTP::Micro;
};

{
   my $file    = 'percona-version-check';
   my $home    = $ENV{HOME} || $ENV{HOMEPATH} || $ENV{USERPROFILE} || '.';
   my @vc_dirs = (
      '/etc/percona',
      '/etc/percona-toolkit',
      '/tmp',
      "$home",
   );

   sub version_check_file {
      foreach my $dir ( @vc_dirs ) {
         if ( -d $dir && -w $dir ) {
            PTDEBUG && _d('Version check file', $file, 'in', $dir);
            return $dir . '/' . $file;
         }
      }
      PTDEBUG && _d('Version check file', $file, 'in', $ENV{PWD});
      return $file;  # in the CWD
   } 
}

sub version_check_time_limit {
   return 60 * 60 * 24;  # one day
}


sub version_check {
   my (%args) = @_;

   my $instances = $args{instances} || [];
   my $instances_to_check;

   PTDEBUG && _d('FindBin::Bin:', $FindBin::Bin);
   if ( !$args{force} ) {
      if ( $FindBin::Bin
           && (-d "$FindBin::Bin/../.bzr" || -d "$FindBin::Bin/../../.bzr") ) {
         PTDEBUG && _d("$FindBin::Bin/../.bzr disables --version-check");
         return;
      }
   }

   eval {
      foreach my $instance ( @$instances ) {
         my ($name, $id) = get_instance_id($instance);
         $instance->{name} = $name;
         $instance->{id}   = $id;
      }

      push @$instances, { name => 'system', id => 0 };

      $instances_to_check = get_instances_to_check(
         instances => $instances,
         vc_file   => $args{vc_file},  # testing
         now       => $args{now},      # testing
      );
      PTDEBUG && _d(scalar @$instances_to_check, 'instances to check');
      return unless @$instances_to_check;

      my $protocol = 'https';  # optimistic, but...
      eval { require IO::Socket::SSL; };
      if ( $EVAL_ERROR ) {
         PTDEBUG && _d($EVAL_ERROR);
         $protocol = 'http';
      }
      PTDEBUG && _d('Using', $protocol);

      my $advice = pingback(
         instances => $instances_to_check,
         protocol  => $protocol,
         url       => $args{url}                       # testing
                   || $ENV{PERCONA_VERSION_CHECK_URL}  # testing
                   || "$protocol://v.percona.com",
      );
      if ( $advice ) {
         PTDEBUG && _d('Advice:', Dumper($advice));
         if ( scalar @$advice > 1) {
            print "\n# " . scalar @$advice . " software updates are "
               . "available:\n";
         }
         else {
            print "\n# A software update is available:\n";
         }
         print join("\n", map { "#   * $_" } @$advice), "\n\n";
      }
   };
   if ( $EVAL_ERROR ) {
      PTDEBUG && _d('Version check failed:', $EVAL_ERROR);
   }

   if ( @$instances_to_check ) {
      eval {
         update_check_times(
            instances => $instances_to_check,
            vc_file   => $args{vc_file},  # testing
            now       => $args{now},      # testing
         );
      };
      if ( $EVAL_ERROR ) {
         PTDEBUG && _d('Error updating version check file:', $EVAL_ERROR);
      }
   }

   if ( $ENV{PTDEBUG_VERSION_CHECK} ) {
      warn "Exiting because the PTDEBUG_VERSION_CHECK "
         . "environment variable is defined.\n";
      exit 255;
   }

   return;
}

sub get_instances_to_check {
   my (%args) = @_;

   my $instances = $args{instances};
   my $now       = $args{now}     || int(time);
   my $vc_file   = $args{vc_file} || version_check_file();

   if ( !-f $vc_file ) {
      PTDEBUG && _d('Version check file', $vc_file, 'does not exist;',
         'version checking all instances');
      return $instances;
   }

   open my $fh, '<', $vc_file or die "Cannot open $vc_file: $OS_ERROR";
   chomp(my $file_contents = do { local $/ = undef; <$fh> });
   PTDEBUG && _d('Version check file', $vc_file, 'contents:', $file_contents);
   close $fh;
   my %last_check_time_for = $file_contents =~ /^([^,]+),(.+)$/mg;

   my $check_time_limit = version_check_time_limit();
   my @instances_to_check;
   foreach my $instance ( @$instances ) {
      my $last_check_time = $last_check_time_for{ $instance->{id} };
      PTDEBUG && _d('Intsance', $instance->{id}, 'last checked',
         $last_check_time, 'now', $now, 'diff', $now - ($last_check_time || 0),
         'hours until next check',
         sprintf '%.2f',
            ($check_time_limit - ($now - ($last_check_time || 0))) / 3600);
      if ( !defined $last_check_time
           || ($now - $last_check_time) >= $check_time_limit ) {
         PTDEBUG && _d('Time to check', Dumper($instance));
         push @instances_to_check, $instance;
      }
   }

   return \@instances_to_check;
}

sub update_check_times {
   my (%args) = @_;

   my $instances = $args{instances};
   my $now       = $args{now}     || int(time);
   my $vc_file   = $args{vc_file} || version_check_file();
   PTDEBUG && _d('Updating last check time:', $now);

   my %all_instances = map {
      $_->{id} => { name => $_->{name}, ts => $now }
   } @$instances;

   if ( -f $vc_file ) {
      open my $fh, '<', $vc_file or die "Cannot read $vc_file: $OS_ERROR";
      my $contents = do { local $/ = undef; <$fh> };
      close $fh;

      foreach my $line ( split("\n", ($contents || '')) ) {
         my ($id, $ts) = split(',', $line);
         if ( !exists $all_instances{$id} ) {
            $all_instances{$id} = { ts => $ts };  # original ts, not updated
         }
      }
   }

   open my $fh, '>', $vc_file or die "Cannot write to $vc_file: $OS_ERROR";
   foreach my $id ( sort keys %all_instances ) {
      PTDEBUG && _d('Updated:', $id, Dumper($all_instances{$id}));
      print { $fh } $id . ',' . $all_instances{$id}->{ts} . "\n";
   }
   close $fh;

   return;
}

sub get_instance_id {
   my ($instance) = @_;

   my $dbh = $instance->{dbh};
   my $dsn = $instance->{dsn};

   my $sql = q{SELECT CONCAT(@@hostname, @@port)};
   PTDEBUG && _d($sql);
   my ($name) = eval { $dbh->selectrow_array($sql) };
   if ( $EVAL_ERROR ) {
      PTDEBUG && _d($EVAL_ERROR);
      $sql = q{SELECT @@hostname};
      PTDEBUG && _d($sql);
      ($name) = eval { $dbh->selectrow_array($sql) };
      if ( $EVAL_ERROR ) {
         PTDEBUG && _d($EVAL_ERROR);
         $name = ($dsn->{h} || 'localhost') . ($dsn->{P} || 3306);
      }
      else {
         $sql = q{SHOW VARIABLES LIKE 'port'};
         PTDEBUG && _d($sql);
         my (undef, $port) = eval { $dbh->selectrow_array($sql) };
         PTDEBUG && _d('port:', $port);
         $name .= $port || '';
      }
   }
   my $id = md5_hex($name);

   PTDEBUG && _d('MySQL instance:', $id, $name, Dumper($dsn));

   return $name, $id;
}


sub pingback {
   my (%args) = @_;
   my @required_args = qw(url instances);
   foreach my $arg ( @required_args ) {
      die "I need a $arg arugment" unless $args{$arg};
   }
   my $url       = $args{url};
   my $instances = $args{instances};

   my $ua = $args{ua} || HTTP::Micro->new( timeout => 3 );

   my $response = $ua->request('GET', $url);
   PTDEBUG && _d('Server response:', Dumper($response));
   die "No response from GET $url"
      if !$response;
   die("GET on $url returned HTTP status $response->{status}; expected 200\n",
       ($response->{content} || '')) if $response->{status} != 200;
   die("GET on $url did not return any programs to check")
      if !$response->{content};

   my $items = parse_server_response(
      response => $response->{content}
   );
   die "Failed to parse server requested programs: $response->{content}"
      if !scalar keys %$items;
      
   my $versions = get_versions(
      items     => $items,
      instances => $instances,
   );
   die "Failed to get any program versions; should have at least gotten Perl"
      if !scalar keys %$versions;

   my $client_content = encode_client_response(
      items      => $items,
      versions   => $versions,
      general_id => md5_hex( hostname() ),
   );

   my $client_response = {
      headers => { "X-Percona-Toolkit-Tool" => File::Basename::basename($0) },
      content => $client_content,
   };
   PTDEBUG && _d('Client response:', Dumper($client_response));

   $response = $ua->request('POST', $url, $client_response);
   PTDEBUG && _d('Server suggestions:', Dumper($response));
   die "No response from POST $url $client_response"
      if !$response;
   die "POST $url returned HTTP status $response->{status}; expected 200"
      if $response->{status} != 200;

   return unless $response->{content};

   $items = parse_server_response(
      response   => $response->{content},
      split_vars => 0,
   );
   die "Failed to parse server suggestions: $response->{content}"
      if !scalar keys %$items;
   my @suggestions = map { $_->{vars} }
                     sort { $a->{item} cmp $b->{item} }
                     values %$items;

   return \@suggestions;
}

sub encode_client_response {
   my (%args) = @_;
   my @required_args = qw(items versions general_id);
   foreach my $arg ( @required_args ) {
      die "I need a $arg arugment" unless $args{$arg};
   }
   my ($items, $versions, $general_id) = @args{@required_args};

   my @lines;
   foreach my $item ( sort keys %$items ) {
      next unless exists $versions->{$item};
      if ( ref($versions->{$item}) eq 'HASH' ) {
         my $mysql_versions = $versions->{$item};
         for my $id ( sort keys %$mysql_versions ) {
            push @lines, join(';', $id, $item, $mysql_versions->{$id});
         }
      }
      else {
         push @lines, join(';', $general_id, $item, $versions->{$item});
      }
   }

   my $client_response = join("\n", @lines) . "\n";
   return $client_response;
}

sub parse_server_response {
   my (%args) = @_;
   my @required_args = qw(response);
   foreach my $arg ( @required_args ) {
      die "I need a $arg arugment" unless $args{$arg};
   }
   my ($response) = @args{@required_args};

   my %items = map {
      my ($item, $type, $vars) = split(";", $_);
      if ( !defined $args{split_vars} || $args{split_vars} ) {
         $vars = [ split(",", ($vars || '')) ];
      }
      $item => {
         item => $item,
         type => $type,
         vars => $vars,
      };
   } split("\n", $response);

   PTDEBUG && _d('Items:', Dumper(\%items));

   return \%items;
}

my %sub_for_type = (
   os_version          => \&get_os_version,
   perl_version        => \&get_perl_version,
   perl_module_version => \&get_perl_module_version,
   mysql_variable      => \&get_mysql_variable,
);

sub valid_item {
   my ($item) = @_;
   return unless $item;
   if ( !exists $sub_for_type{ $item->{type} } ) {
      PTDEBUG && _d('Invalid type:', $item->{type});
      return 0;
   }
   return 1;
}

sub get_versions {
   my (%args) = @_;
   my @required_args = qw(items);
   foreach my $arg ( @required_args ) {
      die "I need a $arg arugment" unless $args{$arg};
   }
   my ($items) = @args{@required_args};

   my %versions;
   foreach my $item ( values %$items ) {
      next unless valid_item($item);
      eval {
         my $version = $sub_for_type{ $item->{type} }->(
            item      => $item,
            instances => $args{instances},
         );
         if ( $version ) {
            chomp $version unless ref($version);
            $versions{$item->{item}} = $version;
         }
      };
      if ( $EVAL_ERROR ) {
         PTDEBUG && _d('Error getting version for', Dumper($item), $EVAL_ERROR);
      }
   }

   return \%versions;
}


sub get_os_version {
   if ( $OSNAME eq 'MSWin32' ) {
      require Win32;
      return Win32::GetOSDisplayName();
   }

  chomp(my $platform = `uname -s`);
  PTDEBUG && _d('platform:', $platform);
  return $OSNAME unless $platform;

   chomp(my $lsb_release
            = `which lsb_release 2>/dev/null | awk '{print \$1}'` || '');
   PTDEBUG && _d('lsb_release:', $lsb_release);

   my $release = "";

   if ( $platform eq 'Linux' ) {
      if ( -f "/etc/fedora-release" ) {
         $release = `cat /etc/fedora-release`;
      }
      elsif ( -f "/etc/redhat-release" ) {
         $release = `cat /etc/redhat-release`;
      }
      elsif ( -f "/etc/system-release" ) {
         $release = `cat /etc/system-release`;
      }
      elsif ( $lsb_release ) {
         $release = `$lsb_release -ds`;
      }
      elsif ( -f "/etc/lsb-release" ) {
         $release = `grep DISTRIB_DESCRIPTION /etc/lsb-release`;
         $release =~ s/^\w+="([^"]+)".+/$1/;
      }
      elsif ( -f "/etc/debian_version" ) {
         chomp(my $rel = `cat /etc/debian_version`);
         $release = "Debian $rel";
         if ( -f "/etc/apt/sources.list" ) {
             chomp(my $code_name = `awk '/^deb/ {print \$3}' /etc/apt/sources.list | awk -F/ '{print \$1}'| awk 'BEGIN {FS="|"} {print \$1}' | sort | uniq -c | sort -rn | head -n1 | awk '{print \$2}'`);
             $release .= " ($code_name)" if $code_name;
         }
      }
      elsif ( -f "/etc/os-release" ) { # openSUSE
         chomp($release = `grep PRETTY_NAME /etc/os-release`);
         $release =~ s/^PRETTY_NAME="(.+)"$/$1/;
      }
      elsif ( `ls /etc/*release 2>/dev/null` ) {
         if ( `grep DISTRIB_DESCRIPTION /etc/*release 2>/dev/null` ) {
            $release = `grep DISTRIB_DESCRIPTION /etc/*release | head -n1`;
         }
         else {
            $release = `cat /etc/*release | head -n1`;
         }
      }
   }
   elsif ( $platform =~ m/(?:BSD|^Darwin)$/ ) {
      my $rel = `uname -r`;
      $release = "$platform $rel";
   }
   elsif ( $platform eq "SunOS" ) {
      my $rel = `head -n1 /etc/release` || `uname -r`;
      $release = "$platform $rel";
   }

   if ( !$release ) {
      PTDEBUG && _d('Failed to get the release, using platform');
      $release = $platform;
   }
   chomp($release);

   $release =~ s/^"|"$//g;

   PTDEBUG && _d('OS version =', $release);
   return $release;
}

sub get_perl_version {
   my (%args) = @_;
   my $item = $args{item};
   return unless $item;

   my $version = sprintf '%vd', $PERL_VERSION;
   PTDEBUG && _d('Perl version', $version);
   return $version;
}

sub get_perl_module_version {
   my (%args) = @_;
   my $item = $args{item};
   return unless $item;

   my $var     = '$' . $item->{item} . '::VERSION';
   my $version = eval "use $item->{item}; $var;";
   PTDEBUG && _d('Perl version for', $var, '=', $version);
   return $version;
}

sub get_mysql_variable {
   return get_from_mysql(
      show => 'VARIABLES',
      @_,
   );
}

sub get_from_mysql {
   my (%args) = @_;
   my $show      = $args{show};
   my $item      = $args{item};
   my $instances = $args{instances};
   return unless $show && $item;

   if ( !$instances || !@$instances ) {
      PTDEBUG && _d('Cannot check', $item,
         'because there are no MySQL instances');
      return;
   }

   my @versions;
   my %version_for;
   foreach my $instance ( @$instances ) {
      next unless $instance->{id};  # special system instance has id=0
      my $dbh = $instance->{dbh};
      local $dbh->{FetchHashKeyName} = 'NAME_lc';
      my $sql = qq/SHOW $show/;
      PTDEBUG && _d($sql);
      my $rows = $dbh->selectall_hashref($sql, 'variable_name');

      my @versions;
      foreach my $var ( @{$item->{vars}} ) {
         $var = lc($var);
         my $version = $rows->{$var}->{value};
         PTDEBUG && _d('MySQL version for', $item->{item}, '=', $version,
            'on', $instance->{name});
         push @versions, $version;
      }
      $version_for{ $instance->{id} } = join(' ', @versions);
   }

   return \%version_for;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End VersionCheck package
# ###########################################################################

# ###########################################################################
# DSNParser package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/DSNParser.pm
#   t/lib/DSNParser.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package DSNParser;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use Data::Dumper;
$Data::Dumper::Indent    = 0;
$Data::Dumper::Quotekeys = 0;

my $dsn_sep = qr/(?<!\\),/;

eval {
   require DBI;
};
my $have_dbi = $EVAL_ERROR ? 0 : 1;

sub new {
   my ( $class, %args ) = @_;
   foreach my $arg ( qw(opts) ) {
      die "I need a $arg argument" unless $args{$arg};
   }
   my $self = {
      opts => {}  # h, P, u, etc.  Should come from DSN OPTIONS section in POD.
   };
   foreach my $opt ( @{$args{opts}} ) {
      if ( !$opt->{key} || !$opt->{desc} ) {
         die "Invalid DSN option: ", Dumper($opt);
      }
      PTDEBUG && _d('DSN option:',
         join(', ',
            map { "$_=" . (defined $opt->{$_} ? ($opt->{$_} || '') : 'undef') }
               keys %$opt
         )
      );
      $self->{opts}->{$opt->{key}} = {
         dsn  => $opt->{dsn},
         desc => $opt->{desc},
         copy => $opt->{copy} || 0,
      };
   }
   return bless $self, $class;
}

sub prop {
   my ( $self, $prop, $value ) = @_;
   if ( @_ > 2 ) {
      PTDEBUG && _d('Setting', $prop, 'property');
      $self->{$prop} = $value;
   }
   return $self->{$prop};
}

sub parse {
   my ( $self, $dsn, $prev, $defaults ) = @_;
   if ( !$dsn ) {
      PTDEBUG && _d('No DSN to parse');
      return;
   }
   PTDEBUG && _d('Parsing', $dsn);
   $prev     ||= {};
   $defaults ||= {};
   my %given_props;
   my %final_props;
   my $opts = $self->{opts};

   foreach my $dsn_part ( split($dsn_sep, $dsn) ) {
      $dsn_part =~ s/\\,/,/g;
      if ( my ($prop_key, $prop_val) = $dsn_part =~  m/^(.)=(.*)$/ ) {
         $given_props{$prop_key} = $prop_val;
      }
      else {
         PTDEBUG && _d('Interpreting', $dsn_part, 'as h=', $dsn_part);
         $given_props{h} = $dsn_part;
      }
   }

   foreach my $key ( keys %$opts ) {
      PTDEBUG && _d('Finding value for', $key);
      $final_props{$key} = $given_props{$key};
      if (   !defined $final_props{$key}
           && defined $prev->{$key} && $opts->{$key}->{copy} )
      {
         $final_props{$key} = $prev->{$key};
         PTDEBUG && _d('Copying value for', $key, 'from previous DSN');
      }
      if ( !defined $final_props{$key} ) {
         $final_props{$key} = $defaults->{$key};
         PTDEBUG && _d('Copying value for', $key, 'from defaults');
      }
   }

   foreach my $key ( keys %given_props ) {
      die "Unknown DSN option '$key' in '$dsn'.  For more details, "
            . "please use the --help option, or try 'perldoc $PROGRAM_NAME' "
            . "for complete documentation."
         unless exists $opts->{$key};
   }
   if ( (my $required = $self->prop('required')) ) {
      foreach my $key ( keys %$required ) {
         die "Missing required DSN option '$key' in '$dsn'.  For more details, "
               . "please use the --help option, or try 'perldoc $PROGRAM_NAME' "
               . "for complete documentation."
            unless $final_props{$key};
      }
   }

   return \%final_props;
}

sub parse_options {
   my ( $self, $o ) = @_;
   die 'I need an OptionParser object' unless ref $o eq 'OptionParser';
   my $dsn_string
      = join(',',
          map  { "$_=".$o->get($_); }
          grep { $o->has($_) && $o->get($_) }
          keys %{$self->{opts}}
        );
   PTDEBUG && _d('DSN string made from options:', $dsn_string);
   return $self->parse($dsn_string);
}

sub as_string {
   my ( $self, $dsn, $props ) = @_;
   return $dsn unless ref $dsn;
   my @keys = $props ? @$props : sort keys %$dsn;
   return join(',',
      map  { "$_=" . ($_ eq 'p' ? '...' : $dsn->{$_}) }
      grep {
         exists $self->{opts}->{$_}
         && exists $dsn->{$_}
         && defined $dsn->{$_}
      } @keys);
}

sub usage {
   my ( $self ) = @_;
   my $usage
      = "DSN syntax is key=value[,key=value...]  Allowable DSN keys:\n\n"
      . "  KEY  COPY  MEANING\n"
      . "  ===  ====  =============================================\n";
   my %opts = %{$self->{opts}};
   foreach my $key ( sort keys %opts ) {
      $usage .= "  $key    "
             .  ($opts{$key}->{copy} ? 'yes   ' : 'no    ')
             .  ($opts{$key}->{desc} || '[No description]')
             . "\n";
   }
   $usage .= "\n  If the DSN is a bareword, the word is treated as the 'h' key.\n";
   return $usage;
}

sub get_cxn_params {
   my ( $self, $info ) = @_;
   my $dsn;
   my %opts = %{$self->{opts}};
   my $driver = $self->prop('dbidriver') || '';
   if ( $driver eq 'Pg' ) {
      $dsn = 'DBI:Pg:dbname=' . ( $info->{D} || '' ) . ';'
         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
                     grep { defined $info->{$_} }
                     qw(h P));
   }
   else {
      $dsn = 'DBI:mysql:' . ( $info->{D} || '' ) . ';'
         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
                     grep { defined $info->{$_} }
                     qw(F h P S A))
         . ';mysql_read_default_group=client'
         . ($info->{L} ? ';mysql_local_infile=1' : '');
   }
   PTDEBUG && _d($dsn);
   return ($dsn, $info->{u}, $info->{p});
}

sub fill_in_dsn {
   my ( $self, $dbh, $dsn ) = @_;
   my $vars = $dbh->selectall_hashref('SHOW VARIABLES', 'Variable_name');
   my ($user, $db) = $dbh->selectrow_array('SELECT USER(), DATABASE()');
   $user =~ s/@.*//;
   $dsn->{h} ||= $vars->{hostname}->{Value};
   $dsn->{S} ||= $vars->{'socket'}->{Value};
   $dsn->{P} ||= $vars->{port}->{Value};
   $dsn->{u} ||= $user;
   $dsn->{D} ||= $db;
}

sub get_dbh {
   my ( $self, $cxn_string, $user, $pass, $opts ) = @_;
   $opts ||= {};
   my $defaults = {
      AutoCommit         => 0,
      RaiseError         => 1,
      PrintError         => 0,
      ShowErrorStatement => 1,
      mysql_enable_utf8 => ($cxn_string =~ m/charset=utf8/i ? 1 : 0),
   };
   @{$defaults}{ keys %$opts } = values %$opts;
   if (delete $defaults->{L}) { # L for LOAD DATA LOCAL INFILE, our own extension
      $defaults->{mysql_local_infile} = 1;
   }

   if ( $opts->{mysql_use_result} ) {
      $defaults->{mysql_use_result} = 1;
   }

   if ( !$have_dbi ) {
      die "Cannot connect to MySQL because the Perl DBI module is not "
         . "installed or not found.  Run 'perl -MDBI' to see the directories "
         . "that Perl searches for DBI.  If DBI is not installed, try:\n"
         . "  Debian/Ubuntu  apt-get install libdbi-perl\n"
         . "  RHEL/CentOS    yum install perl-DBI\n"
         . "  OpenSolaris    pkg install pkg:/SUNWpmdbi\n";

   }

   my $dbh;
   my $tries = 2;
   while ( !$dbh && $tries-- ) {
      PTDEBUG && _d($cxn_string, ' ', $user, ' ', $pass, 
         join(', ', map { "$_=>$defaults->{$_}" } keys %$defaults ));

      $dbh = eval { DBI->connect($cxn_string, $user, $pass, $defaults) };

      if ( !$dbh && $EVAL_ERROR ) {
         if ( $EVAL_ERROR =~ m/locate DBD\/mysql/i ) {
            die "Cannot connect to MySQL because the Perl DBD::mysql module is "
               . "not installed or not found.  Run 'perl -MDBD::mysql' to see "
               . "the directories that Perl searches for DBD::mysql.  If "
               . "DBD::mysql is not installed, try:\n"
               . "  Debian/Ubuntu  apt-get install libdbd-mysql-perl\n"
               . "  RHEL/CentOS    yum install perl-DBD-MySQL\n"
               . "  OpenSolaris    pgk install pkg:/SUNWapu13dbd-mysql\n";
         }
         elsif ( $EVAL_ERROR =~ m/not a compiled character set|character set utf8/ ) {
            PTDEBUG && _d('Going to try again without utf8 support');
            delete $defaults->{mysql_enable_utf8};
         }
         if ( !$tries ) {
            die $EVAL_ERROR;
         }
      }
   }

   if ( $cxn_string =~ m/mysql/i ) {
      my $sql;

      $sql = 'SELECT @@SQL_MODE';
      PTDEBUG && _d($dbh, $sql);
      my ($sql_mode) = eval { $dbh->selectrow_array($sql) };
      if ( $EVAL_ERROR ) {
         die "Error getting the current SQL_MODE: $EVAL_ERROR";
      }

      if ( my ($charset) = $cxn_string =~ m/charset=([\w]+)/ ) {
         $sql = qq{/*!40101 SET NAMES "$charset"*/};
         PTDEBUG && _d($dbh, $sql);
         eval { $dbh->do($sql) };
         if ( $EVAL_ERROR ) {
            die "Error setting NAMES to $charset: $EVAL_ERROR";
         }
         PTDEBUG && _d('Enabling charset for STDOUT');
         if ( $charset eq 'utf8' ) {
            binmode(STDOUT, ':utf8')
               or die "Can't binmode(STDOUT, ':utf8'): $OS_ERROR";
         }
         else {
            binmode(STDOUT) or die "Can't binmode(STDOUT): $OS_ERROR";
         }
      }

      if ( my $vars = $self->prop('set-vars') ) {
         $self->set_vars($dbh, $vars);
      }

      $sql = 'SET @@SQL_QUOTE_SHOW_CREATE = 1'
            . '/*!40101, @@SQL_MODE=\'NO_AUTO_VALUE_ON_ZERO'
            . ($sql_mode ? ",$sql_mode" : '')
            . '\'*/';
      PTDEBUG && _d($dbh, $sql);
      eval { $dbh->do($sql) };
      if ( $EVAL_ERROR ) {
         die "Error setting SQL_QUOTE_SHOW_CREATE, SQL_MODE"
           . ($sql_mode ? " and $sql_mode" : '')
           . ": $EVAL_ERROR";
      }
   }

   PTDEBUG && _d('DBH info: ',
      $dbh,
      Dumper($dbh->selectrow_hashref(
         'SELECT DATABASE(), CONNECTION_ID(), VERSION()/*!50038 , @@hostname*/')),
      'Connection info:',      $dbh->{mysql_hostinfo},
      'Character set info:',   Dumper($dbh->selectall_arrayref(
                     "SHOW VARIABLES LIKE 'character_set%'", { Slice => {}})),
      '$DBD::mysql::VERSION:', $DBD::mysql::VERSION,
      '$DBI::VERSION:',        $DBI::VERSION,
   );

   return $dbh;
}

sub get_hostname {
   my ( $self, $dbh ) = @_;
   if ( my ($host) = ($dbh->{mysql_hostinfo} || '') =~ m/^(\w+) via/ ) {
      return $host;
   }
   my ( $hostname, $one ) = $dbh->selectrow_array(
      'SELECT /*!50038 @@hostname, */ 1');
   return $hostname;
}

sub disconnect {
   my ( $self, $dbh ) = @_;
   PTDEBUG && $self->print_active_handles($dbh);
   $dbh->disconnect;
}

sub print_active_handles {
   my ( $self, $thing, $level ) = @_;
   $level ||= 0;
   printf("# Active %sh: %s %s %s\n", ($thing->{Type} || 'undef'), "\t" x $level,
      $thing, (($thing->{Type} || '') eq 'st' ? $thing->{Statement} || '' : ''))
      or die "Cannot print: $OS_ERROR";
   foreach my $handle ( grep {defined} @{ $thing->{ChildHandles} } ) {
      $self->print_active_handles( $handle, $level + 1 );
   }
}

sub copy {
   my ( $self, $dsn_1, $dsn_2, %args ) = @_;
   die 'I need a dsn_1 argument' unless $dsn_1;
   die 'I need a dsn_2 argument' unless $dsn_2;
   my %new_dsn = map {
      my $key = $_;
      my $val;
      if ( $args{overwrite} ) {
         $val = defined $dsn_1->{$key} ? $dsn_1->{$key} : $dsn_2->{$key};
      }
      else {
         $val = defined $dsn_2->{$key} ? $dsn_2->{$key} : $dsn_1->{$key};
      }
      $key => $val;
   } keys %{$self->{opts}};
   return \%new_dsn;
}

sub set_vars {
   my ($self, $dbh, $vars) = @_;

   return unless $vars;

   foreach my $var ( sort keys %$vars ) {
      my $val = $vars->{$var}->{val};

      (my $quoted_var = $var) =~ s/_/\\_/;
      my ($var_exists, $current_val);
      eval {
         ($var_exists, $current_val) = $dbh->selectrow_array(
            "SHOW VARIABLES LIKE '$quoted_var'");
      };
      my $e = $EVAL_ERROR;
      if ( $e ) {
         PTDEBUG && _d($e);
      }

      if ( $vars->{$var}->{default} && !$var_exists ) {
         PTDEBUG && _d('Not setting default var', $var,
            'because it does not exist');
         next;
      }

      if ( $current_val && $current_val eq $val ) {
         PTDEBUG && _d('Not setting var', $var, 'because its value',
            'is already', $val);
         next;
      }

      my $sql = "SET SESSION $var=$val";
      PTDEBUG && _d($dbh, $sql);
      eval { $dbh->do($sql) };
      if ( my $set_error = $EVAL_ERROR ) {
         chomp($set_error);
         $set_error =~ s/ at \S+ line \d+//;
         my $msg = "Error setting $var: $set_error";
         if ( $current_val ) {
            $msg .= "  The current value for $var is $current_val.  "
                  . "If the variable is read only (not dynamic), specify "
                  . "--set-vars $var=$current_val to avoid this warning, "
                  . "else manually set the variable and restart MySQL.";
         }
         warn $msg . "\n\n";
      }
   }

   return; 
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End DSNParser package
# ###########################################################################

# ###########################################################################
# OptionParser package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/OptionParser.pm
#   t/lib/OptionParser.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package OptionParser;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use List::Util qw(max);
use Getopt::Long;
use Data::Dumper;

my $POD_link_re = '[LC]<"?([^">]+)"?>';

sub new {
   my ( $class, %args ) = @_;
   my @required_args = qw();
   foreach my $arg ( @required_args ) {
      die "I need a $arg argument" unless $args{$arg};
   }

   my ($program_name) = $PROGRAM_NAME =~ m/([.A-Za-z-]+)$/;
   $program_name ||= $PROGRAM_NAME;
   my $home = $ENV{HOME} || $ENV{HOMEPATH} || $ENV{USERPROFILE} || '.';

   my %attributes = (
      'type'       => 1,
      'short form' => 1,
      'group'      => 1,
      'default'    => 1,
      'cumulative' => 1,
      'negatable'  => 1,
   );

   my $self = {
      head1             => 'OPTIONS',        # These args are used internally
      skip_rules        => 0,                # to instantiate another Option-
      item              => '--(.*)',         # Parser obj that parses the
      attributes        => \%attributes,     # DSN OPTIONS section.  Tools
      parse_attributes  => \&_parse_attribs, # don't tinker with these args.

      %args,

      strict            => 1,  # disabled by a special rule
      program_name      => $program_name,
      opts              => {},
      got_opts          => 0,
      short_opts        => {},
      defaults          => {},
      groups            => {},
      allowed_groups    => {},
      errors            => [],
      rules             => [],  # desc of rules for --help
      mutex             => [],  # rule: opts are mutually exclusive
      atleast1          => [],  # rule: at least one opt is required
      disables          => {},  # rule: opt disables other opts 
      defaults_to       => {},  # rule: opt defaults to value of other opt
      DSNParser         => undef,
      default_files     => [
         "/etc/percona-toolkit/percona-toolkit.conf",
         "/etc/percona-toolkit/$program_name.conf",
         "$home/.percona-toolkit.conf",
         "$home/.$program_name.conf",
      ],
      types             => {
         string => 's', # standard Getopt type
         int    => 'i', # standard Getopt type
         float  => 'f', # standard Getopt type
         Hash   => 'H', # hash, formed from a comma-separated list
         hash   => 'h', # hash as above, but only if a value is given
         Array  => 'A', # array, similar to Hash
         array  => 'a', # array, similar to hash
         DSN    => 'd', # DSN
         size   => 'z', # size with kMG suffix (powers of 2^10)
         time   => 'm', # time, with an optional suffix of s/h/m/d
      },
   };

   return bless $self, $class;
}

sub get_specs {
   my ( $self, $file ) = @_;
   $file ||= $self->{file} || __FILE__;
   my @specs = $self->_pod_to_specs($file);
   $self->_parse_specs(@specs);

   open my $fh, "<", $file or die "Cannot open $file: $OS_ERROR";
   my $contents = do { local $/ = undef; <$fh> };
   close $fh;
   if ( $contents =~ m/^=head1 DSN OPTIONS/m ) {
      PTDEBUG && _d('Parsing DSN OPTIONS');
      my $dsn_attribs = {
         dsn  => 1,
         copy => 1,
      };
      my $parse_dsn_attribs = sub {
         my ( $self, $option, $attribs ) = @_;
         map {
            my $val = $attribs->{$_};
            if ( $val ) {
               $val    = $val eq 'yes' ? 1
                       : $val eq 'no'  ? 0
                       :                 $val;
               $attribs->{$_} = $val;
            }
         } keys %$attribs;
         return {
            key => $option,
            %$attribs,
         };
      };
      my $dsn_o = new OptionParser(
         description       => 'DSN OPTIONS',
         head1             => 'DSN OPTIONS',
         dsn               => 0,         # XXX don't infinitely recurse!
         item              => '\* (.)',  # key opts are a single character
         skip_rules        => 1,         # no rules before opts
         attributes        => $dsn_attribs,
         parse_attributes  => $parse_dsn_attribs,
      );
      my @dsn_opts = map {
         my $opts = {
            key  => $_->{spec}->{key},
            dsn  => $_->{spec}->{dsn},
            copy => $_->{spec}->{copy},
            desc => $_->{desc},
         };
         $opts;
      } $dsn_o->_pod_to_specs($file);
      $self->{DSNParser} = DSNParser->new(opts => \@dsn_opts);
   }

   if ( $contents =~ m/^=head1 VERSION\n\n^(.+)$/m ) {
      $self->{version} = $1;
      PTDEBUG && _d($self->{version});
   }

   return;
}

sub DSNParser {
   my ( $self ) = @_;
   return $self->{DSNParser};
};

sub get_defaults_files {
   my ( $self ) = @_;
   return @{$self->{default_files}};
}

sub _pod_to_specs {
   my ( $self, $file ) = @_;
   $file ||= $self->{file} || __FILE__;
   open my $fh, '<', $file or die "Cannot open $file: $OS_ERROR";

   my @specs = ();
   my @rules = ();
   my $para;

   local $INPUT_RECORD_SEPARATOR = '';
   while ( $para = <$fh> ) {
      next unless $para =~ m/^=head1 $self->{head1}/;
      last;
   }

   while ( $para = <$fh> ) {
      last if $para =~ m/^=over/;
      next if $self->{skip_rules};
      chomp $para;
      $para =~ s/\s+/ /g;
      $para =~ s/$POD_link_re/$1/go;
      PTDEBUG && _d('Option rule:', $para);
      push @rules, $para;
   }

   die "POD has no $self->{head1} section" unless $para;

   do {
      if ( my ($option) = $para =~ m/^=item $self->{item}/ ) {
         chomp $para;
         PTDEBUG && _d($para);
         my %attribs;

         $para = <$fh>; # read next paragraph, possibly attributes

         if ( $para =~ m/: / ) { # attributes
            $para =~ s/\s+\Z//g;
            %attribs = map {
                  my ( $attrib, $val) = split(/: /, $_);
                  die "Unrecognized attribute for --$option: $attrib"
                     unless $self->{attributes}->{$attrib};
                  ($attrib, $val);
               } split(/; /, $para);
            if ( $attribs{'short form'} ) {
               $attribs{'short form'} =~ s/-//;
            }
            $para = <$fh>; # read next paragraph, probably short help desc
         }
         else {
            PTDEBUG && _d('Option has no attributes');
         }

         $para =~ s/\s+\Z//g;
         $para =~ s/\s+/ /g;
         $para =~ s/$POD_link_re/$1/go;

         $para =~ s/\.(?:\n.*| [A-Z].*|\Z)//s;
         PTDEBUG && _d('Short help:', $para);

         die "No description after option spec $option" if $para =~ m/^=item/;

         if ( my ($base_option) =  $option =~ m/^\[no\](.*)/ ) {
            $option = $base_option;
            $attribs{'negatable'} = 1;
         }

         push @specs, {
            spec  => $self->{parse_attributes}->($self, $option, \%attribs), 
            desc  => $para
               . (defined $attribs{default} ? " (default $attribs{default})" : ''),
            group => ($attribs{'group'} ? $attribs{'group'} : 'default'),
         };
      }
      while ( $para = <$fh> ) {
         last unless $para;
         if ( $para =~ m/^=head1/ ) {
            $para = undef; # Can't 'last' out of a do {} block.
            last;
         }
         last if $para =~ m/^=item /;
      }
   } while ( $para );

   die "No valid specs in $self->{head1}" unless @specs;

   close $fh;
   return @specs, @rules;
}

sub _parse_specs {
   my ( $self, @specs ) = @_;
   my %disables; # special rule that requires deferred checking

   foreach my $opt ( @specs ) {
      if ( ref $opt ) { # It's an option spec, not a rule.
         PTDEBUG && _d('Parsing opt spec:',
            map { ($_, '=>', $opt->{$_}) } keys %$opt);

         my ( $long, $short ) = $opt->{spec} =~ m/^([\w-]+)(?:\|([^!+=]*))?/;
         if ( !$long ) {
            die "Cannot parse long option from spec $opt->{spec}";
         }
         $opt->{long} = $long;

         die "Duplicate long option --$long" if exists $self->{opts}->{$long};
         $self->{opts}->{$long} = $opt;

         if ( length $long == 1 ) {
            PTDEBUG && _d('Long opt', $long, 'looks like short opt');
            $self->{short_opts}->{$long} = $long;
         }

         if ( $short ) {
            die "Duplicate short option -$short"
               if exists $self->{short_opts}->{$short};
            $self->{short_opts}->{$short} = $long;
            $opt->{short} = $short;
         }
         else {
            $opt->{short} = undef;
         }

         $opt->{is_negatable}  = $opt->{spec} =~ m/!/        ? 1 : 0;
         $opt->{is_cumulative} = $opt->{spec} =~ m/\+/       ? 1 : 0;
         $opt->{is_required}   = $opt->{desc} =~ m/required/ ? 1 : 0;

         $opt->{group} ||= 'default';
         $self->{groups}->{ $opt->{group} }->{$long} = 1;

         $opt->{value} = undef;
         $opt->{got}   = 0;

         my ( $type ) = $opt->{spec} =~ m/=(.)/;
         $opt->{type} = $type;
         PTDEBUG && _d($long, 'type:', $type);


         $opt->{spec} =~ s/=./=s/ if ( $type && $type =~ m/[HhAadzm]/ );

         if ( (my ($def) = $opt->{desc} =~ m/default\b(?: ([^)]+))?/) ) {
            $self->{defaults}->{$long} = defined $def ? $def : 1;
            PTDEBUG && _d($long, 'default:', $def);
         }

         if ( $long eq 'config' ) {
            $self->{defaults}->{$long} = join(',', $self->get_defaults_files());
         }

         if ( (my ($dis) = $opt->{desc} =~ m/(disables .*)/) ) {
            $disables{$long} = $dis;
            PTDEBUG && _d('Deferring check of disables rule for', $opt, $dis);
         }

         $self->{opts}->{$long} = $opt;
      }
      else { # It's an option rule, not a spec.
         PTDEBUG && _d('Parsing rule:', $opt); 
         push @{$self->{rules}}, $opt;
         my @participants = $self->_get_participants($opt);
         my $rule_ok = 0;

         if ( $opt =~ m/mutually exclusive|one and only one/ ) {
            $rule_ok = 1;
            push @{$self->{mutex}}, \@participants;
            PTDEBUG && _d(@participants, 'are mutually exclusive');
         }
         if ( $opt =~ m/at least one|one and only one/ ) {
            $rule_ok = 1;
            push @{$self->{atleast1}}, \@participants;
            PTDEBUG && _d(@participants, 'require at least one');
         }
         if ( $opt =~ m/default to/ ) {
            $rule_ok = 1;
            $self->{defaults_to}->{$participants[0]} = $participants[1];
            PTDEBUG && _d($participants[0], 'defaults to', $participants[1]);
         }
         if ( $opt =~ m/restricted to option groups/ ) {
            $rule_ok = 1;
            my ($groups) = $opt =~ m/groups ([\w\s\,]+)/;
            my @groups = split(',', $groups);
            %{$self->{allowed_groups}->{$participants[0]}} = map {
               s/\s+//;
               $_ => 1;
            } @groups;
         }
         if( $opt =~ m/accepts additional command-line arguments/ ) {
            $rule_ok = 1;
            $self->{strict} = 0;
            PTDEBUG && _d("Strict mode disabled by rule");
         }

         die "Unrecognized option rule: $opt" unless $rule_ok;
      }
   }

   foreach my $long ( keys %disables ) {
      my @participants = $self->_get_participants($disables{$long});
      $self->{disables}->{$long} = \@participants;
      PTDEBUG && _d('Option', $long, 'disables', @participants);
   }

   return; 
}

sub _get_participants {
   my ( $self, $str ) = @_;
   my @participants;
   foreach my $long ( $str =~ m/--(?:\[no\])?([\w-]+)/g ) {
      die "Option --$long does not exist while processing rule $str"
         unless exists $self->{opts}->{$long};
      push @participants, $long;
   }
   PTDEBUG && _d('Participants for', $str, ':', @participants);
   return @participants;
}

sub opts {
   my ( $self ) = @_;
   my %opts = %{$self->{opts}};
   return %opts;
}

sub short_opts {
   my ( $self ) = @_;
   my %short_opts = %{$self->{short_opts}};
   return %short_opts;
}

sub set_defaults {
   my ( $self, %defaults ) = @_;
   $self->{defaults} = {};
   foreach my $long ( keys %defaults ) {
      die "Cannot set default for nonexistent option $long"
         unless exists $self->{opts}->{$long};
      $self->{defaults}->{$long} = $defaults{$long};
      PTDEBUG && _d('Default val for', $long, ':', $defaults{$long});
   }
   return;
}

sub get_defaults {
   my ( $self ) = @_;
   return $self->{defaults};
}

sub get_groups {
   my ( $self ) = @_;
   return $self->{groups};
}

sub _set_option {
   my ( $self, $opt, $val ) = @_;
   my $long = exists $self->{opts}->{$opt}       ? $opt
            : exists $self->{short_opts}->{$opt} ? $self->{short_opts}->{$opt}
            : die "Getopt::Long gave a nonexistent option: $opt";
   $opt = $self->{opts}->{$long};
   if ( $opt->{is_cumulative} ) {
      $opt->{value}++;
   }
   elsif ( ($opt->{type} || '') eq 's' && $val =~ m/^--?(.+)/ ) {
      my $next_opt = $1;
      if (    exists $self->{opts}->{$next_opt}
           || exists $self->{short_opts}->{$next_opt} ) {
         $self->save_error("--$long requires a string value");
         return;
      }
      else {
         $opt->{value} = $val;
      }
   }
   else {
      $opt->{value} = $val;
   }
   $opt->{got} = 1;
   PTDEBUG && _d('Got option', $long, '=', $val);
}

sub get_opts {
   my ( $self ) = @_; 

   foreach my $long ( keys %{$self->{opts}} ) {
      $self->{opts}->{$long}->{got} = 0;
      $self->{opts}->{$long}->{value}
         = exists $self->{defaults}->{$long}       ? $self->{defaults}->{$long}
         : $self->{opts}->{$long}->{is_cumulative} ? 0
         : undef;
   }
   $self->{got_opts} = 0;

   $self->{errors} = [];

   if ( @ARGV && $ARGV[0] eq "--config" ) {
      shift @ARGV;
      $self->_set_option('config', shift @ARGV);
   }
   if ( $self->has('config') ) {
      my @extra_args;
      foreach my $filename ( split(',', $self->get('config')) ) {
         eval {
            push @extra_args, $self->_read_config_file($filename);
         };
         if ( $EVAL_ERROR ) {
            if ( $self->got('config') ) {
               die $EVAL_ERROR;
            }
            elsif ( PTDEBUG ) {
               _d($EVAL_ERROR);
            }
         }
      }
      unshift @ARGV, @extra_args;
   }

   Getopt::Long::Configure('no_ignore_case', 'bundling');
   GetOptions(
      map    { $_->{spec} => sub { $self->_set_option(@_); } }
      grep   { $_->{long} ne 'config' } # --config is handled specially above.
      values %{$self->{opts}}
   ) or $self->save_error('Error parsing options');

   if ( exists $self->{opts}->{version} && $self->{opts}->{version}->{got} ) {
      if ( $self->{version} ) {
         print $self->{version}, "\n";
      }
      else {
         print "Error parsing version.  See the VERSION section of the tool's documentation.\n";
      }
      exit 1;
   }

   if ( @ARGV && $self->{strict} ) {
      $self->save_error("Unrecognized command-line options @ARGV");
   }

   foreach my $mutex ( @{$self->{mutex}} ) {
      my @set = grep { $self->{opts}->{$_}->{got} } @$mutex;
      if ( @set > 1 ) {
         my $err = join(', ', map { "--$self->{opts}->{$_}->{long}" }
                      @{$mutex}[ 0 .. scalar(@$mutex) - 2] )
                 . ' and --'.$self->{opts}->{$mutex->[-1]}->{long}
                 . ' are mutually exclusive.';
         $self->save_error($err);
      }
   }

   foreach my $required ( @{$self->{atleast1}} ) {
      my @set = grep { $self->{opts}->{$_}->{got} } @$required;
      if ( @set == 0 ) {
         my $err = join(', ', map { "--$self->{opts}->{$_}->{long}" }
                      @{$required}[ 0 .. scalar(@$required) - 2] )
                 .' or --'.$self->{opts}->{$required->[-1]}->{long};
         $self->save_error("Specify at least one of $err");
      }
   }

   $self->_check_opts( keys %{$self->{opts}} );
   $self->{got_opts} = 1;
   return;
}

sub _check_opts {
   my ( $self, @long ) = @_;
   my $long_last = scalar @long;
   while ( @long ) {
      foreach my $i ( 0..$#long ) {
         my $long = $long[$i];
         next unless $long;
         my $opt  = $self->{opts}->{$long};
         if ( $opt->{got} ) {
            if ( exists $self->{disables}->{$long} ) {
               my @disable_opts = @{$self->{disables}->{$long}};
               map { $self->{opts}->{$_}->{value} = undef; } @disable_opts;
               PTDEBUG && _d('Unset options', @disable_opts,
                  'because', $long,'disables them');
            }

            if ( exists $self->{allowed_groups}->{$long} ) {

               my @restricted_groups = grep {
                  !exists $self->{allowed_groups}->{$long}->{$_}
               } keys %{$self->{groups}};

               my @restricted_opts;
               foreach my $restricted_group ( @restricted_groups ) {
                  RESTRICTED_OPT:
                  foreach my $restricted_opt (
                     keys %{$self->{groups}->{$restricted_group}} )
                  {
                     next RESTRICTED_OPT if $restricted_opt eq $long;
                     push @restricted_opts, $restricted_opt
                        if $self->{opts}->{$restricted_opt}->{got};
                  }
               }

               if ( @restricted_opts ) {
                  my $err;
                  if ( @restricted_opts == 1 ) {
                     $err = "--$restricted_opts[0]";
                  }
                  else {
                     $err = join(', ',
                               map { "--$self->{opts}->{$_}->{long}" }
                               grep { $_ } 
                               @restricted_opts[0..scalar(@restricted_opts) - 2]
                            )
                          . ' or --'.$self->{opts}->{$restricted_opts[-1]}->{long};
                  }
                  $self->save_error("--$long is not allowed with $err");
               }
            }

         }
         elsif ( $opt->{is_required} ) { 
            $self->save_error("Required option --$long must be specified");
         }

         $self->_validate_type($opt);
         if ( $opt->{parsed} ) {
            delete $long[$i];
         }
         else {
            PTDEBUG && _d('Temporarily failed to parse', $long);
         }
      }

      die "Failed to parse options, possibly due to circular dependencies"
         if @long == $long_last;
      $long_last = @long;
   }

   return;
}

sub _validate_type {
   my ( $self, $opt ) = @_;
   return unless $opt;

   if ( !$opt->{type} ) {
      $opt->{parsed} = 1;
      return;
   }

   my $val = $opt->{value};

   if ( $val && $opt->{type} eq 'm' ) {  # type time
      PTDEBUG && _d('Parsing option', $opt->{long}, 'as a time value');
      my ( $prefix, $num, $suffix ) = $val =~ m/([+-]?)(\d+)([a-z])?$/;
      if ( !$suffix ) {
         my ( $s ) = $opt->{desc} =~ m/\(suffix (.)\)/;
         $suffix = $s || 's';
         PTDEBUG && _d('No suffix given; using', $suffix, 'for',
            $opt->{long}, '(value:', $val, ')');
      }
      if ( $suffix =~ m/[smhd]/ ) {
         $val = $suffix eq 's' ? $num            # Seconds
              : $suffix eq 'm' ? $num * 60       # Minutes
              : $suffix eq 'h' ? $num * 3600     # Hours
              :                  $num * 86400;   # Days
         $opt->{value} = ($prefix || '') . $val;
         PTDEBUG && _d('Setting option', $opt->{long}, 'to', $val);
      }
      else {
         $self->save_error("Invalid time suffix for --$opt->{long}");
      }
   }
   elsif ( $val && $opt->{type} eq 'd' ) {  # type DSN
      PTDEBUG && _d('Parsing option', $opt->{long}, 'as a DSN');
      my $prev = {};
      my $from_key = $self->{defaults_to}->{ $opt->{long} };
      if ( $from_key ) {
         PTDEBUG && _d($opt->{long}, 'DSN copies from', $from_key, 'DSN');
         if ( $self->{opts}->{$from_key}->{parsed} ) {
            $prev = $self->{opts}->{$from_key}->{value};
         }
         else {
            PTDEBUG && _d('Cannot parse', $opt->{long}, 'until',
               $from_key, 'parsed');
            return;
         }
      }
      my $defaults = $self->{DSNParser}->parse_options($self);
      $opt->{value} = $self->{DSNParser}->parse($val, $prev, $defaults);
   }
   elsif ( $val && $opt->{type} eq 'z' ) {  # type size
      PTDEBUG && _d('Parsing option', $opt->{long}, 'as a size value');
      $self->_parse_size($opt, $val);
   }
   elsif ( $opt->{type} eq 'H' || (defined $val && $opt->{type} eq 'h') ) {
      $opt->{value} = { map { $_ => 1 } split(/(?<!\\),\s*/, ($val || '')) };
   }
   elsif ( $opt->{type} eq 'A' || (defined $val && $opt->{type} eq 'a') ) {
      $opt->{value} = [ split(/(?<!\\),\s*/, ($val || '')) ];
   }
   else {
      PTDEBUG && _d('Nothing to validate for option',
         $opt->{long}, 'type', $opt->{type}, 'value', $val);
   }

   $opt->{parsed} = 1;
   return;
}

sub get {
   my ( $self, $opt ) = @_;
   my $long = (length $opt == 1 ? $self->{short_opts}->{$opt} : $opt);
   die "Option $opt does not exist"
      unless $long && exists $self->{opts}->{$long};
   return $self->{opts}->{$long}->{value};
}

sub got {
   my ( $self, $opt ) = @_;
   my $long = (length $opt == 1 ? $self->{short_opts}->{$opt} : $opt);
   die "Option $opt does not exist"
      unless $long && exists $self->{opts}->{$long};
   return $self->{opts}->{$long}->{got};
}

sub has {
   my ( $self, $opt ) = @_;
   my $long = (length $opt == 1 ? $self->{short_opts}->{$opt} : $opt);
   return defined $long ? exists $self->{opts}->{$long} : 0;
}

sub set {
   my ( $self, $opt, $val ) = @_;
   my $long = (length $opt == 1 ? $self->{short_opts}->{$opt} : $opt);
   die "Option $opt does not exist"
      unless $long && exists $self->{opts}->{$long};
   $self->{opts}->{$long}->{value} = $val;
   return;
}

sub save_error {
   my ( $self, $error ) = @_;
   push @{$self->{errors}}, $error;
   return;
}

sub errors {
   my ( $self ) = @_;
   return $self->{errors};
}

sub usage {
   my ( $self ) = @_;
   warn "No usage string is set" unless $self->{usage}; # XXX
   return "Usage: " . ($self->{usage} || '') . "\n";
}

sub descr {
   my ( $self ) = @_;
   warn "No description string is set" unless $self->{description}; # XXX
   my $descr  = ($self->{description} || $self->{program_name} || '')
              . "  For more details, please use the --help option, "
              . "or try 'perldoc $PROGRAM_NAME' "
              . "for complete documentation.";
   $descr = join("\n", $descr =~ m/(.{0,80})(?:\s+|$)/g)
      unless $ENV{DONT_BREAK_LINES};
   $descr =~ s/ +$//mg;
   return $descr;
}

sub usage_or_errors {
   my ( $self, $file, $return ) = @_;
   $file ||= $self->{file} || __FILE__;

   if ( !$self->{description} || !$self->{usage} ) {
      PTDEBUG && _d("Getting description and usage from SYNOPSIS in", $file);
      my %synop = $self->_parse_synopsis($file);
      $self->{description} ||= $synop{description};
      $self->{usage}       ||= $synop{usage};
      PTDEBUG && _d("Description:", $self->{description},
         "\nUsage:", $self->{usage});
   }

   if ( $self->{opts}->{help}->{got} ) {
      print $self->print_usage() or die "Cannot print usage: $OS_ERROR";
      exit 0 unless $return;
   }
   elsif ( scalar @{$self->{errors}} ) {
      print $self->print_errors() or die "Cannot print errors: $OS_ERROR";
      exit 1 unless $return;
   }

   return;
}

sub print_errors {
   my ( $self ) = @_;
   my $usage = $self->usage() . "\n";
   if ( (my @errors = @{$self->{errors}}) ) {
      $usage .= join("\n  * ", 'Errors in command-line arguments:', @errors)
              . "\n";
   }
   return $usage . "\n" . $self->descr();
}

sub print_usage {
   my ( $self ) = @_;
   die "Run get_opts() before print_usage()" unless $self->{got_opts};
   my @opts = values %{$self->{opts}};

   my $maxl = max(
      map {
         length($_->{long})               # option long name
         + ($_->{is_negatable} ? 4 : 0)   # "[no]" if opt is negatable
         + ($_->{type} ? 2 : 0)           # "=x" where x is the opt type
      }
      @opts);

   my $maxs = max(0,
      map {
         length($_)
         + ($self->{opts}->{$_}->{is_negatable} ? 4 : 0)
         + ($self->{opts}->{$_}->{type} ? 2 : 0)
      }
      values %{$self->{short_opts}});

   my $lcol = max($maxl, ($maxs + 3));
   my $rcol = 80 - $lcol - 6;
   my $rpad = ' ' x ( 80 - $rcol );

   $maxs = max($lcol - 3, $maxs);

   my $usage = $self->descr() . "\n" . $self->usage();

   my @groups = reverse sort grep { $_ ne 'default'; } keys %{$self->{groups}};
   push @groups, 'default';

   foreach my $group ( reverse @groups ) {
      $usage .= "\n".($group eq 'default' ? 'Options' : $group).":\n\n";
      foreach my $opt (
         sort { $a->{long} cmp $b->{long} }
         grep { $_->{group} eq $group }
         @opts )
      {
         my $long  = $opt->{is_negatable} ? "[no]$opt->{long}" : $opt->{long};
         my $short = $opt->{short};
         my $desc  = $opt->{desc};

         $long .= $opt->{type} ? "=$opt->{type}" : "";

         if ( $opt->{type} && $opt->{type} eq 'm' ) {
            my ($s) = $desc =~ m/\(suffix (.)\)/;
            $s    ||= 's';
            $desc =~ s/\s+\(suffix .\)//;
            $desc .= ".  Optional suffix s=seconds, m=minutes, h=hours, "
                   . "d=days; if no suffix, $s is used.";
         }
         $desc = join("\n$rpad", grep { $_ } $desc =~ m/(.{0,$rcol}(?!\W))(?:\s+|(?<=\W)|$)/g);
         $desc =~ s/ +$//mg;
         if ( $short ) {
            $usage .= sprintf("  --%-${maxs}s -%s  %s\n", $long, $short, $desc);
         }
         else {
            $usage .= sprintf("  --%-${lcol}s  %s\n", $long, $desc);
         }
      }
   }

   $usage .= "\nOption types: s=string, i=integer, f=float, h/H/a/A=comma-separated list, d=DSN, z=size, m=time\n";

   if ( (my @rules = @{$self->{rules}}) ) {
      $usage .= "\nRules:\n\n";
      $usage .= join("\n", map { "  $_" } @rules) . "\n";
   }
   if ( $self->{DSNParser} ) {
      $usage .= "\n" . $self->{DSNParser}->usage();
   }
   $usage .= "\nOptions and values after processing arguments:\n\n";
   foreach my $opt ( sort { $a->{long} cmp $b->{long} } @opts ) {
      my $val   = $opt->{value};
      my $type  = $opt->{type} || '';
      my $bool  = $opt->{spec} =~ m/^[\w-]+(?:\|[\w-])?!?$/;
      $val      = $bool              ? ( $val ? 'TRUE' : 'FALSE' )
                : !defined $val      ? '(No value)'
                : $type eq 'd'       ? $self->{DSNParser}->as_string($val)
                : $type =~ m/H|h/    ? join(',', sort keys %$val)
                : $type =~ m/A|a/    ? join(',', @$val)
                :                    $val;
      $usage .= sprintf("  --%-${lcol}s  %s\n", $opt->{long}, $val);
   }
   return $usage;
}

sub prompt_noecho {
   shift @_ if ref $_[0] eq __PACKAGE__;
   my ( $prompt ) = @_;
   local $OUTPUT_AUTOFLUSH = 1;
   print $prompt
      or die "Cannot print: $OS_ERROR";
   my $response;
   eval {
      require Term::ReadKey;
      Term::ReadKey::ReadMode('noecho');
      chomp($response = <STDIN>);
      Term::ReadKey::ReadMode('normal');
      print "\n"
         or die "Cannot print: $OS_ERROR";
   };
   if ( $EVAL_ERROR ) {
      die "Cannot read response; is Term::ReadKey installed? $EVAL_ERROR";
   }
   return $response;
}

sub _read_config_file {
   my ( $self, $filename ) = @_;
   open my $fh, "<", $filename or die "Cannot open $filename: $OS_ERROR\n";
   my @args;
   my $prefix = '--';
   my $parse  = 1;

   LINE:
   while ( my $line = <$fh> ) {
      chomp $line;
      next LINE if $line =~ m/^\s*(?:\#|\;|$)/;
      $line =~ s/\s+#.*$//g;
      $line =~ s/^\s+|\s+$//g;
      if ( $line eq '--' ) {
         $prefix = '';
         $parse  = 0;
         next LINE;
      }
      if ( $parse
         && (my($opt, $arg) = $line =~ m/^\s*([^=\s]+?)(?:\s*=\s*(.*?)\s*)?$/)
      ) {
         push @args, grep { defined $_ } ("$prefix$opt", $arg);
      }
      elsif ( $line =~ m/./ ) {
         push @args, $line;
      }
      else {
         die "Syntax error in file $filename at line $INPUT_LINE_NUMBER";
      }
   }
   close $fh;
   return @args;
}

sub read_para_after {
   my ( $self, $file, $regex ) = @_;
   open my $fh, "<", $file or die "Can't open $file: $OS_ERROR";
   local $INPUT_RECORD_SEPARATOR = '';
   my $para;
   while ( $para = <$fh> ) {
      next unless $para =~ m/^=pod$/m;
      last;
   }
   while ( $para = <$fh> ) {
      next unless $para =~ m/$regex/;
      last;
   }
   $para = <$fh>;
   chomp($para);
   close $fh or die "Can't close $file: $OS_ERROR";
   return $para;
}

sub clone {
   my ( $self ) = @_;

   my %clone = map {
      my $hashref  = $self->{$_};
      my $val_copy = {};
      foreach my $key ( keys %$hashref ) {
         my $ref = ref $hashref->{$key};
         $val_copy->{$key} = !$ref           ? $hashref->{$key}
                           : $ref eq 'HASH'  ? { %{$hashref->{$key}} }
                           : $ref eq 'ARRAY' ? [ @{$hashref->{$key}} ]
                           : $hashref->{$key};
      }
      $_ => $val_copy;
   } qw(opts short_opts defaults);

   foreach my $scalar ( qw(got_opts) ) {
      $clone{$scalar} = $self->{$scalar};
   }

   return bless \%clone;     
}

sub _parse_size {
   my ( $self, $opt, $val ) = @_;

   if ( lc($val || '') eq 'null' ) {
      PTDEBUG && _d('NULL size for', $opt->{long});
      $opt->{value} = 'null';
      return;
   }

   my %factor_for = (k => 1_024, M => 1_048_576, G => 1_073_741_824);
   my ($pre, $num, $factor) = $val =~ m/^([+-])?(\d+)([kMG])?$/;
   if ( defined $num ) {
      if ( $factor ) {
         $num *= $factor_for{$factor};
         PTDEBUG && _d('Setting option', $opt->{y},
            'to num', $num, '* factor', $factor);
      }
      $opt->{value} = ($pre || '') . $num;
   }
   else {
      $self->save_error("Invalid size for --$opt->{long}: $val");
   }
   return;
}

sub _parse_attribs {
   my ( $self, $option, $attribs ) = @_;
   my $types = $self->{types};
   return $option
      . ($attribs->{'short form'} ? '|' . $attribs->{'short form'}   : '' )
      . ($attribs->{'negatable'}  ? '!'                              : '' )
      . ($attribs->{'cumulative'} ? '+'                              : '' )
      . ($attribs->{'type'}       ? '=' . $types->{$attribs->{type}} : '' );
}

sub _parse_synopsis {
   my ( $self, $file ) = @_;
   $file ||= $self->{file} || __FILE__;
   PTDEBUG && _d("Parsing SYNOPSIS in", $file);

   local $INPUT_RECORD_SEPARATOR = '';  # read paragraphs
   open my $fh, "<", $file or die "Cannot open $file: $OS_ERROR";
   my $para;
   1 while defined($para = <$fh>) && $para !~ m/^=head1 SYNOPSIS/;
   die "$file does not contain a SYNOPSIS section" unless $para;
   my @synop;
   for ( 1..2 ) {  # 1 for the usage, 2 for the description
      my $para = <$fh>;
      push @synop, $para;
   }
   close $fh;
   PTDEBUG && _d("Raw SYNOPSIS text:", @synop);
   my ($usage, $desc) = @synop;
   die "The SYNOPSIS section in $file is not formatted properly"
      unless $usage && $desc;

   $usage =~ s/^\s*Usage:\s+(.+)/$1/;
   chomp $usage;

   $desc =~ s/\n/ /g;
   $desc =~ s/\s{2,}/ /g;
   $desc =~ s/\. ([A-Z][a-z])/.  $1/g;
   $desc =~ s/\s+$//;

   return (
      description => $desc,
      usage       => $usage,
   );
};

sub set_vars {
   my ($self, $file) = @_;
   $file ||= $self->{file} || __FILE__;

   my %user_vars;
   my $user_vars = $self->has('set-vars') ? $self->get('set-vars') : undef;
   if ( $user_vars ) {
      foreach my $var_val ( @$user_vars ) {
         my ($var, $val) = $var_val =~ m/([^\s=]+)=(\S+)/;
         die "Invalid --set-vars value: $var_val\n" unless $var && defined $val;
         $user_vars{$var} = {
            val     => $val,
            default => 0,
         };
      }
   }

   my %default_vars;
   my $default_vars = $self->read_para_after($file, qr/MAGIC_set_vars/);
   if ( $default_vars ) {
      %default_vars = map {
         my $var_val = $_;
         my ($var, $val) = $var_val =~ m/([^\s=]+)=(\S+)/;
         die "Invalid --set-vars value: $var_val\n" unless $var && defined $val;
         $var => {
            val     => $val,
            default => 1,
         };
      } split("\n", $default_vars);
   }

   my %vars = (
      %default_vars, # first the tool's defaults
      %user_vars,    # then the user's which overwrite the defaults
   );
   PTDEBUG && _d('--set-vars:', Dumper(\%vars));
   return \%vars;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

if ( PTDEBUG ) {
   print STDERR '# ', $^X, ' ', $], "\n";
   if ( my $uname = `uname -a` ) {
      $uname =~ s/\s+/ /g;
      print STDERR "# $uname\n";
   }
   print STDERR '# Arguments: ',
      join(' ', map { my $a = "_[$_]_"; $a =~ s/\n/\n# /g; $a; } @ARGV), "\n";
}

1;
}
# ###########################################################################
# End OptionParser package
# ###########################################################################

# ###########################################################################
# Cxn package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Cxn.pm
#   t/lib/Cxn.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Cxn;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use Scalar::Util qw(blessed);
use constant {
   PTDEBUG => $ENV{PTDEBUG} || 0,
   PERCONA_TOOLKIT_TEST_USE_DSN_NAMES => $ENV{PERCONA_TOOLKIT_TEST_USE_DSN_NAMES} || 0,
};

sub new {
   my ( $class, %args ) = @_;
   my @required_args = qw(DSNParser OptionParser);
   foreach my $arg ( @required_args ) {
      die "I need a $arg argument" unless $args{$arg};
   };
   my ($dp, $o) = @args{@required_args};

   my $dsn_defaults = $dp->parse_options($o);
   my $prev_dsn     = $args{prev_dsn};
   my $dsn          = $args{dsn};
   if ( !$dsn ) {
      $args{dsn_string} ||= 'h=' . ($dsn_defaults->{h} || 'localhost');

      $dsn = $dp->parse(
         $args{dsn_string}, $prev_dsn, $dsn_defaults);
   }
   elsif ( $prev_dsn ) {
      $dsn = $dp->copy($prev_dsn, $dsn);
   }

   my $dsn_name = $dp->as_string($dsn, [qw(h P S)])
               || $dp->as_string($dsn, [qw(F)])
               || '';

   my $self = {
      dsn             => $dsn,
      dbh             => $args{dbh},
      dsn_name        => $dsn_name,
      hostname        => '',
      set             => $args{set},
      NAME_lc         => defined($args{NAME_lc}) ? $args{NAME_lc} : 1,
      dbh_set         => 0,
      ask_pass        => $args{ask_pass},
      DSNParser       => $dp,
      is_cluster_node => undef,
      parent          => $args{parent},
   };

   return bless $self, $class;
}

sub connect {
   my ( $self, %opts ) = @_;
   my $dsn = $opts{dsn} || $self->{dsn};
   my $dp  = $self->{DSNParser};

   my $dbh = $self->{dbh};
   if ( !$dbh || !$dbh->ping() ) {
      if ( $self->{ask_pass} && !$self->{asked_for_pass} ) {
         $dsn->{p} = OptionParser::prompt_noecho("Enter MySQL password: ");
         $self->{asked_for_pass} = 1;
      }
      $dbh = $dp->get_dbh(
         $dp->get_cxn_params($dsn),
         {
            AutoCommit => 1,
            %opts,
         },
      );
   }

   $dbh = $self->set_dbh($dbh);
   if ( $opts{dsn} ) {
      $self->{dsn}      = $dsn;
      $self->{dsn_name} = $dp->as_string($dsn, [qw(h P S)])
                       || $dp->as_string($dsn, [qw(F)])
                       || '';

   }
   PTDEBUG && _d($dbh, 'Connected dbh to', $self->{hostname},$self->{dsn_name});
   return $dbh;
}

sub set_dbh {
   my ($self, $dbh) = @_;

   if ( $self->{dbh} && $self->{dbh} == $dbh && $self->{dbh_set} ) {
      PTDEBUG && _d($dbh, 'Already set dbh');
      return $dbh;
   }

   PTDEBUG && _d($dbh, 'Setting dbh');

   $dbh->{FetchHashKeyName} = 'NAME_lc' if $self->{NAME_lc};

   my $sql = 'SELECT @@server_id /*!50038 , @@hostname*/';
   PTDEBUG && _d($dbh, $sql);
   my ($server_id, $hostname) = $dbh->selectrow_array($sql);
   PTDEBUG && _d($dbh, 'hostname:', $hostname, $server_id);
   if ( $hostname ) {
      $self->{hostname} = $hostname;
   }

   if ( $self->{parent} ) {
      PTDEBUG && _d($dbh, 'Setting InactiveDestroy=1 in parent');
      $dbh->{InactiveDestroy} = 1;
   }

   if ( my $set = $self->{set}) {
      $set->($dbh);
   }

   $self->{dbh}     = $dbh;
   $self->{dbh_set} = 1;
   return $dbh;
}

sub lost_connection {
   my ($self, $e) = @_;
   return 0 unless $e;
   return $e =~ m/MySQL server has gone away/
       || $e =~ m/Lost connection to MySQL server/;
}

sub dbh {
   my ($self) = @_;
   return $self->{dbh};
}

sub dsn {
   my ($self) = @_;
   return $self->{dsn};
}

sub name {
   my ($self) = @_;
   return $self->{dsn_name} if PERCONA_TOOLKIT_TEST_USE_DSN_NAMES;
   return $self->{hostname} || $self->{dsn_name} || 'unknown host';
}

sub remove_duplicate_cxns {
   my ($self, %args) = @_;
   my @cxns     = @{$args{cxns}};
   my $seen_ids = $args{seen_ids} || {};
   PTDEBUG && _d("Removing duplicates from ", join(" ", map { $_->name } @cxns));
   my @trimmed_cxns;

   for my $cxn ( @cxns ) {
      my $dbh  = $cxn->dbh();
      my $sql  = q{SELECT @@server_id};
      PTDEBUG && _d($sql);
      my ($id) = $dbh->selectrow_array($sql);
      PTDEBUG && _d('Server ID for ', $cxn->name, ': ', $id);

      if ( ! $seen_ids->{$id}++ ) {
         push @trimmed_cxns, $cxn
      }
      else {
         PTDEBUG && _d("Removing ", $cxn->name,
                       ", ID ", $id, ", because we've already seen it");
      }
   }

   return \@trimmed_cxns;
}

sub DESTROY {
   my ($self) = @_;

   PTDEBUG && _d('Destroying cxn');

   if ( $self->{parent} ) {
      PTDEBUG && _d($self->{dbh}, 'Not disconnecting dbh in parent');
   }
   elsif ( $self->{dbh}
           && blessed($self->{dbh})
           && $self->{dbh}->can("disconnect") )
   {
      PTDEBUG && _d($self->{dbh}, 'Disconnecting dbh on', $self->{hostname},
         $self->{dsn_name});
      $self->{dbh}->disconnect();
   }

   return;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End Cxn package
# ###########################################################################

# ###########################################################################
# Quoter package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Quoter.pm
#   t/lib/Quoter.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Quoter;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use Data::Dumper;
$Data::Dumper::Indent    = 1;
$Data::Dumper::Sortkeys  = 1;
$Data::Dumper::Quotekeys = 0;

sub new {
   my ( $class, %args ) = @_;
   return bless {}, $class;
}

sub quote {
   my ( $self, @vals ) = @_;
   foreach my $val ( @vals ) {
      $val =~ s/`/``/g;
   }
   return join('.', map { '`' . $_ . '`' } @vals);
}

sub quote_val {
   my ( $self, $val, %args ) = @_;

   return 'NULL' unless defined $val;          # undef = NULL
   return "''" if $val eq '';                  # blank string = ''
   return $val if $val =~ m/^0x[0-9a-fA-F]+$/  # quote hex data
                  && !$args{is_char};          # unless is_char is true

   return $val if $args{is_float};

   $val =~ s/(['\\])/\\$1/g;
   return "'$val'";
}

sub split_unquote {
   my ( $self, $db_tbl, $default_db ) = @_;
   my ( $db, $tbl ) = split(/[.]/, $db_tbl);
   if ( !$tbl ) {
      $tbl = $db;
      $db  = $default_db;
   }
   for ($db, $tbl) {
      next unless $_;
      s/\A`//;
      s/`\z//;
      s/``/`/g;
   }
   
   return ($db, $tbl);
}

sub literal_like {
   my ( $self, $like ) = @_;
   return unless $like;
   $like =~ s/([%_])/\\$1/g;
   return "'$like'";
}

sub join_quote {
   my ( $self, $default_db, $db_tbl ) = @_;
   return unless $db_tbl;
   my ($db, $tbl) = split(/[.]/, $db_tbl);
   if ( !$tbl ) {
      $tbl = $db;
      $db  = $default_db;
   }
   $db  = "`$db`"  if $db  && $db  !~ m/^`/;
   $tbl = "`$tbl`" if $tbl && $tbl !~ m/^`/;
   return $db ? "$db.$tbl" : $tbl;
}

sub serialize_list {
   my ( $self, @args ) = @_;
   PTDEBUG && _d('Serializing', Dumper(\@args));
   return unless @args;

   my @parts;
   foreach my $arg  ( @args ) {
      if ( defined $arg ) {
         $arg =~ s/,/\\,/g;      # escape commas
         $arg =~ s/\\N/\\\\N/g;  # escape literal \N
         push @parts, $arg;
      }
      else {
         push @parts, '\N';
      }
   }

   my $string = join(',', @parts);
   PTDEBUG && _d('Serialized: <', $string, '>');
   return $string;
}

sub deserialize_list {
   my ( $self, $string ) = @_;
   PTDEBUG && _d('Deserializing <', $string, '>');
   die "Cannot deserialize an undefined string" unless defined $string;

   my @parts;
   foreach my $arg ( split(/(?<!\\),/, $string) ) {
      if ( $arg eq '\N' ) {
         $arg = undef;
      }
      else {
         $arg =~ s/\\,/,/g;
         $arg =~ s/\\\\N/\\N/g;
      }
      push @parts, $arg;
   }

   if ( !@parts ) {
      my $n_empty_strings = $string =~ tr/,//;
      $n_empty_strings++;
      PTDEBUG && _d($n_empty_strings, 'empty strings');
      map { push @parts, '' } 1..$n_empty_strings;
   }
   elsif ( $string =~ m/(?<!\\),$/ ) {
      PTDEBUG && _d('Last value is an empty string');
      push @parts, '';
   }

   PTDEBUG && _d('Deserialized', Dumper(\@parts));
   return @parts;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End Quoter package
# ###########################################################################

# ###########################################################################
# VersionParser package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/VersionParser.pm
#   t/lib/VersionParser.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package VersionParser;

use Lmo;
use Scalar::Util qw(blessed);
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use overload (
   '""'     => "version",
   '<=>'    => "cmp",
   'cmp'    => "cmp",
   fallback => 1,
);

use Carp ();

has major => (
    is       => 'ro',
    isa      => 'Int',
    required => 1,
);

has [qw( minor revision )] => (
    is  => 'ro',
    isa => 'Num',
);

has flavor => (
    is      => 'ro',
    isa     => 'Str',
    default => sub { 'Unknown' },
);

has innodb_version => (
    is      => 'ro',
    isa     => 'Str',
    default => sub { 'NO' },
);

sub series {
   my $self = shift;
   return $self->_join_version($self->major, $self->minor);
}

sub version {
   my $self = shift;
   return $self->_join_version($self->major, $self->minor, $self->revision);
}

sub is_in {
   my ($self, $target) = @_;

   return $self eq $target;
}

sub _join_version {
    my ($self, @parts) = @_;

    return join ".", map { my $c = $_; $c =~ s/^0\./0/; $c } grep defined, @parts;
}
sub _split_version {
   my ($self, $str) = @_;
   my @version_parts = map { s/^0(?=\d)/0./; $_ } $str =~ m/(\d+)/g;
   return @version_parts[0..2];
}

sub normalized_version {
   my ( $self ) = @_;
   my $result = sprintf('%d%02d%02d', map { $_ || 0 } $self->major,
                                                      $self->minor,
                                                      $self->revision);
   PTDEBUG && _d($self->version, 'normalizes to', $result);
   return $result;
}

sub comment {
   my ( $self, $cmd ) = @_;
   my $v = $self->normalized_version();

   return "/*!$v $cmd */"
}

my @methods = qw(major minor revision);
sub cmp {
   my ($left, $right) = @_;
   my $right_obj = (blessed($right) && $right->isa(ref($left)))
                   ? $right
                   : ref($left)->new($right);

   my $retval = 0;
   for my $m ( @methods ) {
      last unless defined($left->$m) && defined($right_obj->$m);
      $retval = $left->$m <=> $right_obj->$m;
      last if $retval;
   }
   return $retval;
}

sub BUILDARGS {
   my $self = shift;

   if ( @_ == 1 ) {
      my %args;
      if ( blessed($_[0]) && $_[0]->can("selectrow_hashref") ) {
         PTDEBUG && _d("VersionParser got a dbh, trying to get the version");
         my $dbh = $_[0];
         local $dbh->{FetchHashKeyName} = 'NAME_lc';
         my $query = eval {
            $dbh->selectall_arrayref(q/SHOW VARIABLES LIKE 'version%'/, { Slice => {} })
         };
         if ( $query ) {
            $query = { map { $_->{variable_name} => $_->{value} } @$query };
            @args{@methods} = $self->_split_version($query->{version});
            $args{flavor} = delete $query->{version_comment}
                  if $query->{version_comment};
         }
         elsif ( eval { ($query) = $dbh->selectrow_array(q/SELECT VERSION()/) } ) {
            @args{@methods} = $self->_split_version($query);
         }
         else {
            Carp::confess("Couldn't get the version from the dbh while "
                        . "creating a VersionParser object: $@");
         }
         $args{innodb_version} = eval { $self->_innodb_version($dbh) };
      }
      elsif ( !ref($_[0]) ) {
         @args{@methods} = $self->_split_version($_[0]);
      }

      for my $method (@methods) {
         delete $args{$method} unless defined $args{$method};
      }
      @_ = %args if %args;
   }

   return $self->SUPER::BUILDARGS(@_);
}

sub _innodb_version {
   my ( $self, $dbh ) = @_;
   return unless $dbh;
   my $innodb_version = "NO";

   my ($innodb) =
      grep { $_->{engine} =~ m/InnoDB/i }
      map  {
         my %hash;
         @hash{ map { lc $_ } keys %$_ } = values %$_;
         \%hash;
      }
      @{ $dbh->selectall_arrayref("SHOW ENGINES", {Slice=>{}}) };
   if ( $innodb ) {
      PTDEBUG && _d("InnoDB support:", $innodb->{support});
      if ( $innodb->{support} =~ m/YES|DEFAULT/i ) {
         my $vars = $dbh->selectrow_hashref(
            "SHOW VARIABLES LIKE 'innodb_version'");
         $innodb_version = !$vars ? "BUILTIN"
                         :          ($vars->{Value} || $vars->{value});
      }
      else {
         $innodb_version = $innodb->{support};  # probably DISABLED or NO
      }
   }

   PTDEBUG && _d("InnoDB version:", $innodb_version);
   return $innodb_version;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

no Lmo;
1;
}
# ###########################################################################
# End VersionParser package
# ###########################################################################

# ###########################################################################
# Daemon package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Daemon.pm
#   t/lib/Daemon.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Daemon;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);

use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use POSIX qw(setsid);
use Fcntl qw(:DEFAULT);

sub new {
   my ($class, %args) = @_;
   my $self = {
      log_file       => $args{log_file},
      pid_file       => $args{pid_file},
      daemonize      => $args{daemonize},
      force_log_file => $args{force_log_file},
      parent_exit    => $args{parent_exit},
      pid_file_owner => 0,
   };
   return bless $self, $class;
}

sub run {
   my ($self) = @_;

   my $daemonize      = $self->{daemonize};
   my $pid_file       = $self->{pid_file};
   my $log_file       = $self->{log_file};
   my $force_log_file = $self->{force_log_file};
   my $parent_exit    = $self->{parent_exit};

   PTDEBUG && _d('Starting daemon');

   if ( $pid_file ) {
      eval {
         $self->_make_pid_file(
            pid      => $PID,  # parent's pid
            pid_file => $pid_file,
         );
      };
      die "$EVAL_ERROR\n" if $EVAL_ERROR;
      if ( !$daemonize ) {
         $self->{pid_file_owner} = $PID;  # parent's pid
      }
   }

   if ( $daemonize ) {
      defined (my $child_pid = fork()) or die "Cannot fork: $OS_ERROR";
      if ( $child_pid ) {
         PTDEBUG && _d('Forked child', $child_pid);
         $parent_exit->($child_pid) if $parent_exit;
         exit 0;
      }
 
      POSIX::setsid() or die "Cannot start a new session: $OS_ERROR";
      chdir '/'       or die "Cannot chdir to /: $OS_ERROR";

      if ( $pid_file ) {
         $self->_update_pid_file(
            pid      => $PID,  # child's pid
            pid_file => $pid_file,
         );
         $self->{pid_file_owner} = $PID;
      }
   }

   if ( $daemonize || $force_log_file ) {
      PTDEBUG && _d('Redirecting STDIN to /dev/null');
      close STDIN;
      open  STDIN, '/dev/null'
         or die "Cannot reopen STDIN to /dev/null: $OS_ERROR";
      if ( $log_file ) {
         PTDEBUG && _d('Redirecting STDOUT and STDERR to', $log_file);
         close STDOUT;
         open  STDOUT, '>>', $log_file
            or die "Cannot open log file $log_file: $OS_ERROR";

         close STDERR;
         open  STDERR, ">&STDOUT"
            or die "Cannot dupe STDERR to STDOUT: $OS_ERROR"; 
      }
      else {
         if ( -t STDOUT ) {
            PTDEBUG && _d('No log file and STDOUT is a terminal;',
               'redirecting to /dev/null');
            close STDOUT;
            open  STDOUT, '>', '/dev/null'
               or die "Cannot reopen STDOUT to /dev/null: $OS_ERROR";
         }
         if ( -t STDERR ) {
            PTDEBUG && _d('No log file and STDERR is a terminal;',
               'redirecting to /dev/null');
            close STDERR;
            open  STDERR, '>', '/dev/null'
               or die "Cannot reopen STDERR to /dev/null: $OS_ERROR";
         }
      }

      $OUTPUT_AUTOFLUSH = 1;
   }

   PTDEBUG && _d('Daemon running');
   return;
}

sub _make_pid_file {
   my ($self, %args) = @_;
   my @required_args = qw(pid pid_file);
   foreach my $arg ( @required_args ) {
      die "I need a $arg argument" unless $args{$arg};
   };
   my $pid      = $args{pid};
   my $pid_file = $args{pid_file};

   eval {
      sysopen(PID_FH, $pid_file, O_RDWR|O_CREAT|O_EXCL) or die $OS_ERROR;
      print PID_FH $PID, "\n";
      close PID_FH; 
   };
   if ( my $e = $EVAL_ERROR ) {
      if ( $e =~ m/file exists/i ) {
         my $old_pid = $self->_check_pid_file(
            pid_file => $pid_file,
            pid      => $PID,
         );
         if ( $old_pid ) {
            warn "Overwriting PID file $pid_file because PID $old_pid "
               . "is not running.\n";
         }
         $self->_update_pid_file(
            pid      => $PID,
            pid_file => $pid_file
         );
      }
      else {
         die "Error creating PID file $pid_file: $e\n";
      }
   }

   return;
}

sub _check_pid_file {
   my ($self, %args) = @_;
   my @required_args = qw(pid_file pid);
   foreach my $arg ( @required_args ) {
      die "I need a $arg argument" unless $args{$arg};
   };
   my $pid_file = $args{pid_file};
   my $pid      = $args{pid};

   PTDEBUG && _d('Checking if PID in', $pid_file, 'is running');

   if ( ! -f $pid_file ) {
      PTDEBUG && _d('PID file', $pid_file, 'does not exist');
      return;
   }

   open my $fh, '<', $pid_file
      or die "Error opening $pid_file: $OS_ERROR";
   my $existing_pid = do { local $/; <$fh> };
   chomp($existing_pid) if $existing_pid;
   close $fh
      or die "Error closing $pid_file: $OS_ERROR";

   if ( $existing_pid ) {
      if ( $existing_pid == $pid ) {
         warn "The current PID $pid already holds the PID file $pid_file\n";
         return;
      }
      else {
         PTDEBUG && _d('Checking if PID', $existing_pid, 'is running');
         my $pid_is_alive = kill 0, $existing_pid;
         if ( $pid_is_alive ) {
            die "PID file $pid_file exists and PID $existing_pid is running\n";
         }
      }
   }
   else {
      die "PID file $pid_file exists but it is empty.  Remove the file "
         . "if the process is no longer running.\n";
   }

   return $existing_pid;
}

sub _update_pid_file {
   my ($self, %args) = @_;
   my @required_args = qw(pid pid_file);
   foreach my $arg ( @required_args ) {
      die "I need a $arg argument" unless $args{$arg};
   };
   my $pid      = $args{pid};
   my $pid_file = $args{pid_file};

   open my $fh, '>', $pid_file
      or die "Cannot open $pid_file: $OS_ERROR";
   print { $fh } $pid, "\n"
      or die "Cannot print to $pid_file: $OS_ERROR";
   close $fh
      or warn "Cannot close $pid_file: $OS_ERROR";

   return;
}

sub remove_pid_file {
   my ($self, $pid_file) = @_;
   $pid_file ||= $self->{pid_file};
   if ( $pid_file && -f $pid_file ) {
      unlink $self->{pid_file}
         or warn "Cannot remove PID file $pid_file: $OS_ERROR";
      PTDEBUG && _d('Removed PID file');
   }
   else {
      PTDEBUG && _d('No PID to remove');
   }
   return;
}

sub DESTROY {
   my ($self) = @_;

   if ( $self->{pid_file_owner} == $PID ) {
      $self->remove_pid_file();
   }

   return;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End Daemon package
# ###########################################################################

# ###########################################################################
# Transformers package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Transformers.pm
#   t/lib/Transformers.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Transformers;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use Time::Local qw(timegm timelocal);
use Digest::MD5 qw(md5_hex);
use B qw();

BEGIN {
   require Exporter;
   our @ISA         = qw(Exporter);
   our %EXPORT_TAGS = ();
   our @EXPORT      = ();
   our @EXPORT_OK   = qw(
      micro_t
      percentage_of
      secs_to_time
      time_to_secs
      shorten
      ts
      parse_timestamp
      unix_timestamp
      any_unix_timestamp
      make_checksum
      crc32
      encode_json
   );
}

our $mysql_ts  = qr/(\d\d)(\d\d)(\d\d) +(\d+):(\d+):(\d+)(\.\d+)?/;
our $proper_ts = qr/(\d\d\d\d)-(\d\d)-(\d\d)[T ](\d\d):(\d\d):(\d\d)(\.\d+)?/;
our $n_ts      = qr/(\d{1,5})([shmd]?)/; # Limit \d{1,5} because \d{6} looks

sub micro_t {
   my ( $t, %args ) = @_;
   my $p_ms = defined $args{p_ms} ? $args{p_ms} : 0;  # precision for ms vals
   my $p_s  = defined $args{p_s}  ? $args{p_s}  : 0;  # precision for s vals
   my $f;

   $t = 0 if $t < 0;

   $t = sprintf('%.17f', $t) if $t =~ /e/;

   $t =~ s/\.(\d{1,6})\d*/\.$1/;

   if ($t > 0 && $t <= 0.000999) {
      $f = ($t * 1000000) . 'us';
   }
   elsif ($t >= 0.001000 && $t <= 0.999999) {
      $f = sprintf("%.${p_ms}f", $t * 1000);
      $f = ($f * 1) . 'ms'; # * 1 to remove insignificant zeros
   }
   elsif ($t >= 1) {
      $f = sprintf("%.${p_s}f", $t);
      $f = ($f * 1) . 's'; # * 1 to remove insignificant zeros
   }
   else {
      $f = 0;  # $t should = 0 at this point
   }

   return $f;
}

sub percentage_of {
   my ( $is, $of, %args ) = @_;
   my $p   = $args{p} || 0; # float precision
   my $fmt = $p ? "%.${p}f" : "%d";
   return sprintf $fmt, ($is * 100) / ($of ||= 1);
}

sub secs_to_time {
   my ( $secs, $fmt ) = @_;
   $secs ||= 0;
   return '00:00' unless $secs;

   $fmt ||= $secs >= 86_400 ? 'd'
          : $secs >= 3_600  ? 'h'
          :                   'm';

   return
      $fmt eq 'd' ? sprintf(
         "%d+%02d:%02d:%02d",
         int($secs / 86_400),
         int(($secs % 86_400) / 3_600),
         int(($secs % 3_600) / 60),
         $secs % 60)
      : $fmt eq 'h' ? sprintf(
         "%02d:%02d:%02d",
         int(($secs % 86_400) / 3_600),
         int(($secs % 3_600) / 60),
         $secs % 60)
      : sprintf(
         "%02d:%02d",
         int(($secs % 3_600) / 60),
         $secs % 60);
}

sub time_to_secs {
   my ( $val, $default_suffix ) = @_;
   die "I need a val argument" unless defined $val;
   my $t = 0;
   my ( $prefix, $num, $suffix ) = $val =~ m/([+-]?)(\d+)([a-z])?$/;
   $suffix = $suffix || $default_suffix || 's';
   if ( $suffix =~ m/[smhd]/ ) {
      $t = $suffix eq 's' ? $num * 1        # Seconds
         : $suffix eq 'm' ? $num * 60       # Minutes
         : $suffix eq 'h' ? $num * 3600     # Hours
         :                  $num * 86400;   # Days

      $t *= -1 if $prefix && $prefix eq '-';
   }
   else {
      die "Invalid suffix for $val: $suffix";
   }
   return $t;
}

sub shorten {
   my ( $num, %args ) = @_;
   my $p = defined $args{p} ? $args{p} : 2;     # float precision
   my $d = defined $args{d} ? $args{d} : 1_024; # divisor
   my $n = 0;
   my @units = ('', qw(k M G T P E Z Y));
   while ( $num >= $d && $n < @units - 1 ) {
      $num /= $d;
      ++$n;
   }
   return sprintf(
      $num =~ m/\./ || $n
         ? "%.${p}f%s"
         : '%d',
      $num, $units[$n]);
}

sub ts {
   my ( $time, $gmt ) = @_;
   my ( $sec, $min, $hour, $mday, $mon, $year )
      = $gmt ? gmtime($time) : localtime($time);
   $mon  += 1;
   $year += 1900;
   my $val = sprintf("%d-%02d-%02dT%02d:%02d:%02d",
      $year, $mon, $mday, $hour, $min, $sec);
   if ( my ($us) = $time =~ m/(\.\d+)$/ ) {
      $us = sprintf("%.6f", $us);
      $us =~ s/^0\././;
      $val .= $us;
   }
   return $val;
}

sub parse_timestamp {
   my ( $val ) = @_;
   if ( my($y, $m, $d, $h, $i, $s, $f)
         = $val =~ m/^$mysql_ts$/ )
   {
      return sprintf "%d-%02d-%02d %02d:%02d:"
                     . (defined $f ? '%09.6f' : '%02d'),
                     $y + 2000, $m, $d, $h, $i, (defined $f ? $s + $f : $s);
   }
   elsif ( $val =~ m/^$proper_ts$/ ) {
      return $val;
   }
   return $val;
}

sub unix_timestamp {
   my ( $val, $gmt ) = @_;
   if ( my($y, $m, $d, $h, $i, $s, $us) = $val =~ m/^$proper_ts$/ ) {
      $val = $gmt
         ? timegm($s, $i, $h, $d, $m - 1, $y)
         : timelocal($s, $i, $h, $d, $m - 1, $y);
      if ( defined $us ) {
         $us = sprintf('%.6f', $us);
         $us =~ s/^0\././;
         $val .= $us;
      }
   }
   return $val;
}

sub any_unix_timestamp {
   my ( $val, $callback ) = @_;

   if ( my ($n, $suffix) = $val =~ m/^$n_ts$/ ) {
      $n = $suffix eq 's' ? $n            # Seconds
         : $suffix eq 'm' ? $n * 60       # Minutes
         : $suffix eq 'h' ? $n * 3600     # Hours
         : $suffix eq 'd' ? $n * 86400    # Days
         :                  $n;           # default: Seconds
      PTDEBUG && _d('ts is now - N[shmd]:', $n);
      return time - $n;
   }
   elsif ( $val =~ m/^\d{9,}/ ) {
      PTDEBUG && _d('ts is already a unix timestamp');
      return $val;
   }
   elsif ( my ($ymd, $hms) = $val =~ m/^(\d{6})(?:\s+(\d+:\d+:\d+))?/ ) {
      PTDEBUG && _d('ts is MySQL slow log timestamp');
      $val .= ' 00:00:00' unless $hms;
      return unix_timestamp(parse_timestamp($val));
   }
   elsif ( ($ymd, $hms) = $val =~ m/^(\d{4}-\d\d-\d\d)(?:[T ](\d+:\d+:\d+))?/) {
      PTDEBUG && _d('ts is properly formatted timestamp');
      $val .= ' 00:00:00' unless $hms;
      return unix_timestamp($val);
   }
   else {
      PTDEBUG && _d('ts is MySQL expression');
      return $callback->($val) if $callback && ref $callback eq 'CODE';
   }

   PTDEBUG && _d('Unknown ts type:', $val);
   return;
}

sub make_checksum {
   my ( $val ) = @_;
   my $checksum = uc substr(md5_hex($val), -16);
   PTDEBUG && _d($checksum, 'checksum for', $val);
   return $checksum;
}

sub crc32 {
   my ( $string ) = @_;
   return unless $string;
   my $poly = 0xEDB88320;
   my $crc  = 0xFFFFFFFF;
   foreach my $char ( split(//, $string) ) {
      my $comp = ($crc ^ ord($char)) & 0xFF;
      for ( 1 .. 8 ) {
         $comp = $comp & 1 ? $poly ^ ($comp >> 1) : $comp >> 1;
      }
      $crc = (($crc >> 8) & 0x00FFFFFF) ^ $comp;
   }
   return $crc ^ 0xFFFFFFFF;
}

my $got_json = eval { require JSON };
sub encode_json {
   return JSON::encode_json(@_) if $got_json;
   my ( $data ) = @_;
   return (object_to_json($data) || '');
}


sub object_to_json {
   my ($obj) = @_;
   my $type  = ref($obj);

   if($type eq 'HASH'){
      return hash_to_json($obj);
   }
   elsif($type eq 'ARRAY'){
      return array_to_json($obj);
   }
   else {
      return value_to_json($obj);
   }
}

sub hash_to_json {
   my ($obj) = @_;
   my @res;
   for my $k ( sort { $a cmp $b } keys %$obj ) {
      push @res, string_to_json( $k )
         .  ":"
         . ( object_to_json( $obj->{$k} ) || value_to_json( $obj->{$k} ) );
   }
   return '{' . ( @res ? join( ",", @res ) : '' )  . '}';
}

sub array_to_json {
   my ($obj) = @_;
   my @res;

   for my $v (@$obj) {
      push @res, object_to_json($v) || value_to_json($v);
   }

   return '[' . ( @res ? join( ",", @res ) : '' ) . ']';
}

sub value_to_json {
   my ($value) = @_;

   return 'null' if(!defined $value);

   my $b_obj = B::svref_2object(\$value);  # for round trip problem
   my $flags = $b_obj->FLAGS;
   return $value # as is 
      if $flags & ( B::SVp_IOK | B::SVp_NOK ) and !( $flags & B::SVp_POK ); # SvTYPE is IV or NV?

   my $type = ref($value);

   if( !$type ) {
      return string_to_json($value);
   }
   else {
      return 'null';
   }

}

my %esc = (
   "\n" => '\n',
   "\r" => '\r',
   "\t" => '\t',
   "\f" => '\f',
   "\b" => '\b',
   "\"" => '\"',
   "\\" => '\\\\',
   "\'" => '\\\'',
);

sub string_to_json {
   my ($arg) = @_;

   $arg =~ s/([\x22\x5c\n\r\t\f\b])/$esc{$1}/g;
   $arg =~ s/\//\\\//g;
   $arg =~ s/([\x00-\x08\x0b\x0e-\x1f])/'\\u00' . unpack('H2', $1)/eg;

   utf8::upgrade($arg);
   utf8::encode($arg);

   return '"' . $arg . '"';
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End Transformers package
# ###########################################################################

# ###########################################################################
# Safeguards package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Safeguards.pm
#   t/lib/Safeguards.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Safeguards;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);

use constant PTDEBUG => $ENV{PTDEBUG} || 0;

sub new {
   my ($class, %args) = @_;
   my $self = {
      disk_bytes_free => $args{disk_bytes_free} || 104857600,  # 100 MiB
      disk_pct_free   => $args{disk_pct_free}   || 5,
   };
   return bless $self, $class;
}

sub get_disk_space {
   my ($self, %args) = @_;
   my $filesystem = $args{filesystem} || $ENV{PWD};

   my $disk_space = `df -P -k "$filesystem"`;
   chop($disk_space) if $disk_space;
   PTDEBUG && _d('Disk space on', $filesystem, $disk_space);

   return $disk_space;
}

sub check_disk_space() {
   my ($self, %args) = @_;
   my $disk_space = $args{disk_space};
   PTDEBUG && _d("Checking disk space:\n", $disk_space);

   my ($partition) = $disk_space =~ m/^\s*(\/.+)/m;
   PTDEBUG && _d('Partition:', $partition);
   die "Failed to parse partition from disk space:\n$disk_space"
      unless $partition;

   my (undef, undef, $bytes_used, $bytes_free, $pct_used, undef)
      = $partition =~ m/(\S+)/g;
   PTDEBUG && _d('Bytes used:', $bytes_used, 'free:', $bytes_free,
      'Percentage used:', $pct_used);

   $bytes_used = ($bytes_used || 0) * 1024;
   $bytes_free = ($bytes_free || 0) * 1024;

   $pct_used =~ s/%//;
   my $pct_free = 100 - ($pct_used || 0);

   return $bytes_free >= $self->{disk_bytes_free}
       && $pct_free   >= $self->{disk_pct_free};
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

1;
}
# ###########################################################################
# End Safeguards package
# ###########################################################################

# ###########################################################################
# Percona::Agent::Logger package
# This package is a copy without comments from the original.  The original
# with comments and its test file can be found in the Bazaar repository at,
#   lib/Percona/Agent/Logger.pm
#   t/lib/Percona/Agent/Logger.t
# See https://launchpad.net/percona-toolkit for more information.
# ###########################################################################
{
package Percona::Agent::Logger;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);

use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use POSIX qw(SIGALRM);

use Lmo;
use Transformers;
use Percona::WebAPI::Resource::LogEntry;

Transformers->import(qw(ts));

has 'exit_status' => (
   is       => 'rw',
   isa      => 'ScalarRef',
   required => 1,
);

has 'pid' => (
   is       => 'ro',
   isa      => 'Int',
   required => 1,
);

has 'service' => (
   is       => 'rw',
   isa      => 'Maybe[Str]',
   required => 0,
   default  => sub { return; },
);

has 'data_ts' => (
   is       => 'rw',
   isa      => 'Maybe[Int]',
   required => 0,
   default  => sub { return; },
);

has 'online_logging' => (
   is       => 'ro',
   isa      => 'Bool',
   required => 0,
   default  => sub { return 1 },
);

has 'online_logging_enabled' => (
   is       => 'rw',
   isa      => 'Bool',
   required => 0,
   default  => sub { return 0 },
);

has 'quiet' => (
   is       => 'rw',
   isa      => 'Int',
   required => 0,
   default  => sub { return 0 },
);

has '_buffer' => (
   is       => 'rw',
   isa      => 'ArrayRef',
   required => 0,
   default  => sub { return []; },
);

has '_pipe_write' => (
   is       => 'rw',
   isa      => 'Maybe[FileHandle]',
   required => 0,
);

sub read_stdin {
   my ( $t ) = @_;

   POSIX::sigaction(
      SIGALRM,
      POSIX::SigAction->new(sub { die 'read timeout'; }),
   ) or die "Error setting SIGALRM handler: $OS_ERROR";

   my $timeout = 0;
   my @lines;
   eval {
      alarm $t;
      while(defined(my $line = <STDIN>)) {
         push @lines, $line;
      }
      alarm 0;
   };
   if ( $EVAL_ERROR ) {
      PTDEBUG && _d('Read error:', $EVAL_ERROR);
      die $EVAL_ERROR unless $EVAL_ERROR =~ m/read timeout/;
      $timeout = 1;
   }
   return unless scalar @lines || $timeout;
   return \@lines;
}

sub start_online_logging {
   my ($self, %args) = @_;
   my $client       = $args{client};
   my $log_link     = $args{log_link};
   my $read_timeout = $args{read_timeout} || 3;

   return unless $self->online_logging;

   my $pid = open(my $pipe_write, "|-");

   if ($pid) {
      select $pipe_write;
      $OUTPUT_AUTOFLUSH = 1;
      $self->_pipe_write($pipe_write);
      $self->online_logging_enabled(1);
   }
   else {
      my @log_entries;
      my $n_errors = 0;
      my $oktorun  = 1;
      QUEUE:
      while ($oktorun) {
         my $lines = read_stdin($read_timeout);
         last QUEUE unless $lines;
         LINE:
         while ( defined(my $line = shift @$lines) ) {
            my ($ts, $level, $n_lines, $msg) = $line =~ m/^([^,]+),([^,]+),([^,]+),(.+)/s;
            if ( !$ts || !$level || !$n_lines || !$msg ) {
               warn "$line\n";
               next LINE;
            }
            if ( $n_lines > 1 ) {
               $n_lines--;  # first line
               for ( 1..$n_lines ) {
                  $msg .= shift @$lines;
               }
            }

            push @log_entries, Percona::WebAPI::Resource::LogEntry->new(
               pid       => $self->pid,
               entry_ts  => $ts,
               log_level => $level,
               message   => $msg,
               ($self->service ? (service => $self->service) : ()),
               ($self->data_ts ? (data_ts => $self->data_ts) : ()),
            );
         }  # LINE

         if ( scalar @log_entries ) { 
            eval {
               $client->post(
                  link      => $log_link,
                  resources => \@log_entries,
               );
            };
            if ( my $e = $EVAL_ERROR ) {
               if ( ++$n_errors <= 10 ) {
                  warn "Error sending log entry to API: $e";
                  if ( $n_errors == 10 ) {
                     my $ts = ts(time, 1);  # 1=UTC
                     warn "$ts WARNING $n_errors consecutive errors, no more "
                        . "error messages will be printed until log entries "
                        . "are sent successfully again.\n";
                  }
               }
            }
            else {
               @log_entries = ();
               $n_errors    = 0;
            }
         }  # have log entries

         my $n_log_entries = scalar @log_entries;
         if ( $n_log_entries > 1_000 ) {
            warn "$n_log_entries log entries in send buffer, "
               . "removing first 100 to avoid excessive usage.\n";
            @log_entries = @log_entries[100..($n_log_entries-1)];
         }
      }  # QUEUE

      if ( scalar @log_entries ) {
         my $ts = ts(time, 1);  # 1=UTC
         warn "$ts WARNING Failed to send these log entries "
            . "(timestamps are UTC):\n";
         foreach my $log ( @log_entries ) {
            warn sprintf("%s %s %s\n",
               $log->entry_ts,
               level_name($log->log_level),
               $log->message,
            );
         }
      }

      exit 0;
   } # child

   return;
}

sub level_number {
   my $name = shift;
   die "No log level name given" unless $name;
   my $number = $name eq 'DEBUG'   ? 1
              : $name eq 'INFO'    ? 2
              : $name eq 'WARNING' ? 3
              : $name eq 'ERROR'   ? 4
              : $name eq 'FATAL'   ? 5
              : die "Invalid log level name: $name";
}

sub level_name {
   my $number = shift;
   die "No log level name given" unless $number;
   my $name = $number == 1 ? 'DEBUG'
            : $number == 2 ? 'INFO'
            : $number == 3 ? 'WARNING'
            : $number == 4 ? 'ERROR'
            : $number == 5 ? 'FATAL'
            : die "Invalid log level number: $number";
}

sub debug {
   my $self = shift;
   return if $self->online_logging;
   return $self->_log(0, 'DEBUG', @_);
}

sub info {
   my $self = shift;
   return $self->_log(1, 'INFO', @_);
}

sub warning {
   my $self = shift;
   $self->_set_exit_status();
   return $self->_log(1, 'WARNING', @_);
}

sub error {
   my $self = shift;
   $self->_set_exit_status();
   return $self->_log(1, 'ERROR', @_);
}

sub fatal {
   my $self = shift;
   $self->_set_exit_status();
   $self->_log(1, 'FATAL', @_);
   exit $self->exit_status;
}

sub _set_exit_status {
   my $self = shift;
   my $exit_status = $self->exit_status;  # get ref
   $$exit_status |= 1;                    # deref to set
   $self->exit_status($exit_status);      # save back ref
   return;
}

sub _log {
   my ($self, $online, $level, $msg) = @_;

   my $ts = ts(time, 1);  # 1=UTC
   my $level_number = level_number($level);

   return if $self->quiet && $level_number < $self->quiet;

   chomp($msg);
   my $n_lines = 1;
   $n_lines++ while $msg =~ m/\n/g;

   if ( $online && $self->online_logging_enabled ) {
      while ( defined(my $log_entry = shift @{$self->_buffer}) ) {
         $self->_queue_log_entry(@$log_entry);
      }
      $self->_queue_log_entry($ts, $level_number, $n_lines, $msg);
   }
   else {
      if ( $online && $self->online_logging ) {
         push @{$self->_buffer}, [$ts, $level_number, $n_lines, $msg];
      }

      if ( $level_number >= 3 ) {  # warning
         print STDERR "$ts $level $msg\n";
      }
      else {
         print STDOUT "$ts $level $msg\n";
      }
   }

   return;
}

sub _queue_log_entry {
   my ($self, $ts, $log_level, $n_lines, $msg) = @_;
   print "$ts,$log_level,$n_lines,$msg\n";
   return;
}

sub _d {
   my ($package, undef, $line) = caller 0;
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
        map { defined $_ ? $_ : 'undef' }
        @_;
   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}

no Lmo;
1;
}
# ###########################################################################
# End Percona::Agent::Logger package
# ###########################################################################

# ###########################################################################
# Percona::Agent::Exception::*
# ###########################################################################

{
   package Percona::Agent::Exception::PIDNotFound;

   use Lmo;
   use overload '""' => \&as_string;

   has 'pid_file' => (
      is       => 'ro',
      isa      => 'Maybe[Str]',
      required => 1,
   );

   sub as_string {
      my $self = shift;
      return sprintf "PID file %s does not exist and no matching "
         . "process was found in ps", $self->pid_file;
   }

   no Lmo;
   1;
}

{
   package Percona::Agent::Exception::NoPID;

   use Lmo;
   use overload '""' => \&as_string;

   has 'pid_file' => (
      is       => 'ro',
      isa      => 'Maybe[Str]',
      required => 1,
   );
   
   has 'pid_file_is_empty' => (
      is       => 'ro',
      isa      => 'Bool',
      required => 1,
   );

   sub as_string {
      my $self = shift;
      if ( $self->pid_file_is_empty ) {
         return sprintf "PID file %s is empty", $self->pid_file;
      }
      else {
         return sprintf "PID file %s does not exist and parsing ps output "
            . "failed", $self->pid_file;
      }
   }

   no Lmo;
   1;
}

{
   package Percona::Agent::Exception::PIDNotRunning;

   use Lmo;
   use overload '""' => \&as_string;

   has 'pid' => (
      is       => 'ro',
      isa      => 'Str',
      required => 1,
   );

   sub as_string {
      my $self = shift;
      return sprintf "PID is not running", $self->pid;
   }

   no Lmo;
   1;
}

BEGIN {
   $INC{'Percona/Agent/Exception/PIDNotFound.pm'}   = __FILE__;
   $INC{'Percona/Agent/Exception/NoPID.pm'}         = __FILE__;
   $INC{'Percona/Agent/Exception/PIDNotRunning.pm'} = __FILE__;
}

# ###########################################################################
# This is a combination of modules and programs in one -- a runnable module.
# http://www.perl.com/pub/a/2006/07/13/lightning-articles.html?page=last
# Or, look it up in the Camel book on pages 642 and 643 in the 3rd edition.
#
# Check at the end of this package for the call to main() which actually runs
# the program.
# ###########################################################################
package pt_agent;

use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant PTDEBUG => $ENV{PTDEBUG} || 0;

use Scalar::Util qw(blessed);
use POSIX        qw(signal_h);
use Time::HiRes  qw(sleep time);
use File::Temp   qw(tempfile);
use File::Path;
use File::Basename;
use FindBin;

use Percona::Toolkit; 
use Percona::WebAPI::Client;
use Percona::WebAPI::Exception::Request;
use Percona::WebAPI::Exception::Resource;
use Percona::WebAPI::Resource::Agent;
use Percona::WebAPI::Resource::Config;
use Percona::WebAPI::Resource::Service;
use Percona::WebAPI::Representation;
use Percona::Agent::Exception::PIDNotFound;
use Percona::Agent::Exception::NoPID;
use Percona::Agent::Exception::PIDNotRunning;

Percona::Toolkit->import(qw(_d Dumper have_required_args));
Percona::WebAPI::Representation->import(qw(as_json as_config));
Transformers->import(qw(ts));

use sigtrap 'handler', \&sig_int, 'normal-signals';
use sigtrap 'handler', \&reload_signal, 'USR1';

my $oktorun         = 1;
my $exit_status     = 0;
my $state           = {};
my $exit_on_signals = 0;
my $logger;

use constant MAX_DATA_FILE_SIZE => 15_728_640;  # 15M

my %deps = (
   'DBI'
      => [qw(DBI              libdbi-perl           perl-DBI)],
   'DBD::mysql'
      => [qw(DBD::mysql       libdbd-mysql-perl     perl-DBD-MySQL)],
   'JSON'
      => [qw(JSON             libjson-perl          perl-JSON)],
   'LWP'
      => [qw(LWP              libwww-perl           perl-libwww-perl)],
   'IO::Socket::SSL'
      => [qw(IO::Socket::SSL  libio-socket-ssl-perl perl-IO-Socket-SSL)],
);

# Will check this later.
eval {
   require JSON;
};

sub main {
   local @ARGV   = @_;
   
   # Reset global vars else tests will fail in strange ways.
   $oktorun         = 1;  
   $exit_status     = 0;
   $state           = {};
   $exit_on_signals = 0;

   # ########################################################################
   # Get configuration information.
   # ########################################################################
   my $o = new OptionParser();
   $o->get_specs();
   $o->get_opts();

   my $dp = $o->DSNParser();
   $dp->prop('set-vars', $o->set_vars());

   # We're _not_ running as root, so unless --pid and --log have
   # already been configured, the defaults won't work.  In this
   # case, use tmp values until a new config is received.
   if ( $EUID != 0 ) {
      $o->set('pid', '/tmp/pt-agent.pid') unless $o->got('pid');
      $o->set('log', '/tmp/pt-agent.log') unless $o->got('log');
      $o->set('lib', '/tmp/pt-agent'    ) unless $o->got('lib');
   }

   if ( !$o->get('help') ) {
   }
 
   $o->usage_or_errors();

   if (    $o->get('interactive')
        || $o->get('install')
        || $o->get('uninstall') ) {
      $OUTPUT_AUTOFLUSH = 1 
   }

   # ######################################################################## 
   # Fail-safe: if the agent somehow runs away, i.e. starts to fork-bomb,
   # stop everything.
   # ########################################################################
   my $lib_dir = $o->get('lib');
   if ( too_many_agents(lib_dir => $lib_dir) ) {
      schedule_services(
         services => [],
         lib_dir  => $lib_dir,
      );
      die "Too many agents are running.  Remove the PID files in "
         . "$lib_dir/pids/ if the agents are no longer running.  Else, "
         . "check the log files in $lib_dir/logs/ and online to see "
         . "if the agent is stuck in a loop.  Please contact Percona "
         . "if you need urgent help.\n";
   }

   # ########################################################################
   # Connect to MysSQL later, maybe.
   # ########################################################################
   my $cxn = Cxn->new(
      dsn_string   => '',
      OptionParser => $o,
      DSNParser    => $dp,
      ask_pass     => $o->get('ask-pass'),
   );

   # ########################################################################
   # Make a logger, not online yet.
   # ########################################################################
   $logger = Percona::Agent::Logger->new(
      exit_status    => \$exit_status,
      pid            => $PID,
      online_logging => $o->get('log-api') ? 1 : 0,
   );

   # ########################################################################
   # --(un)install and exit.
   # ########################################################################
   if ( $o->get('install') ) {
      $exit_on_signals = 1;
      install(
         OptionParser => $o,
         Cxn          => $cxn,
         interactive  => $o->get('interactive'),
         flags        => $o->get('install-options'),
      );
      return $exit_status;
   }
   elsif ( $o->get('uninstall') ) {
      $exit_on_signals = 1;
      uninstall(
         OptionParser => $o,
         Cxn          => $cxn,
      );
      return $exit_status;
   }

   # ########################################################################
   # Nothing works without required Perl modules.
   # ########################################################################
   if ( missing_perl_module_deps() ) {
      $logger->fatal("Missing required Perl modules");
   }

   # Check that LWP is new enough
   # https://bugs.launchpad.net/percona-toolkit/+bug/1226721
   if ( $LWP::VERSION < '5.813' ) {
      die "Perl module LWP v5.813 or newer is required; v$LWP::VERSION is installed.  Please upgrade LWP on this server.\n"
   }

   # ########################################################################
   # Nothing works without an API key.
   # ########################################################################
   my $api_key = $o->get('api-key');
   if ( !$api_key ) {
      $logger->fatal("No API key was found or specified.  pt-agent requires a "
         . "Percona Cloud Tools API key.  Put your API key "
         . "in a --config file or specify it with --api-key.");
   }

   # ########################################################################
   # --status, --stop, and --reset
   # ########################################################################
   if ( $o->get('status') ) {
      agent_status(
         api_key  => $o->get('api-key'),
         pid_file => $o->get('pid'),
         lib_dir  => $o->get('lib'),
      );
      return $exit_status;
   }
   elsif ( $o->get('stop') ) {
      stop_agent(
         pid_file => $o->get('pid'),
         lib_dir  => $o->get('lib'),
      );
      $logger->info("Done stopping pt-agent, exit $exit_status");
      return $exit_status;
   }
   elsif ( my $n = $o->get('reset') ) {
      $exit_on_signals = 1;

      my $api_key = $o->get('api-key');
      if ( !$api_key && $n < 2 ) {
         my $config_file = get_config_file();
         if ( -f $config_file ) {
            die "Cannot reset pt-agent because an API key is not set in "
               . "$config_file and --api-key was not specified.  Specify "
               . "--api-key to force the reset.  Else specify --reset "
               . "twice to do a hard reset, after which you will need to "
               . "re-install pt-agent.\n";
         }
         else {
            die "Cannot reset pt-agent because an API key is not set in "
               . "$config_file.  Add 'api-key=<API key>' to $config_file "
               . "or specify it with --api-key.  Else specify --reset "
               . "twice to do a hard reset, after which you will need to "
               . "re-install pt-agent.\n";
         }
      }
      reset_agent(
         pid_file  => $o->get('pid'),  # for stop_agent()
         lib_dir   => $o->get('lib'),
         spool_dir => $o->get('spool'),
         log_file  => $o->get('log'),
         api_key   => $api_key,  # optional
      );
      if ( $exit_status != 0 ) {
         $logger->error("Failed to completely reset pt-agent.  "
            . "Check the warnings and errors and above and try again.");
      }
      else {
         $logger->info("pt-agent has been completely reset.");
      }
      return $exit_status;
   }
   elsif ( $o->get('reload') ) {
      reload_agent(
         pid_file => $o->get('pid'),
      );
      return $exit_status;
   }

   # ########################################################################
   # --ping and exit.
   # ########################################################################
   if ( $o->get('ping') ) {
      my ($client, $entry_links, $logger_client) = get_api_client(
         api_key  => $api_key,
         tries    => 1,
         interval => sub { return },
      );
      if ( !$client || !$entry_links ) {
         die "Failed to initialize the API client.  The API may be down.  Please try again.\n";
      }
      my $api_ok = ping_api(
         client => $client,
      );
      if ( $api_ok ) {
         print $client->{entry_link} . " is up.\n"
      }
      else {
         print $client->{entry_link} . " is down or not reachable.\n";
      }
      exit;
   }

   # ########################################################################
   # --run-service and exit.
   # ########################################################################
   if ( my $service = $o->get('run-service') ) {
      eval {
         run_service(
            agent_api => $o->get('agent-api'),
            api_key   => $api_key,
            service   => $service,
            lib_dir   => $o->get('lib'),
            spool_dir => $o->get('spool'),
            Cxn       => $cxn,
         );
      };
      if ( $EVAL_ERROR ) {
         $logger->fatal("--run-service $service error: $EVAL_ERROR");
      }
      return $exit_status;
   }

   # ########################################################################
   # --send-data and exit.
   # ########################################################################
   if ( my $service = $o->get('send-data') ) {
      eval {
         send_data(
            api_key     => $api_key,
            service     => $service,
            lib_dir     => $o->get('lib'),
            spool_dir   => $o->get('spool'),
            interactive => $o->get('interactive'),
         );
      };
      if ( $EVAL_ERROR ) {
         $logger->fatal("--send-data $service error: $EVAL_ERROR");
      }
      return $exit_status;
   }

   # ########################################################################
   # This is the main pt-agent daemon, a long-running and resilient
   # process.  Only internal errors should cause it to stop.  Else,
   # external errors, like Percona web API not responding, should be
   # retried forever.
   # ########################################################################

   # Check the config file.  This should probably never fail because
   # the config file is $HOME/.pt-agent.conf, so the user should
   # be able to write to their home dir.  --run-service and --send-data
   # don't need to do this because if there's no valid config, they should
   # fail; they'll probably die due to --lib missing, which they verify
   # but don't create.
   my $config_file = get_config_file();
   if ( -f $config_file && !-w $config_file ) {
      $logger->fatal("$config_file exists but is not writable")
   }

   # Start, i.e. init/create/update, the agent.  This forks and daemonizes,
   # so we're the child/daemon process when it returns.  To remember how
   # this differs from run_agent(): first you start a car, then you put it
   # in drive to "run" (drive) it.
   my $running = start_agent(
      api_key   => $api_key,
      Cxn       => $cxn,
      lib_dir   => $o->get('lib'),
      daemonize => $o->get('daemonize'),
      pid_file  => $o->get('pid'),
      log_file  => $o->get('log'),
      # Use default tries and interval: 1440 * 60 = 1 day
   );

   # Wait time between checking for new config and services.
   # Use the tool's built-in default until a config is gotten,
   # then config->{check-interval} will be pass in.
   my $check_interval = $o->get('check-interval');
   my $interval = sub {
      my ($t, $quiet) = @_;
      return unless $oktorun;
      $t ||= $check_interval;
      $logger->debug("Sleeping $t seconds") unless $quiet;
      sleep $t;
   };

   my $safeguards = Safeguards->new(
      disk_bytes_free => $o->get('disk-bytes-free'),
      disk_pct_free   => $o->get('disk-pct-free'),
   );

   # Run the agent's main loop which doesn't return until the service
   # is stopped, killed, or has an internal bug.
   eval {
      run_agent(
         agent      => $running->{agent},
         client     => $running->{client},
         daemon     => $running->{daemon},
         interval   => $interval,
         safeguards => $safeguards,
         Cxn        => $cxn,
         lib_dir    => $o->get('lib'),
      );
   };
   if ( $EVAL_ERROR ) {
      $logger->fatal("Error running agent: $EVAL_ERROR");
   }

   $logger->info("pt-agent exit $exit_status, oktorun $oktorun");

   return $exit_status;
}

# ############################################################################
# Subroutines
# ############################################################################

# ################################################## #
# Percona Web API subs for agent and spool processes #
# ################################################## #

# Create and connect a Percona Web API client.
sub get_api_client {
   my (%args) = @_;

   have_required_args(\%args, qw(
      api_key
      interval
   )) or die;
   my $api_key  = $args{api_key};
   my $interval = $args{interval};

   # Optional args
   my $tries      = $args{tries};
   my $_oktorun   = $args{oktorun}    || sub { return $oktorun };
   my $entry_link = $args{entry_link} || $ENV{PCT_ENTRY_LINK};
   my $quiet      = $args{quiet};

   my $client = Percona::WebAPI::Client->new(
      api_key => $api_key,
      ($entry_link ? (entry_link => $entry_link) : ()),
   );

   my $entry_links;
   while ( $_oktorun->() && (!defined $tries || $tries--) ) {
      if ( !$state->{connecting_to_api}++ ) {
         $logger->debug("Connecting to Percona Web API")  # once
      }
      eval {
         $entry_links = $client->get(link => $client->entry_link);
      };
      if ( my $e = $EVAL_ERROR ) {
         my $code = $client->response->code;
         if ( $code && $code == 503 ) {
            $logger->info("API is down for maintenance (503)");
         }
         else {
            $logger->info("API error: $e");
         }
      }
      elsif (
         !$entry_links
         || (ref($entry_links) || '') ne 'HASH'
         || !scalar keys %$entry_links
      ) {
         $logger->info('API returned invalid entry links: '
            . Dumper($entry_links));
      }
      elsif ( !$entry_links->{agents} ) {
         $logger->info('API did not return agents link: '
            . Dumper($entry_links));
      }
      else {
         $logger->debug("Connected");
         delete $state->{connecting_to_api};
         last;  # success
      }
      if (!defined $tries || $tries > 0) {
         $interval->();  # failure, try again
      }
   }

   # Create another client for Percona::Agent::Logger.  If the primary
   # client was created, then the API key and entry link worked, so
   # just duplicate them for the new logger client.  We don't need to
   # connect the logger client because clients are stateless so knowing
   # the primary client connected ensures that the logger client can/will
   # connect to with the same API and entry link.
   my $logger_client;
   if ( $client && $entry_links ) {
      $logger_client = Percona::WebAPI::Client->new(
         api_key => $api_key,
         ($entry_link ? (entry_link => $entry_link) : ()),
      );
   }

   return $client, $entry_links, $logger_client;
}

sub load_local_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      lib_dir
   )) or die;
   my $lib_dir = $args{lib_dir};

   # Optional args
   my $agent_uuid = $args{agent_uuid};
   my $quiet      = $args{quiet};

   my $agent;
   my $agent_file = $lib_dir . "/agent";
   if ( -f $agent_file ) {
      $logger->debug("Reading saved Agent from $agent_file") unless $quiet;
      my $agent_hashref = JSON::decode_json(slurp($agent_file));
      $agent = Percona::WebAPI::Resource::Agent->new(%$agent_hashref);
      if ( !$agent->uuid ) {
         $logger->fatal("No UUID for Agent in $agent_file.");
      }
   }
   else {
      $logger->debug("No local agent") unless $quiet;
   }

   return $agent;
}

# Initialize the agent, i.e. create and return an Agent resource.
# If there's an agent_id, then its updated (PUT), else a new agent
# is created  (POST).  Doesn't return until successful.
sub init_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      agent
      action
      link
      client
      tries
      interval
   )) or die;
   my $agent    = $args{agent};
   my $action   = $args{action};
   my $link     = $args{link};
   my $client   = $args{client};
   my $tries    = $args{tries};
   my $interval = $args{interval};

   # Optional args
   my $_oktorun = $args{oktorun} || sub { return $oktorun };
   my $actions  = $args{actions};
   my $quiet    = $args{quiet};

   # Update these attribs every time the agent is initialized.
   # Other optional attribs, like versions, are left to the caller.
   chomp(my $who      = `whoami 2>/dev/null`);
   chomp(my $hostname = `hostname`);
   $agent->hostname($hostname);
   $agent->username($ENV{USER} || $ENV{LOGNAME} || $who);

   # Try to create/update the Agent.
   my $success = 0;
   while ( $_oktorun->() && $tries-- ) {
      if ( !$state->{init_action}++ && !$quiet ) {
         $logger->info($action eq 'put' ? "Updating agent " . $agent->name
                                        : "Creating new agent");
      }
      my $agent_uri = eval {
         $client->$action(
            link      => $link,
            resources => $agent,
         );
      };
      if ( $EVAL_ERROR ) {
         my $code = $client->response->code;
         if ( $code && $code == 404 ) {
            my $api_ok = ping_api(
               client => $client,
            );
            if ( $api_ok ) {
               $logger->fatal("API reports agent not found: the agent has been "
                  . "deleted, or its UUID (" . ($agent->uuid || '?') . ") "
                  . "is wrong.  Check https://cloud.percona.com/agents for the "
                  . "list of active agents.");
            }
            else {
               $logger->warning("API is down.");
            }
         }
         elsif ( $code && $code == 403 ) {
            if ( !$state->{too_many_agents}++ ) {
               $logger->warning("API reports too many agents.  Check "
                  . "https://cloud.percona.com/agents for the list of "
                  . "installed agents.  Will try again $tries times, "
                  . "but this warning will not be printed again.");
            }
         }
         else {
            $logger->warning($EVAL_ERROR);
         }
      }
      elsif ( !$agent_uri ) {
         $logger->warning("No URI for Agent " . $agent->name);
      }
      else {
         # The Agent URI will have been returned in the Location header
         # of the POST or PUT response.  GET the Agent (even after PUT)
         # to get a link to the agent's config.
         eval {
            $agent = $client->get(
               link => $agent_uri,
            );
         };
         if ( $EVAL_ERROR ) {
            $logger->warning($EVAL_ERROR);
         }
         else {
            $success = 1;
            last;  # success
         }
      }
      if ( $tries > 0 ) {
         $interval->();  # failure, try again
      }
   }

   delete $state->{init_action};
   delete $state->{too_many_agents};

   return $agent, $success;
}

# Check and init the --lib dir.  This dir is used to save the Agent resource
# (/agent), Service resources (/services/), and crontab for services(/conrtab,
# /crontab.err).
sub init_lib_dir {
   my (%args) = @_;
   have_required_args(\%args, qw(
      lib_dir
   )) or die;
   my $lib_dir = $args{lib_dir};

   # Optiona args
   my $verify = $args{verify};
   my $quiet  = $args{quiet};

   $logger->info(($verify ? 'Verify' : 'Initializing') . " --lib $lib_dir")
      unless $quiet;

   if ( ! -d $lib_dir ) {
      if ( $verify ) {
         die "$lib_dir does not exist\n";
      }
      else {
         $logger->info("$lib_dir does not exist, creating")
            unless $quiet;
         _safe_mkdir($lib_dir);
      }
   }
   elsif ( ! -w $lib_dir ) {
      die "--lib $lib_dir is not writable.\n";
   }

   foreach my $dir ( qw(services logs pids meta) ) {
      my $dir = "$lib_dir/$dir";
      if ( ! -d $dir ) {
         if ( $verify ) {
            die "$dir does not exist\n";
         }
         else {
            $logger->info("$dir does not exist, creating")
               unless $quiet;
            _safe_mkdir($dir);
         }
      }
      elsif ( ! -w $dir ) {
         die "$dir is not writable.\n";
      }
   }

   return;
}

# ################################ #
# Agent (main daemon) process subs #
# ################################ #

sub start_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      api_key
      lib_dir
      Cxn
   )) or die;
   my $api_key  = $args{api_key};
   my $lib_dir  = $args{lib_dir};
   my $cxn      = $args{Cxn};

   # Optional args
   my $agent_uuid    = $args{agent_uuid};
   my $daemonize     = $args{daemonize};
   my $pid_file      = $args{pid_file};
   my $log_file      = $args{log_file};
   my $_oktorun      = $args{oktorun}  || sub { return $oktorun };
   my $tries         = $args{tries}    || 1440;  # 1440 * 60 = 1 day
   my $interval      = $args{interval} || sub { sleep 60; };
   my $versions      = $args{versions};       # for testing
   my $client        = $args{client};         # for testing
   my $entry_links   = $args{entry_links};    # for testing
   my $logger_client = $args{logger_client};  # for testing

   # $logger->info('Starting agent');

   # Daemonize first so all output goes to the --log.
   my $daemon = Daemon->new(
      daemonize   => $daemonize,
      pid_file    => $pid_file,
      log_file    => $log_file,
      parent_exit => sub {
         my $child_pid = shift;
         print "pt-agent has daemonized and is running as PID $child_pid:

  --lib " . ($lib_dir  || '') . "
  --log " . ($log_file || '') . "
  --pid " . ($pid_file || '') . "

These values can change if a different configuration is received.
",
      }
   );
   $daemon->run();
   if ( $daemonize ) {
      # If we daemonized, the parent has already exited and we're the child.
      # We shared a copy of every Cxn with the parent, and the parent's copies
      # were destroyed but the dbhs were not disconnected because the parent
      # attrib was true.  Now, as the child, set it false so the dbhs will be
      # disconnected when our Cxn copies are destroyed.  If we didn't daemonize,
      # then we're not really a parent (since we have no children), so set it
      # false to auto-disconnect the dbhs when our Cxns are destroyed.
      $cxn->{parent} = 0;
   }

   # Make --lib and its subdirectories.
   eval {
      init_lib_dir(
         lib_dir => $lib_dir,
      );
   };
   if ( $EVAL_ERROR ) {
      chomp($EVAL_ERROR);
      $logger->info("Error initializing --lib $lib_dir: $EVAL_ERROR.  "
         . "Configure the agent to use a writeable --lib directory.");
   }

   # Connect to the API and get entry links.
   if ( !$client || !$entry_links ) {
      ($client, $entry_links, $logger_client) = get_api_client(
         api_key  => $api_key,
         tries    => $tries,
         interval => $interval,
      );
   }
   if ( !$client || !$entry_links ) {
      die "Failed to initialize the API client.  The API may be down.  Please try again.\n";
   }
   return unless $_oktorun->();

   # Do a version-check every time the agent starts.  If versions
   # have changed, this can affect how services are implemented.
   # Since this is the only thing we use the Cxn for, get_versions()
   # connects and disconnect it, if possible.  If not possible, the
   # MySQL version isn't sent in hopes that it becomes possible to get
   # it later.
   if ( !$versions || !$versions->{MySQL} ) {
      $versions = get_versions(
         Cxn   => $cxn,
      );
   }
   return unless $_oktorun->();

   # Load and update the local (i.e. existing) agent, or create a new one.
   my $agent;
   my $action;
   my $link;
   if ( $agent_uuid ) {
      $logger->info("Re-creating Agent with UUID $agent_uuid");
      $agent = Percona::WebAPI::Resource::Agent->new(
         uuid     => $agent_uuid,
         versions => $versions,
      );
       $action = 'put';  # update
       $link   = $entry_links->{agents} . '/' . $agent->uuid;
   }
   else {
      # First try to load the local agent.
      $agent = load_local_agent(
         lib_dir => $lib_dir,
      );
      if ( $agent ) {
         # Loaded local agent.
         $action = 'put';  # update
         $link   = $entry_links->{agents} . '/' . $agent->uuid;
         $agent->{versions} = $versions;
      }
      else {
         # No local agent and --agent-uuid wasn't give.
         $agent = Percona::WebAPI::Resource::Agent->new(
            versions => $versions,
         );
         $action = 'post';  # create
         $link   = $entry_links->{agents};
      }
   }

   my $success;
   ($agent, $success) = init_agent(
      agent    => $agent,
      action   => $action,      # put or post
      link     => $link,
      client   => $client,
      tries    => $tries,
      interval => $interval,
      oktorun  => $_oktorun,    # optional
   );
   if ( !$success ) {
      die "Failed to initialize the agent.  The API may be down.  Please try again.\n";
   }

   # Give the logger its client so that it will also POST every log entry
   # to /agent/{uuid}/log.  This is done asynchronously by a thread so a
   # simple info("Hello world!") to STDOUT won't block if the API isn't
   # responding. -- Both client and log_link are required to enable this.
   if ( $logger->online_logging && $agent->links->{log} && $logger_client ) {
      $logger->start_online_logging(
         client   => $logger_client,
         log_link => $agent->links->{log},
      );
      $logger->debug("Log API enabled");
   }

   save_agent(
      agent   => $agent,
      lib_dir => $lib_dir,
   );

   # Remove old service files.  New instance of agent shouldn't inherit
   # anything from previous runs, in case previous runs were bad.
   my $service_files = "$lib_dir/services/*";
   foreach my $service_file ( glob $service_files ) {
      if ( unlink $service_file ) {
         $logger->debug("Removed $service_file");
      }
      else {
         $logger->warning("Cannot remove $service_file: $OS_ERROR");
      }
   }
   eval {
      schedule_services(
         services => [],
         lib_dir  => $lib_dir,
         quiet    => 1,
      );
   };
   if ( $EVAL_ERROR ) {
      $logger->error("Error removing services from crontab: $EVAL_ERROR");
   }

   return {
      agent  => $agent,
      client => $client,
      daemon => $daemon,
   };
}

# Run the agent, i.e. exec the main loop to check/update the config
# and services.  Doesn't return until the service is stopped or killed.
sub run_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      agent
      client
      daemon
      interval
      lib_dir
      safeguards
      Cxn
   )) or die;
   my $agent      = $args{agent};
   my $client     = $args{client};
   my $daemon     = $args{daemon};
   my $interval   = $args{interval};
   my $lib_dir    = $args{lib_dir};
   my $safeguards = $args{safeguards};
   my $cxn        = $args{Cxn};

   # Optional args
   my $_oktorun = $args{oktorun} || sub { return $oktorun };

   $logger->info('Running agent ' . $agent->name);

   # #######################################################################
   # Main agent loop
   # #######################################################################
   $state->{need_mysql_version} = 1;
   $state->{first_config} = 1;
   $state->{ready} = 0;
   my $first_config_interval = 20;
   $logger->info("Checking silently every $first_config_interval seconds"
      . " for the first config");

   my $success;
   my $new_daemon;
   my $config;
   my $services = {};
   while ( $_oktorun->() ) {
      ($config, $lib_dir, $new_daemon, $success) = get_config(
         link    => $agent->links->{config},
         agent   => $agent,
         client  => $client,
         daemon  => $daemon,
         lib_dir => $lib_dir,
         config  => $config,
         quiet   => $state->{first_config},
      );

      # Get services only if we successfully got the config because the services
      # may depened on the current config, specifically the --spool dir.
      if ( $success && $config && $config->links->{services} ) {
         if ( $state->{first_config} ) {
            delete $state->{first_config};
         }

         if ( $new_daemon ) {
            # NOTE: Daemon objects use DESTROY to auto-remove their pid file
            # when they lose scope (i.e. ref count goes to zero).  This
            # assignment destroys (old) $daemon, so it auto-removes the old
            # pid file.  $new_daemon maintains scope and the new pid file
            # by becoming $daemon which was defined in the outer scope so
            # it won't destroy again when we leave this block.  Fancy!
            # About sharing_pid_file: see the comment in apply_config().
            if ( $new_daemon->{sharing_pid_file} ) {
               $daemon->{pid_file_owner} = 0;
               delete $new_daemon->{sharing_pid_file};
            }
            $daemon = $new_daemon;
         }

         # Connect to MySQL, then check stuff.
         my $o = new OptionParser();
         $o->get_specs();
         $o->get_opts();
         my $dp = $o->DSNParser();
         $dp->prop('set-vars', $o->set_vars());
         my $dsn = $dp->parse_options($o);   
         eval {
            $cxn->connect(dsn => $dsn);
         };
         if ( $EVAL_ERROR ) {
            if ( !$state->{mysql_error}++ ) {
               $logger->warning("MySQL connection failure: $EVAL_ERROR");
            }
            else {
               $logger->debug("MySQL connection failure: $EVAL_ERROR");
            }
            $state->{have_mysql} = 0;
            $state->{need_mysql_version} = 1;
         }
         else {
            if ( !$state->{have_mysql} ) {
               $logger->info("MySQL OK");
            }
            $state->{have_mysql} = 1;
            check_if_mysql_restarted(
               dbh => $cxn->dbh,
            );
            if ( $state->{need_mysql_version} ) {
               $logger->debug("Need MySQL version");
               my $versions = get_versions(Cxn => $cxn);
               if ( $versions->{MySQL} ) {
                  $agent->versions($versions);
                  my $updated_agent;
                  ($agent, $updated_agent) = init_agent(
                     agent    => $agent,
                     action   => 'put',
                     link     => $agent->links->{self},
                     client   => $client,
                     tries    => 1,
                     interval => sub { return; },
                     quiet    => 1,
                  );
                  if ( $updated_agent ) {
                     $logger->debug("Got MySQL version");
                     save_agent(
                        agent   => $agent,
                        lib_dir => $lib_dir,
                     );
                     if ( !$state->{ready} || $state->{mysql_error} ) {
                        $logger->info('Agent OK');
                     }
                     delete $state->{need_mysql_version};
                     delete $state->{mysql_error};
                     $state->{ready} = 1;
                  }
               }
               else {
                  if ( !$state->{mysql_error}++ ) {
                     $logger->warning("Failed to get MySQL version");
                  }
                  else {
                     $logger->debug("Failed to get MySQL version");
                  }
               }
            }
            $cxn->dbh->disconnect();
         }

         # Check the safeguards.
         my ($disk_space, $disk_space_ok);
         eval {
            $disk_space = $safeguards->get_disk_space(
               filesystem => $config->options->{spool},
            );
            $disk_space_ok = $safeguards->check_disk_space(
               disk_space => $disk_space,
            );
         };
         if ( $EVAL_ERROR ) {
            $logger->error("Error checking disk space: $EVAL_ERROR");
            $disk_space_ok = 1;
         }
         if ( !$disk_space_ok ) {
            $logger->warning("Disk bytes free/percentage threshold: "
               . $safeguards->{disk_bytes_free}
               . '/'
               . $safeguards->{disk_pct_free});
            $logger->warning("Disk space is low, stopping all services:\n"
               . $disk_space);
            if ( !$state->{all_services_are_stopped} ) {
               stop_all_services(
                  lib_dir => $lib_dir,
               );
            }
            $logger->warning('Services will restart when disk space "
               . "threshold checks pass');
         }
         elsif ( $state->{ready} ) {
            ($services, $success) = get_services(
               link      => $config->links->{services},
               agent     => $agent,
               client    => $client,
               lib_dir   => $lib_dir,
               services  => $services,
               json      => $args{json},    # optional, for testing
               bin_dir   => $args{bin_dir}, # optional, for testing
            );
         }
      }

      # If configured, wait the given interval.  Else, retry more
      # quickly so we're ready to go soon after we're configured.
      $interval->(
          !$state->{ready} ? (20, 1)
         : $config         ? ($config->options->{'check-interval'}, 0)
         :                   ($first_config_interval , 1)  # 1=quiet
      );
   }

   stop_all_services(
      lib_dir => $lib_dir,
   );

   # This shouldn't happen until the service is stopped/killed.
   $logger->info('Agent ' . $agent->name . ' has stopped');
   return;
}

sub get_config {
   my (%args) = @_;
   have_required_args(\%args, qw(
      link
      agent
      client
      daemon
      lib_dir
   )) or die;
   my $link    = $args{link};
   my $agent   = $args{agent};
   my $client  = $args{client};
   my $daemon  = $args{daemon};
   my $lib_dir = $args{lib_dir};

   # Optional args
   my $config = $args{config};   # may not be defined yet
   my $quiet  = $args{quiet};

   my $success = 0;
   my $new_daemon;

   $logger->debug('Getting agent config') unless $quiet;
   my $new_config = eval {
      $client->get(
         link => $link,
      );
   };
   if ( my $e = $EVAL_ERROR ) {
      if (blessed($e)) {
         if ($e->isa('Percona::WebAPI::Exception::Request')) {
            if ( $e->status == 404 ) {
               my $api_ok = ping_api(
                  client => $client,
               );
               if ( $api_ok ) {
                  stop_all_services(
                     lib_dir => $lib_dir,
                  );
                  $logger->fatal("API reports agent not found: the agent has been "
                     . "deleted, or its UUID (" . ($agent->uuid || '?') . ") "
                     . "is wrong.  Check https://cloud.percona.com/agents for a "
                     . "list of active agents.");
               }
               else {
                  # offline warning
                  $logger->_log(0, 'WARNING', "Cannot get agent config: API is down.  "
                        . "Will try again.");
               }
            }
            else {
               # offline warning
               $logger->_log(0, 'WARNING', "Cannot get agent config: API error: $e.  "
                  . "Will try again.")
            }
         }
         elsif ($e->isa('Percona::WebAPI::Exception::Resource')) {
            $logger->error("Invalid agent config: $e");
         }
      }
      else {
         $logger->error("Internal error getting agent config: $e");
      }
   }
   else {
      eval {
         if ( !$quiet ) {
            $logger->debug("Running config: " . ($config ? $config->ts : ''));
            $logger->debug("Current config: " . $new_config->ts);
         }
         if ( !$config || $new_config->ts > $config->ts ) {
            ($lib_dir, $new_daemon) = apply_config(
               agent       => $agent,
               old_config  => $config,
               new_config  => $new_config,
               lib_dir     => $lib_dir,
               daemon      => $daemon,
            );
            $config  = $new_config;
            $success = 1;
            $logger->info('Config ' . $config->ts . ' applied');

            $state->{need_mysql_version} = 1;
         }
         else {
            $success = 1;
            $logger->debug('Config has not changed') unless $quiet;
         }
      };
      if ( $EVAL_ERROR ) {
         chomp $EVAL_ERROR;
         $logger->warning("Failed to apply config " . $new_config->ts
            . ": $EVAL_ERROR  Will try again.");
      }
   }

   return ($config, $lib_dir, $new_daemon, $success);
}

sub apply_config {
   my (%args) = @_;

   have_required_args(\%args, qw(
      agent
      new_config
      lib_dir
      daemon
   )) or die;
   my $agent      = $args{agent};
   my $new_config = $args{new_config};
   my $lib_dir    = $args{lib_dir};
   my $daemon     = $args{daemon};

   # Optional args
   my $old_config = $args{old_config};

   $logger->debug('Applying config ' . $new_config->ts);

   # If the --lib dir has changed, init the new one and re-write
   # the Agent resource in it.
   my $new_lib_dir = $new_config->options->{lib};
   if ( ($new_lib_dir ne $lib_dir) || $state->{first_config} ) {
      $logger->info($state->{first_config} ? "Applying first config"
                                   : "New --lib direcotry: $new_lib_dir");
      init_lib_dir(
         lib_dir => $new_lib_dir,
      );

      # TODO: copy old-lib/services/* to new-lib/services/ ?

      # Save agent as --lib/agent so next time the tool starts it
      # loads the agent from the latest --lib dir.
      save_agent(
         agent   => $agent,
         lib_dir => $new_lib_dir,
      );
   }

   # If --pid or --log has changed, we need to "re-daemonize",
   # i.e. change these files while running, but the program
   # does _not_ actually restart.
   my $new_daemon;
   my $make_new_daemon = 0;
   my $old_pid         = $daemon->{pid_file}         || '';
   my $old_log         = $daemon->{log_file}         || '';
   my $new_pid         = $new_config->options->{pid} || '';
   my $new_log         = $new_config->options->{log} || '';
   if ( $old_pid ne $new_pid ) {
      $logger->info('NOTICE: Changing --pid file from ' . ($old_pid || '(none)')
          . ' to ' . ($new_pid || '(none)'));
      $make_new_daemon = 1;
   }
   if ( $daemon->{daemonize} ) {
      # --log only matters if we're daemonized
      if ( $old_log ne $new_log ) {
         $logger->info('NOTICE: Changing --log file from '
            . ($old_log || '(none)') . ' to ' . ($new_log || '(none)'));
         $make_new_daemon = 1;
      }
   }
   if ( $make_new_daemon ) {
      # We're either already daemonized or we didn't daemonize in the first
      # place, so daemonize => 0 here.  Also, if log hasn't changed, the
      # effect is simply closing and re-opening the same log.
      # TODO: If log changes but pid doesn't? will probably block itself.
      $new_daemon = Daemon->new(
         daemonize      => 0,  
         pid_file       => $new_pid,
         log_file       => $new_log,
         force_log_file => $daemon->{daemonize},
      );
      eval {
         $new_daemon->run();

         if ( $daemon->{daemonize} && $old_log ne $new_log  ) {
            $logger->info('New log file, previous was '
               . ($old_log || 'unset'));
         }
         if ( $old_pid eq $new_pid ) {
            # If the PID file has not, then the old/original daemon and
            # the new daemon are sharing the same pid file. The old one
            # created it, but the new one will continue to hold it when
            # the old one goes away.  Set sharing_pid_file to signal to
            # the caller that they need to set old daemon pid_file_owner=0
            # so it does not auto-remove the shared pid file when it goes
            # away.
            $new_daemon->{sharing_pid_file} = 1;
         }
      };
      if ( $EVAL_ERROR ) {
         die "Error changing --pid and/or --log: $EVAL_ERROR\n";
      }
   }

   # Save config in $HOME/.pt-agent.conf if successful.
   write_config(
      config => $new_config,
   );

   return ($new_lib_dir || $lib_dir), $new_daemon;
}

# Write a Config resource to a Percona Toolkit config file,
# usually $HOME/pt-agent.conf.
sub write_config {
   my (%args) = @_;

   have_required_args(\%args, qw(
      config
   )) or die;
   my $config = $args{config};

   my $file = get_config_file();
   $logger->info("Writing config to $file");

   # Get the api-key line if any; we don't want to/can't clobber this.
   my $api_key;
   my $no_log_api;
   if ( -f $file ) {
      open my $fh, "<", $file
         or die "Error opening $file: $OS_ERROR";
      my $contents = do { local $/ = undef; <$fh> };
      close $fh;
      ($api_key)    = $contents =~ m/^(api-key=\S+)$/m;
      ($no_log_api) = $contents =~ m/^(no-log-api)$/m;
   }

   # Re-write the api-key, if any, then write the config.
   open my $fh, '>', $file
      or die "Error opening $file: $OS_ERROR";
   if ( $api_key ) {
      print { $fh } $api_key, "\n"
         or die "Error writing to $file: $OS_ERROR";
   }
   if ( $no_log_api ) {
      print { $fh } $no_log_api, "\n"
         or die "Error writing to $file: $OS_ERROR";
   }
   print { $fh } as_config($config)
      or die "Error writing to $file: $OS_ERROR";
   close $fh
      or die "Error closing $file: $OS_ERROR";

   return;
}

sub get_services {
   my (%args) = @_;
   have_required_args(\%args, qw(
      link
      agent
      client
      lib_dir
      services
   )) or die;
   my $link          = $args{link};
   my $agent         = $args{agent};
   my $client        = $args{client};
   my $lib_dir       = $args{lib_dir};
   my $prev_services = $args{services};

   my $success = 0;

   eval {
      $logger->debug('Getting services');
      my $curr_services = $client->get(
         link => $link,
      );

      if ( !$curr_services ) {
         $logger->error("GET $link did not return anything, "
            . "expected a list of services");
      }
      elsif ( !scalar @$curr_services && !scalar keys %$prev_services ) {
         $logger->debug("No services are enabled for this agent");

         # Remove these state that no longer matter if there are no services.
         if ( $state->{mysql_restarted} ) {
            $state->{last_uptime}       = 0;
            $state->{last_uptime_check} = 0;
            delete $state->{mysql_restarted};
         }
         if ( $state->{all_services_are_stopped} ) {
            delete $state->{all_services_are_stopped};
         }
      }
      else {
         if ( $state->{all_services_are_stopped} ) {
            $logger->info('Restarting services after safeguard shutdown');
            # If prev_services is empty, then it's like agent startup:
            # get all the latest services and start them, and remove
            # any old services.  We could just start-* the services we
            # already have, but since they were shut down due to a safeguard,
            # maybe (probably) they've changed.
            $prev_services = {};
            delete $state->{all_services_are_stopped};
         }
         elsif ( my $ts = $state->{mysql_restarted} ) {
            $logger->info("Restarting services after MySQL restart at $ts");
            $prev_services              = {};
            $state->{last_uptime}       = 0;
            $state->{last_uptime_check} = 0;
            delete $state->{mysql_restarted};
         }

         # Determine which services are new (added), changed/updated,
         # and removed.
         my $sorted_services = sort_services(
            prev_services => $prev_services,
            curr_services => $curr_services,
         );

         # First, stop and remove services.  Do this before write_services()
         # because this call looks for  --lib/services/stop-service which
         # write_services() removes.  I.e. use the service's stop- meta
         # counterpart (if any) before we remove the service.
         my $removed_ok = apply_services(
            action   => 'stop',
            services => $sorted_services->{removed},
            lib_dir  => $lib_dir,
            bin_dir  => $args{bin_dir},  # optional, for testing
            exec_cmd => $args{exec_cmd}, # optional, for testing
         );

         # Second, save each service in --lib/services/.  Do this before
         # the next calls to apply_services() because those calls look for
         # --lib/services/start-service which won't exist for new services
         # until written by this call.
         write_services(
            sorted_services => $sorted_services,
            lib_dir         => $lib_dir,
            json            => $args{json},  # optional, for testing
         );

         # Start new services and restart existing updated services.
         # Do this before calling schedule_services() so that, for example,
         # start-query-history is ran before query-history is scheduled
         # and starts running.

         # Run services with the run_once flag.  Unlike apply_services(),
         # this call runs the service directly, whether it's meta or not,
         # then it removes it from the services hashref so there's no
         # chance of running it again unless it's received again.
         apply_services_once(
            services => $sorted_services->{services},
            lib_dir  => $lib_dir,
            bin_dir  => $args{bin_dir},  # optional, for testing
            exec_cmd => $args{exec_cmd}, # optional, for testing
         );

         # Start new services.
         my $started_ok = apply_services(
            action   => 'start',
            services => $sorted_services->{added},
            lib_dir  => $lib_dir,
            bin_dir  => $args{bin_dir},  # optional, for testing
            exec_cmd => $args{exec_cmd}, # optional, for testing
         );

         # Restart existing updated services.
         my $restarted_ok = apply_services(
            action   => 'restart',
            services => $sorted_services->{updated},
            lib_dir  => $lib_dir,
            bin_dir  => $args{bin_dir},  # optional, for testing
            exec_cmd => $args{exec_cmd}, # optional, for testing
         );

         # Schedule any services with a run_schedule or spool_schedule.
         # This must be called last, after write_services() and
         # apply_services() because, for example, a service schedule
         # to run at */5 may run effectively immediate if we write
         # the new crontab at 00:04:59, so everything has to be
         # ready to go at this point.
         if (     scalar @$removed_ok
               || scalar @$started_ok
               || scalar @$restarted_ok )
         {
            schedule_services(
               services => [
                  @$started_ok,
                  @$restarted_ok,
                  @{$sorted_services->{unchanged}},
               ],
               lib_dir  => $lib_dir,
               bin_dir  => $args{bin_dir},  # optional, for testing
               exec_cmd => $args{exec_cmd}, # optional, for testing
            );

            $logger->info('Services OK');
         }
         else {
            $logger->debug('Services have not changed');
         }
         # TODO: probably shouldn't keep re-assigning this unless necessary
         $prev_services = $sorted_services->{services};
         $success       = 1;
      }
   };
   if ( $EVAL_ERROR ) {
      $logger->warning($EVAL_ERROR);
   }

   return $prev_services, $success;
}

sub sort_services {
   my (%args) = @_;

   have_required_args(\%args, qw(
      prev_services
      curr_services
   )) or die;
   my $prev_services = $args{prev_services};  # hashref
   my $curr_services = $args{curr_services};  # arrayref

   my $services = {}; # curr_services as hashref keyed on service name
   my @added;
   my @updated;
   my @removed;
   my @unchanged;

   foreach my $service ( @$curr_services ) {
      my $name = $service->name;
      $services->{$name} = $service;

      # apply_services() only needs real services, from which it can infer
      # certain meta-services like "start-foo" for real service "foo",
      # but write_services() needs meta-services too so it can know to
      # remove their files from --lib/services/.

      if ( !exists $prev_services->{$name} ) {
         push @added, $service;
      }
      elsif ( $service->ts > $prev_services->{$name}->ts ) {
         push @updated, $service;
      }
      else {
         push @unchanged, $service;
      }
   }
   if ( scalar keys %$prev_services ) {
      @removed = grep { !exists $services->{$_->name} } values %$prev_services;
   }

   if ( scalar @added ) {
      $logger->info("Added services: "
         . join(', ', map { $_->name } @added));
   }
   if ( scalar @updated ) {
      $logger->info("Services updated: "
         . join(', ', map { $_->name } @updated));
   }
   if ( scalar @removed ) {
      $logger->info("Services removed: "
         . join(', ', map { $_->name } @removed));
   }

   my $sorted_services = {
      services  => $services,
      added     => \@added,
      updated   => \@updated,
      removed   => \@removed,
      unchanged => \@unchanged,
   };
   return $sorted_services;
}

# Write each service to its own file in --lib/.  Remove services
# that are not longer implemented (i.e. not in the services array).
sub write_services {
   my (%args) = @_;

   have_required_args(\%args, qw(
      sorted_services
      lib_dir
   )) or die;
   my $sorted_services = $args{sorted_services};
   my $lib_dir         = $args{lib_dir};

   # Optional args
   my $json = $args{json};  # for testing

   $lib_dir .= '/services';

   $logger->debug("Writing services to $lib_dir");

   # Save current, active services.
   foreach my $service (
      @{$sorted_services->{added}}, @{$sorted_services->{updated}}
   ) {
      my $file   = $lib_dir . '/' . $service->name;
      my $action = -f $file ? 'Updated' : 'Added';
      open my $fh, '>', $file
         or die "Error opening $file: $OS_ERROR";
      print { $fh } as_json($service, with_links => 1, json => $json)
         or die "Error writing to $file: $OS_ERROR";
      close $fh
         or die "Error closing $file: $OS_ERROR";
      $logger->info("$action $file");
   }

   # Remove old services.
   foreach my $service ( @{$sorted_services->{removed}} ) {
      my $file   = $lib_dir . '/' . $service->name;
      if ( -f $file ) {
         unlink $file
            or die "Error removing $file: $OS_ERROR";
         $logger->info("Removed $file");
      }
   }

   return;
}

# Write Service->run_schedule and Service->spool_schedule lines to crontab,
# along with any other non-pt-agent lines, then reload crontab.
sub schedule_services {
   my (%args) = @_;

   have_required_args(\%args, qw(
      services
      lib_dir
   )) or die;
   my $services = $args{services};
   my $lib_dir  = $args{lib_dir};

   # Optional args
   my $quiet    = $args{quiet};
   my $exec_cmd = $args{exec_cmd} || sub { return system(@_) };

   $logger->info("Scheduling services") unless $quiet;

   # Only schedule "periodic" services, i.e. ones that run periodically,
   # not just once.
   my @periodic_services = grep { $_->run_schedule || $_->spool_schedule }
                           @$services;
   my $new_crontab = make_new_crontab(
      %args,
      services => \@periodic_services,
   );
   $logger->info("New crontab:\n" . $new_crontab || '') unless $quiet;

   my $crontab_file = "$lib_dir/crontab";
   open my $fh, '>', $crontab_file
      or die "Error opening $crontab_file: $OS_ERROR";
   print { $fh } $new_crontab
      or die "Error writing to $crontab_file: $OS_ERROR";
   close $fh
      or die "Error closing $crontab_file: $OS_ERROR";

   my $err_file = "$lib_dir/crontab.err";
   my $retval   = $exec_cmd->("crontab $crontab_file > $err_file 2>&1");
   if ( $retval ) {
      my $error =  -f $err_file ? `cat $err_file` : '';
      die "Error setting new crontab: $error";
   }

   return;
}

# Combine Service->run_schedule and (optionally) Service->spool_schedule
# lines with non-pt-agent lines, i.e. don't clobber the user's other
# crontab lines.
sub make_new_crontab {
   my (%args) = @_;

   have_required_args(\%args, qw(
      services
   )) or die;
   my $services = $args{services};

   # Optional args
   my $crontab_list = defined $args{crontab_list} ? $args{crontab_list}
                    :                               `crontab -l 2>/dev/null`;
   my $bin_dir = defined $args{bin_dir} ? $args{bin_dir}
               :                          "$FindBin::Bin/";

   my @other_lines
      = grep { $_ !~ m/pt-agent (?:--run-service|--send-data)/ }
        split("\n", $crontab_list);
   PTDEBUG && _d('Other crontab lines:', Dumper(\@other_lines));

   my $env_vars = env_vars();

   my @pt_agent_lines;
   foreach my $service ( @$services ) {
      if ( $service->run_schedule ) {
         push @pt_agent_lines,
              $service->run_schedule
            . ($env_vars ? " $env_vars" : '')
            . " ${bin_dir}pt-agent --run-service "
            . $service->name;
      }
      if ( $service->spool_schedule ) {
         push @pt_agent_lines,
              $service->spool_schedule
            . ($env_vars ? " $env_vars" : '')
            . " ${bin_dir}pt-agent --send-data "
            . $service->name;
      }
   }
   PTDEBUG && _d('pt-agent crontab lines:', Dumper(\@pt_agent_lines));

   my $new_crontab = join("\n", @other_lines, @pt_agent_lines) . "\n";
   $logger->debug("New crontab: " . ($new_crontab || ''));

   return $new_crontab;
}

# Start real services, i.e. non-meta services.  A real service is like
# "query-history", which probably has meta-services like "start-query-history"
# and "stop-query-history".  We infer these start/stop meta-services
# from the real service's name.  A service doesn't require meta-services;
# there may be nothing to do to start it, in which case the real service
# starts running due to its run_schedule and schedule_services().
sub apply_services {
   my (%args) = @_;
   have_required_args(\%args, qw(
      action
      services
      lib_dir
   )) or die;
   my $action   = $args{action};
   my $services = $args{services};
   my $lib_dir  = $args{lib_dir};

   # Optional args
   my $bin_dir  = defined $args{bin_dir} ? "$args{bin_dir}"
                :                          "$FindBin::Bin/";
   my $exec_cmd = $args{exec_cmd} || sub { return system(@_) };

   $bin_dir .= '/' unless $bin_dir =~ m/\/$/;

   my $env_vars = env_vars();
   my $log      = "$lib_dir/logs/start-stop.log";
   my $cmd_fmt  = ($env_vars ? "$env_vars " : '')
                . $bin_dir . "pt-agent --run-service %s >> $log 2>&1";

   my @applied_ok;
   SERVICE:
   foreach my $service ( @$services ) {
      next if $service->meta;  # only real services

      my $name = $service->name;

      # To restart, one must first stop, then start afterwards.
      if ( $action eq 'stop' || $action eq 'restart' ) {
         if ( -f "$lib_dir/services/stop-$name" ) {
            if ( $action eq 'stop' ) {
               # If all we're doing is stopping services, then always
               # returned them as "applied OK" even if they fail to run
               # because the caller uses returns values to know to
               # update crontab.  So if stop-foo fails, at least we'll
               # still remove --run-service foo from crontab.
               push @applied_ok, $service;
            }
            my $cmd = sprintf $cmd_fmt, "stop-$name";
            $logger->info("Stopping $name: $cmd");
            my $cmd_exit_status = $exec_cmd->($cmd);
            if ( $cmd_exit_status != 0 ) {
               $logger->warning("Error stopping $name, check $log and "
                  . "$lib_dir/logs/$name.run");
               # This doesn't matter for stop, but for restart a failure
               # to first stop means we shouldn't continue and try to start
               # the service (since it hasn't been stopped yet).
               next SERVICE;
            }
         }
      }

      if ( $action eq 'start' || $action eq 'restart' ) {
         # Remove old meta files.  Meta files are generally temporary
         # in any case, persisting info from one interval to the next.
         # If the service has changed (e.g., report interval is longer),
         # there's no easy way to tranistion from old metadata to new,
         # so we just rm the old metadata and start anew.
         my $meta_files = "$lib_dir/meta/$name*";
         foreach my $meta_file ( glob $meta_files ) {
            if ( unlink $meta_file ) {
               $logger->info("Removed $meta_file");
            }
            else {
               $logger->warning("Cannot remove $meta_file: $OS_ERROR");
            }
         }

         # Start the service and wait for it to exit.  If it dies
         # really early (before it really begins), our log file will
         # have the error; else, the service should automatically
         # switch to its default log file ending in ".run".
         if ( -f "$lib_dir/services/start-$name" ) {
            my $cmd = sprintf $cmd_fmt, "start-$name";
            $logger->info("Starting $name: $cmd");
            my $cmd_exit_status = $exec_cmd->($cmd);
            if ( $cmd_exit_status != 0 ) {
               $logger->warning("Error starting $name, check $log and "
                   ."$lib_dir/logs/$name.run");
               next SERVICE;
            }
            push @applied_ok, $service;
            $logger->info("Started $name");
         }
      }
   }

   return \@applied_ok;
}

sub apply_services_once {
   my (%args) = @_;
   have_required_args(\%args, qw(
      services
      lib_dir
   )) or die;
   my $services = $args{services};
   my $lib_dir  = $args{lib_dir};

   # Optional args
   my $bin_dir  = defined $args{bin_dir} ? $args{bin_dir}
                :                          "$FindBin::Bin/";
   my $exec_cmd = $args{exec_cmd} || sub { return system(@_) };

   my $env_vars = env_vars();
   my $log      = "$lib_dir/logs/run-once.log";
   my $cmd_fmt  = ($env_vars ? "$env_vars " : '')
                . $bin_dir . "pt-agent --run-service %s >> $log 2>&1";

   my @ran_ok;
   SERVICE:
   foreach my $name ( sort keys %$services ) {
      my $service = $services->{$name};
      next unless $service->run_once;

      delete $services->{$name};

      my $cmd = sprintf $cmd_fmt, $name;
      $logger->info("Running $name: $cmd");
      my $cmd_exit_status = $exec_cmd->($cmd);
      if ( $cmd_exit_status != 0 ) {
         $logger->error("Error running $name, check $log and "
             ."$lib_dir/logs/$name.run");
         next SERVICE;
      }
      push @ran_ok, $service;
      $logger->info("Ran $name");
   }

   return \@ran_ok;
}

# ########################## #
# --run-service process subs #
# ########################## #

sub run_service {
   my (%args) = @_;

   have_required_args(\%args, qw(
      api_key
      service
      lib_dir
      spool_dir
      Cxn
   )) or die;
   my $api_key   = $args{api_key};
   my $service   = $args{service};
   my $lib_dir   = $args{lib_dir};
   my $spool_dir = $args{spool_dir};
   my $cxn       = $args{Cxn};

   # Optional args
   my $bin_dir     = defined $args{bin_dir} ? $args{bin_dir} : "$FindBin::Bin/";
   my $agent_api   = $args{agent_api};
   my $client      = $args{client};               # for testing
   my $agent       = $args{agent};                # for testing
   my $entry_links = $args{entry_links};          # for testing
   my $json        = $args{json};                 # for testing
   my $prefix      = $args{prefix} || int(time);  # for testing
   my $max_data    = $args{max_data} || MAX_DATA_FILE_SIZE;

   my $start_time = time;

   # Can't do anything with the lib dir.  Since we haven't started
   # logging yet, cron should capture this error and email the user.
   init_lib_dir(
      lib_dir => $lib_dir,
      verify  => 1,  # die unless ok, don't create
      quiet   => 1,
   );

   # Load the Service object from local service JSON file.
   # $service changes from a string scalar to a Service object.
   $service = load_service(
      service => $service,
      lib_dir => $lib_dir,
   );
   my $service_name = $service->name;

   my $daemon = Daemon->new(
      daemonize      => 0,  # no need: we're running from cron
      pid_file       => "$lib_dir/pids/$service_name.$PID",
      log_file       => "$lib_dir/logs/$service_name.run",
      force_log_file => 1,
   );
   $daemon->run();

   if ( $service->meta ) {
      $logger->service($service_name);
   }
   else {
      $logger->service("$service_name run");
   }
   $logger->info("Running $service_name");

   # Connect to Percona, get entry links.
   my $logger_client;
   if ( $agent_api && (!$client || !$entry_links) ) {
      ($client, $entry_links, $logger_client) = get_api_client(
         api_key  => $api_key,
         tries    => 2,
         interval => sub { return 2; },
      );
      if ( !$client || !$entry_links ) {
         # offline warning
         $logger->_log(0, 'WARNING', "Failed to connect to Percona Web API");
      }
   }

   # Load and update the local (i.e. existing) agent, or create a new one.
   if ( !$agent ) {
      # If this fails, there's no local agent, but that shouldn't happen
      # because a local agent originally scheduled this --send-data process.
      # Maybe that agent was deleted from the system but the crontab entry
      # was not and was left running.
      $agent = load_local_agent (
         lib_dir => $lib_dir,
      );
      if ( !$agent ) {
         $logger->fatal("No agent exists ($lib_dir/agent) and --agent-uuid was "
            . "not specified.  Check that the agent is properly installed.");
      }
   }

   # Start online logging, if possible.
   if ( $logger->online_logging && $agent_api && $client && $entry_links && $entry_links->{agents} ) {
      $agent = eval {
         $client->get(
            link => $entry_links->{agents} . '/' . $agent->uuid,
         );
      };
      if ( $EVAL_ERROR ) {
         $logger->info("Failed to get agent for online logging: $EVAL_ERROR");
      }
      else {
         my $log_link = $agent->links->{log};
         $logger->data_ts($prefix) unless $service->meta;
         $logger->start_online_logging(
            client   => $logger_client,
            log_link => $log_link,
         );
         $logger->debug("Log API enabled");
      }
   }
   else {
      $logger->_log(0, 'INFO', "File logging only");
   }

   # Check if any task spools data or uses MySQL.  Any task that spools
   # should also use metadata because all data samples have at least a
   # start_ts and end_ts as metadata.
   my $tasks     = $service->tasks;
   my $use_spool = 0;
   my $use_mysql = 0;
   foreach my $task ( @$tasks ) {
      $use_spool = 1 if ($task->output || '') eq 'spool';
      $use_mysql = 1 if $task->query;
   }

   # $data_dir will be undef if $use_spool is undef; that's ok because
   # only $tmp_dir is always needed.
   my ($data_dir, $tmp_dir) = init_spool_dir(
      spool_dir => $spool_dir,
      service   => $use_spool ? $service->name : undef,
   );

   # Connect to MySQL or quit.
   my $last_error;
   if ( $use_mysql ) {
      $logger->debug("Connecting to MySQL");
      TRY:
      for ( 1..2 ) {
         eval {
            $cxn->connect();
         };
         if ( my $e = $EVAL_ERROR ) {
            $logger->debug("Cannot connect to MySQL: $e");
            $last_error = $e;
            sleep(3);
            next TRY;
         }
         last TRY;
      }
      if ( !$cxn->dbh ) {
         $logger->error("Cannot run " . $service->name . " because it requires "
             . "MySQL but failed to connect to MySQL: " . ($last_error || '(no error)'));
         return;
      }
   }

   # Run the tasks, spool any data.
   my @output_files;
   my $recursive_service = '--run-service ' . $service->name;
   my $data_file         = $prefix . '.' . $service->name . '.data';
   my $tmp_data_file     = "$tmp_dir/$data_file";
   my $taskno            = 0;
   my $metadata          = { data_ts => $prefix };
   my $store             = {};
   my $env_vars          = env_vars();

   TASK:
   foreach my $task ( @$tasks ) {
      # Set up the output file, i.e. where this run puts its results.
      # Runs can access each other's output files.  E.g. run0 may
      # write to fileX, then subsequent tasks can access that file
      # with the special var __RUN_N_OUTPUT__ where N=0.  Output files
      # have this format: (prefix.)service.type(.n), where prefix is
      # an optional unique ID for this run (usually a Unix ts); service
      # is the service name; type is "data", "tmp", "meta", etc.; and
      # n is an optional ID or instance of the type.  The .data is the
      # only file required: it's the data sent by send_data().
      my $task_output_file = "$tmp_dir/$prefix."
                           . $service->name
                           . ".output.$taskno";
      my $append           = 0;
      my $output_file;
      my $join_char;
      my ($store_key, $store_key_value_tuple);

      my $output = $task->output || '';
      if ( $output eq 'spool' ) {
         $output_file = $tmp_data_file;
      }
      elsif ( $output =~ m/^stage:(\S+)/ ) {
         my $file_suffix = $1;
         $output_file = "$tmp_dir/$prefix." . $service->name . "$file_suffix";
      }
      elsif ( $output =~ m/^meta:(\S+)/ ) {
         my $attrib = $1;
         $output_file = "$lib_dir/meta/" . $service->name . ".meta.$attrib";
      }
      elsif ( $output =~ m/^join:(.)$/ ) {
         $join_char = $1;
         $output_file = $task_output_file;
      }
      elsif ( $output =~ m/store:key:([\w-]+)/ ) {
         $store_key = $1;
         $output_file = $task_output_file;
      }
      elsif ( $output eq 'store:output' ) {
         $store_key = $taskno;
         $output_file = $task_output_file;
      }
      elsif ( $output eq 'store:key_value_tuple' ) {
         $store_key_value_tuple = 1;
      }
      elsif ( $output eq 'store:output' ) {
         $store_key = $taskno;
         $output_file = $task_output_file;
      }
      elsif ( $output =~ m/append:(\S+)/ ) {
         $output_file = $1;
         $append = 1;
      }
      elsif ( $output eq 'tmp' ) {
         $output_file = $task_output_file;
      }

      if ( !$output_file ) {
         $output_file = '/dev/null';
         push @output_files, undef,
      }
      else {
         push @output_files, $output_file;
      }
      PTDEBUG && _d("Task $taskno output:", Dumper(\@output_files));

      if ( my $query = $task->query ) {
         $query = replace_special_vars(
            cmd          => $query,
            spool_dir    => $spool_dir,
            output_files => \@output_files,  # __RUN_n_OUTPUT__
            service      => $service->name,  # __SERVICE__
            lib_dir      => $lib_dir,        # __LIB__
            meta_dir     => "$lib_dir/meta", # __META__
            stage_dir    => $tmp_dir,        # __STAGE__
            store        => $store,          # __STORE_key__
            ts           => $prefix,         # __TS__
            bin_dir      => $bin_dir,        # __BIN_DIR__
            env          => $env_vars,       # __ENV__
         );
         $logger->info($task->name . ": $query");
         my $rows;
         my $t0 = time;
         eval {
            if ( $join_char || $store_key_value_tuple ) {
               $rows = $cxn->dbh->selectall_arrayref($query);
            }
            else {
               $cxn->dbh->do($query);
            }
         };
         if ( $EVAL_ERROR ) {
            $logger->error("Error executing $query: $EVAL_ERROR");
            last TASK;
         }

         if ( $rows ) {
            $logger->debug('Query returned ' . scalar @$rows . ' rows');
            if ( $join_char ) {
               my $fh;
               if ( !open($fh, '>', $output_file) ) {
                  $logger->error("Cannot open $output_file: $OS_ERROR");
                  last TASK;
               }
               foreach my $row ( @$rows ) {
                  print { $fh } join($join_char,
                        map { defined $_  ? $_ : 'NULL' } @$row), "\n"
                     or $logger->error("Cannot write to $output_file: $OS_ERROR");
               }
               close $fh
                  or $logger->warning("Cannot close $output_file: $OS_ERROR");
            }
            elsif ( $store_key_value_tuple ) {
               foreach my $row ( @$rows ) {
                  $store->{$row->[0]} = defined $row->[1] ? $row->[1] : 'NULL';
               }
            }
         }
         my $t1 = time;

         push @{$metadata->{tasks}}, {
            start_ts => ts($t0, 1),
            end_ts   => ts($t1, 1),
            run_time => sprintf('%.6f', $t1 - $t0),
         };
      }
      elsif ( my $program = $task->program ) {
         # Create the full command line to execute, replacing any
         # special vars like __RUN_N_OUTPUT__, __TMPDIR__, etc.
         my $cmd = join(' ',
            $task->program,
            ($append ? '>>' : '>'),
            $output_file,
         );
         $cmd = replace_special_vars(
            cmd          => $cmd,
            spool_dir    => $spool_dir,
            output_files => \@output_files,  # __RUN_n_OUTPUT__
            service      => $service->name,  # __SERVICE__
            lib_dir      => $lib_dir,        # __LIB__
            meta_dir     => "$lib_dir/meta", # __META__
            stage_dir    => $tmp_dir,        # __STAGE__
            store        => $store,          # __STORE_key__
            ts           => $prefix,         # __TS__
            bin_dir      => $bin_dir,        # __BIN_DIR__
            env          => $env_vars,       # __ENV__
         );
         $logger->debug("Task $taskno command: $cmd");

         if ( $cmd =~ m/$recursive_service/ ) {
            $logger->fatal("Recursive service detected: $cmd");
            die;  # fatal() should die, but just in case
         }

         # Execute this run.
         my $t0 = time;
         system($cmd);
         if ( $store_key ) {
            my $value = slurp($output_file);
            chomp($value) if $value;
            $store->{$store_key} = $value;
         }
         my $t1 = time;
         my $run_time = sprintf('%.6f', $t1 - $t0);
         my $cmd_exit_status = $CHILD_ERROR >> 8;
         $logger->info($task->name . ": $cmd (runtime: $run_time exit: $cmd_exit_status)");
         $exit_status |= $cmd_exit_status;

         push @{$metadata->{tasks}}, {
            start_ts    => ts($t0, 1),
            end_ts      => ts($t1, 1),
            run_time    => $run_time,
            exit_status => $cmd_exit_status,
         };

         if ( $cmd_exit_status == 200 && !$service->meta ) {
            # Equivalent to 0 for meta-services that need to stop early
            # but let the non-meta, i.e. real, parent service continue.
         }
         elsif ( $cmd_exit_status == 253 ) {
            $logger->error($task->name . ' exit status not zero, '
               . 'stopping ' . $service->name . ' service');
            stop_service(
               service => $service->name,
               lib_dir => $lib_dir,
            );
            last TASK;
         }
         elsif ( $cmd_exit_status == 254 ) {
            $logger->error($task->name . ' exit status not zero, '
               . 'stopping all services');
            stop_all_services(
               lib_dir => $lib_dir
            );
            last TASK;
         }
         elsif ( $cmd_exit_status != 0 ) {
            $logger->info($task->name . ' exit status not zero, '
               . 'stopping tasks');
            last TASK;
         }
      }
      else {
         $logger->error('Invalid Task resource:', Dumper($task));
         last TASK;
      }

      $taskno++;
   }

   # Move the spool file from --spool/.tmp/ to --spool/<service>/
   # if 1) the service spools data and 2) there is data.
   my $file_size = (-s $tmp_data_file) || 0;
   $logger->debug("$tmp_data_file size: " . ($file_size || 0) . " bytes");
   if ( $use_spool && $file_size ) {
      # Save metadata about this sample _first_, because --send-data looks
      # for the data file first, then for a corresponding .meta file.  If
      # we write the data file first, then we create a race condition: while
      # we're writing, --send-data could see the data file but not the
      # .meta file because we haven't written it yet.  So writing the .meta
      # file first guarantees that if --send-data sees a data file, the
      # .meta already exists.  (And there's no race condition on writing
      # the data file because we use a quasi-atomic system mv.)
      read_metadata(
         service   => $service->name,
         prefix    => $prefix,
         metadata  => $metadata,
         stage_dir => $tmp_dir,
      );
      $metadata->{run_time} = sprintf('%.6f', time - $start_time);
      (my $meta_file = $data_file) =~ s/\.data/\.meta/;
      my $json_metadata = as_json($metadata, json => $json);
      write_to_file(
         data => $json_metadata,
         file => "$data_dir/$meta_file",
      );

      if ( $file_size <= $max_data ) {
         # Use system mv instead of Perl File::Copy::move() because it's
         # unknown if the Perl version will do an optimized move, i.e.
         # simply move the inode, _not_ copy the file.  A system mv on
         # the same filesystem is pretty much guaranteed to do an optimized,
         # i.e. quasi-atomic, move.
         my $cmd = "mv $tmp_data_file $data_dir";
         $logger->debug($cmd);
         system($cmd);
         my $cmd_exit_status = $CHILD_ERROR >> 8;
         if ( $cmd_exit_status != 0 ) {
            $logger->error("Move failed: $cmd") 
         }
         $exit_status |= $cmd_exit_status;
      }
      else {
         $logger->error("Data file is larger than $max_data, skipping: "
            . ($json_metadata || ''));
      }
   }

   # Remove staged files.  Anything to save should have been moved
   # from staging by a task.
   foreach my $file ( glob "$tmp_dir/$prefix." . $service->name . ".*" ) {
      unlink $file
         or $logger->warning("Error removing $file: $OS_ERROR");
   }

   $logger->info("Exit: $exit_status");
   return $exit_status;  # returning global var for testing
}

sub load_service {
   my (%args) = @_;

   have_required_args(\%args, qw(
      service
      lib_dir
   )) or die;
   my $service   = $args{service};
   my $lib_dir   = $args{lib_dir};

   my $service_file = "$lib_dir/services/$service";
   if ( ! -f $service_file ) {
      $logger->fatal("Cannot load $service: $service_file does not exist.");
   }

   my $service_obj;
   eval {
      my $service_hash = JSON::decode_json(slurp($service_file));
      $service_obj  = Percona::WebAPI::Resource::Service->new(%$service_hash);
   };
   if ( $EVAL_ERROR ) {
      $logger->fatal("Cannot load $service: $EVAL_ERROR");
   }

   return $service_obj;
}

sub replace_special_vars {
   my (%args) = @_;

   have_required_args(\%args, qw(
      cmd
      spool_dir
      output_files
      service
      lib_dir
      meta_dir
      stage_dir
      store
      ts
      bin_dir
   )) or die;
   my $cmd          = $args{cmd};
   my $spool_dir    = $args{spool_dir};
   my $output_files = $args{output_files};
   my $service      = $args{service};
   my $lib_dir      = $args{lib_dir};
   my $meta_dir     = $args{meta_dir};
   my $stage_dir    = $args{stage_dir};
   my $store        = $args{store};
   my $ts           = $args{ts};
   my $bin_dir      = $args{bin_dir};

   # Optional args
   my $env = $args{env} || '';

   my $word;
   my $new_cmd;
   eval {
      $new_cmd = join(' ',
         map {
            $word = $_;
            $word =~ s/__RUN_(\d+)_OUTPUT__/$output_files->[$1]/g;
            $word =~ s/__STORE_([\w-]+)__/$store->{$1}/g;
            $word =~ s/__TS__/$ts/g;
            $word =~ s/__LIB__/$lib_dir/g;
            $word =~ s/__META__/$meta_dir/g;
            $word =~ s/__STAGE__/$stage_dir/g;
            $word =~ s/__SERVICE__/$service/g;
            $word =~ s/__STAGE_FILE__/$stage_dir\/$ts.$service/g;
            $word =~ s/__META_FILE__/$meta_dir\/$service.meta/g;
            $word =~ s/__BIN_DIR__/$bin_dir/g;
            $word =~ s/__TRASH__/$spool_dir\/.trash/g;
            $word =~ s/__ENV__/$env/g;
            $word;
         }
         split(/\s+/, $cmd)
      );
   };
   if ( $EVAL_ERROR ) {
      $logger->fatal("Error replacing " . ($word || '')
         . " in $cmd: $EVAL_ERROR");
   }

   return $new_cmd;
}

sub init_spool_dir {
   my (%args) = @_;

   have_required_args(\%args, qw(
      spool_dir
   )) or die;
   my $spool_dir = $args{spool_dir};

   # Optional args
   my $service = $args{service};
   my $quiet   = $args{quiet};

   if ( !-d $spool_dir ) {
      $logger->info("$spool_dir does not exist, creating")
         unless $quiet;
      _safe_mkdir($spool_dir);
   }
   elsif ( !-w $spool_dir ) {
      die "$spool_dir is not writeable\n";
   }

   foreach my $subdir ( $service, '.tmp', '.trash' ) {
      next unless $subdir;  # service may be undef
      my $dir = "$spool_dir/$subdir";
      if ( ! -d $dir ) {
         $logger->info("$dir does not exist, creating")
            unless $quiet;
         _safe_mkdir($dir);
      }
      elsif ( !-w $dir ) {
         die "$dir is not writeable\n";
      }
   }

   my $data_dir = $service ? "$spool_dir/$service" : undef;
   my $tmp_dir  = "$spool_dir/.tmp";

   return $data_dir, $tmp_dir;
}

sub read_metadata {
   my (%args) = @_;

   have_required_args(\%args, qw(
      service
      prefix
      metadata
      stage_dir
   )) or die;
   my $service   = $args{service};
   my $prefix    = $args{prefix};
   my $metadata  = $args{metadata};
   my $stage_dir = $args{stage_dir};

   # Example filename: 123456.query-history.meta.stop_offset
   foreach my $file ( glob "$stage_dir/$prefix.$service.meta.*" ) {
      PTDEBUG && _d('metadata file:', $file);
      my ($attrib) = $file =~ m/\.meta\.(\S+)$/;
      my $value    = slurp($file);
      chomp($value) if $value;
      PTDEBUG && _d('metadata', $attrib, '=', $value);
      $metadata->{$attrib} = $value;
      unlink $file
         or $logger->warning("Cannot rm $file: $OS_ERROR");
   }

   return;
}

# ######################## #
# --send-data process subs #
# ######################## #

# Send every file or directory in each service's directory in --spool/.
# E.g. --spool/query-monitor should contain files with pt-query-digest
# output.  The per-service dirs are created in run_service(). 
sub send_data {
   my (%args) = @_;

   have_required_args(\%args, qw(
      api_key
      service
      lib_dir
      spool_dir
   )) or die;
   my $api_key   = $args{api_key};
   my $service   = $args{service};
   my $lib_dir   = $args{lib_dir};
   my $spool_dir = $args{spool_dir};

   # Optional args
   my $interactive = $args{interactive};
   my $max_data    = $args{max_data} || MAX_DATA_FILE_SIZE;
   my $agent       = $args{agent};        # for testing
   my $client      = $args{client};       # for testing
   my $entry_links = $args{entry_links};  # for testing
   my $json        = $args{json};         # for testing
   my $delay       = defined $args{delay} ? $args{delay} : rand(30);  # for testing

   # Can't do anything with the lib dir.  Since we haven't started
   # logging yet, cron should capture this error and email the user.
   init_lib_dir(
      lib_dir => $lib_dir,
      verify  => 1,
      quiet   => 1,
   );

   # Load the Service object from local service JSON file.
   # $service changes from a string scalar to a Service object.
   $service = load_service(
      service => $service,
      lib_dir => $lib_dir,
   );
   my $service_name = $service->name;

   my ($service_dir) = init_spool_dir(
      spool_dir => $spool_dir,
      service   => $service->name,
   );

   my @data_files = glob "$service_dir/*.data";
   if ( scalar @data_files == 0 ) {
      $logger->debug("No $service_name data files to send");
      return;
   }

   # Log all output to a file.
   my $daemon = Daemon->new(
      daemonize      => 0,  # no need: we're running from cron
      pid_file       => "$lib_dir/pids/$service_name.send",
      log_file       => "$lib_dir/logs/$service_name.send",
      force_log_file => $interactive ? 0 : 1,
   );
   $daemon->run();

   # Spool time is +/- [0, 1] minute from API, we randomize that further
   # by a few seconds to prevent all agents from sending at exactly
   # 00:01:00, 00:00:00, or 00:02:00.
   sleep $delay if $delay;

   $logger->service("$service_name send");
   my $data_link = $service->links->{data};
   $logger->info("Sending " . scalar @data_files . " data files ($data_link)");

   # Connect to Percona, get entry links.
   my $logger_client;
   if ( !$client || !$entry_links ) {
      ($client, $entry_links, $logger_client) = get_api_client(
         api_key  => $api_key,
         tries    => 3,
         interval => sub { sleep 10 },
      );
      if ( !$client || !$entry_links ) {
         $logger->fatal("Failed to connect to Percona Web API")
      }
   }

   # Load and update the local (i.e. existing) agent, or create a new one.
   if ( !$agent ) {
      # If this fails, there's no local agent, but that shouldn't happen
      # because a local agent originally scheduled this --send-data process.
      # Maybe that agent was deleted from the system but the crontab entry
      # was not and was left running.
      $agent = load_local_agent (
         lib_dir => $lib_dir,
      );
      if ( !$agent ) {
         $logger->fatal("No agent exists ($lib_dir/agent) and --agent-uuid was "
            . "not specified.  Check that the agent is properly installed.");
      }
   }

   $agent = eval {
      $client->get(
         link => $entry_links->{agents} . '/' . $agent->uuid,
      );
   };
   if ( $EVAL_ERROR ) {
      $logger->fatal("Failed to get the agent: $EVAL_ERROR");
   }
   my $log_link = $agent->links->{log};
   if ( $log_link ) {
      $logger->start_online_logging(
         client   => $logger_client,
         log_link => $log_link,
      );
      $logger->debug("Log API enabled");
   }

   # Send data files in the service's spool dir.
   # Only iterator over data files because run_service() writes
   # them last to avoid a race condition with us.  See the code
   # comment about writing the .meta file first in run_service().
   DATA_FILE:
   foreach my $data_file ( @data_files ) {
      (my $meta_file = $data_file) =~ s/\.data/.meta/;

      if ( $interactive ) {
         my $key;
         PROMPT:
         while ( !$key ) {
            print "\n", `ls -l $data_file`;
            print "Send [Ynaq]: ";
            $key = <STDIN>;
            chomp($key);
            $key = lc($key);
            last DATA_FILE if $key eq 'q'; 
            next DATA_FILE if $key eq 'n';
            last PROMPT    if $key eq 'y';
            if ( $key eq 'a' ) {
               print "Sending all remaining files...\n";
               $interactive = 0;
               last PROMPT;
            }
            warn "Invalid response: $key\n";
         }
      }

      my $data_file_size = (-s $data_file) || 0;
      if ( $data_file_size > $max_data ) {
         $logger->error("Not sending $data_file because it is too large: "
            . "$data_file_size > $max_data.  This should not happen; "
            . "please contact Percona or file a bug, and verify that "
            . "all services are running properly.");
         next DATA_FILE;
      }

      eval {
         # Send the file as-is.  The --run-service process should
         # have written the data in a format that's ready to send.
         send_file(
            client    => $client,
            agent     => $agent,
            meta_file => $meta_file,
            data_file => $data_file,
            link      => $data_link,
            json      => $json,
         );
      };
      if ( my $e = $EVAL_ERROR ) {
         if (blessed($e) && $e->isa('Percona::WebAPI::Exception::Request')) {
            my $error_msg = $client->response->content;
            $logger->warning('Error ' . $e->status . " sending $data_file ($data_file_size bytes): "
               . ($error_msg || '(No error message from server)'));
         }
         else {
            chomp $e;
            $logger->warning("Error sending $data_file ($data_file_size bytes): $e");
         } 
         next DATA_FILE;
      }

      # Data file sent successfully; now remove it.  Failure here
      # is an error, not a warning, because if we can't remove the
      # file then we risk re-sending it, and currently we have no
      # way to determine if a file has been sent or not other than
      # whether it exists or not.
      eval {
         unlink $data_file or die $OS_ERROR;
      };
      if ( $EVAL_ERROR ) {
         chomp $EVAL_ERROR;
         $logger->warning("Sent $data_file but failed to remove it: "
            . $EVAL_ERROR);
         last DATA_FILE;
      }

      if ( -f $meta_file ) {
         unlink $meta_file or $logger->warning($OS_ERROR);
      }

      $logger->info("Sent: $data_file ($data_file_size bytes)");
   }

   $logger->info("Exit: $exit_status");
   return $exit_status;  # returning global var for testing
}

# Send the Agent and file's contents as-is as a multi-part POST.
sub send_file {
   my (%args) = @_;

   have_required_args(\%args, qw(
      client
      agent
      data_file
      link
   )) or die;
   my $client    = $args{client};
   my $agent     = $args{agent};
   my $data_file = $args{data_file};
   my $link      = $args{link};

   # Optional args
   my $meta_file = $args{meta_file};
   my $json      = $args{json};  # for testing

   # Create a multi-part resource: first the Agent, so Percona knows
   # from whom the sample data is coming, then metadata about the sample,
   # then the actual sample data.  Each part is separated by a special
   # boundary value.  The contents of the data file are sent as-is
   # because here we don't know or care about the data; that's a job
   # for the PWS server.
   my $boundary = 'Ym91bmRhcnk'; # "boundary" in base64, without a trailing =

   my $agent_json = as_json($agent, json => $json);
   chomp($agent_json);

   my $meta = -f $meta_file && -s $meta_file ? slurp($meta_file) : '';
   $meta =~ s/^\s+//;
   $meta =~ s/\s+$//;

   my $data = -s $data_file ? slurp($data_file) : '';
   $data =~ s/^\s+//;
   $data =~ s/\s+$//;

   # Put it all together:
   my $resource   = <<CONTENT;
--$boundary
Content-Disposition: form-data; name="agent"

$agent_json
--$boundary
Content-Disposition: form-data; name="meta"

$meta
--$boundary
Content-Disposition: form-data; name="data"

$data
--$boundary
CONTENT

   # This will die if the server response isn't 2xx or 3xx.  The caller,
   # send_data(), should catch this.
   $client->post(
      link      => $link,
      resources => $resource,
      headers   => {
         'Content-Type' => "multipart/form-data; boundary=$boundary",
      }
   );

   return;
}

# ############################################ #
# --status, --stop, --reload, and --reset subs #
# ############################################ #

sub agent_status {
   my (%args) = @_;

   have_required_args(\%args, qw(
      pid_file
      lib_dir
   )) or die;
   my $pid_file = $args{pid_file};
   my $lib_dir  = $args{lib_dir};

   # Optional args
   my $api_key      = $args{api_key};
   my $crontab_list = defined $args{crontab_list} ? $args{crontab_list}
                    :                               `crontab -l 2>/dev/null`;
   my $bin_dir = defined $args{bin_dir} ? $args{bin_dir}
               :                          "$FindBin::Bin/";

   # Check if pt-agent is running.
   my $pid = eval {
      get_agent_pid(
         pid_file => $pid_file,
      );
   };
   if ( my $e = $EVAL_ERROR ) {
      if ( !blessed($e) ) {
         $logger->warning("Sorry, an error occured while getting the pt-agent PID: $e");
      }
      elsif ( $e->isa('Percona::Agent::Exception::PIDNotFound') ) {
         $logger->info("pt-agent is not running");
      }
      elsif ( $e->isa('Percona::Agent::Exception::PIDNotRunning') ) {
         $logger->warning("$e.  pt-agent may have stopped unexpectedly or crashed.");
      }
      else {  # unhandled exception
         $logger->warning("Sorry, an unknown exception occured while getting "
            . "the pt-agent PID: $e");
      }
   }
   else {
      $logger->info("pt-agent is running as PID $pid")
   }

   if ( $api_key ) {
      $logger->info("API key: " . ($api_key || ''));
   }
   else {
      $logger->warning("No API key is set");
   }

   # Get the agent's info.
   if ( -f "$lib_dir/agent" ) {
      my $agent = JSON::decode_json(slurp("$lib_dir/agent"));
      foreach my $attrib ( qw(uuid hostname username) ) {
         $logger->info("Agent $attrib: " . ($agent->{$attrib} || ''));
      }
   }
   else {
      $logger->warning("$lib_dir/agent does not exist");
   }

   # Parse pt-agent lines from crontab to see what's scheduled/running.
   my %scheduled = map {
      my $line = $_;
      my ($service) = $line =~ m/pt-agent (?:--run-service|--send-data) (\S+)/;
      $service => 1;
   }
   grep { $_ =~ m/pt-agent (?:--run-service|--send-data)/ }
   split("\n", $crontab_list);

   my %have_service;
   if ( -d "$lib_dir/services" ) {
      SERVICE:
      foreach my $service_file ( glob "$lib_dir/services/*" ) {
         my $service = eval {
            JSON::decode_json(slurp($service_file));
         };
         if ( $EVAL_ERROR ) {
            $logger->warning("$service_file is corrupt");
            next SERVICE;
         }
         $service = Percona::WebAPI::Resource::Service->new(%$service);
         next if $service->meta;  # only real services
         $have_service{$service->name} = 1;
         if ( $scheduled{$service->name} ) {
            if ( $pid ) {
               $logger->info($service->name . " is running");
            }
            else {
               $logger->warning($service->name . " is running but pt-agent is not");
            }
         }
         else {
            if ( $pid ) {
               $logger->warning($service->name . " is not running");
            }
            else {
               $logger->info($service->name . " has stopped");
            }
         }
      }
   }
   else {
      $logger->warning("$lib_dir/services does not exist");
   }

   # Look for services that are still scheduled/running but that we'll
   # don't/shouldn't have.  This can happen if the crontab gets messed
   # up, --stop fails, etc.
   foreach my $scheduled_service ( sort keys %scheduled ) {
      if ( !$have_service{$scheduled_service} ) {
         $logger->warning("$scheduled_service is running but "
            . "$lib_dir/services/$scheduled_service does not exist");
      }
   }

   return;
}

sub stop_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      pid_file
      lib_dir
   )) or die;
   my $pid_file = $args{pid_file};
   my $lib_dir  = $args{lib_dir};

   my $stopped = 0;

   # Get the agent's PID and kill it.  If the PID file doesn't
   # exist for some reason, get_agent_pid() will attempt to find
   # pt-agent --daemonize in ps.  And if pt-agent doesn't respond
   # to the TERM signal after a short while, we kill it with
   # the KILL signal.
   my $pid = eval {
      get_agent_pid(
         pid_file => $pid_file,
      );
   };
   if ( my $e = $EVAL_ERROR ) {
      if ( !blessed($e) ) {
         $logger->warning("Sorry, an error occured while getting the pt-agent PID: $e");
      }
      elsif ( $e->isa('Percona::Agent::Exception::PIDNotFound') ) {
         $logger->info("pt-agent is not running");
         $stopped = 1;
      }
      elsif ( $e->isa('Percona::Agent::Exception::PIDNotRunning') ) {
         $logger->warning("$e.  pt-agent may have stopped unexpectedly or crashed.");
         $stopped = 1;
      }
      else {  # unhandled exception
         $logger->warning("Sorry, an unknown exception occured while getting "
            . "the pt-agent PID: $e");
      }
   }
   else {
      $logger->info("Stopping pt-agent...");
      kill 15, $pid;
      my $running;
      for (1..5) {
         $running = kill 0, $pid;
         last if !$running;
         sleep 0.5;
      }
      $running = kill 0, $pid;
      if ( $running ) {
         $logger->warning("pt-agent did not respond to the TERM signal, using "
            . "the KILL signal...");
         kill 9, $pid;
         for (1..2) {
            $running = kill 0, $pid;
            last if !$running;
            sleep 0.5;
         }
         $running = kill 0, $pid;
         if ( $running ) {
            # Shouldn't happen:
            $logger->warning("pt-agent did not response to the KILL signal");
         }
         else {
            $logger->info("Killed pt-agent");
            $stopped = 1;
         }
      }
      else {
         $logger->info("pt-agent has stopped");
         $stopped = 1;
      }

      # pt-agent should remove its own PID file, but in case it didn't,
      # (e.g we had to kill -9 it), we remove the PID file manually.
      if ( -f $pid_file ) {
         unlink $pid_file
            or $logger->warning("Cannot remove $pid_file: $OS_ERROR.  Remove "
               . "this file manually.");
      }
   }

   stop_all_services(
      lib_dir => $lib_dir,
   );

   # TODO: kill --lib/pids/*

   return $stopped;
}

sub stop_all_services {
   my (%args) = @_;

   have_required_args(\%args, qw(
      lib_dir
   )) or die;
   my $lib_dir  = $args{lib_dir};
   
   # Optional args
   my $bin_dir = defined $args{bin_dir} ? $args{bin_dir}
               :                          "$FindBin::Bin/";

   # Un-schedule all services, i.e. remove them from the user's crontab,
   # leaving the user's other tasks untouched.
   $logger->info("Removing all services from crontab...");
   eval {
      schedule_services(
         services => [],
         lib_dir  => $lib_dir,
         quiet    => 1,
      );
   };
   if ( $EVAL_ERROR ) {
      $logger->error("Error removing services from crontab: $EVAL_ERROR");
   }

   # Stop all real services by running their stop-<service> meta-service.
   # If a real service doesn't have a stop-<service> meta-service, then
   # presumably nothing needs to be done to stop it other than un-scheduling
   # it, which we've already done.
   if ( -d "$lib_dir/services" ) {
      my $env_vars = env_vars();

      SERVICE:
      foreach my $file ( glob "$lib_dir/services/stop-*" ) {
         my $service  = basename($file);
         my $stop_log = "$lib_dir/logs/$service.stop";
         my $run_log  = "$lib_dir/logs/$service.run";
         my $cmd      = ($env_vars ? "$env_vars " : '')
                      . "${bin_dir}pt-agent --run-service $service --no-agent-api"
                      . " </dev/null"
                      . " >$stop_log 2>&1";
         $logger->info("Stopping $service...");
         PTDEBUG && _d($cmd);
         system($cmd);
         my $cmd_exit_status = $CHILD_ERROR >> 8;
         if ( $cmd_exit_status != 0 ) {
            my $err = -f $run_log ? slurp($run_log) : '';
            $logger->error("Error stopping $service.  Check $stop_log and the "
               . "online logs for details.  The service may still be running.");
            next SERVICE;
         }
         unlink $stop_log
            or $logger->warning("Cannot remove $stop_log: $OS_ERROR");
      }
   }
   else {
      $logger->info("$lib_dir/services does not exist, no services to stop")
   }

   $state->{all_services_are_stopped} = 1;

   return;
}

sub stop_service {
   my (%args) = @_;

   have_required_args(\%args, qw(
      service
      lib_dir
   )) or die;
   my $service = $args{service};
   my $lib_dir = $args{lib_dir};
   
   # Optional args
   my $bin_dir = defined $args{bin_dir} ? $args{bin_dir}
               :                          "$FindBin::Bin/";

   if ( -d "$lib_dir/services" ) {
      my $stop_service_file = "$lib_dir/services/stop-$service";
      if ( -f $stop_service_file ) {
         my $stop_service = basename($stop_service_file);
         my $env_vars     = env_vars();
         my $stop_log     = "$lib_dir/logs/$service.stop";
         my $run_log      = "$lib_dir/logs/$service.run";
         my $cmd          = ($env_vars ? "$env_vars " : '')
                          . "${bin_dir}pt-agent --run-service $stop_service"
                          . " </dev/null"
                          . " >$stop_log 2>&1";
         $logger->info("Stopping $service...");
         PTDEBUG && _d($cmd);
         system($cmd);
         my $cmd_exit_status = $CHILD_ERROR >> 8;
         if ( $cmd_exit_status != 0 ) {
            my $err = -f $run_log ? slurp($run_log) : '';
            $logger->error("Error stopping $service.  Check $stop_log, "
               . "$run_log, and the online online logs for details.  "
               . "$service may still be running.");
         }
         else {
            unlink $stop_log
               or $logger->warning("Cannot remove $stop_log: $OS_ERROR");
         }
      }
      else {
         $logger->warning("$stop_service_file does not exist, cannot stop $service");
      }
   }
   else {
      $logger->warning("$lib_dir/services does not exist, cannot stop $service");
   }

   return;
}

sub reset_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      pid_file
      lib_dir
      spool_dir
      log_file
   )) or die;
   my $pid_file  = $args{pid_file};  # for stop_agent()
   my $lib_dir   = $args{lib_dir};
   my $spool_dir = $args{spool_dir};
   my $log_file  = $args{log_file};

   # Optional args
   my $api_key = $args{api_key};

   if ( -t STDIN ) {
      print "\nWARNING: All services will stop and all data in $spool_dir/ "
         ."will be deleted.  Are you sure you want to reset pt-agent?\n\n"
         . "Press any key to continue, else Ctrl-C to abort.\n";
      my $confirmation = <STDIN>;
   }

   $logger->info('Stopping pt-agent...');
   my $stopped = stop_agent(
      pid_file => $pid_file,
      lib_dir  => $lib_dir,
   );
   if ( !$stopped ) {
      $logger->warning('Failed to stop pt-agent.  Stop the agent, or verify that '
         . 'it is no longer running, and try again.');
      return;
   }

   my $agent = load_local_agent(
      lib_dir => $lib_dir,
      quiet   => 1,
   );
   if ( !$agent ) {
      $logger->warning("$lib_dir/agent does not exist. You will need to re-install "
         . "pt-agent after the reset.");
   }

   $logger->info("Removing $lib_dir/...");
   rmtree($lib_dir)
      or $logger->warning("Cannot remove $lib_dir/: $OS_ERROR");
   init_lib_dir(
      lib_dir => $lib_dir,
   );

   if ( $agent ) {
      my $new_agent = Percona::WebAPI::Resource::Agent->new(
         uuid => $agent->uuid,
      );
      save_agent(
         lib_dir => $lib_dir,
         agent   => $new_agent,
      );
   }

   $logger->info("Removing $spool_dir/...");
   rmtree($spool_dir)
      or $logger->warning("Cannot remove $spool_dir/: $OS_ERROR");
   init_spool_dir(
      spool_dir => $spool_dir,
   );

   my $config_file = get_config_file();
   my $config      = -f $config_file ? slurp($config_file) : '';
   $logger->info("Resetting $config_file...");
   open my $fh, '>', $config_file
      or $logger->error("Cannot write to $config_file: $OS_ERROR");
   if ( $api_key ) {
      print { $fh } "api-key=$api_key\n";
   }
   foreach my $line ( split("\n", $config) ) {
      next unless $line =~ m/^\s*(?:user|host|password|socket|defaults-file|port)/;
      print { $fh } $line, "\n";
   }
   close $fh
      or $logger->warning("Cannot close $config_file: $OS_ERROR");

   if ( -f $log_file ) {
      $logger->info("Removing $log_file...");
      unlink $log_file
         or $logger->warning("Cannot remove $log_file: $OS_ERROR");
   }

   return;
}

sub get_agent_pid {
   my (%args) = @_;
   my $pid_file = $args{pid_file};

   my $pid;
   if ( -f $pid_file ) {
      PTDEBUG && _d('Reading PID from', $pid_file);
      chop($pid = slurp($pid_file));
   }
   else {
      my $ps_output = `ps ax | grep 'pt-agent --daemonize' | grep -v grep`;
      PTDEBUG && _d('Reading PID from ps', $ps_output);
      if ( !$ps_output ) {
         die Percona::Agent::Exception::PIDNotFound->new(
            pid_file  => $pid_file,
         );
      }
      # Match the first digits, which should be the PID.
      ($pid) = $ps_output =~ m/(\d+)/;
   }

   if ( !$pid ) {
      die Percona::Agent::Exception::NoPID->new(
         pid_file          => $pid_file,
         pid_file_is_empty => -f $pid_file,
      );
   }

   my $running = kill 0, $pid;
   if ( !$running ) {
      die Percona::Agent::Exception::PIDNotRunning->new(
         pid => $pid,
      );
   }

   return $pid;
}

sub reload_signal {
   my ( $signal ) = @_;
   print STDERR "\n# Caught SIG$signal, reloading configuration.\n";
   $state->{reload} = 1;
   return;
}

sub reload_agent {
   my (%args) = @_;

   have_required_args(\%args, qw(
      pid_file
   )) or die;
   my $pid_file = $args{pid_file};
   my $lib_dir  = $args{lib_dir};

   my $pid = eval {
      get_agent_pid(
         pid_file => $pid_file,
      );
   };
   if ( my $e = $EVAL_ERROR ) {
      if ( !blessed($e) ) {
         $logger->warning("Sorry, an error occured while getting the pt-agent PID: $e");
      }
      elsif ( $e->isa('Percona::Agent::Exception::PIDNotFound') ) {
         $logger->warning("pt-agent is not running");
      }
      elsif ( $e->isa('Percona::Agent::Exception::PIDNotRunning') ) {
         $logger->warning("$e.  pt-agent may have stopped unexpectedly or crashed.");
      }
      else {  # unhandled exception
         $logger->warning("Sorry, an unknown exception occured while getting "
            . "the pt-agent PID: $e");
      }
   }
   else {
      kill 10, $pid;  # SIGUSR1, caught in reload_signal()
      $logger->info("Sent reload signal (SIGUSR1) to pt-agent PID $pid");
   }

   return;
}

# ############## #
# --install subs #
# ############## #

sub install {
   my (%args) = @_;
   have_required_args(\%args, qw(
      OptionParser
      Cxn
      flags
   )) or die;
   my $o     = $args{OptionParser};
   my $cxn   = $args{Cxn};
   my $flags = $args{flags};

   # Optional args
   my $interactive = $args{interactive};

   $logger->quiet(Percona::Agent::Logger::level_number('ERROR'));

   my $agent_my_cnf = '/etc/percona/agent/my.cnf';
   my $config_file  = get_config_file();
   my $lib_dir      = $o->get('lib');

   my $step_result;
   my $stepno   = 0;
   my $skip     = 0;
   my $step_fmt = "Step %d of %d: %s: ";
   my @steps  = (
      "Verify the user is root",
      "Check Perl module dependencies",
      "Check for crontab",
      "Verify pt-agent is not installed",
      "Verify the API key",
      "Connect to MySQL",
      "Check if MySQL is a slave",
      "Create a MySQL user for the agent",
      "Initialize $agent_my_cnf",
      "Initialize $config_file",
      "Create the agent",
      "Run the agent",
   );
   my $n_steps = scalar @steps;
   my $next_step = sub {
      my (%args) = @_;
      my $repeat = $args{repeat};
      my $done   = $args{done};
      # Result of the previous step
      my $result = 'OK';
      if ( $step_result ) {
         $result = $step_result;
         $step_result = undef;
      }
      print "$result\n" if $stepno && !$repeat;
      while ( $skip ) {
         printf $step_fmt,
            $stepno + ($repeat ? 0 : 1),
            $n_steps,
            $steps[$repeat ? $stepno - 1 : $stepno];
         $stepno++;
         print "SKIP\n";
         $skip--;
      }
      if ( $done ) {
         print "INSTALLATION COMPLETE\n";
         return;
      }
      # This step
      printf $step_fmt,
         $stepno + ($repeat ? 0 : 1),
         $n_steps,
         $steps[$repeat ? $stepno - 1 : $stepno];
      $stepno++ unless $repeat;
   };

   # ########################################################################
   # Pre-install checklist
   # ######################################################################## 
   
   # Must be root for --install.
   $next_step->();
   if ( $EUID != 0 ) {
      die "You must run pt-agent --install as root.\n";
   }

   # Check Perl module dependencies
   $next_step->();
   exit 1 if missing_perl_module_deps();

   # Check that LWP is new enough
   # https://bugs.launchpad.net/percona-toolkit/+bug/1226721
   if ( $LWP::VERSION < '5.813' ) {
      die "LWP v5.813 or newer is required; v$LWP::VERSION is installed.  Please upgrade LWP on this server and try again.\n"
   }

   # Check for crontab
   $next_step->();
   my $crontab = `which crontab 2>/dev/null`;
   if ( !$crontab ) {
      die "cron is not installed, or crontab is not in your PATH.\n";
   }

   # Verify pt-agent is not installed
   $next_step->();
   my @install_files = ($config_file, "$lib_dir/agent");
   my @have_files;
   foreach my $file (@install_files) {
      push @have_files, $file if -f $file;
   }
   if ( scalar @have_files ) {
      print "FAIL\n";
      die "It looks like pt-agent is already installed because these files exist:\n"
         . join("\n", map { "  $_" } @have_files)
         . "\nRun pt-agent --uninstall to remove these files.  To upgrade pt-agent, "
         . "install the new version, run pt-agent --stop, then pt-agent --daemonize "
         . "to restart pt-agent with the new version.\n";
   }

   # Must have a valid API key.
   $next_step->();
   my $got_api_key = 0;
   my $api_key = $o->get('api-key');
   if ( !$api_key ) {
      print "\n";
      if ( $interactive || -t STDIN ) {
         while ( !$api_key ) {
            print "Enter your API key: ";
            $api_key = <STDIN>;
            chomp($api_key) if $api_key;
            if ( !$api_key || length($api_key) < 32 ) {
               warn "Invalid API key; it should be at least 32 characters long.  Please try again.\n";
               $api_key = '';
            }
         }
      }
      else {
         die "Please specify your --api-key.\n";
      }
      $got_api_key = 1;
   }

   my $client;
   my $entry_links;
   if ( $flags->{offline} ) {
      $skip++;
   }
   else {
      if ($got_api_key) {
         $next_step->(repeat => 1);
      }
      eval {
         ($client, $entry_links) = get_api_client(
            api_key  => $api_key,
            interval => sub { return; },
            tries    => 1,
         );
      };
      if ( my $e = $EVAL_ERROR ) {
         die "Sorry, an error occurred while verifying the API key: $e";
      }
      elsif ( !$entry_links ) {
         if ( $client->response->code && $client->response->code == 401 ) {
            die "Sorry, the API key $api_key is not valid.  Please check the API key and try again.\n";
         }
         else {
            my $err = $client->response->message || 'Unknown error';
            die "Sorry, an error occured while verifying the API key: $err\n";
         }
      }
   }

   #  Must be able to connect to MySQL to create pt_agent user.
   $next_step->();
   eval {
      $cxn->connect();
   };
   if ( $EVAL_ERROR ) {
      chomp $EVAL_ERROR;
      die "Cannot connect to MySQL: $EVAL_ERROR\n"
         . "Please re-run pt-agent --install and specify MySQL connection "
         . "options like --user and --host to connect to MySQL as a user "
         . "with sufficient privileges to create MySQL users.\n";
   }

   # Check if MySQL is a slave
   $next_step->();
   my $slave = $cxn->dbh->selectrow_hashref("SHOW SLAVE STATUS");
   if ( $slave ) {
      $step_result = 'YES, TO MASTER ' . $slave->{master_host} || '?';
   }
   else {
      $step_result = 'NO';
   }

   # ########################################################################
   # Do the install
   # ########################################################################

   # Create a MySQL user for the agent
   $next_step->();
   if ( -f $agent_my_cnf ) {
      $step_result = "NO, USE EXISTING $agent_my_cnf";
   }
   else {
      if ( !$slave ) {  # master
         create_mysql_user($cxn, $agent_my_cnf);
      }
      else { # slave
         if ( $flags->{force_dangerous_slave_install} ) {
            create_mysql_user($cxn, $agent_my_cnf);
         }
         else {
            die "Sorry, cannot install the agent because MySQL is a slave "
               . "and $agent_my_cnf does not exist.  It is not safe to "
               . "write to a slave, so a MySQL user for the agent cannot "
               . "be created.  First install the agent on the master, then "
               . "copy $agent_my_cnf from the master to this server.  "
               . "See SLAVE INSTALL in the docs for more information.\n";
         }
      }
   }

   # Save the API key and defaults file in ~/.pt-agent.conf.
   $next_step->();
   eval {
      write_to_file(
         data => "api-key=$api_key\ndefaults-file=$agent_my_cnf\n",
         file => $config_file,
      );
   };
   if ( $EVAL_ERROR ) {
      die "Sorry, an error occured while initializing $config_file: "
         . $EVAL_ERROR;
   }

   # Init --lib and --spool.  pt-agent would do this itself, but we'll
   # do it now in case there are problems. 
   $next_step->();
   init_lib_dir(
      lib_dir => $lib_dir,
   );
   init_spool_dir(
      spool_dir => $o->get('spool'),
   );

   # 8. Start the agent, don't run it yet.  Normally this forks in
   # anticipation of run_agent() being called next, but we don't do
   # this during install; we run the agent manually later.
   if ( $flags->{offline} ) {
      $skip++;  # Init agent
      $skip++;  # Run agent
   }
   else {
      $next_step->();
      my $running = eval {
         start_agent(
            api_key     => $api_key,
            lib_dir     => $o->get('lib'),
            Cxn         => $cxn,
            client      => $client,
            entry_links => $entry_links,
            agent_uuid  => $o->get('agent-uuid'),
            daemonize   => 0,
            pid_file    => undef,
            log_file    => undef,
            tries       => 2,
            interval    => sub { sleep 2; },
         );
      };
      if ( $EVAL_ERROR ) {
         if ( $client->response->code && $client->response->code == 403 ) {
            die "The maximum number of agents for this organization has been reached; "
               . "no more agents can be created.  Delete unused agents from "
               . "https://cloud.percona.com/agents and try again.\n";
         }
         else {
            die "Sorry, an error occurred while starting the agent: $EVAL_ERROR";
         }
      }

      # 9. Run the agent daemon.  If all the previous worked, the agent
      # should be able to start without problems.  It will get and apply
      # the default config, then get and apply any services (probably won't
      # have any yet).
      $next_step->();
      my $env = env_vars();
      my $cmd = "$env $FindBin::Bin/pt-agent --daemonize";
      my $ret = system($cmd);
      if ( $ret >> 8 ) {
         die "Sorry, an error occured while starting pt-agent.\n";
      }
   }

   # ########################################################################
   # Done installing
   # ########################################################################
   $next_step->(done => 1);

   my $hostname = `hostname`;
   chomp($hostname);

   if ( $flags->{offline} ) {
      print "The agent has been installed, but it was not started.  "
         . "Run pt-agent --daemonize to start the agent, then go to "
         . "https://cloud.percona.com/agents#$hostname to enable services "
         . "for the agent.\n";
   }
   else {
      print "The agent has been installed and started, but it is not "
         . "running any services yet.  Go to "
         . "https://cloud.percona.com/agents#$hostname to enable services "
         . "for the agent.\n";
   }

   return;
}

sub create_mysql_user {
   my ($cxn, $agent_my_cnf, $user, $pass) = @_;

   if ( !$user || !$pass ) {
      $user = 'pt_agent';
      $pass = pseudo_random_password();

      my $sql = "GRANT SUPER,USAGE ON *.* TO 'pt_agent'\@'localhost' "
              . "IDENTIFIED BY '$pass'";
      eval {
         $cxn->dbh->do($sql);
      };
      if ( $EVAL_ERROR ) {
         die "Sorry, an error occurred while creating a MySQL user for the agent: "
            . $EVAL_ERROR;
      }
      $cxn->dbh->disconnect();
   }

   # Init $agent_my_cnf
   # We could set user= and pass= in ~/.pt-agent.conf, but each new agent
   # has a different MySQL password but shares the same default agent
   # config, so if we set pass=foo, the next agent would set it to
   # pass=bar, etc.  Instead, every agent sets/uses
   # defaults-file=/etc/percona/agent/my.cnf in the default config, but
   # the contents of that file is different for each agent.

   if ( !-d '/etc/percona' ) {
      _safe_mkdir('/etc/percona');
   }
   if ( !-d '/etc/percona/agent' ) {
      _safe_mkdir('/etc/percona/agent');
   }
   my $my_cnf = "[client]\nuser=$user\npass=$pass\n";
   my $dsn = $cxn->dsn;
   if ( $dsn->{h} ) {
      $my_cnf .= "host=$dsn->{h}\n";
   }
   if ( $dsn->{P} ) {
      $my_cnf .= "port=$dsn->{P}\n";
   }
   if ( $dsn->{S} ) {
      $my_cnf .= "socket=$dsn->{S}\n";
   }
   eval {
      write_to_file(
         data => $my_cnf,
         file => $agent_my_cnf,
      );
   };
   if ( $EVAL_ERROR ) {
      die "Sorry, an error occured while initializing $agent_my_cnf: "
         . $EVAL_ERROR;
   } 

   return;
}

sub pseudo_random_password {
   my @chars = ("A".."Z", "a".."z", "0".."9");
   my $string;
   $string .= $chars[rand @chars] for 1..9;
   return $string;
}

sub missing_perl_module_deps {
   my @missing_deps;
   foreach my $pm ( sort keys %deps ) {
      my $dep = $deps{$pm};
      eval "require $dep->[0]";
      if ( $EVAL_ERROR ) {
         push @missing_deps, $dep;
      }
   }
   if ( @missing_deps ) {
      warn "These Perl modules need to be installed:\n\n";
      foreach my $dep ( @missing_deps ) {
         warn "$dep->[0]\n  apt-get install $dep->[1]\n  yum install $dep->[2]\n\n";
      }
   }
   return scalar @missing_deps;
}

# ################ #
# --uninstall subs #
# ################ #

sub uninstall {
   my (%args) = @_;
   have_required_args(\%args, qw(
      OptionParser
      Cxn
   )) or die;
   my $o     = $args{OptionParser};
   my $cxn   = $args{Cxn};
   my $flags = $args{flags};
   
   if ( $EUID != 0 ) {
      die "You must run pt-agent --uninstall as root.\n";
   }

   my $config_file = get_config_file();
   my $lib_dir     = $o->get('lib');
   my $spool_dir   = $o->get('spool');
   
   print "Uninstalling pt-agent...\n";

   # Stop the agent.  This must succeed else it's not safe to remove its
   # files and dirs while it's running.
   my $stopped = stop_agent(
      pid_file => $o->get('pid'),
      lib_dir  => $o->get('lib'),
   );
   if ( !$stopped ) {
      $logger->fatal("Failed to stop pt-agent.");
   }

   # Agent is stopped so now it's safe to remove all our files and dirs.
   my @shell_cmds;
   if ( -d $lib_dir ) {
      push @shell_cmds, "rm -rf $lib_dir";
   }
   if ( -d $spool_dir ) {
      push @shell_cmds, "rm -rf $spool_dir"
   }
   if ( -d "/etc/percona/agent" ) {
      push @shell_cmds, "rm -rf /etc/percona/agent/";
   }
   if ( -f $config_file ) {
      push @shell_cmds, "rm -f $config_file"
   }

   my $rm_files_ok;
   if ( scalar @shell_cmds ) {
      print "Are you sure you want to run these command " 
         . "to uninstall pt-agent?\n"
         . join("\n", map { "  $_" } @shell_cmds) . "\n";
      while ( !$rm_files_ok ) {
         print "Enter 'yes' to run these commands, or CTRL-C to abort: ";
         $rm_files_ok = <STDIN>;
         chomp($rm_files_ok) if $rm_files_ok;
         if ( $rm_files_ok && $rm_files_ok eq 'yes' ) {
            last;
         }
         else {
            $rm_files_ok = 0;
         }
      }
      # CTRL-C should prevent us from getting here, but just in case:
      return if @shell_cmds && !$rm_files_ok;
      foreach my $cmd ( @shell_cmds ) {
         print "$cmd\n";
         system($cmd);
         if ( $CHILD_ERROR ) {
            warn "Command failed: $cmd\n";
            $rm_files_ok = 0;
         }
      }
   }
   else {
      warn "ERROR: No pt-agent files or directories found.  You can ignore this "
         . "error if the agent is not installed, or if it has already been "
         . "removed.  Else, verify that the values in $config_file are "
         . "correct and try again.\n";
   }

   eval {
      $cxn->connect();
   };
   if ( $EVAL_ERROR ) {
      chomp $EVAL_ERROR;
      die "ERROR: Cannot connect to MySQL: $EVAL_ERROR\n"
         . "Please re-run pt-agent --uninstall and specify MySQL connection "
         . "options like --user and --host to connect to MySQL as a user "
         . "with sufficient privileges to drop MySQL users.\n";
   }

   my $drop_mysql_user_ok;
   eval {
      $cxn->dbh->selectall_arrayref("SHOW GRANTS FOR 'pt_agent'\@'localhost'");
   };
   if ( !$EVAL_ERROR ) {
      my $sql = "DROP USER 'pt_agent'\@'localhost'";
      print "Are you sure you want to execute this statement " 
         . "to remove the pt-agent MySQL user?\n$sql\n";
      while ( !$drop_mysql_user_ok ) {
         print "Enter 'yes' to execute this statment, or CTRL-C to abort: ";
         $drop_mysql_user_ok = <STDIN>;
         chomp($drop_mysql_user_ok) if $drop_mysql_user_ok;
         if ( $drop_mysql_user_ok && $drop_mysql_user_ok eq 'yes' ) {
            last;
         }
         else {
            $drop_mysql_user_ok = 0;
         }
      }
      # CTRL-C should prevent us from getting here, but just in case:
      return unless $drop_mysql_user_ok;
      eval {
         $cxn->dbh->do($sql);
      };
      if ( $EVAL_ERROR ) {
         warn "Error dropping the pt-agent MySQL user: $EVAL_ERROR";
         $drop_mysql_user_ok = 0;
      }
   }
   else {
      warn "ERROR: No pt-agent MySQL user found.  You can ignore this "
         . "error if the agent is not installed, or if it has already been "
         . "removed.  Else, verify that the values in $config_file are "
         . "correct and try again.\n";
   }

   print "\n";
   if ( $rm_files_ok && $drop_mysql_user_ok ) {
      print "pt-agent and all its data has been removed from this server, "
      . "but the agent and any data it sent has not been deleted from "
      . "Percona Cloud Tools.  Go to https://cloud.percona.com/agents "
      . "to delete the agent.\n";
   }
   else {
      warn "Uninstalling pt-agent failed.  See previous output for errors "
         . "and try again.  Contact Percona if you need help.\n";
   }

   return
}

# ################## #
# Misc and util subs #
# ################## #

sub get_config_file {
   my $home_dir    = $ENV{HOME} || $ENV{HOMEPATH} || $ENV{USERPROFILE} || '.';
   my $config_file = "$home_dir/.pt-agent.conf";
   return $config_file;
}

sub save_agent {
   my (%args) = @_;
   have_required_args(\%args, qw(
      agent
      lib_dir
   )) or die;
   my $agent   = $args{agent};
   my $lib_dir = $args{lib_dir};
   my $file    = $lib_dir . '/agent';
   $logger->debug("Saving Agent to $file");
   eval {
      open my $fh, '>', $file
         or die "Error opening $file: $OS_ERROR";
      print { $fh } as_json($agent)
         or die "Error writing to $file: $OS_ERROR";
      close $fh
         or die "Error closing $file: $OS_ERROR";
   };
   if ( $EVAL_ERROR ) {
      if ( !$state->{save_agent_error}++ ) {
         chomp($EVAL_ERROR);
         $logger->warning("Cannot save agent to $lib_dir: $EVAL_ERROR.  "
            . "Configure the agent to use a writeable --lib directory.  "
            . "This warning will not be printed again.");
      }
   }
   delete $state->{save_agent_error};
   return;
}

sub slurp {
   my ($file) = @_;
   return unless -f $file;
   open my $fh, '<', $file
      or die "Error opening $file: $OS_ERROR";
   my $data = do {
      local $INPUT_RECORD_SEPARATOR = undef;
      <$fh>;
   };
   close $fh;
   return $data;
}

sub write_to_file {
   my (%args) = @_;
   my $data = $args{data};
   my $file = $args{file};
   die "No file" unless $file;
   open my $fh, '>', $file
      or die "Error opening $file: $OS_ERROR";
   print { $fh } $data;
   close $fh;
   return;
}

sub _set_logger {
   my $new_logger = shift;
   $logger = $new_logger;
   return;
}

sub get_versions {
   my (%args) = @_;
   my $cxn      = $args{Cxn};
   my $tries    = $args{tries}    || 1;
   my $interval = $args{interval} || sub { return; };

   # This is currently the actual response from GET v.percona.com
   my $fake_response = <<EOL;
OS;os_version
MySQL;mysql_variable;version_comment,version
Perl;perl_version
DBD::mysql;perl_module_version
Percona::Toolkit;perl_module_version
JSON;perl_module_version
LWP;perl_module_version
IO::Socket::SSL;perl_module_version
DBD::mysql;perl_module_version
EOL

   my $items = VersionCheck::parse_server_response(
      response => $fake_response,
   );

   my $instances = [
      { name => 'system', id => 0, },
   ];

   my $have_mysql = -1;
   if ( !$cxn->dbh || !$cxn->dbh->ping() ) {
      $logger->debug("Connecting to MySQL");
      eval {
         $cxn->connect();
      };
      if ( $EVAL_ERROR ) {
         $logger->debug("Cannot connect to MySQL: $EVAL_ERROR");
         $have_mysql = 0;
      }
      else {
         $have_mysql = 1;
      }
   }

   if ( $have_mysql ) {
      $logger->debug("Have MySQL connection");
      my ($name, $id) = VersionCheck::get_instance_id(
         { dbh => $cxn->dbh, dsn => $cxn->dsn },
      );
      push @$instances,
         { name => $name, id => $id, dbh => $cxn->dbh, dsn => $cxn->dsn };

      # Disconnect MySQL if we connected it.
      if ( $have_mysql == 1 ) {
         $logger->debug("Disconnecting MySQL");
         eval {
            $cxn->dbh->disconnect();
         };
         if ( $EVAL_ERROR ) {
            $logger->debug($EVAL_ERROR);
         }
      }
   }

   my $versions = VersionCheck::get_versions(
      items     => $items,
      instances => $instances,
   );

   my %version_for;
   foreach my $item ( sort keys %$items ) {
      next unless exists $versions->{$item};
      if ( ref($versions->{$item}) eq 'HASH' ) {
         my $mysql_versions = $versions->{$item};
         for my $id ( sort keys %$mysql_versions ) {
            $version_for{$item} = $mysql_versions->{$id};
         }
      }
      else {
         $version_for{$item} = $versions->{$item};
      }
   }

   PTDEBUG && _d('Versions:', Dumper(\%version_for));
   return \%version_for;
}

sub env_vars {
   my @vars;
   foreach my $var ( qw(
      PTDEBUG
      PERCONA_TOOLKIT_TEST_USE_DSN_NAMES
      PCT_ENTRY_LINK
   )) {
      if ( my $val = $ENV{$var} ) {
         push @vars, "$var=\"$val\"";
      }
   }
   return join(' ', @vars);
}

sub _safe_mkdir {
   my $dir = shift;

   # Multiple processes are running at once, all running the same code,
   # all trying to init pt-agent's various directories if necessary, so
   # race conditions abound.  Another process may have created the dir
   # between -d checking for it and now, so if mkdir throws a "File exists"
   # error and the dir does now exist, then that's ok.  Else, it's an error.
   eval { 
      mkdir $dir or die $OS_ERROR;
   };
   if ( my $e = $EVAL_ERROR ) {
      if ( $e =~ /exists/i && -d $dir ) {
         PTDEBUG && _d('Another process created', $dir);
      }
      else {
         die "Cannot mkdir $dir: $e";
      }
   }
   return;
}

sub check_if_mysql_restarted {
   my (%args) = @_;
   have_required_args(\%args, qw(
      dbh
   )) or die;
   my $dbh = $args{dbh};

   # Optional args
   my $uptime = $args{uptime};  # for testing
   my $margin = $args{margin} || 5;

   if ( !$uptime ) {
      my $sql = "SHOW STATUS LIKE 'uptime'";
      eval {
         (undef, $uptime) = $dbh->selectrow_array($sql);
      };
      if ( $EVAL_ERROR ) {
         $logger->error("$sql: $EVAL_ERROR");
         return;
      }
   }

   my $now = int(time);

   if ( !$state->{last_uptime} || !$state->{last_uptime_check} ) {
      $logger->debug("MySQL uptime: $uptime");
      delete $state->{mysql_restarted};
   }
   elsif ( !$state->{mysql_restarted} ) {
      my $elapsed_time     = $now - $state->{last_uptime_check};
      my $exepected_uptime = $state->{last_uptime} + $elapsed_time;
      my $mysql_restarted  = $uptime > ($exepected_uptime - $margin) && $uptime < ($exepected_uptime + $margin) ? 0 : 1;
      $logger->debug("MySQL uptime check: last=$state->{last_uptime} elapsed=$elapsed_time expected=$exepected_uptime "
         . "+/- ${margin}s actual=$uptime");
      if ( $mysql_restarted ) {
         $logger->warning("MySQL restarted: last=$state->{last_uptime} "
            . "elapsed=$elapsed_time expected=$exepected_uptime "
            . "+/- ${margin}s actual=$uptime");
         $state->{mysql_restarted} = ts(time, 1);  # 1=UTC
         $state->{need_mysql_version} = 1;
      }
   }

   $state->{last_uptime}       = $uptime;
   $state->{last_uptime_check} = $now;

   return;
}

sub too_many_agents {
   my (%args) = @_;
   have_required_args(\%args, qw(
      lib_dir
   )) or die;
   my $lib_dir = $args{lib_dir};
   return unless -d "$lib_dir/pids";
   my @pids = glob "$lib_dir/pids/*";
   return scalar @pids > 10 ? 1 : 0;
}

sub ping_api {
   my (%args) = @_;
   have_required_args(\%args, qw(
      client
   )) or die;
   my $client = $args{client};
   my $ping_link = $client->entry_link . '/ping';
   $ping_link =~ s!//ping!/ping!g;  # //ping doesn't work
   eval {
      $client->get(
         link => $ping_link,
      );
   };
   return $EVAL_ERROR ? 0 : 1;
}

sub _logger {
   my $_logger = shift;
   $logger = $_logger if $_logger;
   return $logger;
}

sub _state {
   my $_state = shift;
   $state = $_state if $_state;
   return $state;
}

# Catches signals so we can exit gracefully.
sub sig_int {
   my ( $signal ) = @_;
   $oktorun = 0;
   if ( $exit_on_signals ) {
      print STDERR "\n# Caught SIG$signal, exiting.\n";
      exit 1;
   }
   print STDERR "# Caught SIG$signal.  Use 'kill -ABRT $PID' if "
      . "the tool does not exit normally in a few seconds.\n";
   return;
}

# ############################################################################
# Run the program.
# ############################################################################

if ( !caller ) { exit main(@ARGV); }

1; # Because this is a module as well as a script.

# ############################################################################
# Documentation
# ############################################################################
=pod

=head1 NAME

pt-agent - Agent for Percona Cloud Tools

=head1 SYNOPSIS

Usage: pt-agent [OPTIONS]

pt-agent is the client-side agent for Percona Cloud Tools.  It is not
a general command line tool like other tools in Percona Toolkit, it is
configured and controlled through the web at https://cloud.percona.com.
Visit https://cloud.percona.com for more information and to sign up.

=head1 DESCRIPTION

pt-agent is the client-side agent for Percona Cloud Tools (PCT).  It is
controlled and configured through the web app at https://cloud.percona.com.
Visit https://cloud.percona.com for more information and to sign up.

pt-agent, or "the agent", is a single, unique instance of the tool running
on a server.  Two agents cannot run on the same server (see L<"--pid">).

The agent is a daemon that runs as root.  It should be started with
L<"--daemonize">.  It connects periodically to Percona to update
its configuration and services, and it schedules L<"--run-service"> and
L<"--send-data"> instances of itself using cron.  Other than L<"INSTALLING">
and starting the agent locally, all control and configuration is done through
the web at https://cloud.percona.com.

=head1 INSTALLING

pt-agent must be installed and ran as root.  It is possible to run as
a non-root user, but this requires a more complicated and manual installation.
Please contact Percona for help if you need to run pt-agent as a non-root user.

Installing the agent as root is very simple:

  # pt-agent --install

The agent will prompt you for your Percona Cloud Tools API key.  Then it
will verify the API key, create a MySQL user for the agent, and run the agent.
When the install process is complete, go to https://cloud.percona.com to enable
services for agent.

Please contact Percona if you need help installing the agent.

=head2 SLAVE INSTALL

There are two ways to install pt-agent on a slave.  The first and best way
is to install the agent on the master so that the L<"MYSQL USER"> is created
on the master and replicates to slaves.  This is best because it avoids
writing to the slave.  Then create the C</etc/percona/agent/> directory on
the slave and copy in to it C</etc/percona/agent/my.cnf> from the master.
Run L<"--install"> on the slave and pt-agent will automatically detect and
use the MySQL user and password in C</etc/percona/agent/my.cnf>.  Repeat the
process for other slaves.

The second way to install pt-agent on a slave is not safe because it writes
directly to the slave: specify L<"--install-options">
C<force_dangerous_slave_install> in addition to L<"--install">.  As the
install option name implies, this is dangerous, but it forces pt-agent
to ignore that MySQL is a slave.

=head2 Percona XtraDB Cluster (PXC) INSTALL

Installing pt-agent on Percona XtraDB Cluster (PXC) nodes is the same as
installing it safely on slaves.  First install the agent on any node.  This
will create the L<"MYSQL USER"> that will replicate to all other nodes.
Then create the C</etc/percona/agent/> directory on another node and copy in
to it C</etc/percona/agent/my.cnf> from the first node where pt-agent was
installed.  Run L<"--install"> on the node and pt-agent will automatically
detect and use the MySQL user and password in C</etc/percona/agent/my.cnf>.
Repeat the process for other nodes.

=head1 MYSQL USER

During L<"--install">, pt-agent creates the following MySQL user:

  GRANT SUPER, USAGE ON *.* TO 'pt_agent'@'localhost' IDENTIFIED BY 'pass'

C<pass> is a random string.  MySQL options for the agent are stored in
C</etc/percona/agent/my.cnf>.  The C<SUPER> privilege is required so that
the agent can set global MySQL variables like C<long_query_time>.

=head1 EXIT STATUS

pt-agent exists zero if no errors or warnings occurred, else it exits non-zero.

=head1 OPTIONS

L<"--run-service"> and L<"--send-data"> are mutually exclusive.

L<"--status">, L<"--stop">, and L<"--reset"> are mutually exclusive.

=over

=item --[no]agent-api

default: yes

Enable the agent API; do not use this option manually.  This option is used
internally to allow the agent to stop itself and shutdown quickly.

=item --agent-uuid

type: string

Existing agent UUID for re-installing an agent.

=item --api-key

type: string

Your secret Percona Cloud Tools API key.

=item --ask-pass

Prompt for MySQL password.

=item --check-interval

type: time; default: 1m

How often to check for a new configuration and services.

=item --config

type: Array

Read this comma-separated list of config files; if specified, this must be the
first option on the command line.

See the L<"--help"> output for a list of default config files.

=item --daemonize

Daemonize the agent.  This causes the agent to fork into the background and
L<"--log"> all output.

Fork to the background and detach from the shell.  POSIX operating systems only.

=item --defaults-file

short form: -F; type: string

Only read MySQL options from the given file.  You must give an absolute
pathname.

=item --disk-bytes-free

type: size; default: 100M

Stop all services if the disk has less than this much free space.
This prevents the agent from filling up the disk with service data.

Valid size value suffixes are k, M, G, and T.

=item --disk-pct-free

type: int; default: 5

Stop all services if the disk has less than this percent free space.
This prevents the agent from filling up the disk with service data.

This option works similarly to L<"--disk-bytes-free"> but specifies a
percentage margin of safety instead of a bytes margin of safety.
The agent honors both options, and will not collect any data unless both
margins are satisfied.

=item --help

Print the agent's help and exit.

=item --host

short form: -h; type: string; default: localhost

MySQL host.

=item --install

Install pt-agent as root.

=item --install-options

type: Hash

Comma-separated list of L<"--install"> options.  Options are:

=over

=item offline

Do not verify the API key or start the agent.

=item force_dangerous_slave_install

Like the option's name suggests: this forces a dangerous slave install,
so you should not use this option unless you are aware of the potential
consequences.  To install the agent on a slave, C</etc/percona/agent/my.cnf>
must exist because it is not safe to create the agent's MySQL user on
a slave.  The agent should be installed on the master first, then
C</etc/percona/agent/my.cnf> copied from the master server to the slave
server.  Using this option forces the agent to create the agent's MySQL
user on the slave.  B<WARNING>: writing to a slave is dangerous and could
cause replication to crash.

=back

=item --interactive

Run in interactive mode (disables L<"--[no]log-api">).

=item --lib

type: string; default: /var/lib/pt-agent

Directory in which to save local data.  pt-agent is remotely controlled and
configured, but it also saves data locally.  These files should not be edited
manually.

=item --log

type: string; default: /var/log/pt-agent.log

Log all output to this file when daemonized.

=item --[no]log-api

default: yes

Log everything through the Percona Cloud Tools API.

=item --password

short form: -p; type: string

MySQL password.

=item --pid

type: string; default: /var/run/pt-agent.pid

Create the given PID file.  The file contains the process ID of the script.
The PID file is removed when the script exits.  Before starting, the script
checks if the PID file already exists.  If it does not, then the script creates
and writes its own PID to it.  If it does, then the script checks the following:
if the file contains a PID and a process is running with that PID, then
the script dies; or, if there is no process running with that PID, then the
script overwrites the file with its own PID and starts; else, if the file
contains no PID, then the script dies.

=item --ping

Ping the Percona Cloud Tools API and exit.

=item --port

short form: -P; type: int

MySQL port number.

=item --reload

Force pt-agent to reload its configuration immediately.

=item --reset

cumulative: yes; default: 0

Reset pt-agent to a clean post-install state.

B<WARNING>: all L<"--spool"> data will be deleted.

=item --run-service

type: string

Run a service and spool its data for L<"--send-data">.  I<You do not need
to run pt-agent with this option.>  The main pt-agent daemon schedules
instances of itself with this option.

=item --send-data

type: string

Send data for a service to Percona.  I<You do not need to run pt-agent with
this option.>  The main pt-agent daemon schedules instances of itself with
this option.

=item --set-vars

type: Array

Set the MySQL variables in this comma-separated list of C<variable=value> pairs.

By default, the agent sets:

=for comment ignore-pt-internal-value
MAGIC_set_vars

   wait_timeout=10000

Variables specified on the command line override these defaults.  For
example, specifying C<--set-vars wait_timeout=500> overrides the default
value of C<10000>.

The agent prints a warning and continues if a variable cannot be set.

=item --socket

short form: -S; type: string

MySQL socket file.

=item --spool

type: string; default: /var/spool/pt-agent

Directory in which to save service data before sending to Percona.
L<"--run-service"> saves data in this directory, and L<"--send-data">
reads data from this directory.  Each service has its own subdirectory,
like C<--spool/query-history> for the Query History service.  Data
is removed by L<"--send-data"> after it is successfully sent to Percona.

=item --status

Print the status of pt-agent.

=item --stop

Stop pt-agent and all services.

=item --uninstall

Completely remove pt-agent and all its data from the server.  This does not
delete the agent from https://cloud.percona.com.

=item --user

short form: -u; type: string

MySQL user, if not the current system user.

=item --version

Print the agent's version and exit.

=back

=head1 DSN OPTIONS

These DSN options are used to create a DSN.  Each option is given like
C<option=value>.  The options are case-sensitive, so P and p are not the
same option.  There cannot be whitespace before or after the C<=> and
if the value contains whitespace it must be quoted.  DSN options are
comma-separated.  See the L<percona-toolkit> manpage for full details.

=over

=item * A

dsn: charset; copy: yes

Default character set.

=item * D

copy: no

Default database when connecting.

=item * F

dsn: mysql_read_default_file; copy: yes

Defaults file for connection values.

=item * h

dsn: host; copy: yes

MySQL host.

=item * p

dsn: password; copy: yes

MySQL password.

=item * P

dsn: port; copy: yes

MySQL port number.

=item * S

dsn: mysql_socket; copy: no

MySQL socket file.

=item * u

dsn: user; copy: yes

MySQL user, if not the current system user.

=back

=head1 ENVIRONMENT

The environment variable C<PTDEBUG> enables verbose debugging output to STDERR.
To enable debugging and capture all output to a file, run the tool like:

   PTDEBUG=1 pt-agent ... > FILE 2>&1

Be careful: debugging output is voluminous and can generate several megabytes
of output.

=head1 SYSTEM REQUIREMENTS

pt-agent requires:

=over

=item * A Percona Cloud Tools account (https://cloud.percona.com)

=item * Access to https://cloud-api.percona.com

=item * Perl 5.8 or newer

=item * Standard Linux bin tools (grep, awk, stat, etc.)

=item * cron

=item * A Bash shell

=item * Core Perl modules

=item * DBD::mysql Perl module

=item * JSON Perl module

=item * LWP (>= v5.813) Perl module

=item * IO::Socket::SSL Perl module

=back

=head1 BUGS

For a list of known bugs, see L<http://www.percona.com/bugs/pt-agent>.

Please report bugs at L<https://bugs.launchpad.net/percona-toolkit>.
Include the following information in your bug report:

=over

=item * Complete command-line used to run the tool

=item * Tool L<"--version">

=item * MySQL version of all servers involved

=item * Output from the tool including STDERR

=item * Input files (log/dump/config files, etc.)

=back

If possible, include debugging output by running the tool with C<PTDEBUG>;
see L<"ENVIRONMENT">.

=head1 DOWNLOADING

Visit L<http://www.percona.com/software/percona-toolkit/> to download the
latest release of Percona Toolkit.

=head1 AUTHORS

Daniel Nichter

=head1 ABOUT PERCONA TOOLKIT

This tool is part of Percona Toolkit, a collection of advanced command-line
tools developed by Percona for MySQL support and consulting.  Percona Toolkit
was forked from two projects in June, 2011: Maatkit and Aspersa.  Those
projects were created by Baron Schwartz and developed primarily by him and
Daniel Nichter, both of whom are employed by Percona.  Visit
L<http://www.percona.com/software/> for more software developed by Percona.

=head1 COPYRIGHT, LICENSE, AND WARRANTY

This program is copyright 2013-2014 Percona LLC and/or its affiliates.

THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.

This program is free software; you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, version 2; OR the Perl Artistic License.  On UNIX and similar
systems, you can issue `man perlgpl' or `man perlartistic' to read these
licenses.

You should have received a copy of the GNU General Public License along with
this program; if not, write to the Free Software Foundation, Inc., 59 Temple
Place, Suite 330, Boston, MA  02111-1307  USA.

=head1 VERSION

pt-agent 2.2.7

=cut