lib/MogileFS/IOStatWatcher.pm

   1 package MogileFS::IOStatWatcher;
   2 use strict;
   3 use Sys::Syscall 0.22; # We use it indirectly, and trigger bugs in earlier versions.
   4 use Danga::Socket;
   5 use IO::Socket::INET;
   6
   7 =head1 Methods
   8
   9 =head2 $iow = MogileFS::IOStatWatcher->new()
  10
  11 Returns a new IOStatWatcher object.
  12
  13 =cut
  14
  15 sub new {
  16     my ($class) = @_;
  17     my $self = bless {
  18         hosts => {},
  19     }, $class;
  20     $self->on_stats; # set an empty handler.
  21     return $self;
  22 }
  23
  24 =head2 $iow->set_hosts( host1 [, host2 [, ...] ] )
  25
  26 Sets the list of hosts to connect to for collecting IOStat information. This call can block if you
  27 pass it hostnames instead of ip addresses.
  28
  29 Upon successful connection, the on_stats callback will be called each time the statistics are
  30 collected. Error states (failed connections, etc.) will trigger retries on 60 second intervals, and
  31 disconnects will trigger an immediate reconnect.
  32
  33 =cut
  34
  35 sub set_hosts {
  36     my ($self, @ips) = @_;
  37     my $old_hosts = $self->{hosts};
  38     my $new_hosts = {};
  39     foreach my $host (@ips) {
  40         $new_hosts->{$host} = (delete $old_hosts->{$host}) || MogileFS::IOStatWatch::Client->new($host, $self);
  41     }
  42     # TODO: close hosts that were removed (things in %$old_hosts)
  43     $self->{hosts} = $new_hosts;
  44 }
  45
  46 =head2 $iow->on_stats( coderef )
  47
  48 Sets the coderef called for the C<on_stats> callback.
  49
  50 =cut
  51
  52 sub on_stats {
  53     my ($self, $cb) = @_;
  54
  55     unless (ref $cb eq 'CODE') {
  56         $cb = sub {};
  57     }
  58
  59     $self->{on_stats} = $cb;
  60 }
  61
  62 =head1 Callbacks
  63
  64 =head2 on_stats->( host, stats )
  65
  66 Called each time device use statistics are collected. The C<host>
  67 argument is the value passed in to the C<set_hosts> method. The
  68 C<stats> object is a hashref of mogile device numbers (without leading
  69 "dev") to their corresponding utilization percentages.
  70
  71 =cut
  72
  73 # Everything beyond here is internal.
  74
  75 sub got_stats {
  76     my ($self, $host, $stats) = @_;
  77     $self->{on_stats}->($host, $stats);
  78 }
  79
  80 sub restart_monitoring_if_needed {
  81     my ($self, $host) = @_;
  82     return unless $self->{hosts}->{$host} && $self->{hosts}->{$host}->{closed};
  83     $self->{hosts}->{$host} = MogileFS::IOStatWatch::Client->new($host, $self);
  84 }
  85
  86 sub got_error {
  87     my ($self, $host) = @_;
  88     Danga::Socket->AddTimer(60, sub {
  89         $self->restart_monitoring_if_needed($host);
  90     });
  91 }
  92
  93 sub got_disconnect {
  94     my ($self, $host) = @_;
  95     $self->{hosts}->{$host} = MogileFS::IOStatWatch::Client->new($host, $self);
  96 }
  97
  98 # Support class that does the communication with individual hosts.
  99 package MogileFS::IOStatWatch::Client;
 100
 101 use strict;
 102 use warnings;
 103
 104 use base 'Danga::Socket';
 105 use fields qw(host watcher buffer active);
 106
 107 sub new {
 108     my MogileFS::IOStatWatch::Client $self = shift;
 109     my $hostspec = shift;
 110     my $watcher = shift;
 111
 112     my $sock = IO::Socket::INET->new(
 113                                      PeerAddr => $hostspec,
 114                                      PeerPort => MogileFS->config("mogstored_stream_port"),
 115                                      Proto    => 'tcp',
 116                                      Blocking => 0,
 117                                      );
 118     return unless $sock;
 119
 120     $self = fields::new($self) unless ref $self;
 121     $self->SUPER::new($sock);
 122     $self->watch_write(1);
 123     $self->watch_read(1);
 124
 125     $self->{watcher} = $watcher;
 126     $self->{buffer} = '';
 127     $self->{host} = $hostspec;
 128
 129     return $self;
 130 }
 131
 132 sub event_write {
 133     my MogileFS::IOStatWatch::Client $self = shift;
 134     $self->{active} = 1;
 135     $self->write("watch\n");
 136     $self->watch_write(0); # I hope I can safely assume that 6 characters will write properly.
 137 }
 138
 139 sub event_read {
 140     my MogileFS::IOStatWatch::Client $self = shift;
 141
 142     my $bref = $self->read(10240);
 143     return $self->close unless defined $bref;
 144
 145     $self->{buffer} .= $$bref;
 146
 147     if ($self->{buffer} =~ m/^ERR\s+(.*?)\s* $ /x) {
 148         # There was an error on the way to watching this machine, close it and stay quiet.
 149         $self->close;
 150     }
 151
 152     # If we can yank off lines till there is one by itself with a . on it, we've gotten a full set of stats.
 153     while ($self->{buffer} =~ s/^(.*?\n)?\.\n//s) {
 154         my %stats;
 155         foreach my $line (split /\n+/, $1) {
 156             next unless $line;
 157             my ($devnum, $util) = split /\s+/, $line;
 158             $stats{$devnum} = $util;
 159         }
 160         $self->{watcher}->got_stats($self->{host}, \%stats);
 161     }
 162 }
 163
 164 sub event_err {
 165     my MogileFS::IOStatWatch::Client $self = shift;
 166     $self->{watcher}->got_error($self->{host});
 167 }
 168
 169 sub event_hup {
 170     my MogileFS::IOStatWatch::Client $self = shift;
 171     $self->{watcher}->got_error($self->{host});
 172 }
 173
 174 sub close {
 175     my MogileFS::IOStatWatch::Client $self = shift;
 176     if ($self->{active}) {
 177         $self->{watcher}->got_disconnect($self->{host});
 178     } else {
 179         $self->{watcher}->got_error($self->{host});
 180     }
 181     $self->SUPER::close(@_);
 182 }
 183 1;
 184