[svn:qpsmtpd] r555 - trunk/lib/Danga/DNS
[email protected] 26 Oct 2005 19:09:04 -0000
| Newsgroups | perl.cvs.qpsmtpd |
|---|---|
| Message-ID | <[email protected]> |
Author: msergeant
Date: Wed Oct 26 12:09:04 2005
New Revision: 555
Modified:
trunk/lib/Danga/DNS/Resolver.pm
Log:
Support all resolvers in resolv.conf, and issue retries on errors the same
way gethostbyname() does.
Modified: trunk/lib/Danga/DNS/Resolver.pm
==============================================================================
--- trunk/lib/Danga/DNS/Resolver.pm (original)
+++ trunk/lib/Danga/DNS/Resolver.pm Wed Oct 26 12:09:04 2005
@@ -3,7 +3,7 @@
package Danga::DNS::Resolver;
use base qw(Danga::Socket);
-use fields qw(res dst id_to_asker id_to_query timeout cache cache_timeout);
+use fields qw(res dst cache cache_timeout queries);
use Net::DNS;
use Socket;
@@ -30,16 +30,16 @@ sub new {
) || die "Cannot create socket: $!";
IO::Handle::blocking($sock, 0);
- trace(2, "Using nameserver $res->{nameservers}->[0]:$res->{port}\n");
- my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton($res->{'nameservers'}->[0]));
- #my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton('127.0.0.1'));
- #my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton('10.2.1.20'));
+ $self->{dst} = [];
+
+ foreach my $ns (@{ $res->{nameservers} }) {
+ trace(2, "Using nameserver $ns:$res->{port}\n");
+ my $dst_sockaddr = sockaddr_in($res->{'port'}, inet_aton($ns));
+ push @{$self->{dst}}, $dst_sockaddr;
+ }
$self->{res} = $res;
- $self->{dst} = $dst_sockaddr;
- $self->{id_to_asker} = {};
- $self->{id_to_query} = {};
- $self->{timeout} = {};
+ $self->{queries} = {};
$self->{cache} = {};
$self->{cache_timeout} = {};
@@ -52,10 +52,17 @@ sub new {
return $self;
}
+sub ns {
+ my Danga::DNS::Resolver $self = shift;
+ my $index = shift;
+ return if $index > $#{$self->{dst}};
+ return $self->{dst}->[$index];
+}
+
sub pending {
my Danga::DNS::Resolver $self = shift;
- return keys(%{$self->{id_to_asker}});
+ return keys(%{$self->{queries}});
}
sub _query {
@@ -73,20 +80,14 @@ sub _query {
}
my $packet = $self->{res}->make_query_packet($host, $type);
- my $packet_data = $packet->data;
- my $h = $packet->header;
- my $id = $h->id;
-
- if (!$self->sock->send($packet_data, 0, $self->{dst})) {
- return;
- }
+ my $packet_data = $packet->data;
+ my $id = $packet->header->id;
- trace(2, "Query: $host ($id)\n");
-
- $self->{id_to_asker}->{$id} = $asker;
- $self->{id_to_query}->{$id} = $host;
- $self->{timeout}->{$id} = $now;
+ my $query = Danga::DNS::Resolver::Query->new(
+ $self, $asker, $host, $type, $now, $id, $packet_data,
+ ) or return;
+ $self->{queries}->{$id} = $query;
return 1;
}
@@ -97,15 +98,12 @@ sub query_txt {
my $now = time();
- trace(2, "[" . keys(%{$self->{id_to_asker}}) . "] trying to resolve TXT: @hosts\n");
+ trace(2, "trying to resolve TXT: @hosts\n");
foreach my $host (@hosts) {
$self->_query($asker, $host, 'TXT', $now) || return;
}
- #print "+Pending queries: " . keys(%{$self->{id_to_asker}}) .
- # " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
-
return 1;
}
@@ -115,15 +113,12 @@ sub query_mx {
my $now = time();
- trace(2, "[" . keys(%{$self->{id_to_asker}}) . "] trying to resolve MX: @hosts\n");
+ trace(2, "trying to resolve MX: @hosts\n");
foreach my $host (@hosts) {
$self->_query($asker, $host, 'MX', $now) || return;
}
- #print "+Pending queries: " . keys(%{$self->{id_to_asker}}) .
- # " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
-
return 1;
}
@@ -133,15 +128,12 @@ sub query {
my $now = time();
- trace(2, "[" . keys(%{$self->{id_to_asker}}) . "] trying to resolve A/PTR: @hosts\n");
+ trace(2, "trying to resolve A/PTR: @hosts\n");
foreach my $host (@hosts) {
$self->_query($asker, $host, 'A', $now) || return;
}
- #print "+Pending queries: " . keys(%{$self->{id_to_asker}}) .
- # " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
-
return 1;
}
@@ -154,20 +146,20 @@ sub _do_cleanup {
my $idle = $self->max_idle_time;
my @to_delete;
- while (my ($id, $t) = each(%{$self->{timeout}})) {
- if ($t < ($now - $idle)) {
+ while (my ($id, $obj) = each(%{$self->{queries}})) {
+ if ($obj->{timeout} < ($now - $idle)) {
push @to_delete, $id;
}
}
foreach my $id (@to_delete) {
- delete $self->{timeout}{$id};
- my $asker = delete $self->{id_to_asker}{$id};
- my $query = delete $self->{id_to_query}{$id};
- $asker->run_callback("NXDOMAIN", $query);
+ my $query = delete $self->{queries}{$id};
+ $query->timeout() and next;
+ # add back in if timeout caused us to loop to next server
+ $self->{queries}->{$id} = $query;
}
- foreach my $type ('A', 'TXT') {
+ foreach my $type ('A', 'TXT', 'MX') {
@to_delete = ();
while (my ($query, $t) = each(%{$self->{cache_timeout}{$type}})) {
@@ -199,17 +191,14 @@ sub event_read {
my $header = $packet->header;
my $id = $header->id;
- my $asker = delete $self->{id_to_asker}->{$id};
- my $query = delete $self->{id_to_query}->{$id};
- delete $self->{timeout}{$id};
-
- #print "-Pending queries: " . keys(%{$self->{id_to_asker}}) .
- # " / Cache Size: " . keys(%{$self->{cache}}) . "\n";
- if (!$asker) {
- trace(1, "No asker for id: $id\n");
+ my $qobj = delete $self->{queries}->{$id};
+ if (!$qobj) {
+ trace(1, "No query for id: $id\n");
return;
}
+ my $query = $qobj->{host};
+
my $now = time();
my @questions = $packet->question;
#print STDERR "response to ", $questions[0]->string, "\n";
@@ -217,61 +206,64 @@ sub event_read {
# my $q = shift @questions;
if ($rr->type eq "PTR") {
my $rdns = $rr->ptrdname;
- if ($query) {
- # NB: Cached as an "A" lookup as there's no overlap and they
- # go through the same query() function above
- $self->{cache}{A}{$query} = $rdns;
- $self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
- }
- $asker->run_callback($rdns, $query);
+ # NB: Cached as an "A" lookup as there's no overlap and they
+ # go through the same query() function above
+ $self->{cache}{A}{$query} = $rdns;
+ # $self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
+ $self->{cache_timeout}{A}{$query} = $now + $rr->ttl;
+ $qobj->run_callback($rdns);
}
elsif ($rr->type eq "A") {
my $ip = $rr->address;
- if ($query) {
- $self->{cache}{A}{$query} = $ip;
- $self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
- }
- $asker->run_callback($ip, $query);
+ $self->{cache}{A}{$query} = $ip;
+ # $self->{cache_timeout}{A}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
+ $self->{cache_timeout}{A}{$query} = $now + $rr->ttl;
+ $qobj->run_callback($ip);
}
elsif ($rr->type eq "TXT") {
my $txt = $rr->txtdata;
- if ($query) {
- $self->{cache}{TXT}{$query} = $txt;
- $self->{cache_timeout}{TXT}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
- }
- $asker->run_callback($txt, $query);
+ $self->{cache}{TXT}{$query} = $txt;
+ # $self->{cache_timeout}{TXT}{$query} = $now + 60; # should use $rr->ttl but that would cache for too long
+ $self->{cache_timeout}{TXT}{$query} = $now + $rr->ttl;
+ $qobj->run_callback($txt);
+ }
+ elsif ($rr->type eq "MX") {
+ my $host = $rr->exchange;
+ my $preference = $rr->preference;
+ $self->{cache}{MX}{$query} = [$host, $preference];
+ $self->{cache_timeout}{MX}{$query} = $now + $rr->ttl;
+ $qobj->run_callback([$host, $preference]);
}
else {
# came back, but not a PTR or A record
- $asker->run_callback("unknown", $query);
+ $qobj->run_callback("UNKNOWN");
}
$answers++;
}
if (!$answers) {
if ($err eq "NXDOMAIN") {
# trace("found => NXDOMAIN\n");
- $asker->run_callback("NXDOMAIN", $query);
+ $qobj->run_callback("NXDOMAIN");
}
elsif ($err eq "SERVFAIL") {
# try again???
- print "SERVFAIL looking for $query (Pending: " . keys(%{$self->{id_to_asker}}) . ")\n";
+ print "SERVFAIL looking for $query\n";
#$self->query($asker, $query);
- $asker->run_callback($err, $query);
- #$self->{id_to_asker}->{$id} = $asker;
- #$self->{id_to_query}->{$id} = $query;
- #$self->{timeout}{$id} = time();
-
+ $qobj->error($err) and next;
+ # add back in if error() resulted in query being re-issued
+ $self->{queries}->{$id} = $qobj;
}
elsif ($err eq "NOERROR") {
- $asker->run_callback($err, $query);
+ $qobj->run_callback($err);
}
elsif($err) {
print("error: $err\n");
- $asker->run_callback($err, $query);
+ $qobj->error($err) and next;
+ $self->{queries}->{$id} = $qobj;
}
else {
# trace("no answers\n");
- $asker->run_callback("NXDOMAIN", $query);
+ $qobj->run_callback("NOANSWER");
}
}
}
@@ -286,6 +278,116 @@ sub close {
# confess "Danga::DNS::Resolver socket should never be closed!";
}
+package Danga::DNS::Resolver::Query;
+
+use constant MAX_QUERIES => 10;
+
+sub trace {
+ my $level = shift;
+ print ("$::DEBUG/$level [$$] dns lookup: @_") if $::DEBUG >= $level;
+}
+
+sub new {
+ my ($class, $res, $asker, $host, $type, $now, $id, $data) = @_;
+
+ my $self = {
+ resolver => $res,
+ asker => $asker,
+ host => $host,
+ type => $type,
+ timeout => $now,
+ id => $id,
+ data => $data,
+ repeat => 2, # number of retries
+ ns => 0,
+ nqueries => 0,
+ };
+
+ trace(2, "NS Query: $host ($id)\n");
+
+ bless $self, $class;
+
+ $self->send_query || return;
+
+ return $self;
+}
+
+#sub DESTROY {
+# my $self = shift;
+# trace(2, "DESTROY $self\n");
+#}
+
+sub timeout {
+ my $self = shift;
+
+ trace(2, "NS Query timeout. Trying next host\n");
+ if ($self->send_query) {
+ # had another NS to send to, reset timeout
+ $self->{timeout} = time();
+ return;
+ }
+
+ # can we loop/repeat?
+ if (($self->{nqueries} <= MAX_QUERIES) &&
+ ($self->{repeat} > 1))
+ {
+ trace(2, "NS Query timeout. Next host failed. Trying loop\n");
+ $self->{repeat}--;
+ $self->{ns} = 0;
+ return $self->timeout();
+ }
+
+ trace(2, "NS Query timeout. All failed. Running callback(TIMEOUT)\n");
+ # otherwise we really must timeout.
+ $self->run_callback("TIMEOUT");
+ return 1;
+}
+
+sub error {
+ my ($self, $error) = @_;
+
+ trace(2, "NS Query error. Trying next host\n");
+ if ($self->send_query) {
+ # had another NS to send to, reset timeout
+ $self->{timeout} = time();
+ return;
+ }
+
+ # can we loop/repeat?
+ if (($self->{nqueries} <= MAX_QUERIES) &&
+ ($self->{repeat} > 1))
+ {
+ trace(2, "NS Query error. Next host failed. Trying loop\n");
+ $self->{repeat}--;
+ $self->{ns} = 0;
+ return $self->error($error);
+ }
+
+ trace(2, "NS Query error. All failed. Running callback($error)\n");
+ # otherwise we really must timeout.
+ $self->run_callback($error);
+ return 1;
+}
+
+sub run_callback {
+ my ($self, $response) = @_;
+ trace(2, "NS Query callback($self->{host} = $response\n");
+ $self->{asker}->run_callback($response, $self->{host});
+}
+
+sub send_query {
+ my ($self) = @_;
+
+ my $dst = $self->{resolver}->ns($self->{ns}++);
+ return unless defined $dst;
+ if (!$self->{resolver}->sock->send($self->{data}, 0, $dst)) {
+ return;
+ }
+
+ $self->{nqueries}++;
+ return 1;
+}
+
1;
=head1 NAME