regression-tests.dnsdist/test_OOOR.py

   1 #!/usr/bin/env python
   2 import dns
   3 import socket
   4 import struct
   5 import time
   6 import threading
   7 from dnsdisttests import DNSDistTest, pickAvailablePort
   8
   9 class OOORTCPResponder(object):
  10
  11     def handleConnection(self, conn):
  12         try:
  13
  14             while True:
  15                 try:
  16                     data = conn.recv(2)
  17                 except socket.timeout:
  18                     data = None
  19
  20                 if not data:
  21                     conn.close()
  22                     break
  23
  24                 (datalen,) = struct.unpack("!H", data)
  25                 data = conn.recv(datalen)
  26
  27                 # computing the correct ID for the response
  28                 request = dns.message.from_wire(data)
  29                 #print("got a query for %s" % (request.question[0].name))
  30                 if request.question[0].name == "0.simple.ooor.tests.powerdns.com":
  31                     time.sleep(1)
  32
  33                 response = dns.message.make_response(request)
  34
  35                 wire = response.to_wire()
  36                 conn.send(struct.pack("!H", len(wire)))
  37                 conn.send(wire)
  38
  39         except ConnectionError as err:
  40             print("Error in the thread handling reverse OOOR connections: %s" % (err))
  41         finally:
  42             conn.close()
  43
  44     def __init__(self, port):
  45         OOORTCPResponder.numberOfConnections = 0
  46
  47         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  48         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  49         sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
  50         try:
  51             sock.bind(("127.0.0.1", port))
  52         except socket.error as e:
  53             print("Error binding in the TCP responder: %s" % str(e))
  54             sys.exit(1)
  55
  56         sock.listen(100)
  57         while True:
  58             (conn, _) = sock.accept()
  59             conn.settimeout(5.0)
  60
  61             OOORTCPResponder.numberOfConnections = OOORTCPResponder.numberOfConnections + 1
  62             thread = threading.Thread(name='Connection Handler',
  63                                       target=self.handleConnection,
  64                                       args=[conn])
  65             thread.daemon = True
  66             thread.start()
  67
  68         sock.close()
  69
  70 class ReverseOOORTCPResponder(OOORTCPResponder):
  71
  72     def handleConnection(self, conn):
  73         try:
  74             # short timeout since we want to answer only after receiving 5 requests
  75             # or a timeout
  76             conn.settimeout(0.2)
  77
  78             queuedResponses = []
  79             while True:
  80                 timedout = False
  81                 try:
  82                     data = conn.recv(2)
  83                 except socket.timeout:
  84                     data = None
  85                     timedout = True
  86
  87                 if timedout or len(queuedResponses) >= 5:
  88                     queuedResponses.reverse()
  89                     for response in queuedResponses:
  90                         wire = response.to_wire()
  91                         conn.send(struct.pack("!H", len(wire)))
  92                         conn.send(wire)
  93                     queuedResponses = []
  94                     if timedout:
  95                         continue
  96                 elif not data:
  97                     conn.close()
  98                     break
  99
 100                 (datalen,) = struct.unpack("!H", data)
 101                 data = conn.recv(datalen)
 102
 103                 # computing the correct ID for the response
 104                 request = dns.message.from_wire(data)
 105                 #print("got a query for %s" % (request.question[0].name))
 106
 107                 response = dns.message.make_response(request)
 108                 queuedResponses.append(response)
 109
 110         except ConnectionError as err:
 111             print("Error in the thread handling reverse OOOR connections: %s" % (err))
 112         finally:
 113             conn.close()
 114
 115     def __init__(self, port):
 116         ReverseOOORTCPResponder.numberOfConnections = 0
 117
 118         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 119         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
 120         sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 121         try:
 122             sock.bind(("127.0.0.1", port))
 123         except socket.error as e:
 124             print("Error binding in the TCP responder: %s" % str(e))
 125             sys.exit(1)
 126
 127         sock.listen(100)
 128         while True:
 129             (conn, _) = sock.accept()
 130
 131             ReverseOOORTCPResponder.numberOfConnections = ReverseOOORTCPResponder.numberOfConnections + 1
 132             thread = threading.Thread(name='Connection Handler',
 133                                       target=self.handleConnection,
 134                                       args=[conn])
 135             thread.daemon = True
 136             thread.start()
 137
 138         sock.close()
 139
 140
 141 OOORResponderPort = pickAvailablePort()
 142 ooorTCPResponder = threading.Thread(name='TCP Responder', target=OOORTCPResponder, args=[OOORResponderPort])
 143 ooorTCPResponder.daemon = True
 144 ooorTCPResponder.start()
 145
 146 ReverseOOORResponderPort = pickAvailablePort()
 147 ReverseOoorTCPResponder = threading.Thread(name='TCP Responder', target=ReverseOOORTCPResponder, args=[ReverseOOORResponderPort])
 148 ReverseOoorTCPResponder.daemon = True
 149 ReverseOoorTCPResponder.start()
 150
 151 class TestOOORWithClientNotBackend(DNSDistTest):
 152     # this test suite uses a different responder port
 153     _testServerPort = OOORResponderPort
 154
 155     _concurrentQueriesFromClient = 10
 156     _config_template = """
 157     newServer{address="127.0.0.1:%d", maxInFlight=0, pool={""}}:setUp()
 158     newServer{address="127.0.0.1:%d", maxInFlight=0, pool={"more-queries"}}:setUp()
 159     -- route these queries to a different backend so we don't reuse the connection from a previous test
 160     addAction("more-queries.ooor.tests.powerdns.com.", PoolAction("more-queries"))
 161     setLocal("%s:%d", {maxInFlight=%d})
 162     """
 163     _config_params = ['_testServerPort', '_testServerPort', '_dnsDistListeningAddr', '_dnsDistPort', '_concurrentQueriesFromClient']
 164     _verboseMode = True
 165     _skipListeningOnCL = True
 166
 167     @classmethod
 168     def startResponders(cls):
 169         return
 170
 171     def testSimple(self):
 172         """
 173         OOOR: 5 queries
 174         """
 175         names = []
 176         OOORTCPResponder.numberOfConnections = 0
 177
 178         for idx in range(5):
 179             names.append('%d.simple.ooor.tests.powerdns.com.' % (idx))
 180
 181         conn = self.openTCPConnection()
 182
 183         counter = 0
 184         for name in names:
 185             query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
 186             query.id = counter
 187             counter = counter + 1
 188
 189             self.sendTCPQueryOverConnection(conn, query)
 190
 191         receivedResponses = {}
 192
 193         for name in names:
 194             receivedResponse = self.recvTCPResponseOverConnection(conn)
 195             self.assertTrue(receivedResponse)
 196             receivedResponses[str(receivedResponse.question[0].name)] = (receivedResponse)
 197
 198         self.assertEqual(len(receivedResponses), 5)
 199         for idx in range(5):
 200             self.assertIn('%d.simple.ooor.tests.powerdns.com.' % (idx), receivedResponses)
 201
 202         # we can get a response to one of the first query before they all have
 203         # been read, reusing a backend connection
 204         self.assertLessEqual(OOORTCPResponder.numberOfConnections, 5)
 205
 206     def testMoreQueriesThanAllowedInFlight(self):
 207         """
 208         OOOR: 100 queries, 10 in flight
 209         """
 210         names = []
 211         OOORTCPResponder.numberOfConnections = 0
 212
 213         for idx in range(100):
 214             names.append('%d.more-queries.ooor.tests.powerdns.com.' % (idx))
 215
 216         conn = self.openTCPConnection()
 217
 218         counter = 0
 219         for name in names:
 220             query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
 221             query.id = counter
 222             counter = counter + 1
 223
 224             self.sendTCPQueryOverConnection(conn, query)
 225
 226         receivedResponses = {}
 227
 228         for name in names:
 229             receivedResponse = self.recvTCPResponseOverConnection(conn)
 230             self.assertTrue(receivedResponse)
 231             receivedResponses[str(receivedResponse.question[0].name)] = (receivedResponse)
 232
 233         self.assertEqual(len(receivedResponses), 100)
 234         for idx in range(5):
 235             self.assertIn('%d.more-queries.ooor.tests.powerdns.com.' % (idx), receivedResponses)
 236
 237         self.assertLessEqual(OOORTCPResponder.numberOfConnections, self._concurrentQueriesFromClient)
 238
 239 class TestOOORWithClientAndBackend(DNSDistTest):
 240     # this test suite uses a different responder port
 241     _testServerPort = ReverseOOORResponderPort
 242
 243     _concurrentQueriesFromClient = 10
 244     _concurrentQueriesToServer = 5
 245     _config_template = """
 246     newServer{address="127.0.0.1:%d", maxInFlight=%d, pool={""}}:setUp()
 247     newServer{address="127.0.0.1:%d", maxInFlight=%d, pool={"more-queries"}}:setUp()
 248     -- route these queries to a different backend so we don't reuse the connection from a previous test
 249     addAction("more-queries.reverse-ooor.tests.powerdns.com.", PoolAction("more-queries"))
 250     setLocal("%s:%d", {maxInFlight=%d})
 251     """
 252     _config_params = ['_testServerPort', '_concurrentQueriesToServer', '_testServerPort', '_concurrentQueriesToServer', '_dnsDistListeningAddr', '_dnsDistPort', '_concurrentQueriesFromClient']
 253     _verboseMode = True
 254     _skipListeningOnCL = True
 255
 256     @classmethod
 257     def startResponders(cls):
 258         return
 259
 260     def testSimple(self):
 261         """
 262         OOOR Reverse: 5 queries
 263         """
 264         names = []
 265         ReverseOOORTCPResponder.numberOfConnections = 0
 266
 267         for idx in range(5):
 268             names.append('%d.simple.reverse-ooor.tests.powerdns.com.' % (idx))
 269
 270         conn = self.openTCPConnection()
 271
 272         counter = 0
 273         for name in names:
 274             query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
 275             query.id = counter
 276             counter = counter + 1
 277
 278             self.sendTCPQueryOverConnection(conn, query)
 279
 280         receivedResponses = {}
 281
 282         for name in names:
 283             receivedResponse = self.recvTCPResponseOverConnection(conn)
 284             self.assertTrue(receivedResponse)
 285             receivedResponses[str(receivedResponse.question[0].name)] = (receivedResponse)
 286
 287         self.assertEqual(len(receivedResponses), 5)
 288         for idx in range(5):
 289             self.assertIn('%d.simple.reverse-ooor.tests.powerdns.com.' % (idx), receivedResponses)
 290
 291         self.assertEqual(ReverseOOORTCPResponder.numberOfConnections, 1)
 292
 293     def testMoreQueriesThanAllowedInFlight(self):
 294         """
 295         OOOR Reverse: 100 queries, 10 in flight, 5 per backend
 296         """
 297         names = []
 298         ReverseOOORTCPResponder.numberOfConnections = 0
 299
 300         for idx in range(100):
 301             names.append('%d.more-queries.reverse-ooor.tests.powerdns.com.' % (idx))
 302
 303         conn = self.openTCPConnection()
 304
 305         counter = 0
 306         for name in names:
 307             query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
 308             query.id = counter
 309             counter = counter + 1
 310
 311             self.sendTCPQueryOverConnection(conn, query)
 312
 313         receivedResponses = {}
 314
 315         for name in names:
 316             receivedResponse = self.recvTCPResponseOverConnection(conn)
 317             self.assertTrue(receivedResponse)
 318             receivedResponses[str(receivedResponse.question[0].name)] = (receivedResponse)
 319             #print("Received a response for %s" % (receivedResponse.question[0].name))
 320
 321         self.assertEqual(len(receivedResponses), 100)
 322         for idx in range(5):
 323             self.assertIn('%d.more-queries.reverse-ooor.tests.powerdns.com.' % (idx), receivedResponses)
 324
 325         # in theory they could all be handled by the same backend if we get the responses
 326         # fast enough, but over 100 queries that's very, very unlikely
 327         self.assertEqual(ReverseOOORTCPResponder.numberOfConnections, 2)