lib/MogileFS/Server.pm

   1 package MogileFS::Server;
   2 use strict;
   3 use warnings;
   4 use vars qw($VERSION);
   5 $VERSION = "2.59";
   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::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         $server->sockopt(SO_KEEPALIVE, 1);
 134
 135         # save sub to accept a client
 136         push @servers, $server;
 137         Danga::Socket->AddOtherFds( fileno($server) => sub {
 138                 while (my $csock = $server->accept) {
 139                     MogileFS::Connection::Client->new($csock);
 140                 }
 141             } );
 142     }
 143
 144     MogileFS::ProcManager->push_pre_fork_cleanup(sub {
 145         # so children don't hold server connection open
 146         close($_) foreach @servers;
 147     });
 148
 149     # setup the post event loop callback to spawn jobs, and the timeout
 150     Danga::Socket->DebugLevel(3);
 151     Danga::Socket->SetLoopTimeout( 250 ); # 250 milliseconds
 152     Danga::Socket->SetPostLoopCallback(MogileFS::ProcManager->PostEventLoopChecker);
 153
 154     # and now, actually start listening for events
 155     eval {
 156         print( "Starting event loop for frontend job on pid $$.\n" ) if $DEBUG;
 157         Danga::Socket->EventLoop();
 158     };
 159
 160     if ($@) {
 161         Mgd::log('err', "crash log: $@");
 162         exit 1;
 163     }
 164     Mgd::log('info', 'ending run');
 165     Sys::Syslog::closelog();
 166     exit(0);
 167 }
 168
 169 # --------------------------------------------------------------------------
 170
 171 package MogileFS;
 172 # just so MogileFS->config($key) will work:
 173 use MogileFS::Config qw(config);
 174
 175 my %hooks;
 176
 177 sub register_worker_command {
 178     # just pass this through to the Worker class
 179     return MogileFS::Worker::Query::register_command(@_);
 180 }
 181
 182 sub register_global_hook {
 183     $hooks{$_[0]} = $_[1];
 184     return 1;
 185 }
 186
 187 sub unregister_global_hook {
 188     delete $hooks{$_[0]};
 189     return 1;
 190 }
 191
 192 sub run_global_hook {
 193     my $hookname = shift;
 194     my $ref = $hooks{$hookname};
 195     return $ref->(@_) if defined $ref;
 196     return undef;
 197 }
 198
 199 # --------------------------------------------------------------------------
 200
 201 package Mgd;  # conveniently short name
 202 use strict;
 203 use warnings;
 204 use MogileFS::Config;
 205 use MogileFS::Util qw(error fatal debug); # for others calling Mgd::foo()
 206
 207 sub server {
 208     return MogileFS::Server->server;
 209 }
 210
 211 # database checking/connecting
 212 sub validate_dbh {
 213     my $sto = Mgd::get_store();
 214     my $had_dbh = $sto->have_dbh;
 215     $sto->recheck_dbh();
 216     my $dbh;
 217     eval { $dbh = $sto->dbh };
 218     # Doesn't matter what the failure was; workers should retry later.
 219     error("Error validating master DB: $@") if $@ && $had_dbh;
 220     return $dbh;
 221 }
 222 sub get_dbh      { return Mgd::get_store()->dbh  }
 223
 224 # the eventual replacement for callers asking for a dbh directly:
 225 # they'll ask for the current store, which is a database abstraction
 226 # layer.
 227 my ($store, $store_pid);
 228 sub get_store {
 229     return $store if $store && $store_pid == $$;
 230     $store_pid = $$;
 231     return $store = MogileFS::Store->new;
 232 }
 233
 234 sub close_store {
 235     if ($store) {
 236         $store->dbh->disconnect();
 237         $store = undef;
 238         return 1;
 239     }
 240     return 0;
 241 }
 242
 243 # only for t/ scripts to explicitly set a store, without loading in a config
 244 sub set_store {
 245     my ($s) = @_;
 246     $store = $s;
 247     $store_pid = $$;
 248 }
 249
 250 sub domain_factory {
 251     return MogileFS::Factory::Domain->get_factory;
 252 }
 253
 254 sub class_factory {
 255     return MogileFS::Factory::Class->get_factory;
 256 }
 257
 258 sub host_factory {
 259     return MogileFS::Factory::Host->get_factory;
 260 }
 261
 262 sub device_factory {
 263     return MogileFS::Factory::Device->get_factory;
 264 }
 265
 266 # log stuff to syslog or the screen
 267 sub log {
 268     # simple logging functionality
 269     if (! $MogileFS::Config::daemonize) {
 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