lib/MogileFS/Server.pm

   1 package MogileFS::Server;
   2 use strict;
   3 use warnings;
   4 use vars qw($VERSION);
   5 $VERSION = "2.62";
   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 qw(SO_KEEPALIVE);
  31
  32 use MogileFS::Util qw(daemonize);
  33 use MogileFS::Config;
  34
  35 use MogileFS::ProcManager;
  36 use MogileFS::Connection::Client;
  37 use MogileFS::Connection::Worker;
  38
  39 use MogileFS::Worker::Query;
  40 use MogileFS::Worker::Delete;
  41 use MogileFS::Worker::Replicate;
  42 use MogileFS::Worker::Reaper;
  43 use MogileFS::Worker::Monitor;
  44 use MogileFS::Worker::Fsck;
  45 use MogileFS::Worker::JobMaster;
  46
  47 use MogileFS::Factory::Domain;
  48 use MogileFS::Factory::Class;
  49 use MogileFS::Factory::Host;
  50 use MogileFS::Factory::Device;
  51 use MogileFS::Domain;
  52 use MogileFS::Class;
  53 use MogileFS::Host;
  54 use MogileFS::Device;
  55
  56 use MogileFS::HTTPFile;
  57 use MogileFS::FID;
  58 use MogileFS::DevFID;
  59
  60 use MogileFS::Store;
  61
  62 use MogileFS::ReplicationPolicy::MultipleHosts;
  63
  64 my $server; # server singleton
  65 sub server {
  66     my ($pkg) = @_;
  67     return $server ||= bless {}, $pkg;
  68 }
  69
  70 # --------------------------------------------------------------------------
  71 # instance methods:
  72 # --------------------------------------------------------------------------
  73
  74 sub run {
  75     my $self = shift;
  76
  77     MogileFS::Config->load_config;
  78
  79     # don't run as root
  80     die "mogilefsd cannot be run as root\n"
  81         if $< == 0 && MogileFS->config('user') ne "root";
  82
  83     MogileFS::Config->check_database;
  84     daemonize() if MogileFS->config("daemonize");
  85
  86     MogileFS::ProcManager->set_min_workers('monitor'     => 1);
  87
  88     # open up our log
  89     Sys::Syslog::openlog('mogilefsd', 'pid', 'daemon');
  90     Mgd::log('info', 'beginning run');
  91
  92     unless (MogileFS::ProcManager->write_pidfile) {
  93         Mgd::log('info', "Couldn't write pidfile, ending run");
  94         Sys::Syslog::closelog();
  95         exit 1;
  96     }
  97
  98     # Install signal handlers.
  99     $SIG{TERM}  = sub {
 100         my @children = MogileFS::ProcManager->child_pids;
 101         print STDERR scalar @children, " children to kill.\n" if $DEBUG;
 102         my $count = kill( 'TERM' => @children );
 103         print STDERR "Sent SIGTERM to $count children.\n" if $DEBUG;
 104         MogileFS::ProcManager->remove_pidfile;
 105         Mgd::log('info', 'ending run due to SIGTERM');
 106         Sys::Syslog::closelog();
 107
 108         exit 0;
 109     };
 110
 111     $SIG{INT}  = sub {
 112         my @children = MogileFS::ProcManager->child_pids;
 113         print STDERR scalar @children, " children to kill.\n" if $DEBUG;
 114         my $count = kill( 'INT' => @children );
 115         print STDERR "Sent SIGINT to $count children.\n" if $DEBUG;
 116         MogileFS::ProcManager->remove_pidfile;
 117         Mgd::log('info', 'ending run due to SIGINT');
 118         exit 0;
 119     };
 120     $SIG{PIPE} = 'IGNORE';  # catch them by hand
 121
 122     # setup server sockets to listen for client connections
 123     my @servers;
 124     foreach my $listen (@{ MogileFS->config('listen') }) {
 125         my $server = IO::Socket::INET->new(LocalAddr => $listen,
 126                                            Type      => SOCK_STREAM,
 127                                            Proto     => 'tcp',
 128                                            Blocking  => 0,
 129                                            Reuse     => 1,
 130                                            Listen    => 1024 )
 131             or die "Error creating socket: $@\n";
 132         $server->sockopt(SO_KEEPALIVE, 1);
 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         $| = 1;
 270         # syslog acts like printf so we have to use printf and append a \n
 271         shift; # ignore the first parameter (info, warn, critical, etc)
 272         my $mask = shift; # format string
 273         $mask .= "\n" unless $mask =~ /\n$/;
 274         my $message = @_ ? sprintf($mask, @_) : $mask;
 275         print '[', scalar localtime(), '] ', $message;
 276     } else {
 277         # just pass the parameters to syslog
 278         Sys::Syslog::syslog(@_);
 279     }
 280 }
 281
 282 1;
 283 __END__
 284 #Just for MakeMaker's kinda lame regexp for ABSTRACT_FROM
 285 =dummypod
 286 mogilefs::server - MogileFS (distributed filesystem) server.
 287