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