ceph/src/ceph_osd.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
  15 #include <sys/types.h>
  16 #include <sys/stat.h>
  17 #include <fcntl.h>
  18 #include <boost/scoped_ptr.hpp>
  19
  20 #include <iostream>
  21 #include <string>
  22
  23 #include "osd/OSD.h"
  24 #include "os/ObjectStore.h"
  25 #include "mon/MonClient.h"
  26 #include "include/ceph_features.h"
  27
  28 #include "common/config.h"
  29
  30 #include "mon/MonMap.h"
  31
  32 #include "msg/Messenger.h"
  33
  34 #include "common/Throttle.h"
  35 #include "common/Timer.h"
  36 #include "common/TracepointProvider.h"
  37 #include "common/ceph_argparse.h"
  38 #include "common/numa.h"
  39
  40 #include "global/global_init.h"
  41 #include "global/signal_handler.h"
  42
  43 #include "include/color.h"
  44 #include "common/errno.h"
  45 #include "common/pick_address.h"
  46
  47 #include "perfglue/heap_profiler.h"
  48
  49 #include "include/ceph_assert.h"
  50
  51 #include "common/Preforker.h"
  52
  53 #define dout_context g_ceph_context
  54 #define dout_subsys ceph_subsys_osd
  55
  56 namespace {
  57
  58 TracepointProvider::Traits osd_tracepoint_traits("libosd_tp.so",
  59                                                  "osd_tracing");
  60 TracepointProvider::Traits os_tracepoint_traits("libos_tp.so",
  61                                                 "osd_objectstore_tracing");
  62 TracepointProvider::Traits bluestore_tracepoint_traits("libbluestore_tp.so",
  63                                                        "bluestore_tracing");
  64 #ifdef WITH_OSD_INSTRUMENT_FUNCTIONS
  65 TracepointProvider::Traits cyg_profile_traits("libcyg_profile_tp.so",
  66                                                  "osd_function_tracing");
  67 #endif
  68
  69 } // anonymous namespace
  70
  71 OSD *osdptr = nullptr;
  72
  73 void handle_osd_signal(int signum)
  74 {
  75   if (osdptr)
  76     osdptr->handle_signal(signum);
  77 }
  78
  79 static void usage()
  80 {
  81   cout << "usage: ceph-osd -i <ID> [flags]\n"
  82        << "  --osd-data PATH data directory\n"
  83        << "  --osd-journal PATH\n"
  84        << "                    journal file or block device\n"
  85        << "  --mkfs            create a [new] data directory\n"
  86        << "  --mkkey           generate a new secret key. This is normally used in combination with --mkfs\n"
  87        << "  --monmap          specify the path to the monitor map. This is normally used in combination with --mkfs\n"
  88        << "  --osd-uuid        specify the OSD's fsid. This is normally used in combination with --mkfs\n"
  89        << "  --keyring         specify a path to the osd keyring. This is normally used in combination with --mkfs\n"
  90        << "  --convert-filestore\n"
  91        << "                    run any pending upgrade operations\n"
  92        << "  --flush-journal   flush all data out of journal\n"
  93        << "  --osdspec-affinity\n"
  94        << "                    set affinity to an osdspec\n"
  95        << "  --dump-journal    dump all data of journal\n"
  96        << "  --mkjournal       initialize a new journal\n"
  97        << "  --check-wants-journal\n"
  98        << "                    check whether a journal is desired\n"
  99        << "  --check-allows-journal\n"
 100        << "                    check whether a journal is allowed\n"
 101        << "  --check-needs-journal\n"
 102        << "                    check whether a journal is required\n"
 103        << "  --debug_osd <N>   set debug level (e.g. 10)\n"
 104        << "  --get-device-fsid PATH\n"
 105        << "                    get OSD fsid for the given block device\n"
 106        << std::endl;
 107   generic_server_usage();
 108 }
 109
 110 int main(int argc, const char **argv)
 111 {
 112   vector<const char*> args;
 113   argv_to_vec(argc, argv, args);
 114   if (args.empty()) {
 115     cerr << argv[0] << ": -h or --help for usage" << std::endl;
 116     exit(1);
 117   }
 118   if (ceph_argparse_need_usage(args)) {
 119     usage();
 120     exit(0);
 121   }
 122
 123   map<string,string> defaults = {
 124     // We want to enable leveldb's log, while allowing users to override this
 125     // option, therefore we will pass it as a default argument to global_init().
 126     { "leveldb_log", "" }
 127   };
 128   auto cct = global_init(
 129     &defaults,
 130     args, CEPH_ENTITY_TYPE_OSD,
 131     CODE_ENVIRONMENT_DAEMON,
 132     0, "osd_data");
 133   ceph_heap_profiler_init();
 134
 135   Preforker forker;
 136
 137   // osd specific args
 138   bool mkfs = false;
 139   bool mkjournal = false;
 140   bool check_wants_journal = false;
 141   bool check_allows_journal = false;
 142   bool check_needs_journal = false;
 143   bool mkkey = false;
 144   bool flushjournal = false;
 145   bool dump_journal = false;
 146   bool convertfilestore = false;
 147   bool get_osd_fsid = false;
 148   bool get_cluster_fsid = false;
 149   bool get_journal_fsid = false;
 150   bool get_device_fsid = false;
 151   string device_path;
 152   std::string dump_pg_log;
 153   std::string osdspec_affinity;
 154
 155   std::string val;
 156   for (std::vector<const char*>::iterator i = args.begin(); i != args.end(); ) {
 157     if (ceph_argparse_double_dash(args, i)) {
 158       break;
 159     } else if (ceph_argparse_flag(args, i, "--mkfs", (char*)NULL)) {
 160       mkfs = true;
 161     } else if (ceph_argparse_witharg(args, i, &val, "--osdspec-affinity", (char*)NULL)) {
 162      osdspec_affinity = val;
 163     } else if (ceph_argparse_flag(args, i, "--mkjournal", (char*)NULL)) {
 164       mkjournal = true;
 165     } else if (ceph_argparse_flag(args, i, "--check-allows-journal", (char*)NULL)) {
 166       check_allows_journal = true;
 167     } else if (ceph_argparse_flag(args, i, "--check-wants-journal", (char*)NULL)) {
 168       check_wants_journal = true;
 169     } else if (ceph_argparse_flag(args, i, "--check-needs-journal", (char*)NULL)) {
 170       check_needs_journal = true;
 171     } else if (ceph_argparse_flag(args, i, "--mkkey", (char*)NULL)) {
 172       mkkey = true;
 173     } else if (ceph_argparse_flag(args, i, "--flush-journal", (char*)NULL)) {
 174       flushjournal = true;
 175     } else if (ceph_argparse_flag(args, i, "--convert-filestore", (char*)NULL)) {
 176       convertfilestore = true;
 177     } else if (ceph_argparse_witharg(args, i, &val, "--dump-pg-log", (char*)NULL)) {
 178       dump_pg_log = val;
 179     } else if (ceph_argparse_flag(args, i, "--dump-journal", (char*)NULL)) {
 180       dump_journal = true;
 181     } else if (ceph_argparse_flag(args, i, "--get-cluster-fsid", (char*)NULL)) {
 182       get_cluster_fsid = true;
 183     } else if (ceph_argparse_flag(args, i, "--get-osd-fsid", "--get-osd-uuid", (char*)NULL)) {
 184       get_osd_fsid = true;
 185     } else if (ceph_argparse_flag(args, i, "--get-journal-fsid", "--get-journal-uuid", (char*)NULL)) {
 186       get_journal_fsid = true;
 187     } else if (ceph_argparse_witharg(args, i, &device_path,
 188                                      "--get-device-fsid", (char*)NULL)) {
 189       get_device_fsid = true;
 190     } else {
 191       ++i;
 192     }
 193   }
 194   if (!args.empty()) {
 195     cerr << "unrecognized arg " << args[0] << std::endl;
 196     exit(1);
 197   }
 198
 199   if (global_init_prefork(g_ceph_context) >= 0) {
 200     std::string err;
 201     int r = forker.prefork(err);
 202     if (r < 0) {
 203       cerr << err << std::endl;
 204       return r;
 205     }
 206     if (forker.is_parent()) {
 207       g_ceph_context->_log->start();
 208       if (forker.parent_wait(err) != 0) {
 209         return -ENXIO;
 210       }
 211       return 0;
 212     }
 213     setsid();
 214     global_init_postfork_start(g_ceph_context);
 215   }
 216   common_init_finish(g_ceph_context);
 217   global_init_chdir(g_ceph_context);
 218
 219   if (get_journal_fsid) {
 220     device_path = g_conf().get_val<std::string>("osd_journal");
 221     get_device_fsid = true;
 222   }
 223   if (get_device_fsid) {
 224     uuid_d uuid;
 225     int r = ObjectStore::probe_block_device_fsid(g_ceph_context, device_path,
 226                                                  &uuid);
 227     if (r < 0) {
 228       cerr << "failed to get device fsid for " << device_path
 229            << ": " << cpp_strerror(r) << std::endl;
 230       forker.exit(1);
 231     }
 232     cout << uuid << std::endl;
 233     forker.exit(0);
 234   }
 235
 236   if (!dump_pg_log.empty()) {
 237     common_init_finish(g_ceph_context);
 238     bufferlist bl;
 239     std::string error;
 240
 241     if (bl.read_file(dump_pg_log.c_str(), &error) >= 0) {
 242       pg_log_entry_t e;
 243       auto p = bl.cbegin();
 244       while (!p.end()) {
 245         uint64_t pos = p.get_off();
 246         try {
 247           decode(e, p);
 248         }
 249         catch (const buffer::error &e) {
 250           derr << "failed to decode LogEntry at offset " << pos << dendl;
 251           forker.exit(1);
 252         }
 253         derr << pos << ":\t" << e << dendl;
 254       }
 255     } else {
 256       derr << "unable to open " << dump_pg_log << ": " << error << dendl;
 257     }
 258     forker.exit(0);
 259   }
 260
 261   // whoami
 262   char *end;
 263   const char *id = g_conf()->name.get_id().c_str();
 264   int whoami = strtol(id, &end, 10);
 265   std::string data_path = g_conf().get_val<std::string>("osd_data");
 266   if (*end || end == id || whoami < 0) {
 267     derr << "must specify '-i #' where # is the osd number" << dendl;
 268     forker.exit(1);
 269   }
 270
 271   if (data_path.empty()) {
 272     derr << "must specify '--osd-data=foo' data path" << dendl;
 273     forker.exit(1);
 274   }
 275
 276   // the store
 277   std::string store_type;
 278   {
 279     char fn[PATH_MAX];
 280     snprintf(fn, sizeof(fn), "%s/type", data_path.c_str());
 281     int fd = ::open(fn, O_RDONLY|O_CLOEXEC);
 282     if (fd >= 0) {
 283       bufferlist bl;
 284       bl.read_fd(fd, 64);
 285       if (bl.length()) {
 286         store_type = string(bl.c_str(), bl.length() - 1);  // drop \n
 287         dout(5) << "object store type is " << store_type << dendl;
 288       }
 289       ::close(fd);
 290     } else if (mkfs) {
 291       store_type = g_conf().get_val<std::string>("osd_objectstore");
 292     } else {
 293       // hrm, infer the type
 294       snprintf(fn, sizeof(fn), "%s/current", data_path.c_str());
 295       struct stat st;
 296       if (::stat(fn, &st) == 0 &&
 297           S_ISDIR(st.st_mode)) {
 298         derr << "missing 'type' file, inferring filestore from current/ dir"
 299              << dendl;
 300         store_type = "filestore";
 301       } else {
 302         snprintf(fn, sizeof(fn), "%s/block", data_path.c_str());
 303         if (::stat(fn, &st) == 0 &&
 304             S_ISLNK(st.st_mode)) {
 305           derr << "missing 'type' file, inferring bluestore from block symlink"
 306                << dendl;
 307           store_type = "bluestore";
 308         } else {
 309           derr << "missing 'type' file and unable to infer osd type" << dendl;
 310           forker.exit(1);
 311         }
 312       }
 313     }
 314   }
 315
 316   std::string journal_path = g_conf().get_val<std::string>("osd_journal");
 317   uint32_t flags = g_conf().get_val<uint64_t>("osd_os_flags");
 318   ObjectStore *store = ObjectStore::create(g_ceph_context,
 319                                            store_type,
 320                                            data_path,
 321                                            journal_path,
 322                                            flags);
 323   if (!store) {
 324     derr << "unable to create object store" << dendl;
 325     forker.exit(-ENODEV);
 326   }
 327
 328
 329   if (mkkey) {
 330     common_init_finish(g_ceph_context);
 331     KeyRing keyring;
 332
 333     EntityName ename{g_conf()->name};
 334     EntityAuth eauth;
 335
 336     std::string keyring_path = g_conf().get_val<std::string>("keyring");
 337     int ret = keyring.load(g_ceph_context, keyring_path);
 338     if (ret == 0 &&
 339         keyring.get_auth(ename, eauth)) {
 340       derr << "already have key in keyring " << keyring_path << dendl;
 341     } else {
 342       eauth.key.create(g_ceph_context, CEPH_CRYPTO_AES);
 343       keyring.add(ename, eauth);
 344       bufferlist bl;
 345       keyring.encode_plaintext(bl);
 346       int r = bl.write_file(keyring_path.c_str(), 0600);
 347       if (r)
 348         derr << TEXT_RED << " ** ERROR: writing new keyring to "
 349              << keyring_path << ": " << cpp_strerror(r) << TEXT_NORMAL
 350              << dendl;
 351       else
 352         derr << "created new key in keyring " << keyring_path << dendl;
 353     }
 354   }
 355
 356   if (mkfs) {
 357     common_init_finish(g_ceph_context);
 358
 359     if (g_conf().get_val<uuid_d>("fsid").is_zero()) {
 360       derr << "must specify cluster fsid" << dendl;
 361       forker.exit(-EINVAL);
 362     }
 363
 364     int err = OSD::mkfs(g_ceph_context, store, g_conf().get_val<uuid_d>("fsid"),
 365                         whoami, osdspec_affinity);
 366     if (err < 0) {
 367       derr << TEXT_RED << " ** ERROR: error creating empty object store in "
 368            << data_path << ": " << cpp_strerror(-err) << TEXT_NORMAL << dendl;
 369       forker.exit(1);
 370     }
 371     dout(0) << "created object store " << data_path
 372             << " for osd." << whoami
 373             << " fsid " << g_conf().get_val<uuid_d>("fsid")
 374             << dendl;
 375   }
 376   if (mkfs || mkkey) {
 377     forker.exit(0);
 378   }
 379   if (mkjournal) {
 380     common_init_finish(g_ceph_context);
 381     int err = store->mkjournal();
 382     if (err < 0) {
 383       derr << TEXT_RED << " ** ERROR: error creating fresh journal "
 384            << journal_path << " for object store " << data_path << ": "
 385            << cpp_strerror(-err) << TEXT_NORMAL << dendl;
 386       forker.exit(1);
 387     }
 388     derr << "created new journal " << journal_path
 389          << " for object store " << data_path << dendl;
 390     forker.exit(0);
 391   }
 392   if (check_wants_journal) {
 393     if (store->wants_journal()) {
 394       cout << "wants journal: yes" << std::endl;
 395       forker.exit(0);
 396     } else {
 397       cout << "wants journal: no" << std::endl;
 398       forker.exit(1);
 399     }
 400   }
 401   if (check_allows_journal) {
 402     if (store->allows_journal()) {
 403       cout << "allows journal: yes" << std::endl;
 404       forker.exit(0);
 405     } else {
 406       cout << "allows journal: no" << std::endl;
 407       forker.exit(1);
 408     }
 409   }
 410   if (check_needs_journal) {
 411     if (store->needs_journal()) {
 412       cout << "needs journal: yes" << std::endl;
 413       forker.exit(0);
 414     } else {
 415       cout << "needs journal: no" << std::endl;
 416       forker.exit(1);
 417     }
 418   }
 419   if (flushjournal) {
 420     common_init_finish(g_ceph_context);
 421     int err = store->mount();
 422     if (err < 0) {
 423       derr << TEXT_RED << " ** ERROR: error flushing journal " << journal_path
 424            << " for object store " << data_path
 425            << ": " << cpp_strerror(-err) << TEXT_NORMAL << dendl;
 426       goto flushjournal_out;
 427     }
 428     store->umount();
 429     derr << "flushed journal " << journal_path
 430          << " for object store " << data_path
 431          << dendl;
 432 flushjournal_out:
 433     delete store;
 434     forker.exit(err < 0 ? 1 : 0);
 435   }
 436   if (dump_journal) {
 437     common_init_finish(g_ceph_context);
 438     int err = store->dump_journal(cout);
 439     if (err < 0) {
 440       derr << TEXT_RED << " ** ERROR: error dumping journal " << journal_path
 441            << " for object store " << data_path
 442            << ": " << cpp_strerror(-err) << TEXT_NORMAL << dendl;
 443       forker.exit(1);
 444     }
 445     derr << "dumped journal " << journal_path
 446          << " for object store " << data_path
 447          << dendl;
 448     forker.exit(0);
 449   }
 450
 451   if (convertfilestore) {
 452     int err = store->mount();
 453     if (err < 0) {
 454       derr << TEXT_RED << " ** ERROR: error mounting store " << data_path
 455            << ": " << cpp_strerror(-err) << TEXT_NORMAL << dendl;
 456       forker.exit(1);
 457     }
 458     err = store->upgrade();
 459     store->umount();
 460     if (err < 0) {
 461       derr << TEXT_RED << " ** ERROR: error converting store " << data_path
 462            << ": " << cpp_strerror(-err) << TEXT_NORMAL << dendl;
 463       forker.exit(1);
 464     }
 465     forker.exit(0);
 466   }
 467
 468   string magic;
 469   uuid_d cluster_fsid, osd_fsid;
 470   ceph_release_t require_osd_release = ceph_release_t::unknown;
 471   int w;
 472   int r = OSD::peek_meta(store, &magic, &cluster_fsid, &osd_fsid, &w,
 473                          &require_osd_release);
 474   if (r < 0) {
 475     derr << TEXT_RED << " ** ERROR: unable to open OSD superblock on "
 476          << data_path << ": " << cpp_strerror(-r)
 477          << TEXT_NORMAL << dendl;
 478     if (r == -ENOTSUP) {
 479       derr << TEXT_RED << " **        please verify that underlying storage "
 480            << "supports xattrs" << TEXT_NORMAL << dendl;
 481     }
 482     forker.exit(1);
 483   }
 484   if (w != whoami) {
 485     derr << "OSD id " << w << " != my id " << whoami << dendl;
 486     forker.exit(1);
 487   }
 488   if (strcmp(magic.c_str(), CEPH_OSD_ONDISK_MAGIC)) {
 489     derr << "OSD magic " << magic << " != my " << CEPH_OSD_ONDISK_MAGIC
 490          << dendl;
 491     forker.exit(1);
 492   }
 493
 494   if (get_cluster_fsid) {
 495     cout << cluster_fsid << std::endl;
 496     forker.exit(0);
 497   }
 498   if (get_osd_fsid) {
 499     cout << osd_fsid << std::endl;
 500     forker.exit(0);
 501   }
 502
 503   {
 504     auto from_release = require_osd_release;
 505     ostringstream err;
 506     if (!can_upgrade_from(from_release, "require_osd_release", err)) {
 507       derr << err.str() << dendl;
 508       forker.exit(1);
 509     }
 510   }
 511
 512   // consider objectstore numa node
 513   int os_numa_node = -1;
 514   r = store->get_numa_node(&os_numa_node, nullptr, nullptr);
 515   if (r >= 0 && os_numa_node >= 0) {
 516     dout(1) << " objectstore numa_node " << os_numa_node << dendl;
 517   }
 518   int iface_preferred_numa_node = -1;
 519   if (g_conf().get_val<bool>("osd_numa_prefer_iface")) {
 520     iface_preferred_numa_node = os_numa_node;
 521   }
 522
 523   // messengers
 524   std::string msg_type = g_conf().get_val<std::string>("ms_type");
 525   std::string public_msg_type =
 526     g_conf().get_val<std::string>("ms_public_type");
 527   std::string cluster_msg_type =
 528     g_conf().get_val<std::string>("ms_cluster_type");
 529
 530   public_msg_type = public_msg_type.empty() ? msg_type : public_msg_type;
 531   cluster_msg_type = cluster_msg_type.empty() ? msg_type : cluster_msg_type;
 532   uint64_t nonce = Messenger::get_pid_nonce();
 533   Messenger *ms_public = Messenger::create(g_ceph_context, public_msg_type,
 534                                            entity_name_t::OSD(whoami), "client",
 535                                            nonce,
 536                                            Messenger::HAS_HEAVY_TRAFFIC |
 537                                            Messenger::HAS_MANY_CONNECTIONS);
 538   Messenger *ms_cluster = Messenger::create(g_ceph_context, cluster_msg_type,
 539                                             entity_name_t::OSD(whoami), "cluster",
 540                                             nonce,
 541                                             Messenger::HAS_HEAVY_TRAFFIC |
 542                                             Messenger::HAS_MANY_CONNECTIONS);
 543   Messenger *ms_hb_back_client = Messenger::create(g_ceph_context, cluster_msg_type,
 544                                              entity_name_t::OSD(whoami), "hb_back_client",
 545                                              nonce, Messenger::HEARTBEAT);
 546   Messenger *ms_hb_front_client = Messenger::create(g_ceph_context, public_msg_type,
 547                                              entity_name_t::OSD(whoami), "hb_front_client",
 548                                              nonce, Messenger::HEARTBEAT);
 549   Messenger *ms_hb_back_server = Messenger::create(g_ceph_context, cluster_msg_type,
 550                                                    entity_name_t::OSD(whoami), "hb_back_server",
 551                                                    nonce, Messenger::HEARTBEAT);
 552   Messenger *ms_hb_front_server = Messenger::create(g_ceph_context, public_msg_type,
 553                                                     entity_name_t::OSD(whoami), "hb_front_server",
 554                                                     nonce, Messenger::HEARTBEAT);
 555   Messenger *ms_objecter = Messenger::create(g_ceph_context, public_msg_type,
 556                                              entity_name_t::OSD(whoami), "ms_objecter",
 557                                              nonce, 0);
 558   if (!ms_public || !ms_cluster || !ms_hb_front_client || !ms_hb_back_client || !ms_hb_back_server || !ms_hb_front_server || !ms_objecter)
 559     forker.exit(1);
 560   ms_cluster->set_cluster_protocol(CEPH_OSD_PROTOCOL);
 561   ms_hb_front_client->set_cluster_protocol(CEPH_OSD_PROTOCOL);
 562   ms_hb_back_client->set_cluster_protocol(CEPH_OSD_PROTOCOL);
 563   ms_hb_back_server->set_cluster_protocol(CEPH_OSD_PROTOCOL);
 564   ms_hb_front_server->set_cluster_protocol(CEPH_OSD_PROTOCOL);
 565
 566   dout(0) << "starting osd." << whoami
 567           << " osd_data " << data_path
 568           << " " << ((journal_path.empty()) ?
 569                     "(no journal)" : journal_path)
 570           << dendl;
 571
 572   uint64_t message_size =
 573     g_conf().get_val<Option::size_t>("osd_client_message_size_cap");
 574   boost::scoped_ptr<Throttle> client_byte_throttler(
 575     new Throttle(g_ceph_context, "osd_client_bytes", message_size));
 576
 577   // All feature bits 0 - 34 should be present from dumpling v0.67 forward
 578   uint64_t osd_required =
 579     CEPH_FEATURE_UID |
 580     CEPH_FEATURE_PGID64 |
 581     CEPH_FEATURE_OSDENC;
 582
 583   ms_public->set_default_policy(Messenger::Policy::stateless_registered_server(0));
 584   ms_public->set_policy_throttlers(entity_name_t::TYPE_CLIENT,
 585                                    client_byte_throttler.get(),
 586                                    nullptr);
 587   ms_public->set_policy(entity_name_t::TYPE_MON,
 588                         Messenger::Policy::lossy_client(osd_required));
 589   ms_public->set_policy(entity_name_t::TYPE_MGR,
 590                         Messenger::Policy::lossy_client(osd_required));
 591
 592   ms_cluster->set_default_policy(Messenger::Policy::stateless_server(0));
 593   ms_cluster->set_policy(entity_name_t::TYPE_MON, Messenger::Policy::lossy_client(0));
 594   ms_cluster->set_policy(entity_name_t::TYPE_OSD,
 595                          Messenger::Policy::lossless_peer(osd_required));
 596   ms_cluster->set_policy(entity_name_t::TYPE_CLIENT,
 597                          Messenger::Policy::stateless_server(0));
 598
 599   ms_hb_front_client->set_policy(entity_name_t::TYPE_OSD,
 600                           Messenger::Policy::lossy_client(0));
 601   ms_hb_back_client->set_policy(entity_name_t::TYPE_OSD,
 602                           Messenger::Policy::lossy_client(0));
 603   ms_hb_back_server->set_policy(entity_name_t::TYPE_OSD,
 604                                 Messenger::Policy::stateless_server(0));
 605   ms_hb_front_server->set_policy(entity_name_t::TYPE_OSD,
 606                                  Messenger::Policy::stateless_server(0));
 607
 608   ms_objecter->set_default_policy(Messenger::Policy::lossy_client(CEPH_FEATURE_OSDREPLYMUX));
 609
 610   entity_addrvec_t public_addrs, cluster_addrs;
 611   r = pick_addresses(g_ceph_context, CEPH_PICK_ADDRESS_PUBLIC, &public_addrs,
 612                      iface_preferred_numa_node);
 613   if (r < 0) {
 614     derr << "Failed to pick public address." << dendl;
 615     forker.exit(1);
 616   }
 617   r = pick_addresses(g_ceph_context, CEPH_PICK_ADDRESS_CLUSTER, &cluster_addrs,
 618                      iface_preferred_numa_node);
 619   if (r < 0) {
 620     derr << "Failed to pick cluster address." << dendl;
 621     forker.exit(1);
 622   }
 623
 624   if (ms_public->bindv(public_addrs) < 0)
 625     forker.exit(1);
 626
 627   if (ms_cluster->bindv(cluster_addrs) < 0)
 628     forker.exit(1);
 629
 630   bool is_delay = g_conf().get_val<bool>("osd_heartbeat_use_min_delay_socket");
 631   if (is_delay) {
 632     ms_hb_front_client->set_socket_priority(SOCKET_PRIORITY_MIN_DELAY);
 633     ms_hb_back_client->set_socket_priority(SOCKET_PRIORITY_MIN_DELAY);
 634     ms_hb_back_server->set_socket_priority(SOCKET_PRIORITY_MIN_DELAY);
 635     ms_hb_front_server->set_socket_priority(SOCKET_PRIORITY_MIN_DELAY);
 636   }
 637
 638   entity_addrvec_t hb_front_addrs = public_addrs;
 639   for (auto& a : hb_front_addrs.v) {
 640     a.set_port(0);
 641   }
 642   if (ms_hb_front_server->bindv(hb_front_addrs) < 0)
 643     forker.exit(1);
 644   if (ms_hb_front_client->client_bind(hb_front_addrs.front()) < 0)
 645     forker.exit(1);
 646
 647   entity_addrvec_t hb_back_addrs = cluster_addrs;
 648   for (auto& a : hb_back_addrs.v) {
 649     a.set_port(0);
 650   }
 651   if (ms_hb_back_server->bindv(hb_back_addrs) < 0)
 652     forker.exit(1);
 653   if (ms_hb_back_client->client_bind(hb_back_addrs.front()) < 0)
 654     forker.exit(1);
 655
 656   // install signal handlers
 657   init_async_signal_handler();
 658   register_async_signal_handler(SIGHUP, sighup_handler);
 659
 660   TracepointProvider::initialize<osd_tracepoint_traits>(g_ceph_context);
 661   TracepointProvider::initialize<os_tracepoint_traits>(g_ceph_context);
 662   TracepointProvider::initialize<bluestore_tracepoint_traits>(g_ceph_context);
 663 #ifdef WITH_OSD_INSTRUMENT_FUNCTIONS
 664   TracepointProvider::initialize<cyg_profile_traits>(g_ceph_context);
 665 #endif
 666
 667   srand(time(NULL) + getpid());
 668
 669   MonClient mc(g_ceph_context);
 670   if (mc.build_initial_monmap() < 0)
 671     return -1;
 672   global_init_chdir(g_ceph_context);
 673
 674   if (global_init_preload_erasure_code(g_ceph_context) < 0) {
 675     forker.exit(1);
 676   }
 677
 678   osdptr = new OSD(g_ceph_context,
 679                    store,
 680                    whoami,
 681                    ms_cluster,
 682                    ms_public,
 683                    ms_hb_front_client,
 684                    ms_hb_back_client,
 685                    ms_hb_front_server,
 686                    ms_hb_back_server,
 687                    ms_objecter,
 688                    &mc,
 689                    data_path,
 690                    journal_path);
 691
 692   int err = osdptr->pre_init();
 693   if (err < 0) {
 694     derr << TEXT_RED << " ** ERROR: osd pre_init failed: " << cpp_strerror(-err)
 695          << TEXT_NORMAL << dendl;
 696     forker.exit(1);
 697   }
 698
 699   ms_public->start();
 700   ms_hb_front_client->start();
 701   ms_hb_back_client->start();
 702   ms_hb_front_server->start();
 703   ms_hb_back_server->start();
 704   ms_cluster->start();
 705   ms_objecter->start();
 706
 707   // start osd
 708   err = osdptr->init();
 709   if (err < 0) {
 710     derr << TEXT_RED << " ** ERROR: osd init failed: " << cpp_strerror(-err)
 711          << TEXT_NORMAL << dendl;
 712     forker.exit(1);
 713   }
 714
 715   // -- daemonize --
 716
 717   if (g_conf()->daemonize) {
 718     global_init_postfork_finish(g_ceph_context);
 719     forker.daemonize();
 720   }
 721
 722
 723   register_async_signal_handler_oneshot(SIGINT, handle_osd_signal);
 724   register_async_signal_handler_oneshot(SIGTERM, handle_osd_signal);
 725
 726   osdptr->final_init();
 727
 728   if (g_conf().get_val<bool>("inject_early_sigterm"))
 729     kill(getpid(), SIGTERM);
 730
 731   ms_public->wait();
 732   ms_hb_front_client->wait();
 733   ms_hb_back_client->wait();
 734   ms_hb_front_server->wait();
 735   ms_hb_back_server->wait();
 736   ms_cluster->wait();
 737   ms_objecter->wait();
 738
 739   unregister_async_signal_handler(SIGHUP, sighup_handler);
 740   unregister_async_signal_handler(SIGINT, handle_osd_signal);
 741   unregister_async_signal_handler(SIGTERM, handle_osd_signal);
 742   shutdown_async_signal_handler();
 743
 744   // done
 745   delete osdptr;
 746   delete ms_public;
 747   delete ms_hb_front_client;
 748   delete ms_hb_back_client;
 749   delete ms_hb_front_server;
 750   delete ms_hb_back_server;
 751   delete ms_cluster;
 752   delete ms_objecter;
 753
 754   client_byte_throttler.reset();
 755
 756   // cd on exit, so that gmon.out (if any) goes into a separate directory for each node.
 757   char s[20];
 758   snprintf(s, sizeof(s), "gmon/%d", getpid());
 759   if ((mkdir(s, 0755) == 0) && (chdir(s) == 0)) {
 760     dout(0) << "ceph-osd: gmon.out should be in " << s << dendl;
 761   }
 762
 763   return 0;
 764 }