lib/DBIx/Class/Storage/DBI/Replicated/Pool.pm

   1 package DBIx::Class::Storage::DBI::Replicated::Pool;
   2
   3 use Moo;
   4 use DBIx::Class::Storage::DBI::Replicated::Replicant;
   5 use List::Util 'sum';
   6 use Scalar::Util 'reftype';
   7 use DBI ();
   8 use Types::Standard qw/Num Int ClassName HashRef Object/;
   9 use DBIx::Class::Storage::DBI::Replicated::Types 'DBICStorageDBI';
  10 use Try::Tiny;
  11
  12 use namespace::clean -except => 'meta';
  13
  14 =head1 NAME
  15
  16 DBIx::Class::Storage::DBI::Replicated::Pool - Manage a pool of replicants
  17
  18 =head1 SYNOPSIS
  19
  20 This class is used internally by L<DBIx::Class::Storage::DBI::Replicated>.  You
  21 shouldn't need to create instances of this class.
  22
  23 =head1 DESCRIPTION
  24
  25 In a replicated storage type, there is at least one replicant to handle the
  26 read-only traffic.  The Pool class manages this replicant, or list of
  27 replicants, and gives some methods for querying information about their status.
  28
  29 =head1 ATTRIBUTES
  30
  31 This class defines the following attributes.
  32
  33 =head2 maximum_lag ($num)
  34
  35 This is a number which defines the maximum allowed lag returned by the
  36 L<DBIx::Class::Storage::DBI/lag_behind_master> method.  The default is 0.  In
  37 general, this should return a larger number when the replicant is lagging
  38 behind its master, however the implementation of this is database specific, so
  39 don't count on this number having a fixed meaning.  For example, MySQL will
  40 return a number of seconds that the replicating database is lagging.
  41
  42 =cut
  43
  44 has 'maximum_lag' => (
  45   is=>'rw',
  46   isa=>Num,
  47   lazy=>1,
  48   default=>0,
  49 );
  50
  51 =head2 last_validated
  52
  53 This is an integer representing a time since the last time the replicants were
  54 validated. It's nothing fancy, just an integer provided via the perl L<time|perlfunc/time>
  55 built-in.
  56
  57 =cut
  58
  59 has 'last_validated' => (
  60   is=>'rw',
  61   isa=>Int,
  62   reader=>'last_validated',
  63   writer=>'_last_validated',
  64   lazy=>1,
  65   default=>0,
  66 );
  67
  68 =head2 replicant_type ($classname)
  69
  70 Base class used to instantiate replicants that are in the pool.  Unless you
  71 need to subclass L<DBIx::Class::Storage::DBI::Replicated::Replicant> you should
  72 just leave this alone.
  73
  74 =cut
  75
  76 has 'replicant_type' => (
  77   is=>'ro',
  78   isa=>ClassName,
  79   required=>1,
  80   default=>'DBIx::Class::Storage::DBI',
  81   handles=>{
  82     'create_replicant' => 'new',
  83   },
  84 );
  85
  86 =head2 replicants
  87
  88 A hashref of replicant, with the key being the dsn and the value returning the
  89 actual replicant storage.  For example, if the $dsn element is something like:
  90
  91   "dbi:SQLite:dbname=dbfile"
  92
  93 You could access the specific replicant via:
  94
  95   $schema->storage->replicants->{'dbname=dbfile'}
  96
  97 This attributes also supports the following helper methods:
  98
  99 =over 4
 100
 101 =item set_replicant($key=>$storage)
 102
 103 Pushes a replicant onto the HashRef under $key
 104
 105 =item get_replicant($key)
 106
 107 Retrieves the named replicant
 108
 109 =item has_replicants
 110
 111 Returns true if the Pool defines replicants.
 112
 113 =item num_replicants
 114
 115 The number of replicants in the pool
 116
 117 =item delete_replicant ($key)
 118
 119 Removes the replicant under $key from the pool
 120
 121 =back
 122
 123 =cut
 124
 125 has 'replicants' => (
 126   is=>'rw',
 127   isa=>HashRef[Object],
 128   default=>sub {{}},
 129 );
 130
 131 sub set_replicant { $_[0]->replicants->{$_[1]} = $_[2] }
 132 sub get_replicant { $_[0]->replicants->{$_[1]} }
 133 sub has_replicants { !!keys %{$_[0]->replicants} }
 134 sub num_replicants { 0+keys %{$_[0]->replicants} }
 135 sub delete_replicant { delete $_[0]->replicants->{$_[1]} }
 136 sub all_replicant_storages { values %{$_[0]->replicants} }
 137
 138 has next_unknown_replicant_id => (
 139   is => 'rw',
 140   isa => Int,
 141   default => 1,
 142 );
 143
 144 sub inc_unknown_replicant_id {
 145   my ($self) = @_;
 146   $self->next_unknown_replicant_id($self->next_unknown_replicant_id + 1);
 147 }
 148
 149 =head2 master
 150
 151 Reference to the master Storage.
 152
 153 =cut
 154
 155 has master => (is => 'rw', isa => DBICStorageDBI, weak_ref => 1);
 156
 157 =head1 METHODS
 158
 159 This class defines the following methods.
 160
 161 =head2 connect_replicants ($schema, Array[$connect_info])
 162
 163 Given an array of $dsn or connect_info structures suitable for connected to a
 164 database, create an L<DBIx::Class::Storage::DBI::Replicated::Replicant> object
 165 and store it in the L</replicants> attribute.
 166
 167 =cut
 168
 169 sub connect_replicants {
 170   my $self = shift @_;
 171   my $schema = shift @_;
 172
 173   my @newly_created = ();
 174   foreach my $connect_info (@_) {
 175     $connect_info = [ $connect_info ]
 176       if reftype $connect_info ne 'ARRAY';
 177
 178     my $connect_coderef =
 179       (reftype($connect_info->[0])||'') eq 'CODE' ? $connect_info->[0]
 180         : (reftype($connect_info->[0])||'') eq 'HASH' &&
 181           $connect_info->[0]->{dbh_maker};
 182
 183     my $dsn;
 184     my $replicant = do {
 185 # yes this is evil, but it only usually happens once (for coderefs)
 186 # this will fail if the coderef does not actually DBI::connect
 187       no warnings 'redefine';
 188       my $connect = \&DBI::connect;
 189       local *DBI::connect = sub {
 190         $dsn = $_[1];
 191         goto $connect;
 192       };
 193       $self->connect_replicant($schema, $connect_info);
 194     };
 195
 196     my $key;
 197
 198     if (!$dsn) {
 199       if (!$connect_coderef) {
 200         $dsn = $connect_info->[0];
 201         $dsn = $dsn->{dsn} if (reftype($dsn)||'') eq 'HASH';
 202       }
 203       else {
 204         # all attempts to get the DSN failed
 205         $key = "UNKNOWN_" . $self->next_unknown_replicant_id;
 206         $self->inc_unknown_replicant_id;
 207       }
 208     }
 209     if ($dsn) {
 210       $replicant->dsn($dsn);
 211       ($key) = ($dsn =~ m/^dbi\:.+\:(.+)$/i);
 212     }
 213
 214     $replicant->id($key);
 215     $self->set_replicant($key => $replicant);
 216
 217     push @newly_created, $replicant;
 218   }
 219
 220   return @newly_created;
 221 }
 222
 223 =head2 connect_replicant ($schema, $connect_info)
 224
 225 Given a schema object and a hashref of $connect_info, connect the replicant
 226 and return it.
 227
 228 =cut
 229
 230 sub connect_replicant {
 231   my ($self, $schema, $connect_info) = @_;
 232   my $replicant = $self->create_replicant($schema);
 233   $replicant->connect_info($connect_info);
 234
 235 ## It is undesirable for catalyst to connect at ->conect_replicants time, as
 236 ## connections should only happen on the first request that uses the database.
 237 ## So we try to set the driver without connecting, however this doesn't always
 238 ## work, as a driver may need to connect to determine the DB version, and this
 239 ## may fail.
 240 ##
 241 ## Why this is necessary at all, is that we need to have the final storage
 242 ## class to apply the Replicant role.
 243
 244   $self->_safely($replicant, '->_determine_driver', sub {
 245     $replicant->_determine_driver
 246   });
 247
 248   Moo::Role->apply_roles_to_object(
 249     $replicant,
 250     'DBIx::Class::Storage::DBI::Replicated::Replicant',
 251   );
 252
 253   # link back to master
 254   $replicant->master($self->master);
 255
 256   return $replicant;
 257 }
 258
 259 =head2 _safely_ensure_connected ($replicant)
 260
 261 The standard ensure_connected method with throw an exception should it fail to
 262 connect.  For the master database this is desirable, but since replicants are
 263 allowed to fail, this behavior is not desirable.  This method wraps the call
 264 to ensure_connected in an eval in order to catch any generated errors.  That
 265 way a slave can go completely offline (e.g. the box itself can die) without
 266 bringing down your entire pool of databases.
 267
 268 =cut
 269
 270 sub _safely_ensure_connected {
 271   my ($self, $replicant, @args) = @_;
 272
 273   return $self->_safely($replicant, '->ensure_connected', sub {
 274     $replicant->ensure_connected(@args)
 275   });
 276 }
 277
 278 =head2 _safely ($replicant, $name, $code)
 279
 280 Execute C<$code> for operation C<$name> catching any exceptions and printing an
 281 error message to the C<<$replicant->debugobj>>.
 282
 283 Returns 1 on success and undef on failure.
 284
 285 =cut
 286
 287 sub _safely {
 288   my ($self, $replicant, $name, $code) = @_;
 289
 290   return try {
 291     $code->();
 292     1;
 293   } catch {
 294     $replicant->debugobj->print(sprintf(
 295       "Exception trying to $name for replicant %s, error is %s",
 296       $replicant->_dbi_connect_info->[0], $_)
 297     );
 298     undef;
 299   };
 300 }
 301
 302 =head2 connected_replicants
 303
 304 Returns true if there are connected replicants.  Actually is overloaded to
 305 return the number of replicants.  So you can do stuff like:
 306
 307   if( my $num_connected = $storage->has_connected_replicants ) {
 308     print "I have $num_connected connected replicants";
 309   } else {
 310     print "Sorry, no replicants.";
 311   }
 312
 313 This method will actually test that each replicant in the L</replicants> hashref
 314 is actually connected, try not to hit this 10 times a second.
 315
 316 =cut
 317
 318 sub connected_replicants {
 319   my $self = shift @_;
 320   return sum( map {
 321     $_->connected ? 1:0
 322   } $self->all_replicants );
 323 }
 324
 325 =head2 active_replicants
 326
 327 This is an array of replicants that are considered to be active in the pool.
 328 This does not check to see if they are connected, but if they are not, DBIC
 329 should automatically reconnect them for us when we hit them with a query.
 330
 331 =cut
 332
 333 sub active_replicants {
 334   my $self = shift @_;
 335   return ( grep {$_} map {
 336     $_->active ? $_:0
 337   } $self->all_replicants );
 338 }
 339
 340 =head2 all_replicants
 341
 342 Just a simple array of all the replicant storages.  No particular order to the
 343 array is given, nor should any meaning be derived.
 344
 345 =cut
 346
 347 sub all_replicants {
 348   my $self = shift @_;
 349   return values %{$self->replicants};
 350 }
 351
 352 =head2 validate_replicants
 353
 354 This does a check to see if 1) each replicate is connected (or reconnectable),
 355 2) that is ->is_replicating, and 3) that it is not exceeding the lag amount
 356 defined by L</maximum_lag>.  Replicants that fail any of these tests are set to
 357 inactive, and thus removed from the replication pool.
 358
 359 This tests L</all_replicants>, since a replicant that has been previous marked
 360 as inactive can be reactivated should it start to pass the validation tests again.
 361
 362 See L<DBIx::Class::Storage::DBI> for more about checking if a replicating
 363 connection is not following a master or is lagging.
 364
 365 Calling this method will generate queries on the replicant databases so it is
 366 not recommended that you run them very often.
 367
 368 This method requires that your underlying storage engine supports some sort of
 369 native replication mechanism.  Currently only MySQL native replication is
 370 supported.  Your patches to make other replication types work are welcomed.
 371
 372 =cut
 373
 374 sub validate_replicants {
 375   my $self = shift @_;
 376   foreach my $replicant($self->all_replicants) {
 377     if($self->_safely_ensure_connected($replicant)) {
 378       my $is_replicating = $replicant->is_replicating;
 379       unless(defined $is_replicating) {
 380         $replicant->debugobj->print("Storage Driver ".ref($self)." Does not support the 'is_replicating' method.  Assuming you are manually managing.\n");
 381         next;
 382       } else {
 383         if($is_replicating) {
 384           my $lag_behind_master = $replicant->lag_behind_master;
 385           unless(defined $lag_behind_master) {
 386             $replicant->debugobj->print("Storage Driver ".ref($self)." Does not support the 'lag_behind_master' method.  Assuming you are manually managing.\n");
 387             next;
 388           } else {
 389             if($lag_behind_master <= $self->maximum_lag) {
 390               $replicant->active(1);
 391             } else {
 392               $replicant->active(0);
 393             }
 394           }
 395         } else {
 396           $replicant->active(0);
 397         }
 398       }
 399     } else {
 400       $replicant->active(0);
 401     }
 402   }
 403   ## Mark that we completed this validation.
 404   $self->_last_validated(time);
 405 }
 406
 407 =head1 FURTHER QUESTIONS?
 408
 409 Check the list of L<additional DBIC resources|DBIx::Class/GETTING HELP/SUPPORT>.
 410
 411 =head1 COPYRIGHT AND LICENSE
 412
 413 This module is free software L<copyright|DBIx::Class/COPYRIGHT AND LICENSE>
 414 by the L<DBIx::Class (DBIC) authors|DBIx::Class/AUTHORS>. You can
 415 redistribute it and/or modify it under the same terms as the
 416 L<DBIx::Class library|DBIx::Class/COPYRIGHT AND LICENSE>.
 417
 418 =cut
 419
 420 1;