lib/MogileFS/Host.pm

   1 package MogileFS::Host;
   2 use strict;
   3 use warnings;
   4 use MogileFS::Util qw(throw);
   5 use Net::Netmask;
   6 use Carp qw(croak);
   7 use MogileFS::Connection::Mogstored;
   8 use MogileFS::Connection::HTTP;
   9 use MogileFS::ConnectionPool;
  10 our $http_pool;
  11
  12 =head1
  13
  14 MogileFS::Host - host class
  15
  16 =cut
  17
  18 # Centralized here instead of three places.
  19 my @observed_fields = qw/observed_state/;
  20 my @fields = (qw/hostid hostname hostip status http_port http_get_port altip altmask/,
  21     @observed_fields);
  22
  23 # TODO: Validate a few things: state, observed state.
  24 sub new_from_args {
  25     my ($class, $args, $dev_factory) = @_;
  26     my $self = bless {
  27         dev_factory => $dev_factory,
  28         %{$args},
  29     }, $class;
  30
  31     $self->{mask} = ($self->{altip} && $self->{altmask}) ?
  32         Net::Netmask->new2($self->{altmask}) : undef;
  33
  34     return $self;
  35 }
  36
  37 sub valid_state {
  38     my ($class, $state) = @_;
  39     return $state && $state =~ /\A(?:alive|dead|down)\z/;
  40 }
  41
  42 # Instance methods:
  43
  44 sub id        { $_[0]{hostid} }
  45 sub name      { $_[0]{hostname} }
  46 sub hostname  { $_[0]{hostname} }
  47 sub hostip    { $_[0]{hostip} }
  48 sub status    { $_[0]{status} }
  49 sub http_port { $_[0]{http_port} }
  50
  51 sub http_get_port {
  52     return $_[0]->{http_get_port} || $_[0]->{http_port};
  53 }
  54
  55 sub ip {
  56     my $self = shift;
  57     if ($self->{mask} && $self->{altip} &&
  58         ($MogileFS::REQ_altzone || ($MogileFS::REQ_client_ip &&
  59          $self->{mask}->match($MogileFS::REQ_client_ip)))) {
  60         return $self->{altip};
  61     } else {
  62         return $self->{hostip};
  63     }
  64 }
  65
  66 sub fields {
  67     my $self = shift;
  68     my @tofetch = @_ ? @_ : @fields;
  69     return { map { $_ => $self->{$_} } @tofetch };
  70 }
  71
  72 sub observed_fields {
  73     return $_[0]->fields(@observed_fields);
  74 }
  75
  76 sub alive {
  77     return $_[0]->status eq 'alive';
  78 }
  79
  80 sub observed_reachable {
  81     my $self = shift;
  82     return $self->{observed_state} && $self->{observed_state} eq 'reachable';
  83 }
  84
  85 sub observed_unreachable {
  86     my $self = shift;
  87     return $self->{observed_state} && $self->{observed_state} eq 'unreachable';
  88 }
  89
  90 # returns/creates a MogileFS::Connection::Mogstored object to the
  91 # host's mogstored management/side-channel port (which starts
  92 # unconnected, and only connects when you ask it to, with its sock
  93 # method)
  94 sub mogstored_conn {
  95     my $self = shift;
  96     return $self->{mogstored_conn} ||=
  97       MogileFS::Connection::Mogstored->new($self->ip, $self->sidechannel_port);
  98 }
  99
 100 sub sidechannel_port {
 101     # TODO: let this be configurable per-host?  currently it's configured
 102     # once for all machines.
 103     MogileFS->config("mogstored_stream_port");
 104 }
 105
 106 # starts an HTTP request on the given $port with $method to $path
 107 # Calls cb with an HTTP::Response object when done
 108 sub _http_conn {
 109     my ($self, $port, $method, $path, $opts, $cb) = @_;
 110     _init_pools();
 111
 112     $http_pool->start($opts->{ip} || $self->ip, $port, sub {
 113         $_[0]->start($method, $path, $opts, $cb);
 114     });
 115 }
 116
 117 # Returns a ready, blocking HTTP connection
 118 # This is only used by replicate
 119 sub http_conn_get {
 120     my ($self, $opts) = @_;
 121     my $ip = $opts->{ip} || $self->ip;
 122     my $port = $opts->{port} || $self->http_port;
 123
 124     _init_pools();
 125     my $conn = $http_pool->conn_get($ip, $port);
 126     $conn->sock->blocking(1);
 127     return $conn;
 128 }
 129
 130 # Returns a blocking HTTP connection back to the pool.
 131 # This is the inverse of http_conn_get, and should be called when
 132 # done using a connection (iff the connection is really still alive)
 133 # (and makes it non-blocking for future use)
 134 # This is only used by replicate.
 135 sub http_conn_put {
 136     my ($self, $conn) = @_;
 137     $conn->sock->blocking(0);
 138     $http_pool->conn_put($conn);
 139 }
 140
 141 sub http_get {
 142     my ($self, $method, $path, $opts, $cb) = @_;
 143     $opts ||= {};
 144     $self->_http_conn($self->http_get_port, $method, $path, $opts, $cb);
 145 }
 146
 147 sub http {
 148     my ($self, $method, $path, $opts, $cb) = @_;
 149     $opts ||= {};
 150     my $port = delete $opts->{port} || $self->http_port;
 151     $self->_http_conn($port, $method, $path, $opts, $cb);
 152 }
 153
 154 # FIXME - make these customizable
 155 sub _init_pools {
 156     return if $http_pool;
 157
 158     $http_pool = MogileFS::ConnectionPool->new("MogileFS::Connection::HTTP");
 159 }
 160
 161 1;