[svn:qpsmtpd] rev 410 - in branches/high_perf: . lib/Qpsmtpd plugins
[email protected] 28 Apr 2005 21:38:44 -0000
| Newsgroups | perl.cvs.qpsmtpd |
|---|---|
| Message-ID | <[email protected]> |
Author: msergeant
Date: Thu Apr 28 14:38:43 2005
New Revision: 410
Added:
branches/high_perf/lib/Qpsmtpd/Stats.pm
branches/high_perf/plugins/stats
Modified:
branches/high_perf/lib/Qpsmtpd/ConfigServer.pm
branches/high_perf/lib/Qpsmtpd/PollServer.pm
branches/high_perf/qpsmtpd
Log:
Much improved config server, especially the stats
Modified: branches/high_perf/lib/Qpsmtpd/ConfigServer.pm
==============================================================================
--- branches/high_perf/lib/Qpsmtpd/ConfigServer.pm (original)
+++ branches/high_perf/lib/Qpsmtpd/ConfigServer.pm Thu Apr 28 14:38:43 2005
@@ -3,9 +3,11 @@
package Qpsmtpd::ConfigServer;
use base ('Danga::Client');
+use Qpsmtpd::Constants;
+
+use strict;
use fields qw(
- commands
_auth
_commands
_config_cache
@@ -15,16 +17,19 @@
_extras
);
+my $PROMPT = "Enter command: ";
+
sub new {
my Qpsmtpd::ConfigServer $self = shift;
$self = fields::new($self) unless ref $self;
$self->SUPER::new( @_ );
- $self->{commands} = { help => 1, status => 1, };
- $self->write("Enter command:\n");
+ $self->write($PROMPT);
return $self;
}
+sub max_idle_time { 3600 } # one hour
+
sub process_line {
my $self = shift;
my $line = shift || return;
@@ -55,7 +60,8 @@
my $self = shift;
my ($msg) = shift || "program fault - command not performed";
print STDERR "$0 [$$]: $msg ($!)\n";
- return $self->respond("Error - " . $msg, "Enter command:");
+ $self->respond("Error - " . $msg);
+ return $PROMPT;
}
sub _process_line {
@@ -64,18 +70,18 @@
$line =~ s/\r?\n//;
my ($cmd, @params) = split(/ +/, $line);
- my $meth = lc($cmd);
- if (my $lookup = $self->{commands}->{$meth} && $self->can($meth)) {
+ my $meth = "cmd_" . lc($cmd);
+ if (my $lookup = $self->can($meth)) {
my $resp = eval {
$lookup->($self, @params);
};
if ($@) {
my $error = $@;
chomp($error);
- $self->log(LOGERROR, "Command Error: $error");
+ Qpsmtpd->log(LOGERROR, "Command Error: $error");
return $self->fault("command '$cmd' failed unexpectedly");
}
- return $resp . "\nEnter command:\n";
+ return "$resp\n$PROMPT";
}
else {
# No such method - i.e. unrecognized command
@@ -84,24 +90,74 @@
}
my %helptext = (
- all => "Available Commands:\n\nSTATUS\nHELP [CMD]",
+ help => "HELP [CMD] - Get help on all commands or a specific command",
status => "STATUS - Returns status information about current connections",
+ list => "LIST [LIMIT] - List the connections, specify limit or negative limit to shrink list",
+ kill => "KILL (\$IP | \$REF) - Disconnect all connections from \$IP or connection reference \$REF",
+ pause => "PAUSE - Stop accepting new connections",
+ continue => "CONTINUE - Resume accepting connections",
+ reload => "RELOAD - Reload all plugins and config",
+ quit => "QUIT - Exit the config server",
);
-sub help {
+sub cmd_help {
my $self = shift;
my ($subcmd) = @_;
- $subcmd ||= 'all';
+ $subcmd ||= 'help';
$subcmd = lc($subcmd);
- my $txt = $helptext{$subcmd} || "Unrecognised help option. Try 'help all'";
- warn "help returning: $txt\n";
- return $txt . "\n";
+ if ($subcmd eq 'help') {
+ my $txt = join("\n", map { substr($_, 0, index($_, "-")) } sort values(%helptext));
+ return "Available Commands:\n\n$txt\n";
+ }
+ my $txt = $helptext{$subcmd} || "Unrecognised help option. Try 'help' for a full list.";
+ return "$txt\n";
}
-sub status {
+sub cmd_quit {
my $self = shift;
+ $self->close;
+}
+
+sub cmd_pause {
+ my $self = shift;
+
+ my $other_fds = $self->OtherFds;
+
+ $self->{other_fds} = { %$other_fds };
+ %$other_fds = ();
+ return "PAUSED";
+}
+
+sub cmd_status {
+ my $self = shift;
+
+# Status should show:
+# - Total time running
+# - Total number of mails received
+# - Total number of mails rejected (5xx)
+# - Total number of mails tempfailed (5xx)
+# - Avg number of mails/minute
+# - Number of current connections
+# - Number of outstanding DNS queries
+
+ my $output = "Current Status as of " . gmtime() . " GMT\n\n";
+
+ if ($INC{'Qpsmtpd/Stats.pm'}) {
+ # Stats plugin is loaded
+ my $uptime = Qpsmtpd::Stats->uptime;
+ my $recvd = Qpsmtpd::Stats->mails_received;
+ my $reject = Qpsmtpd::Stats->mails_rejected;
+ my $soft = Qpsmtpd::Stats->mails_tempfailed;
+ my $rate = Qpsmtpd::Stats->mails_per_sec;
+ $output .= sprintf(" Uptime: %0.2f sec\n".
+ " Mails Received: % 10d\n".
+ " 5xx: % 10d\n".
+ " 4xx: % 10d\n".
+ "Mails per second: %0.2f\n",
+ $uptime, $recvd, $reject, $soft, $rate);
+ }
my $descriptors = Danga::Socket->DescriptorMap;
@@ -117,9 +173,99 @@
}
}
- return
-" Current Connections: $current_connections
- Current DNS Queries: $current_dns";
+ $output .= "Curr Connections: $current_connections\n".
+ "Curr DNS Queries: $current_dns";
+
+ return $output;
+}
+
+sub cmd_list {
+ my $self = shift;
+ my ($count) = @_;
+
+ my $descriptors = Danga::Socket->DescriptorMap;
+
+ my $list = "Current" . ($count ? (($count > 0) ? " Oldest $count" : " Newest ".-$count) : "") . " Connections: \n\n";
+ my @all;
+ foreach my $fd (keys %$descriptors) {
+ my $pob = $descriptors->{$fd};
+ if ($pob->isa("Qpsmtpd::PollServer")) {
+ next unless $pob->connection->remote_ip; # haven't even started yet
+ push @all, [$pob+0, $pob->connection->remote_ip,
+ $pob->connection->remote_host, $pob->uptime];
+ }
+ }
+
+ @all = sort { $a->[3] <=> $b->[3] } @all;
+ if ($count) {
+ if ($count > 0) {
+ @all = @all[$#all-($count-1) .. $#all];
+ }
+ else {
+ @all = @all[0..(abs($count) - 1)];
+ }
+ }
+ foreach my $item (@all) {
+ $list .= sprintf("%x : %s [%s] Connected %0.2fs\n", @$item);
+ }
+
+ return $list;
+}
+
+sub cmd_kill {
+ my $self = shift;
+ my ($match) = @_;
+
+ return "SYNTAX: KILL (\$IP | \$REF)\n" unless $match;
+
+ my $descriptors = Danga::Socket->DescriptorMap;
+
+ my $killed = 0;
+ my $is_ip = (index($match, '.') >= 0);
+ foreach my $fd (keys %$descriptors) {
+ my $pob = $descriptors->{$fd};
+ if ($pob->isa("Qpsmtpd::PollServer")) {
+ if ($is_ip) {
+ next unless $pob->connection->remote_ip; # haven't even started yet
+ if ($pob->connection->remote_ip eq $match) {
+ $pob->write("550 Your connection has been killed by an administrator\r\n");
+ $pob->disconnect;
+ $killed++;
+ }
+ }
+ else {
+ # match by ID
+ if ($pob+0 == hex($match)) {
+ $pob->write("550 Your connection has been killed by an administrator\r\n");
+ $pob->disconnect;
+ $killed++;
+ }
+ }
+ }
+ }
+
+ return "Killed $killed connection" . ($killed > 1 ? "s" : "") . "\n";
+}
+
+sub cmd_dump {
+ my $self = shift;
+ my ($ref) = @_;
+
+ return "SYNTAX: DUMP \$REF\n" unless $ref;
+ require Data::Dumper;
+ $Data::Dumper::Indent=1;
+
+ my $descriptors = Danga::Socket->DescriptorMap;
+ foreach my $fd (keys %$descriptors) {
+ my $pob = $descriptors->{$fd};
+ if ($pob->isa("Qpsmtpd::PollServer")) {
+ if ($pob+0 == hex($ref)) {
+ return Data::Dumper::Dumper($pob);
+ }
+ }
+ }
+
+ return "Unable to find the connection: $ref. Try the LIST command\n";
}
1;
Modified: branches/high_perf/lib/Qpsmtpd/PollServer.pm
==============================================================================
--- branches/high_perf/lib/Qpsmtpd/PollServer.pm (original)
+++ branches/high_perf/lib/Qpsmtpd/PollServer.pm Thu Apr 28 14:38:43 2005
@@ -13,6 +13,7 @@
data_size
max_size
hooks
+ start_time
_auth
_commands
_config_cache
@@ -28,6 +29,7 @@
use Mail::Header;
use POSIX qw(strftime);
use Socket qw(inet_aton AF_INET CRLF);
+use Time::HiRes qw(time);
use strict;
sub input_sock {
@@ -41,10 +43,17 @@
$self = fields::new($self) unless ref $self;
$self->SUPER::new( @_ );
+ $self->{start_time} = time;
$self->load_plugins;
return $self;
}
+sub uptime {
+ my Qpsmtpd::PollServer $self = shift;
+
+ return (time() - $self->{start_time});
+}
+
sub reset_for_next_message {
my $self = shift;
$self->SUPER::reset_for_next_message(@_);
Added: branches/high_perf/lib/Qpsmtpd/Stats.pm
==============================================================================
--- (empty file)
+++ branches/high_perf/lib/Qpsmtpd/Stats.pm Thu Apr 28 14:38:43 2005
@@ -0,0 +1,35 @@
+# $Id$
+
+package Qpsmtpd::Stats;
+
+use strict;
+use Qpsmtpd;
+use Qpsmtpd::Constants;
+use Time::HiRes qw(time);
+
+my $START_TIME = time;
+our $MAILS_RECEIVED = 0;
+our $MAILS_REJECTED = 0;
+our $MAILS_TEMPFAIL = 0;
+
+sub uptime {
+ return (time() - $START_TIME);
+}
+
+sub mails_received {
+ return $MAILS_RECEIVED;
+}
+
+sub mails_rejected {
+ return $MAILS_REJECTED;
+}
+
+sub mails_tempfailed {
+ return $MAILS_TEMPFAIL;
+}
+
+sub mails_per_sec {
+ return ($MAILS_RECEIVED / uptime());
+}
+
+1;
\ No newline at end of file
Added: branches/high_perf/plugins/stats
==============================================================================
--- (empty file)
+++ branches/high_perf/plugins/stats Thu Apr 28 14:38:43 2005
@@ -0,0 +1,31 @@
+#!/usr/bin/perl -w
+
+use Qpsmtpd::Stats;
+
+sub register {
+ my ($self) = @_;
+
+ $self->register_hook('deny', 'increment_deny');
+ $self->register_hook('queue', 'increment_mails');
+}
+
+sub increment_deny {
+ my ($self, $level) = @_;
+
+ if ($level == DENY or $level == DENY_DISCONNECT) {
+ $Qpsmtpd::Stats::MAILS_REJECTED++;
+ }
+ elsif ($level == DENYSOFT or $level == DENYSOFT_DISCONNECT) {
+ $Qpsmtpd::Stats::MAILS_TEMPFAIL++;
+ }
+
+ return DECLINED;
+}
+
+sub increment_mails {
+ my $self = shift;
+
+ $Qpsmtpd::Stats::MAILS_RECEIVED++;
+
+ return DECLINED;
+}
\ No newline at end of file
Modified: branches/high_perf/qpsmtpd
==============================================================================
--- branches/high_perf/qpsmtpd (original)
+++ branches/high_perf/qpsmtpd Thu Apr 28 14:38:43 2005
@@ -43,6 +43,7 @@
my $MAXCONN = 15; # max simultaneous connections
my $USER = 'smtpd'; # user to suid to
my $MAXCONNIP = 5; # max simultaneous connections from one IP
+my $PAUSED = 0;
sub help {
print <<EOT;
@@ -97,8 +98,8 @@
my $POLL = "with " . ($Danga::Socket::HaveEpoll ? "epoll()" :
$Danga::Socket::HaveKQueue ? "kqueue()" : "poll()");
-my $server;
-my $config_server;
+my $SERVER;
+my $CONFIG_SERVER;
# Code for inetd/tcpserver mode
if ($ENV{REMOTE_HOST} or $ENV{TCPREMOTEHOST}) {
@@ -143,7 +144,7 @@
$SIG{CHLD} = "DEFAULT";
- Qpsmtpd::PollServer->OtherFds(fileno($server) => \&accept_handler);
+ Qpsmtpd::PollServer->OtherFds(fileno($SERVER) => \&accept_handler);
Qpsmtpd::PollServer->EventLoop();
exit;
}
@@ -199,7 +200,7 @@
sub run_as_server {
# establish SERVER socket, bind and listen.
- $server = IO::Socket::INET->new(LocalPort => $PORT,
+ $SERVER = IO::Socket::INET->new(LocalPort => $PORT,
LocalAddr => $LOCALADDR,
Type => SOCK_STREAM,
Proto => IPPROTO_TCP,
@@ -208,10 +209,10 @@
Listen => 10 )
or die "Error creating server $LOCALADDR:$PORT : $@\n";
- IO::Handle::blocking($server, 0);
- binmode($server, ':raw');
+ IO::Handle::blocking($SERVER, 0);
+ binmode($SERVER, ':raw');
- $config_server = IO::Socket::INET->new(LocalPort => $CONFIG_PORT,
+ $CONFIG_SERVER = IO::Socket::INET->new(LocalPort => $CONFIG_PORT,
LocalAddr => $CONFIG_LOCALADDR,
Type => SOCK_STREAM,
Proto => IPPROTO_TCP,
@@ -220,8 +221,8 @@
Listen => 1 )
or die "Error creating server $CONFIG_LOCALADDR:$CONFIG_PORT : $@\n";
- IO::Handle::blocking($config_server, 0);
- binmode($config_server, ':raw');
+ IO::Handle::blocking($CONFIG_SERVER, 0);
+ binmode($CONFIG_SERVER, ':raw');
# Drop priviledges
my (undef, undef, $quid, $qgid) = getpwnam $USER or
@@ -258,8 +259,8 @@
}
::log(LOGDEBUG, "Listening on $PORT with single process $POLL" .
($LineMode ? " (forking server)" : ""));
- Qpsmtpd::PollServer->OtherFds(fileno($server) => \&accept_handler,
- fileno($config_server) => \&config_handler,
+ Qpsmtpd::PollServer->OtherFds(fileno($SERVER) => \&accept_handler,
+ fileno($CONFIG_SERVER) => \&config_handler,
);
while (1) {
Qpsmtpd::PollServer->EventLoop();
@@ -270,7 +271,7 @@
}
sub config_handler {
- my $csock = $config_server->accept();
+ my $csock = $CONFIG_SERVER->accept();
if (!$csock) {
warn("accept failed on config server: $!");
return;
@@ -295,7 +296,7 @@
return;
}
- my $csock = $server->accept();
+ my $csock = $SERVER->accept();
if (!$csock) {
# warn("accept() failed: $!");
return;
@@ -313,6 +314,12 @@
my $client = Qpsmtpd::PollServer->new($csock);
my $rem_ip = $client->peer_ip_string;
+ if ($PAUSED) {
+ $client->write("451 Sorry, this server is currently paused\r\n");
+ $client->close;
+ return;
+ }
+
if ($MAXCONNIP) {
my $num_conn = 1; # seed with current value
@@ -370,7 +377,7 @@
return $csock->close();
}
- $server->close(); # make sure the child doesn't accept() new connections
+ $SERVER->close(); # make sure the child doesn't accept() new connections
$SIG{$_} = 'DEFAULT' for keys %SIG;
@@ -406,3 +413,7 @@
warn("$$ $message\n");
}
+sub pause {
+ my ($pause) = @_;
+ $PAUSED = $pause;
+}