lib/MogileFS/Server.pm

   1 package MogileFS::Server;
   2 use strict;
   3 use warnings;
   4 use vars qw($VERSION);
   5 $VERSION = "2.57";
   6
   7 =head1 NAME
   8
   9 MogileFS::Server - MogileFS (distributed filesystem) server
  10
  11 =head1 SYNOPSIS
  12
  13  $s = MogileFS::Server->server;
  14  $s->run;
  15
  16 =cut
  17
  18 use IO::Socket;
  19 use Symbol;
  20 use POSIX;
  21 use File::Copy ();
  22 use Carp;
  23 use File::Basename ();
  24 use File::Path ();
  25 use Sys::Syslog ();
  26 use Time::HiRes ();
  27 use Net::Netmask;
  28 use LWP::UserAgent;
  29 use List::Util;
  30 use Socket ();
  31
  32 use MogileFS::Util qw(daemonize);
  33 use MogileFS::Sys;
  34 use MogileFS::Config;
  35
  36 use MogileFS::ProcManager;
  37 use MogileFS::Connection::Client;
  38 use MogileFS::Connection::Worker;
  39
  40 use MogileFS::Worker::Query;
  41 use MogileFS::Worker::Delete;
  42 use MogileFS::Worker::Replicate;
  43 use MogileFS::Worker::Reaper;
  44 use MogileFS::Worker::Monitor;
  45 use MogileFS::Worker::Fsck;
  46 use MogileFS::Worker::JobMaster;
  47
  48 use MogileFS::Factory::Domain;
  49 use MogileFS::Factory::Class;
  50 use MogileFS::Factory::Host;
  51 use MogileFS::Factory::Device;
  52 use MogileFS::Domain;
  53 use MogileFS::Class;
  54 use MogileFS::Host;
  55 use MogileFS::Device;
  56
  57 use MogileFS::HTTPFile;
  58 use MogileFS::FID;
  59 use MogileFS::DevFID;
  60
  61 use MogileFS::Store;
  62
  63 use MogileFS::ReplicationPolicy::MultipleHosts;
  64
  65 my $server; # server singleton
  66 sub server {
  67     my ($pkg) = @_;
  68     return $server ||= bless {}, $pkg;
  69 }
  70
  71 # --------------------------------------------------------------------------
  72 # instance methods:
  73 # --------------------------------------------------------------------------
  74
  75 sub run {
  76     my $self = shift;
  77
  78     MogileFS::Config->load_config;
  79
  80     # don't run as root
  81     die "mogilefsd cannot be run as root\n"
  82         if $< == 0 && MogileFS->config('user') ne "root";
  83
  84     MogileFS::Config->check_database;
  85     daemonize() if MogileFS->config("daemonize");
  86
  87     MogileFS::ProcManager->set_min_workers('monitor'     => 1);
  88
  89     # open up our log
  90     Sys::Syslog::openlog('mogilefsd', 'pid', 'daemon');
  91     Mgd::log('info', 'beginning run');
  92
  93     unless (MogileFS::ProcManager->write_pidfile) {
  94         Mgd::log('info', "Couldn't write pidfile, ending run");
  95         Sys::Syslog::closelog();
  96         exit 1;
  97     }
  98
  99     # Install signal handlers.
 100     $SIG{TERM}  = sub {
 101         my @children = MogileFS::ProcManager->child_pids;
 102         print STDERR scalar @children, " children to kill.\n" if $DEBUG;
 103         my $count = kill( 'TERM' => @children );
 104         print STDERR "Sent SIGTERM to $count children.\n" if $DEBUG;
 105         MogileFS::ProcManager->remove_pidfile;
 106         Mgd::log('info', 'ending run due to SIGTERM');
 107         Sys::Syslog::closelog();
 108
 109         exit 0;
 110     };
 111
 112     $SIG{INT}  = sub {
 113         my @children = MogileFS::ProcManager->child_pids;
 114         print STDERR scalar @children, " children to kill.\n" if $DEBUG;
 115         my $count = kill( 'INT' => @children );
 116         print STDERR "Sent SIGINT to $count children.\n" if $DEBUG;
 117         MogileFS::ProcManager->remove_pidfile;
 118         Mgd::log('info', 'ending run due to SIGINT');
 119         exit 0;
 120     };
 121     $SIG{PIPE} = 'IGNORE';  # catch them by hand
 122
 123     # setup server sockets to listen for client connections
 124     my @servers;
 125     foreach my $listen (@{ MogileFS->config('listen') }) {
 126         my $server = IO::Socket::INET->new(LocalAddr => $listen,
 127                                            Type      => SOCK_STREAM,
 128                                            Proto     => 'tcp',
 129                                            Blocking  => 0,
 130                                            Reuse     => 1,
 131                                            Listen    => 1024 )
 132             or die "Error creating socket: $@\n";
 133
 134         # save sub to accept a client
 135         push @servers, $server;
 136         Danga::Socket->AddOtherFds( fileno($server) => sub {
 137                 while (my $csock = $server->accept) {
 138                     MogileFS::Connection::Client->new($csock);
 139                 }
 140             } );
 141     }
 142
 143     MogileFS::ProcManager->push_pre_fork_cleanup(sub {
 144         # so children don't hold server connection open
 145         close($_) foreach @servers;
 146     });
 147
 148     # setup the post event loop callback to spawn jobs, and the timeout
 149     Danga::Socket->DebugLevel(3);
 150     Danga::Socket->SetLoopTimeout( 250 ); # 250 milliseconds
 151     Danga::Socket->SetPostLoopCallback(MogileFS::ProcManager->PostEventLoopChecker);
 152
 153     # and now, actually start listening for events
 154     eval {
 155         print( "Starting event loop for frontend job on pid $$.\n" ) if $DEBUG;
 156         Danga::Socket->EventLoop();
 157     };
 158
 159     if ($@) {
 160         Mgd::log('err', "crash log: $@");
 161         exit 1;
 162     }
 163     Mgd::log('info', 'ending run');
 164     Sys::Syslog::closelog();
 165     exit(0);
 166 }
 167
 168 # --------------------------------------------------------------------------
 169
 170 package MogileFS;
 171 # just so MogileFS->config($key) will work:
 172 use MogileFS::Config qw(config);
 173
 174 my %hooks;
 175
 176 sub register_worker_command {
 177     # just pass this through to the Worker class
 178     return MogileFS::Worker::Query::register_command(@_);
 179 }
 180
 181 sub register_global_hook {
 182     $hooks{$_[0]} = $_[1];
 183     return 1;
 184 }
 185
 186 sub unregister_global_hook {
 187     delete $hooks{$_[0]};
 188     return 1;
 189 }
 190
 191 sub run_global_hook {
 192     my $hookname = shift;
 193     my $ref = $hooks{$hookname};
 194     return $ref->(@_) if defined $ref;
 195     return undef;
 196 }
 197
 198 # --------------------------------------------------------------------------
 199
 200 package Mgd;  # conveniently short name
 201 use strict;
 202 use warnings;
 203 use MogileFS::Config;
 204 use MogileFS::Util qw(error fatal debug); # for others calling Mgd::foo()
 205
 206 sub server {
 207     return MogileFS::Server->server;
 208 }
 209
 210 # database checking/connecting
 211 sub validate_dbh {
 212     my $sto = Mgd::get_store();
 213     my $had_dbh = $sto->have_dbh;
 214     $sto->recheck_dbh();
 215     my $dbh;
 216     eval { $dbh = $sto->dbh };
 217     # Doesn't matter what the failure was; workers should retry later.
 218     error("Error validating master DB: $@") if $@ && $had_dbh;
 219     return $dbh;
 220 }
 221 sub get_dbh      { return Mgd::get_store()->dbh  }
 222
 223 # the eventual replacement for callers asking for a dbh directly:
 224 # they'll ask for the current store, which is a database abstraction
 225 # layer.
 226 my ($store, $store_pid);
 227 sub get_store {
 228     return $store if $store && $store_pid == $$;
 229     $store_pid = $$;
 230     return $store = MogileFS::Store->new;
 231 }
 232
 233 sub close_store {
 234     if ($store) {
 235         $store->dbh->disconnect();
 236         $store = undef;
 237         return 1;
 238     }
 239     return 0;
 240 }
 241
 242 # only for t/ scripts to explicitly set a store, without loading in a config
 243 sub set_store {
 244     my ($s) = @_;
 245     $store = $s;
 246     $store_pid = $$;
 247 }
 248
 249 sub domain_factory {
 250     return MogileFS::Factory::Domain->get_factory;
 251 }
 252
 253 sub class_factory {
 254     return MogileFS::Factory::Class->get_factory;
 255 }
 256
 257 sub host_factory {
 258     return MogileFS::Factory::Host->get_factory;
 259 }
 260
 261 sub device_factory {
 262     return MogileFS::Factory::Device->get_factory;
 263 }
 264
 265 # log stuff to syslog or the screen
 266 sub log {
 267     # simple logging functionality
 268     if (! $MogileFS::Config::daemonize) {
 269         # syslog acts like printf so we have to use printf and append a \n
 270         shift; # ignore the first parameter (info, warn, critical, etc)
 271         my $mask = shift; # format string
 272         $mask .= "\n" unless $mask =~ /\n$/;
 273         my $message = @_ ? sprintf($mask, @_) : $mask;
 274         print '[', scalar localtime(), '] ', $message;
 275     } else {
 276         # just pass the parameters to syslog
 277         Sys::Syslog::syslog(@_);
 278     }
 279 }
 280
 281 1;
 282 __END__
 283 #Just for MakeMaker's kinda lame regexp for ABSTRACT_FROM
 284 =dummypod
 285 mogilefs::server - MogileFS (distributed filesystem) server.
 286