Reduce number of concurrent DNS lookups in Machine Tracker

changelog.d/+2669.fixed.md

@@ -0,0 +1 @@
+Fix Machine Tracker DNS search crashing from exhausting all available file descriptors

python/nav/asyncdns.py

@@ -33,7 +33,7 @@
 from IPy import IP
 from twisted.names import dns
 from twisted.names import client
-from twisted.internet import defer
+from twisted.internet import defer, task
 
 # pylint: disable=E1101
 from twisted.internet import reactor
@@ -46,6 +46,9 @@
 from twisted.names.error import DNSNotImplementedError, DNSQueryRefusedError
 
 
+BATCH_SIZE = 100
+
+
 def reverse_lookup(addresses):
     """Runs parallel reverse DNS lookups for addresses.
 
@@ -75,21 +78,31 @@
         )
         self.results = defaultdict(list)
         self._finished = False
+        self._errors = []
 
     def resolve(self, names):
         """Resolves DNS names in parallel"""
-        self._finished = False
         self.results = defaultdict(list)
-
-        deferred_list = []
-        for name in names:
-            for deferred in self.lookup(name):
-                deferred.addCallback(self._extract_records, name)
-                deferred.addErrback(self._errback, name)
-                deferred_list.append(deferred)
-
-        deferred_list = defer.DeferredList(deferred_list)
-        deferred_list.addCallback(self._parse_result)
+        self._finished = False
+        self._errors = []
+
+        def lookup_names():
+            for name in names:
+                for deferred in self.lookup(name):
+                    deferred.addCallback(self._extract_records, name)
+                    deferred.addErrback(self._errback, name)
+                    deferred.addCallback(self._save_result)
+                    yield deferred
+
+        # Limits the number of parallel requests to BATCH_SIZE
+        coop = task.Cooperator()
+        work = lookup_names()
+        deferred_list = defer.DeferredList(
+            [
+                coop.coiterate(work).addErrback(self._save_error)
+                for _ in range(BATCH_SIZE)
+            ]
+        )
         deferred_list.addCallback(self._finish)
 
         while not self._finished:
@@ -98,6 +111,10 @@
         # iteration to ensure the resolver library closes its UDP sockets
         reactor.iterate()
 
+        # raise first error if any occurred
+        for error in self._errors:
+            raise error
+
         return dict(self.results)
 
     def lookup(self, name):
@@ -108,19 +125,22 @@
     def _extract_records(result, name):
         raise NotImplementedError
 
-    def _parse_result(self, result):
-        """Parses the result to the correct format"""
-        for _success, (name, response) in result:
-            if isinstance(response, Exception):
-                self.results[name] = response
-            else:
-                self.results[name].extend(response)
+    def _save_result(self, result):
+        name, response = result
+        if isinstance(response, Exception):
+            self.results[name] = response
+        else:
+            self.results[name].extend(response)
 
     @staticmethod
     def _errback(failure, host):
         """Errback"""
         return host, failure.value
 
+    def _save_error(self, failure):
+        """Errback for coiterator. Saves error so it can be raised later"""
+        self._errors.append(failure.value)
+
     def _finish(self, _):
         self._finished = True