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