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