ceph/src/test/mon/test_mon_workloadgen.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14 #include "acconfig.h"
  15
  16 #ifdef HAVE_SYS_MOUNT_H
  17 #include <sys/mount.h>
  18 #endif
  19
  20 #ifdef HAVE_SYS_PARAM_H
  21 #include <sys/param.h>
  22 #endif
  23
  24 #ifdef HAVE_SYS_VFS_H
  25 #include <sys/vfs.h>
  26 #endif
  27
  28 #include <iostream>
  29 #include <string>
  30 #include <map>
  31
  32 #include <boost/scoped_ptr.hpp>
  33 #include <boost/random/mersenne_twister.hpp>
  34 #include <boost/random/uniform_int.hpp>
  35
  36
  37 #include "osd/osd_types.h"
  38 #include "osdc/Objecter.h"
  39 #include "mon/MonClient.h"
  40 #include "msg/Dispatcher.h"
  41 #include "msg/Messenger.h"
  42 #include "common/Timer.h"
  43 #include "common/ceph_argparse.h"
  44 #include "global/global_init.h"
  45 #include "global/signal_handler.h"
  46 #include "common/config.h"
  47 #include "common/debug.h"
  48 #include "common/errno.h"
  49 #include "common/ceph_mutex.h"
  50 #include "common/strtol.h"
  51 #include "common/LogEntry.h"
  52 #include "auth/KeyRing.h"
  53 #include "auth/AuthAuthorizeHandler.h"
  54 #include "include/uuid.h"
  55 #include "include/ceph_assert.h"
  56
  57 #include "messages/MOSDBoot.h"
  58 #include "messages/MOSDAlive.h"
  59 #include "messages/MOSDPGCreate.h"
  60 #include "messages/MOSDPGRemove.h"
  61 #include "messages/MOSDMap.h"
  62 #include "messages/MPGStats.h"
  63 #include "messages/MLog.h"
  64 #include "messages/MOSDPGTemp.h"
  65
  66 using namespace std;
  67
  68 #define dout_context g_ceph_context
  69 #define dout_subsys ceph_subsys_
  70 #undef dout_prefix
  71 #define dout_prefix _prefix(_dout, get_name())
  72 static ostream& _prefix(std::ostream *_dout, const string &n) {
  73   return *_dout << " stub(" << n << ") ";
  74 }
  75
  76
  77 typedef boost::mt11213b rngen_t;
  78 typedef boost::scoped_ptr<Messenger> MessengerRef;
  79 typedef boost::scoped_ptr<Objecter> ObjecterRef;
  80
  81 class TestStub : public Dispatcher
  82 {
  83  protected:
  84   MessengerRef messenger;
  85   MonClient monc;
  86
  87   ceph::mutex lock;
  88   ceph::condition_variable cond;
  89   SafeTimer timer;
  90
  91   bool do_shutdown;
  92   double tick_seconds;
  93
  94   struct C_Tick : public Context {
  95     TestStub *s;
  96     explicit C_Tick(TestStub *stub) : s(stub) {}
  97     void finish(int r) override {
  98       generic_dout(20) << "C_Tick::" << __func__ << dendl;
  99       if (r == -ECANCELED) {
 100         generic_dout(20) << "C_Tick::" << __func__
 101                         << " shutdown" << dendl;
 102         return;
 103       }
 104       s->tick();
 105     }
 106   };
 107
 108   bool ms_dispatch(Message *m) override = 0;
 109   void ms_handle_connect(Connection *con) override = 0;
 110   void ms_handle_remote_reset(Connection *con) override = 0;
 111   virtual int _shutdown() = 0;
 112   // courtesy method to be implemented by the stubs at their
 113   // own discretion
 114   virtual void _tick() { }
 115   // different stubs may have different needs; if a stub needs
 116   // to tick, then it must call this function.
 117   void start_ticking(double t=1.0) {
 118     tick_seconds = t;
 119     if (t <= 0) {
 120       stop_ticking();
 121       return;
 122     }
 123     dout(20) << __func__ << " adding tick timer" << dendl;
 124     timer.add_event_after(tick_seconds, new C_Tick(this));
 125   }
 126   // If we have a function to start ticking that the stubs can
 127   // use at their own discretion, then we should also have a
 128   // function to disable said ticking to be used the same way.
 129   // Just in case.
 130   // For simplicity's sake, we don't cancel the tick right off
 131   // the bat; instead, we wait for the next tick to kick in and
 132   // disable itself.
 133   void stop_ticking() {
 134     dout(20) << __func__ << " disable tick" << dendl;
 135     tick_seconds = 0;
 136   }
 137
 138  public:
 139   void tick() {
 140     std::cout << __func__ << std::endl;
 141     if (do_shutdown || (tick_seconds <= 0)) {
 142       std::cout << __func__ << " "
 143                 << (do_shutdown ? "shutdown" : "stop ticking")
 144                 << std::endl;
 145       return;
 146     }
 147     _tick();
 148     timer.add_event_after(tick_seconds, new C_Tick(this));
 149   }
 150
 151   virtual const string get_name() = 0;
 152   virtual int init() = 0;
 153
 154   virtual int shutdown() {
 155     std::lock_guard l{lock};
 156     do_shutdown = true;
 157     int r = _shutdown();
 158     if (r < 0) {
 159       dout(10) << __func__ << " error shutting down: "
 160                << cpp_strerror(-r) << dendl;
 161       return r;
 162     }
 163     monc.shutdown();
 164     timer.shutdown();
 165     messenger->shutdown();
 166     return 0;
 167   }
 168
 169   virtual void print(ostream &out) {
 170     out << "stub(" << get_name() << ")";
 171   }
 172
 173   void wait() {
 174     if (messenger != NULL)
 175       messenger->wait();
 176   }
 177
 178   TestStub(CephContext *cct, string who)
 179     : Dispatcher(cct),
 180       monc(cct),
 181       lock(ceph::make_mutex(who.append("::lock"))),
 182       timer(cct, lock),
 183       do_shutdown(false),
 184       tick_seconds(0.0) { }
 185 };
 186
 187 class ClientStub : public TestStub
 188 {
 189   ObjecterRef objecter;
 190   rngen_t gen;
 191
 192  protected:
 193   bool ms_dispatch(Message *m) override {
 194     std::lock_guard l{lock};
 195     dout(1) << "client::" << __func__ << " " << *m << dendl;
 196     switch (m->get_type()) {
 197     case CEPH_MSG_OSD_MAP:
 198       objecter->handle_osd_map((MOSDMap*)m);
 199       cond.notify_all();
 200       break;
 201     }
 202     return true;
 203   }
 204
 205   void ms_handle_connect(Connection *con) override {
 206     dout(1) << "client::" << __func__ << " " << con << dendl;
 207     std::lock_guard l{lock};
 208     objecter->ms_handle_connect(con);
 209   }
 210
 211   void ms_handle_remote_reset(Connection *con) override {
 212     dout(1) << "client::" << __func__ << " " << con << dendl;
 213     std::lock_guard l{lock};
 214     objecter->ms_handle_remote_reset(con);
 215   }
 216
 217   bool ms_handle_reset(Connection *con) override {
 218     dout(1) << "client::" << __func__ << dendl;
 219     std::lock_guard l{lock};
 220     objecter->ms_handle_reset(con);
 221     return false;
 222   }
 223
 224   bool ms_handle_refused(Connection *con) override {
 225     return false;
 226   }
 227
 228   const string get_name() override {
 229     return "client";
 230   }
 231
 232   int _shutdown() override {
 233     if (objecter) {
 234       objecter->shutdown();
 235     }
 236     return 0;
 237   }
 238
 239  public:
 240   explicit ClientStub(CephContext *cct)
 241     : TestStub(cct, "client"),
 242       gen((int) time(NULL))
 243   { }
 244
 245   int init() override {
 246     int err;
 247     err = monc.build_initial_monmap();
 248     if (err < 0) {
 249       derr << "ClientStub::" << __func__ << " ERROR: build initial monmap: "
 250            << cpp_strerror(err) << dendl;
 251       return err;
 252     }
 253
 254     messenger.reset(Messenger::create_client_messenger(cct, "stubclient"));
 255     ceph_assert(messenger.get() != NULL);
 256
 257     messenger->set_default_policy(
 258         Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX));
 259     dout(10) << "ClientStub::" << __func__ << " starting messenger at "
 260             << messenger->get_myaddrs() << dendl;
 261
 262     objecter.reset(new Objecter(cct, messenger.get(), &monc, NULL, 0, 0));
 263     ceph_assert(objecter.get() != NULL);
 264     objecter->set_balanced_budget();
 265
 266     monc.set_messenger(messenger.get());
 267     objecter->init();
 268     messenger->add_dispatcher_head(this);
 269     messenger->start();
 270     monc.set_want_keys(CEPH_ENTITY_TYPE_MON|CEPH_ENTITY_TYPE_OSD);
 271
 272     err = monc.init();
 273     if (err < 0) {
 274       derr << "ClientStub::" << __func__ << " monc init error: "
 275            << cpp_strerror(-err) << dendl;
 276       return err;
 277     }
 278
 279     err = monc.authenticate();
 280     if (err < 0) {
 281       derr << "ClientStub::" << __func__ << " monc authenticate error: "
 282            << cpp_strerror(-err) << dendl;
 283       monc.shutdown();
 284       return err;
 285     }
 286     monc.wait_auth_rotating(30.0);
 287
 288     objecter->set_client_incarnation(0);
 289     objecter->start();
 290
 291     lock.lock();
 292     timer.init();
 293     monc.renew_subs();
 294
 295     lock.unlock();
 296
 297     objecter->wait_for_osd_map();
 298
 299     dout(10) << "ClientStub::" << __func__ << " done" << dendl;
 300     return 0;
 301   }
 302 };
 303
 304 class OSDStub : public TestStub
 305 {
 306   int whoami;
 307   OSDSuperblock sb;
 308   OSDMap osdmap;
 309   osd_stat_t osd_stat;
 310
 311   map<pg_t,pg_stat_t> pgs;
 312   set<pg_t> pgs_changes;
 313
 314   rngen_t gen;
 315   boost::uniform_int<> mon_osd_rng;
 316
 317   utime_t last_boot_attempt;
 318   static const double STUB_BOOT_INTERVAL;
 319
 320
 321  public:
 322
 323   enum {
 324     STUB_MON_OSD_ALIVE    = 1,
 325     STUB_MON_OSD_PGTEMP   = 2,
 326     STUB_MON_OSD_FAILURE  = 3,
 327     STUB_MON_OSD_PGSTATS  = 4,
 328     STUB_MON_LOG          = 5,
 329
 330     STUB_MON_OSD_FIRST    = STUB_MON_OSD_ALIVE,
 331     STUB_MON_OSD_LAST     = STUB_MON_LOG,
 332   };
 333
 334   struct C_CreatePGs : public Context {
 335     OSDStub *s;
 336     explicit C_CreatePGs(OSDStub *stub) : s(stub) {}
 337     void finish(int r) override {
 338       if (r == -ECANCELED) {
 339         generic_dout(20) << "C_CreatePGs::" << __func__
 340                         << " shutdown" << dendl;
 341         return;
 342       }
 343       generic_dout(20) << "C_CreatePGs::" << __func__ << dendl;
 344       s->auto_create_pgs();
 345     }
 346   };
 347
 348
 349   OSDStub(int _whoami, CephContext *cct)
 350     : TestStub(cct, "osd"),
 351       whoami(_whoami),
 352       gen(whoami),
 353       mon_osd_rng(STUB_MON_OSD_FIRST, STUB_MON_OSD_LAST)
 354   {
 355     dout(20) << __func__ << " auth supported: "
 356              << cct->_conf->auth_supported << dendl;
 357     stringstream ss;
 358     ss << "client-osd" << whoami;
 359     std::string public_msgr_type = cct->_conf->ms_public_type.empty() ? cct->_conf.get_val<std::string>("ms_type") : cct->_conf->ms_public_type;
 360     messenger.reset(Messenger::create(cct, public_msgr_type, entity_name_t::OSD(whoami),
 361                                       ss.str().c_str(), getpid(), 0));
 362
 363     Throttle throttler(g_ceph_context, "osd_client_bytes",
 364         g_conf()->osd_client_message_size_cap);
 365
 366     messenger->set_default_policy(
 367         Messenger::Policy::stateless_server(0));
 368     messenger->set_policy_throttlers(entity_name_t::TYPE_CLIENT,
 369                                     &throttler, NULL);
 370     messenger->set_policy(entity_name_t::TYPE_MON,
 371         Messenger::Policy::lossy_client(
 372           CEPH_FEATURE_UID |
 373           CEPH_FEATURE_PGID64 |
 374           CEPH_FEATURE_OSDENC));
 375     messenger->set_policy(entity_name_t::TYPE_OSD,
 376         Messenger::Policy::stateless_server(0));
 377
 378     dout(10) << __func__ << " public addr " << g_conf()->public_addr << dendl;
 379     int err = messenger->bind(g_conf()->public_addr);
 380     if (err < 0)
 381       exit(1);
 382
 383     if (monc.build_initial_monmap() < 0)
 384       exit(1);
 385
 386     messenger->start();
 387     monc.set_messenger(messenger.get());
 388   }
 389
 390   int init() override {
 391     dout(10) << __func__ << dendl;
 392     std::lock_guard l{lock};
 393
 394     dout(1) << __func__ << " fsid " << monc.monmap.fsid
 395             << " osd_fsid " << g_conf()->osd_uuid << dendl;
 396     dout(1) << __func__ << " name " << g_conf()->name << dendl;
 397
 398     timer.init();
 399     messenger->add_dispatcher_head(this);
 400     monc.set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD);
 401
 402     int err = monc.init();
 403     if (err < 0) {
 404       derr << __func__ << " monc init error: "
 405            << cpp_strerror(-err) << dendl;
 406       return err;
 407     }
 408
 409     err = monc.authenticate();
 410     if (err < 0) {
 411       derr << __func__ << " monc authenticate error: "
 412            << cpp_strerror(-err) << dendl;
 413       monc.shutdown();
 414       return err;
 415     }
 416     ceph_assert(!monc.get_fsid().is_zero());
 417
 418     monc.wait_auth_rotating(30.0);
 419
 420
 421     dout(10) << __func__ << " creating osd superblock" << dendl;
 422     sb.cluster_fsid = monc.monmap.fsid;
 423     sb.osd_fsid.generate_random();
 424     sb.whoami = whoami;
 425     sb.compat_features = CompatSet();
 426     dout(20) << __func__ << " " << sb << dendl;
 427     dout(20) << __func__ << " osdmap " << osdmap << dendl;
 428
 429     update_osd_stat();
 430
 431     start_ticking();
 432     // give a chance to the mons to inform us of what PGs we should create
 433     timer.add_event_after(30.0, new C_CreatePGs(this));
 434
 435     return 0;
 436   }
 437
 438   int _shutdown() override {
 439
 440     return 0;
 441   }
 442
 443   void boot() {
 444     dout(1) << __func__ << " boot?" << dendl;
 445
 446     utime_t now = ceph_clock_now();
 447     if ((last_boot_attempt > 0.0)
 448         && ((now - last_boot_attempt)) <= STUB_BOOT_INTERVAL) {
 449       dout(1) << __func__ << " backoff and try again later." << dendl;
 450       return;
 451     }
 452
 453     dout(1) << __func__ << " boot!" << dendl;
 454     MOSDBoot *mboot = new MOSDBoot;
 455     mboot->sb = sb;
 456     last_boot_attempt = now;
 457     monc.send_mon_message(mboot);
 458   }
 459
 460   void add_pg(pg_t pgid, epoch_t epoch, pg_t parent) {
 461
 462     utime_t now = ceph_clock_now();
 463
 464     pg_stat_t s;
 465     s.created = epoch;
 466     s.last_epoch_clean = epoch;
 467     s.parent = parent;
 468     s.state |= PG_STATE_CLEAN | PG_STATE_ACTIVE;
 469     s.last_fresh = now;
 470     s.last_change = now;
 471     s.last_clean = now;
 472     s.last_active = now;
 473     s.last_unstale = now;
 474
 475     pgs[pgid] = s;
 476     pgs_changes.insert(pgid);
 477   }
 478
 479   void auto_create_pgs() {
 480     bool has_pgs = !pgs.empty();
 481     dout(10) << __func__
 482              << ": " << (has_pgs ? "has pgs; ignore" : "create pgs") << dendl;
 483     if (has_pgs)
 484       return;
 485
 486     if (!osdmap.get_epoch()) {
 487       dout(1) << __func__
 488               << " still don't have osdmap; reschedule pg creation" << dendl;
 489       timer.add_event_after(10.0, new C_CreatePGs(this));
 490       return;
 491     }
 492
 493     auto& osdmap_pools = osdmap.get_pools();
 494     for (auto pit = osdmap_pools.begin(); pit != osdmap_pools.end(); ++pit) {
 495       const int64_t pool_id = pit->first;
 496       const pg_pool_t &pool = pit->second;
 497       int ruleno = pool.get_crush_rule();
 498
 499       if (!osdmap.crush->rule_exists(ruleno)) {
 500         dout(20) << __func__
 501                  << " no crush rule for pool id " << pool_id
 502                  << " rule no " << ruleno << dendl;
 503         continue;
 504       }
 505
 506       epoch_t pool_epoch = pool.get_last_change();
 507       dout(20) << __func__
 508                << " pool num pgs " << pool.get_pg_num()
 509                << " epoch " << pool_epoch << dendl;
 510
 511       for (ps_t ps = 0; ps < pool.get_pg_num(); ++ps) {
 512         pg_t pgid(ps, pool_id);
 513         pg_t parent;
 514         dout(20) << __func__
 515                  << " pgid " << pgid << " parent " << parent << dendl;
 516         add_pg(pgid, pool_epoch, parent);
 517       }
 518     }
 519   }
 520
 521   void update_osd_stat() {
 522     struct statfs stbuf;
 523     int ret = statfs(".", &stbuf);
 524     if (ret < 0) {
 525       ret = -errno;
 526       dout(0) << __func__
 527               << " cannot statfs ." << cpp_strerror(ret) << dendl;
 528       return;
 529     }
 530
 531     osd_stat.statfs.total = stbuf.f_blocks * stbuf.f_bsize;
 532     osd_stat.statfs.available = stbuf.f_bavail * stbuf.f_bsize;
 533     osd_stat.statfs.internally_reserved = 0;
 534   }
 535
 536   void send_pg_stats() {
 537     dout(10) << __func__
 538              << " pgs " << pgs.size() << " osdmap " << osdmap << dendl;
 539     MPGStats *mstats = new MPGStats(monc.get_fsid(), osdmap.get_epoch());
 540
 541     mstats->set_tid(1);
 542     mstats->osd_stat = osd_stat;
 543
 544     set<pg_t>::iterator it;
 545     for (it = pgs_changes.begin(); it != pgs_changes.end(); ++it) {
 546       pg_t pgid = (*it);
 547       if (pgs.count(pgid) == 0) {
 548         derr << __func__
 549              << " pgid " << pgid << " not on our map" << dendl;
 550         ceph_abort_msg("pgid not on our map");
 551       }
 552       pg_stat_t &s = pgs[pgid];
 553       mstats->pg_stat[pgid] = s;
 554
 555       JSONFormatter f(true);
 556       s.dump(&f);
 557       dout(20) << __func__
 558                << " pg " << pgid << " stats:\n";
 559       f.flush(*_dout);
 560       *_dout << dendl;
 561
 562     }
 563     dout(10) << __func__ << " send " << *mstats << dendl;
 564     monc.send_mon_message(mstats);
 565   }
 566
 567   void modify_pg(pg_t pgid) {
 568     dout(10) << __func__ << " pg " << pgid << dendl;
 569     ceph_assert(pgs.count(pgid) > 0);
 570
 571     pg_stat_t &s = pgs[pgid];
 572     utime_t now = ceph_clock_now();
 573
 574     if (now - s.last_change < 10.0) {
 575       dout(10) << __func__
 576                << " pg " << pgid << " changed in the last 10s" << dendl;
 577       return;
 578     }
 579
 580     s.state ^= PG_STATE_CLEAN;
 581     if (s.state & PG_STATE_CLEAN)
 582       s.last_clean = now;
 583     s.last_change = now;
 584     s.reported_seq++;
 585
 586     pgs_changes.insert(pgid);
 587   }
 588
 589   void modify_pgs() {
 590     dout(10) << __func__ << dendl;
 591
 592     if (pgs.empty()) {
 593       dout(1) << __func__
 594               << " no pgs available! don't attempt to modify." << dendl;
 595       return;
 596     }
 597
 598     boost::uniform_int<> pg_rng(0, pgs.size()-1);
 599     set<int> pgs_pos;
 600
 601     int num_pgs = pg_rng(gen);
 602     while ((int)pgs_pos.size() < num_pgs)
 603       pgs_pos.insert(pg_rng(gen));
 604
 605     map<pg_t,pg_stat_t>::iterator it = pgs.begin();
 606     set<int>::iterator pos_it = pgs_pos.begin();
 607
 608     int pgs_at = 0;
 609     while (pos_it != pgs_pos.end()) {
 610       int at = *pos_it;
 611       dout(20) << __func__ << " pg at pos " << at << dendl;
 612       while ((pgs_at != at) && (it != pgs.end())) {
 613         ++it;
 614         ++pgs_at;
 615       }
 616       ceph_assert(it != pgs.end());
 617       dout(20) << __func__
 618                << " pg at pos " << at << ": " << it->first << dendl;
 619       modify_pg(it->first);
 620       ++pos_it;
 621     }
 622   }
 623
 624   void op_alive() {
 625     dout(10) << __func__ << dendl;
 626     if (!osdmap.exists(whoami)) {
 627       dout(0) << __func__ << " I'm not in the osdmap!!\n";
 628       JSONFormatter f(true);
 629       osdmap.dump(&f);
 630       f.flush(*_dout);
 631       *_dout << dendl;
 632     }
 633     if (osdmap.get_epoch() == 0) {
 634       dout(1) << __func__ << " wait for osdmap" << dendl;
 635       return;
 636     }
 637     epoch_t up_thru = osdmap.get_up_thru(whoami);
 638     dout(10) << __func__ << "up_thru: " << osdmap.get_up_thru(whoami) << dendl;
 639
 640     monc.send_mon_message(new MOSDAlive(osdmap.get_epoch(), up_thru));
 641   }
 642
 643   void op_pgtemp() {
 644     if (osdmap.get_epoch() == 0) {
 645       dout(1) << __func__ << " wait for osdmap" << dendl;
 646       return;
 647     }
 648     dout(10) << __func__ << dendl;
 649     MOSDPGTemp *m = new MOSDPGTemp(osdmap.get_epoch());
 650     monc.send_mon_message(m);
 651   }
 652
 653   void op_failure() {
 654     dout(10) << __func__ << dendl;
 655   }
 656
 657   void op_pgstats() {
 658     dout(10) << __func__ << dendl;
 659
 660     modify_pgs();
 661     if (!pgs_changes.empty())
 662       send_pg_stats();
 663     monc.sub_want("osd_pg_creates", 0, CEPH_SUBSCRIBE_ONETIME);
 664     monc.renew_subs();
 665
 666     dout(20) << __func__ << " pg pools:\n";
 667
 668     JSONFormatter f(true);
 669     f.open_array_section("pools");
 670     auto& osdmap_pools = osdmap.get_pools();
 671     for (auto pit = osdmap_pools.begin(); pit != osdmap_pools.end(); ++pit) {
 672       const int64_t pool_id = pit->first;
 673       const pg_pool_t &pool = pit->second;
 674       f.open_object_section("pool");
 675       f.dump_int("pool_id", pool_id);
 676       f.open_object_section("pool_dump");
 677       pool.dump(&f);
 678       f.close_section();
 679       f.close_section();
 680     }
 681     f.close_section();
 682     f.flush(*_dout);
 683     *_dout << dendl;
 684   }
 685
 686   void op_log() {
 687     dout(10) << __func__ << dendl;
 688
 689     MLog *m = new MLog(monc.get_fsid());
 690
 691     boost::uniform_int<> log_rng(1, 10);
 692     size_t num_entries = log_rng(gen);
 693     dout(10) << __func__
 694              << " send " << num_entries << " log messages" << dendl;
 695
 696     utime_t now = ceph_clock_now();
 697     int seq = 0;
 698     for (; num_entries > 0; --num_entries) {
 699       LogEntry e;
 700       e.rank = messenger->get_myname();
 701       e.addrs = messenger->get_myaddrs();
 702       e.stamp = now;
 703       e.seq = seq++;
 704       e.prio = CLOG_DEBUG;
 705       e.msg = "OSDStub::op_log";
 706       m->entries.push_back(e);
 707     }
 708
 709     monc.send_mon_message(m);
 710   }
 711
 712   void _tick() override {
 713     if (!osdmap.exists(whoami)) {
 714       std::cout << __func__ << " not in the cluster; boot!" << std::endl;
 715       boot();
 716       return;
 717     }
 718
 719     update_osd_stat();
 720
 721     boost::uniform_int<> op_rng(STUB_MON_OSD_FIRST, STUB_MON_OSD_LAST);
 722     int op = op_rng(gen);
 723     switch (op) {
 724     case STUB_MON_OSD_ALIVE:
 725       op_alive();
 726       break;
 727     case STUB_MON_OSD_PGTEMP:
 728       op_pgtemp();
 729       break;
 730     case STUB_MON_OSD_FAILURE:
 731       op_failure();
 732       break;
 733     case STUB_MON_OSD_PGSTATS:
 734       op_pgstats();
 735       break;
 736     case STUB_MON_LOG:
 737       op_log();
 738       break;
 739     }
 740   }
 741
 742   void handle_pg_create(MOSDPGCreate *m) {
 743     ceph_assert(m != NULL);
 744     if (m->epoch < osdmap.get_epoch()) {
 745       std::cout << __func__ << " epoch " << m->epoch << " < "
 746                << osdmap.get_epoch() << "; dropping" << std::endl;
 747       m->put();
 748       return;
 749     }
 750
 751     for (map<pg_t,pg_create_t>::iterator it = m->mkpg.begin();
 752          it != m->mkpg.end(); ++it) {
 753       pg_create_t &c = it->second;
 754       std::cout << __func__ << " pg " << it->first
 755               << " created " << c.created
 756               << " parent " << c.parent << std::endl;
 757       if (pgs.count(it->first)) {
 758         std::cout << __func__ << " pg " << it->first
 759                  << " exists; skipping" << std::endl;
 760         continue;
 761       }
 762
 763       pg_t pgid = it->first;
 764       add_pg(pgid, c.created, c.parent);
 765     }
 766     send_pg_stats();
 767   }
 768
 769   void handle_osd_map(MOSDMap *m) {
 770     dout(1) << __func__ << dendl;
 771     if (m->fsid != monc.get_fsid()) {
 772       dout(0) << __func__
 773               << " message fsid " << m->fsid << " != " << monc.get_fsid()
 774               << dendl;
 775       dout(0) << __func__ << " " << m
 776               << " from " << m->get_source_inst()
 777               << dendl;
 778       dout(0) << monc.get_monmap() << dendl;
 779     }
 780     ceph_assert(m->fsid == monc.get_fsid());
 781
 782     epoch_t first = m->get_first();
 783     epoch_t last = m->get_last();
 784     dout(5) << __func__
 785             << " epochs [" << first << "," << last << "]"
 786             << " current " << osdmap.get_epoch() << dendl;
 787
 788     if (last <= osdmap.get_epoch()) {
 789       dout(5) << __func__ << " no new maps here; dropping" << dendl;
 790       m->put();
 791       return;
 792     }
 793
 794     if (first > osdmap.get_epoch() + 1) {
 795       dout(5) << __func__
 796               << osdmap.get_epoch() + 1 << ".." << (first-1) << dendl;
 797       if ((m->oldest_map < first && osdmap.get_epoch() == 0) ||
 798           m->oldest_map <= osdmap.get_epoch()) {
 799         monc.sub_want("osdmap", osdmap.get_epoch()+1,
 800                        CEPH_SUBSCRIBE_ONETIME);
 801         monc.renew_subs();
 802         m->put();
 803         return;
 804       }
 805     }
 806
 807     epoch_t start_full = std::max(osdmap.get_epoch() + 1, first);
 808
 809     if (m->maps.size() > 0) {
 810       map<epoch_t,bufferlist>::reverse_iterator rit;
 811       rit = m->maps.rbegin();
 812       if (start_full <= rit->first) {
 813         start_full = rit->first;
 814         dout(5) << __func__
 815                 << " full epoch " << start_full << dendl;
 816         bufferlist &bl = rit->second;
 817         auto p = bl.cbegin();
 818         osdmap.decode(p);
 819       }
 820     }
 821
 822     for (epoch_t e = start_full; e <= last; e++) {
 823       map<epoch_t,bufferlist>::iterator it;
 824       it = m->incremental_maps.find(e);
 825       if (it == m->incremental_maps.end())
 826         continue;
 827
 828       dout(20) << __func__
 829                << " incremental epoch " << e
 830                << " on full epoch " << start_full << dendl;
 831       OSDMap::Incremental inc;
 832       bufferlist &bl = it->second;
 833       auto p = bl.cbegin();
 834       inc.decode(p);
 835
 836       int err = osdmap.apply_incremental(inc);
 837       if (err < 0) {
 838         derr << "osd." << whoami << "::" << __func__
 839              << "** ERROR: applying incremental: "
 840              << cpp_strerror(err) << dendl;
 841         ceph_abort_msg("error applying incremental");
 842       }
 843     }
 844     dout(30) << __func__ << "\nosdmap:\n";
 845     JSONFormatter f(true);
 846     osdmap.dump(&f);
 847     f.flush(*_dout);
 848     *_dout << dendl;
 849
 850     if (osdmap.is_up(whoami) &&
 851         osdmap.get_addrs(whoami) == messenger->get_myaddrs()) {
 852       dout(1) << __func__
 853               << " got into the osdmap and we're up!" << dendl;
 854     }
 855
 856     if (m->newest_map && m->newest_map > last) {
 857       dout(1) << __func__
 858               << " they have more maps; requesting them!" << dendl;
 859       monc.sub_want("osdmap", osdmap.get_epoch()+1, CEPH_SUBSCRIBE_ONETIME);
 860       monc.renew_subs();
 861     }
 862
 863     dout(10) << __func__ << " done" << dendl;
 864     m->put();
 865   }
 866
 867   bool ms_dispatch(Message *m) override {
 868     dout(1) << __func__ << " " << *m << dendl;
 869
 870     switch (m->get_type()) {
 871     case MSG_OSD_PG_CREATE:
 872       handle_pg_create((MOSDPGCreate*)m);
 873       break;
 874     case CEPH_MSG_OSD_MAP:
 875       handle_osd_map((MOSDMap*)m);
 876       break;
 877     default:
 878       m->put();
 879       break;
 880     }
 881     return true;
 882   }
 883
 884   void ms_handle_connect(Connection *con) override {
 885     dout(1) << __func__ << " " << con << dendl;
 886     if (con->get_peer_type() == CEPH_ENTITY_TYPE_MON) {
 887       dout(10) << __func__ << " on mon" << dendl;
 888     }
 889   }
 890
 891   void ms_handle_remote_reset(Connection *con) override {}
 892
 893   bool ms_handle_reset(Connection *con) override {
 894     dout(1) << __func__ << dendl;
 895     return con->get_priv().get();
 896   }
 897
 898   bool ms_handle_refused(Connection *con) override {
 899     return false;
 900   }
 901
 902   const string get_name() override {
 903     stringstream ss;
 904     ss << "osd." << whoami;
 905     return ss.str();
 906   }
 907 };
 908
 909 double const OSDStub::STUB_BOOT_INTERVAL = 10.0;
 910
 911 #undef dout_prefix
 912 #define dout_prefix *_dout << "main "
 913
 914 const char *our_name = NULL;
 915 vector<TestStub*> stubs;
 916 ceph::mutex shutdown_lock = ceph::make_mutex("main::shutdown_lock");
 917 ceph::condition_variable shutdown_cond;
 918 Context *shutdown_cb = NULL;
 919 SafeTimer *shutdown_timer = NULL;
 920
 921 struct C_Shutdown : public Context
 922 {
 923   void finish(int r) override {
 924     generic_dout(10) << "main::shutdown time has ran out" << dendl;
 925     shutdown_cond.notify_all();
 926   }
 927 };
 928
 929 void handle_test_signal(int signum)
 930 {
 931   if ((signum != SIGINT) && (signum != SIGTERM))
 932     return;
 933
 934   std::cerr << "*** Got signal " << sig_str(signum) << " ***" << std::endl;
 935   std::lock_guard l{shutdown_lock};
 936   if (shutdown_timer) {
 937     shutdown_timer->cancel_all_events();
 938     shutdown_cond.notify_all();
 939   }
 940 }
 941
 942 void usage() {
 943   ceph_assert(our_name != NULL);
 944
 945   std::cout << "usage: " << our_name
 946             << " <--stub-id ID> [--stub-id ID...]"
 947             << std::endl;
 948   std::cout << "\n\
 949 Global Options:\n\
 950   -c FILE                   Read configuration from FILE\n\
 951   --keyring FILE            Read keyring from FILE\n\
 952   --help                    This message\n\
 953 \n\
 954 Test-specific Options:\n\
 955   --stub-id ID1..ID2        Interval of OSD ids for multiple stubs to mimic.\n\
 956   --stub-id ID              OSD id a stub will mimic to be\n\
 957                             (same as --stub-id ID..ID)\n\
 958 " << std::endl;
 959 }
 960
 961 int get_id_interval(int &first, int &last, string &str)
 962 {
 963   size_t found = str.find("..");
 964   string first_str, last_str;
 965   if (found == string::npos) {
 966     first_str = last_str = str;
 967   } else {
 968     first_str = str.substr(0, found);
 969     last_str = str.substr(found+2);
 970   }
 971
 972   string err;
 973   first = strict_strtol(first_str.c_str(), 10, &err);
 974   if ((first == 0) && (!err.empty())) {
 975     std::cerr << err << std::endl;
 976     return -1;
 977   }
 978
 979   last = strict_strtol(last_str.c_str(), 10, &err);
 980   if ((last == 0) && (!err.empty())) {
 981     std::cerr << err << std::endl;
 982     return -1;
 983   }
 984   return 0;
 985 }
 986
 987 int main(int argc, const char *argv[])
 988 {
 989   vector<const char*> args;
 990   our_name = argv[0];
 991   argv_to_vec(argc, argv, args);
 992
 993   auto cct = global_init(NULL, args,
 994                          CEPH_ENTITY_TYPE_OSD, CODE_ENVIRONMENT_UTILITY,
 995                          CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
 996
 997   common_init_finish(g_ceph_context);
 998   g_ceph_context->_conf.apply_changes(nullptr);
 999
1000   set<int> stub_ids;
1001   double duration = 300.0;
1002
1003   for (std::vector<const char*>::iterator i = args.begin(); i != args.end();) {
1004     string val;
1005
1006     if (ceph_argparse_double_dash(args, i)) {
1007       break;
1008     } else if (ceph_argparse_witharg(args, i, &val,
1009         "--stub-id", (char*) NULL)) {
1010       int first = -1, last = -1;
1011       if (get_id_interval(first, last, val) < 0) {
1012         std::cerr << "** error parsing stub id '" << val << "'" << std::endl;
1013         exit(1);
1014       }
1015
1016       for (; first <= last; ++first)
1017         stub_ids.insert(first);
1018     } else if (ceph_argparse_witharg(args, i, &val,
1019         "--duration", (char*) NULL)) {
1020       string err;
1021       duration = (double) strict_strtol(val.c_str(), 10, &err);
1022       if ((duration == 0) && (!err.empty())) {
1023         std::cerr << "** error parsing '--duration " << val << "': '"
1024                   << err << std::endl;
1025         exit(1);
1026       }
1027     } else if (ceph_argparse_flag(args, i, "--help", (char*) NULL)) {
1028       usage();
1029       exit(0);
1030     } else {
1031       std::cerr << "unknown argument '" << *i << "'" << std::endl;
1032       return 1;
1033     }
1034   }
1035
1036   if (stub_ids.empty()) {
1037     std::cerr << "** error: must specify at least one '--stub-id <ID>'"
1038          << std::endl;
1039     usage();
1040     return 1;
1041   }
1042
1043   for (set<int>::iterator i = stub_ids.begin(); i != stub_ids.end(); ++i) {
1044     int whoami = *i;
1045
1046     std::cout << __func__ << " starting stub." << whoami << std::endl;
1047     OSDStub *stub = new OSDStub(whoami, g_ceph_context);
1048     int err = stub->init();
1049     if (err < 0) {
1050       std::cerr << "** osd stub error: " << cpp_strerror(-err) << std::endl;
1051       return 1;
1052     }
1053     stubs.push_back(stub);
1054   }
1055
1056   std::cout << __func__ << " starting client stub" << std::endl;
1057   ClientStub *cstub = new ClientStub(g_ceph_context);
1058   int err = cstub->init();
1059   if (err < 0) {
1060     std::cerr << "** client stub error: " << cpp_strerror(-err) << std::endl;
1061     return 1;
1062   }
1063   stubs.push_back(cstub);
1064
1065   init_async_signal_handler();
1066   register_async_signal_handler_oneshot(SIGINT, handle_test_signal);
1067   register_async_signal_handler_oneshot(SIGTERM, handle_test_signal);
1068
1069   {
1070     unique_lock locker{shutdown_lock};
1071     shutdown_timer = new SafeTimer(g_ceph_context, shutdown_lock);
1072     shutdown_timer->init();
1073     if (duration != 0) {
1074       std::cout << __func__
1075                 << " run test for " << duration << " seconds" << std::endl;
1076       shutdown_timer->add_event_after((double) duration, new C_Shutdown);
1077     }
1078     shutdown_cond.wait(locker);
1079     shutdown_timer->shutdown();
1080     delete shutdown_timer;
1081     shutdown_timer = NULL;
1082   }
1083   unregister_async_signal_handler(SIGINT, handle_test_signal);
1084   unregister_async_signal_handler(SIGTERM, handle_test_signal);
1085
1086   std::cout << __func__ << " waiting for stubs to finish" << std::endl;
1087   vector<TestStub*>::iterator it;
1088   int i;
1089   for (i = 0, it = stubs.begin(); it != stubs.end(); ++it, ++i) {
1090     if (*it != NULL) {
1091       (*it)->shutdown();
1092       (*it)->wait();
1093       std::cout << __func__ << " finished " << (*it)->get_name() << std::endl;
1094       delete (*it);
1095       (*it) = NULL;
1096     }
1097   }
1098
1099   return 0;
1100 }