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