pdns/calidns.cc

   1 /*
   2  * This file is part of PowerDNS or dnsdist.
   3  * Copyright -- PowerDNS.COM B.V. and its contributors
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of version 2 of the GNU General Public License as
   7  * published by the Free Software Foundation.
   8  *
   9  * In addition, for the avoidance of any doubt, permission is granted to
  10  * link this program with OpenSSL and to (re)distribute the binaries
  11  * produced as the result of such linking.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21  */
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif
  25
  26 #include <atomic>
  27 #include <iostream>
  28 #include <fstream>
  29 #include <memory>
  30 #include <poll.h>
  31 #include <thread>
  32
  33 #include <boost/program_options.hpp>
  34
  35 #include "dns_random.hh"
  36 #include "dnsparser.hh"
  37 #include "dnswriter.hh"
  38 #include "dnsrecords.hh"
  39 #include "ednsoptions.hh"
  40 #include "ednssubnet.hh"
  41 #include "misc.hh"
  42 #include "sstuff.hh"
  43 #include "statbag.hh"
  44
  45 using std::thread;
  46 using std::unique_ptr;
  47
  48 StatBag S;
  49
  50 static std::atomic<unsigned int> g_recvcounter, g_recvbytes;
  51 static volatile bool g_done;
  52
  53 namespace po = boost::program_options;
  54 static po::variables_map g_vm;
  55
  56 static bool g_quiet;
  57
  58 //NOLINTNEXTLINE(performance-unnecessary-value-param): we do want a copy to increase the reference count, thank you very much
  59 static void recvThread(const std::shared_ptr<std::vector<std::unique_ptr<Socket>>> sockets)
  60 {
  61   vector<pollfd> rfds, fds;
  62   for (const auto& s : *sockets) {
  63     if (s == nullptr) {
  64       continue;
  65     }
  66     struct pollfd pfd;
  67     pfd.fd = s->getHandle();
  68     pfd.events = POLLIN;
  69     pfd.revents = 0;
  70     rfds.push_back(pfd);
  71   }
  72
  73   int err;
  74
  75 #ifdef HAVE_RECVMMSG
  76   vector<struct mmsghdr> buf(100);
  77   for(auto& m : buf) {
  78     cmsgbuf_aligned *cbuf = new cmsgbuf_aligned;
  79     fillMSGHdr(&m.msg_hdr, new struct iovec, cbuf, sizeof(*cbuf), new char[1500], 1500, new ComboAddress("127.0.0.1"));
  80   }
  81 #else
  82   struct msghdr buf;
  83   cmsgbuf_aligned *cbuf = new cmsgbuf_aligned;
  84   fillMSGHdr(&buf, new struct iovec, cbuf, sizeof(*cbuf), new char[1500], 1500, new ComboAddress("127.0.0.1"));
  85 #endif
  86
  87   while(!g_done) {
  88     fds=rfds;
  89
  90     err = poll(&fds[0], fds.size(), -1);
  91     if (err < 0) {
  92       if (errno == EINTR)
  93         continue;
  94       unixDie("Unable to poll for new UDP events");
  95     }
  96
  97     for(auto &pfd : fds) {
  98       if (pfd.revents & POLLIN) {
  99 #ifdef HAVE_RECVMMSG
 100         if ((err=recvmmsg(pfd.fd, &buf[0], buf.size(), MSG_WAITFORONE, 0)) < 0 ) {
 101           if(errno != EAGAIN)
 102             unixDie("recvmmsg");
 103           continue;
 104         }
 105         g_recvcounter+=err;
 106         for(int n=0; n < err; ++n)
 107           g_recvbytes += buf[n].msg_len;
 108 #else
 109         if ((err = recvmsg(pfd.fd, &buf, 0)) < 0) {
 110           if (errno != EAGAIN)
 111             unixDie("recvmsg");
 112           continue;
 113         }
 114         g_recvcounter++;
 115         for (decltype(buf.msg_iovlen) i = 0; i < buf.msg_iovlen; i++)
 116           g_recvbytes += buf.msg_iov[i].iov_len;
 117 #endif
 118       }
 119     }
 120   }
 121 }
 122
 123 static ComboAddress getRandomAddressFromRange(const Netmask& ecsRange)
 124 {
 125   ComboAddress result = ecsRange.getMaskedNetwork();
 126   uint8_t bits = ecsRange.getBits();
 127   if (bits > 0) {
 128     uint32_t mod = 1 << (32 - bits);
 129     result.sin4.sin_addr.s_addr = result.sin4.sin_addr.s_addr + htonl(dns_random(mod));
 130   }
 131   else {
 132     result.sin4.sin_addr.s_addr = dns_random_uint32();
 133   }
 134
 135   return result;
 136 }
 137
 138 static void replaceEDNSClientSubnet(vector<uint8_t>* packet, const Netmask& ecsRange)
 139 {
 140   /* the last 4 bytes of the packet are the IPv4 address */
 141   ComboAddress rnd = getRandomAddressFromRange(ecsRange);
 142   uint32_t addr = rnd.sin4.sin_addr.s_addr;
 143
 144   const auto packetSize = packet->size();
 145   if (packetSize < sizeof(addr)) {
 146     return;
 147   }
 148
 149   memcpy(&packet->at(packetSize - sizeof(addr)), &addr, sizeof(addr));
 150 }
 151
 152 static void sendPackets(const vector<std::unique_ptr<Socket>>& sockets, const vector<vector<uint8_t>* >& packets, uint32_t qps, ComboAddress dest, const Netmask& ecsRange)
 153 {
 154   unsigned int burst=100;
 155   const auto nsecPerBurst=1*(unsigned long)(burst*1000000000.0/qps);
 156   struct timespec nsec;
 157   nsec.tv_sec=0;
 158   nsec.tv_nsec=0;
 159   int count=0;
 160   unsigned int nBursts=0;
 161   DTime dt;
 162   dt.set();
 163
 164   struct Unit {
 165     struct msghdr msgh;
 166     struct iovec iov;
 167     cmsgbuf_aligned cbuf;
 168   };
 169   vector<unique_ptr<Unit> > units;
 170
 171   for(const auto& p : packets) {
 172     count++;
 173
 174     Unit u;
 175
 176     if (!ecsRange.empty()) {
 177       replaceEDNSClientSubnet(p, ecsRange);
 178     }
 179
 180     fillMSGHdr(&u.msgh, &u.iov, nullptr, 0, (char*)&(*p)[0], p->size(), &dest);
 181
 182     auto socketHandle = sockets[count % sockets.size()]->getHandle();
 183     ssize_t sendmsgRet = sendmsg(socketHandle, &u.msgh, 0);
 184     if (sendmsgRet != 0) {
 185       if (sendmsgRet < 0) {
 186         unixDie("sendmsg");
 187       }
 188     }
 189
 190     if(!(count%burst)) {
 191       nBursts++;
 192       // Calculate the time in nsec we need to sleep to the next burst.
 193       // If this is negative, it means that we are not achieving the requested
 194       // target rate, in which case we skip the sleep.
 195       int toSleep = nBursts*nsecPerBurst - 1000*dt.udiffNoReset();
 196       if (toSleep > 0) {
 197         nsec.tv_nsec = toSleep;
 198         nanosleep(&nsec, 0);
 199       }
 200     }
 201   }
 202 }
 203
 204 static void usage(po::options_description &desc) {
 205   cerr<<"Syntax: calidns [OPTIONS] QUERY_FILE DESTINATION INITIAL_QPS HITRATE"<<endl;
 206   cerr<<desc<<endl;
 207 }
 208
 209 namespace {
 210 void parseQueryFile(const std::string& queryFile, vector<std::shared_ptr<vector<uint8_t>>>& unknown, bool useECSFromFile, bool wantRecursion, bool addECS)
 211 {
 212   ifstream ifs(queryFile);
 213   string line;
 214   std::vector<std::string> fields;
 215   fields.reserve(3);
 216
 217   while (getline(ifs, line)) {
 218     vector<uint8_t> packet;
 219     DNSPacketWriter::optvect_t ednsOptions;
 220     boost::trim(line);
 221     if (line.empty() || line.at(0) == '#') {
 222       continue;
 223     }
 224
 225     fields.clear();
 226     stringtok(fields, line, "\t ");
 227     if ((useECSFromFile && fields.size() < 3) || fields.size() < 2) {
 228       cerr<<"Skipping invalid line '"<<line<<", it does not contain enough values"<<endl;
 229       continue;
 230     }
 231
 232     const std::string& qname = fields.at(0);
 233     const std::string& qtype = fields.at(1);
 234     std::string subnet;
 235
 236     if (useECSFromFile) {
 237       subnet = fields.at(2);
 238     }
 239
 240     DNSPacketWriter packetWriter(packet, DNSName(qname), DNSRecordContent::TypeToNumber(qtype));
 241     packetWriter.getHeader()->rd = wantRecursion;
 242     packetWriter.getHeader()->id = dns_random_uint16();
 243
 244     if (!subnet.empty() || addECS) {
 245       EDNSSubnetOpts opt;
 246       opt.source = Netmask(subnet.empty() ? "0.0.0.0/32" : subnet);
 247       ednsOptions.emplace_back(EDNSOptionCode::ECS, makeEDNSSubnetOptsString(opt));
 248     }
 249
 250     if (!ednsOptions.empty() || (packetWriter.getHeader()->id % 2) != 0) {
 251       packetWriter.addOpt(1500, 0, EDNSOpts::DNSSECOK, ednsOptions);
 252       packetWriter.commit();
 253     }
 254     unknown.push_back(std::make_shared<vector<uint8_t>>(packet));
 255   }
 256
 257   shuffle(unknown.begin(), unknown.end(), pdns::dns_random_engine());
 258 }
 259 }
 260
 261 /*
 262   New plan. Set cache hit percentage, which we achieve on a per second basis.
 263   So we start with 10000 qps for example, and for 90% cache hit ratio means
 264   we take 1000 unique queries and each send them 10 times.
 265
 266   We then move the 1000 unique queries to the 'known' pool.
 267
 268   For the next second, say 20000 qps, we know we are going to need 2000 new queries,
 269   so we take 2000 from the unknown pool. Then we need 18000 cache hits. We can get 1000 from
 270   the known pool, leaving us down 17000. Or, we have 3000 in total now and we need 2000. We simply
 271   repeat the 3000 mix we have ~7 times. The 2000 can now go to the known pool too.
 272
 273   For the next second, say 30000 qps, we'll need 3000 cache misses, which we get from
 274   the unknown pool. To this we add 3000 queries from the known pool. Next up we repeat this batch 5
 275   times.
 276
 277   In general the algorithm therefore is:
 278
 279   1) Calculate number of cache misses required, get them from the unknown pool
 280   2) Move those to the known pool
 281   3) Fill up to amount of queries we need with random picks from the known pool
 282
 283 */
 284
 285 int main(int argc, char** argv)
 286 try
 287 {
 288   po::options_description desc("Options");
 289   desc.add_options()
 290     ("help,h", "Show this helpful message")
 291     ("version", "Show the version number")
 292     ("ecs", po::value<string>(), "Add EDNS Client Subnet option to outgoing queries using random addresses from the specified range (IPv4 only)")
 293     ("ecs-from-file", "Read IP or subnet values from the query file and add them as EDNS Client Subnet options to outgoing queries")
 294     ("increment", po::value<float>()->default_value(1.1),  "Set the factor to increase the QPS load per run")
 295     ("maximum-qps", po::value<uint32_t>(), "Stop incrementing once this rate has been reached, to provide a stable load")
 296     ("minimum-success-rate", po::value<double>()->default_value(0), "Stop the test as soon as the success rate drops below this value, in percent")
 297     ("plot-file", po::value<string>(), "Write results to the specific file")
 298     ("quiet", "Whether to run quietly, outputting only the maximum QPS reached. This option is mostly useful when used with --minimum-success-rate")
 299     ("want-recursion", "Set the Recursion Desired flag on queries");
 300   po::options_description alloptions;
 301   po::options_description hidden("hidden options");
 302   hidden.add_options()
 303     ("query-file", po::value<string>(), "File with queries")
 304     ("destination", po::value<string>(), "Destination address")
 305     ("initial-qps", po::value<uint32_t>(), "Initial number of queries per second")
 306     ("hitrate", po::value<double>(), "Aim this percent cache hitrate");
 307
 308   alloptions.add(desc).add(hidden);
 309   po::positional_options_description p;
 310   p.add("query-file", 1);
 311   p.add("destination", 1);
 312   p.add("initial-qps", 1);
 313   p.add("hitrate", 1);
 314
 315   po::store(po::command_line_parser(argc, argv).options(alloptions).positional(p).run(), g_vm);
 316   po::notify(g_vm);
 317
 318   if (g_vm.count("help")) {
 319     usage(desc);
 320     return EXIT_SUCCESS;
 321   }
 322
 323   if (g_vm.count("version")) {
 324     cerr<<"calidns "<<VERSION<<endl;
 325     return EXIT_SUCCESS;
 326   }
 327
 328   if (!(g_vm.count("query-file") && g_vm.count("destination") && g_vm.count("initial-qps") && g_vm.count("hitrate"))) {
 329     usage(desc);
 330     return EXIT_FAILURE;
 331   }
 332
 333   float increment = 1.1;
 334   try {
 335     increment = g_vm["increment"].as<float>();
 336   }
 337   catch(...) {
 338   }
 339
 340   bool wantRecursion = g_vm.count("want-recursion");
 341   bool useECSFromFile = g_vm.count("ecs-from-file");
 342   g_quiet = g_vm.count("quiet");
 343
 344   double hitrate = g_vm["hitrate"].as<double>();
 345   if (hitrate > 100 || hitrate < 0) {
 346     cerr<<"hitrate must be between 0 and 100, not "<<hitrate<<endl;
 347     return EXIT_FAILURE;
 348   }
 349   hitrate /= 100;
 350   uint32_t qpsstart = g_vm["initial-qps"].as<uint32_t>();
 351
 352   uint32_t maximumQps = std::numeric_limits<uint32_t>::max();
 353   if (g_vm.count("maximum-qps")) {
 354     maximumQps = g_vm["maximum-qps"].as<uint32_t>();
 355   }
 356
 357   double minimumSuccessRate = g_vm["minimum-success-rate"].as<double>();
 358   if (minimumSuccessRate > 100.0 || minimumSuccessRate < 0.0) {
 359     cerr<<"Minimum success rate must be between 0 and 100, not "<<minimumSuccessRate<<endl;
 360     return EXIT_FAILURE;
 361   }
 362
 363   Netmask ecsRange;
 364   if (g_vm.count("ecs")) {
 365
 366     try {
 367       ecsRange = Netmask(g_vm["ecs"].as<string>());
 368       if (!ecsRange.empty()) {
 369
 370         if (!ecsRange.isIPv4()) {
 371           cerr<<"Only IPv4 ranges are supported for ECS at the moment!"<<endl;
 372           return EXIT_FAILURE;
 373         }
 374
 375         if (!g_quiet) {
 376           cout<<"Adding ECS option to outgoing queries with random addresses from the "<<ecsRange.toString()<<" range"<<endl;
 377         }
 378       }
 379     }
 380     catch (const NetmaskException& e) {
 381       cerr<<"Error while parsing the ECS netmask: "<<e.reason<<endl;
 382       return EXIT_FAILURE;
 383     }
 384   }
 385
 386   struct sched_param param;
 387   param.sched_priority=99;
 388
 389 #ifdef HAVE_SCHED_SETSCHEDULER
 390   if(sched_setscheduler(0, SCHED_FIFO, &param) < 0) {
 391     if (!g_quiet) {
 392       cerr<<"Unable to set SCHED_FIFO: "<<stringerror()<<endl;
 393     }
 394   }
 395 #endif
 396
 397   reportAllTypes();
 398   vector<std::shared_ptr<vector<uint8_t>>> unknown;
 399   vector<std::shared_ptr<vector<uint8_t>>> known;
 400   parseQueryFile(g_vm["query-file"].as<string>(), unknown, useECSFromFile, wantRecursion, !ecsRange.empty());
 401
 402   if (!g_quiet) {
 403     cout<<"Generated "<<unknown.size()<<" ready to use queries"<<endl;
 404   }
 405
 406   auto sockets = std::make_shared<std::vector<std::unique_ptr<Socket>>>();
 407   ComboAddress dest;
 408   try {
 409     dest = ComboAddress(g_vm["destination"].as<string>(), 53);
 410   }
 411   catch (PDNSException &e) {
 412     cerr<<e.reason<<endl;
 413     return EXIT_FAILURE;
 414   }
 415   for(int i=0; i < 24; ++i) {
 416     auto sock = make_unique<Socket>(dest.sin4.sin_family, SOCK_DGRAM);
 417     //    sock->connect(dest);
 418     try {
 419       setSocketSendBuffer(sock->getHandle(), 2000000);
 420     }
 421     catch (const std::exception& e) {
 422       if (!g_quiet) {
 423         cerr<<e.what()<<endl;
 424       }
 425     }
 426     try {
 427       setSocketReceiveBuffer(sock->getHandle(), 2000000);
 428     }
 429     catch (const std::exception& e) {
 430       if (!g_quiet) {
 431         cerr<<e.what()<<endl;
 432       }
 433     }
 434
 435     sockets->push_back(std::move(sock));
 436   }
 437
 438   {
 439     std::thread receiver(recvThread, sockets);
 440     receiver.detach();
 441   }
 442
 443   uint32_t qps;
 444
 445   ofstream plot;
 446   if (g_vm.count("plot-file")) {
 447     plot.open(g_vm["plot-file"].as<string>());
 448     if (!plot) {
 449       cerr<<"Error opening "<<g_vm["plot-file"].as<string>()<<" for writing: "<<stringerror()<<endl;
 450       return EXIT_FAILURE;
 451     }
 452   }
 453
 454   double bestQPS = 0.0;
 455   double bestPerfectQPS = 0.0;
 456
 457   for(qps=qpsstart;;) {
 458     double seconds=1;
 459     if (!g_quiet) {
 460       cout<<"Aiming at "<<qps<< "qps (RD="<<wantRecursion<<") for "<<seconds<<" seconds at cache hitrate "<<100.0*hitrate<<"%";
 461     }
 462     unsigned int misses=(1-hitrate)*qps*seconds;
 463     unsigned int total=qps*seconds;
 464     if (misses == 0) {
 465       misses = 1;
 466     }
 467     if (!g_quiet) {
 468       cout<<", need "<<misses<<" misses, "<<total<<" queries, have "<<unknown.size()<<" unknown left!"<<endl;
 469     }
 470
 471     if (misses > unknown.size()) {
 472       cerr<<"Not enough queries remaining (need at least "<<misses<<" and got "<<unknown.size()<<", please add more to the query file), exiting."<<endl;
 473       return EXIT_FAILURE;
 474     }
 475     vector<vector<uint8_t>*> toSend;
 476     unsigned int n;
 477     for(n=0; n < misses; ++n) {
 478       auto ptr=unknown.back();
 479       unknown.pop_back();
 480       toSend.push_back(ptr.get());
 481       known.push_back(ptr);
 482     }
 483     for(;n < total; ++n) {
 484       toSend.push_back(known[dns_random(known.size())].get());
 485     }
 486
 487     shuffle(toSend.begin(), toSend.end(), pdns::dns_random_engine());
 488     g_recvcounter.store(0);
 489     g_recvbytes=0;
 490     DTime dt;
 491     dt.set();
 492
 493     sendPackets(*sockets, toSend, qps, dest, ecsRange);
 494
 495     const auto udiff = dt.udiffNoReset();
 496     const auto realqps=toSend.size()/(udiff/1000000.0);
 497     if (!g_quiet) {
 498       cout<<"Achieved "<<realqps<<" qps over "<< udiff/1000000.0<<" seconds"<<endl;
 499     }
 500
 501     usleep(50000);
 502     const auto received = g_recvcounter.load();
 503     const auto udiffReceived = dt.udiff();
 504     const auto realReceivedQPS = received/(udiffReceived/1000000.0);
 505     double perc=received*100.0/toSend.size();
 506      if (!g_quiet) {
 507        cout<<"Received "<<received<<" packets over "<< udiffReceived/1000000.0<<" seconds ("<<perc<<"%, adjusted received rate "<<realReceivedQPS<<" qps)"<<endl;
 508      }
 509
 510     if (plot) {
 511       plot<<qps<<" "<<realqps<<" "<<perc<<" "<<received/(udiff/1000000.0)<<" " << 8*g_recvbytes.load()/(udiff/1000000.0)<<endl;
 512       plot.flush();
 513     }
 514
 515     if (qps < maximumQps) {
 516       qps *= increment;
 517     }
 518     else {
 519       qps = maximumQps;
 520     }
 521
 522     if (minimumSuccessRate > 0.0 && perc < minimumSuccessRate) {
 523       if (g_quiet) {
 524         cout<<bestQPS<<endl;
 525       }
 526       else {
 527         cout<<"The latest success rate ("<<perc<<") dropped below the minimum success rate of "<<minimumSuccessRate<<", stopping."<<endl;
 528         cout<<"The final rate reached before failing was "<<bestQPS<<" qps (best rate at 100% was "<<bestPerfectQPS<<" qps)"<<endl;
 529       }
 530       break;
 531     }
 532
 533     bestQPS = std::max(bestQPS, realReceivedQPS);
 534     if (perc >= 100.0) {
 535       bestPerfectQPS = std::max(bestPerfectQPS, realReceivedQPS);
 536     }
 537   }
 538
 539   if (plot) {
 540     plot.flush();
 541   }
 542
 543   // t1.detach();
 544 }
 545 catch (const std::exception& exp)
 546 {
 547   cerr<<"Fatal error: "<<exp.what()<<endl;
 548   return EXIT_FAILURE;
 549 }
 550 catch (const NetmaskException& exp)
 551 {
 552   cerr<<"Fatal error: "<<exp.reason<<endl;
 553   return EXIT_FAILURE;
 554 }