ceph/src/tools/ceph_objectstore_tool.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) 2013 Inktank
   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 <boost/program_options/variables_map.hpp>
  16 #include <boost/program_options/parsers.hpp>
  17 #include <boost/scoped_ptr.hpp>
  18 #include <boost/optional.hpp>
  19
  20 #include <stdlib.h>
  21
  22 #include "common/Formatter.h"
  23 #include "common/errno.h"
  24 #include "common/ceph_argparse.h"
  25
  26 #include "global/global_init.h"
  27
  28 #include "os/ObjectStore.h"
  29 #include "os/filestore/FileJournal.h"
  30 #include "os/filestore/FileStore.h"
  31 #ifdef HAVE_LIBFUSE
  32 #include "os/FuseStore.h"
  33 #endif
  34
  35 #include "osd/PGLog.h"
  36 #include "osd/OSD.h"
  37 #include "osd/PG.h"
  38
  39 #include "json_spirit/json_spirit_value.h"
  40 #include "json_spirit/json_spirit_reader.h"
  41
  42 #include "rebuild_mondb.h"
  43 #include "ceph_objectstore_tool.h"
  44 #include "include/compat.h"
  45 #include "include/util.h"
  46
  47 namespace po = boost::program_options;
  48 using namespace std;
  49
  50 #ifdef INTERNAL_TEST
  51 CompatSet get_test_compat_set() {
  52   CompatSet::FeatureSet ceph_osd_feature_compat;
  53   CompatSet::FeatureSet ceph_osd_feature_ro_compat;
  54   CompatSet::FeatureSet ceph_osd_feature_incompat;
  55   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_BASE);
  56   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_PGINFO);
  57   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_OLOC);
  58   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_LEC);
  59   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_CATEGORIES);
  60   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_HOBJECTPOOL);
  61   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_BIGINFO);
  62   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_LEVELDBINFO);
  63   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_LEVELDBLOG);
  64 #ifdef INTERNAL_TEST2
  65   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_SNAPMAPPER);
  66   ceph_osd_feature_incompat.insert(CEPH_OSD_FEATURE_INCOMPAT_SHARDS);
  67 #endif
  68   return CompatSet(ceph_osd_feature_compat, ceph_osd_feature_ro_compat,
  69                    ceph_osd_feature_incompat);
  70 }
  71 #endif
  72
  73 const ssize_t max_read = 1024 * 1024;
  74 const int fd_none = INT_MIN;
  75 bool outistty;
  76 bool dry_run;
  77
  78 struct action_on_object_t {
  79   virtual ~action_on_object_t() {}
  80   virtual int call(ObjectStore *store, coll_t coll, ghobject_t &ghobj, object_info_t &oi) = 0;
  81 };
  82
  83 int _action_on_all_objects_in_pg(ObjectStore *store, coll_t coll, action_on_object_t &action, bool debug)
  84 {
  85   unsigned LIST_AT_A_TIME = 100;
  86   ghobject_t next;
  87   while (!next.is_max()) {
  88     vector<ghobject_t> list;
  89     int r = store->collection_list(
  90                                    coll,
  91                                    next,
  92                                    ghobject_t::get_max(),
  93                                    LIST_AT_A_TIME,
  94                                    &list,
  95                                    &next);
  96     if (r < 0) {
  97       cerr << "Error listing collection: " << coll << ", "
  98            << cpp_strerror(r) << std::endl;
  99       return r;
 100     }
 101     for (vector<ghobject_t>::iterator obj = list.begin();
 102          obj != list.end();
 103          ++obj) {
 104       if (obj->is_pgmeta())
 105         continue;
 106       object_info_t oi;
 107       if (coll != coll_t::meta()) {
 108         bufferlist attr;
 109         r = store->getattr(coll, *obj, OI_ATTR, attr);
 110         if (r < 0) {
 111           cerr << "Error getting attr on : " << make_pair(coll, *obj) << ", "
 112                << cpp_strerror(r) << std::endl;
 113           continue;
 114         }
 115         bufferlist::iterator bp = attr.begin();
 116         try {
 117           ::decode(oi, bp);
 118         } catch (...) {
 119           r = -EINVAL;
 120           cerr << "Error getting attr on : " << make_pair(coll, *obj) << ", "
 121                << cpp_strerror(r) << std::endl;
 122           continue;
 123         }
 124       }
 125       r = action.call(store, coll, *obj, oi);
 126       if (r < 0)
 127         return r;
 128     }
 129   }
 130   return 0;
 131 }
 132
 133 int action_on_all_objects_in_pg(ObjectStore *store, string pgidstr, action_on_object_t &action, bool debug)
 134 {
 135   spg_t pgid;
 136   // Scan collections in case this is an ec pool but no shard specified
 137   unsigned scanned = 0;
 138   int r = 0;
 139   vector<coll_t> colls_to_check;
 140   vector<coll_t> candidates;
 141   r = store->list_collections(candidates);
 142   if (r < 0) {
 143     cerr << "Error listing collections: " << cpp_strerror(r) << std::endl;
 144     return r;
 145   }
 146   pgid.parse(pgidstr.c_str());
 147   for (vector<coll_t>::iterator i = candidates.begin();
 148        i != candidates.end();
 149        ++i) {
 150     spg_t cand_pgid;
 151     if (!i->is_pg(&cand_pgid))
 152       continue;
 153
 154     // If an exact match or treat no shard as any shard
 155     if (cand_pgid == pgid ||
 156         (pgid.is_no_shard() && pgid.pgid == cand_pgid.pgid)) {
 157       colls_to_check.push_back(*i);
 158     }
 159   }
 160
 161   if (debug)
 162     cerr << colls_to_check.size() << " pgs to scan" << std::endl;
 163   for (vector<coll_t>::iterator i = colls_to_check.begin();
 164        i != colls_to_check.end();
 165        ++i, ++scanned) {
 166     if (debug)
 167       cerr << "Scanning " << *i << ", " << scanned << "/"
 168            << colls_to_check.size() << " completed" << std::endl;
 169     r = _action_on_all_objects_in_pg(store, *i, action, debug);
 170     if (r < 0)
 171       break;
 172   }
 173   return r;
 174 }
 175
 176 int action_on_all_objects_in_exact_pg(ObjectStore *store, coll_t coll, action_on_object_t &action, bool debug)
 177 {
 178   int r = _action_on_all_objects_in_pg(store, coll, action, debug);
 179   return r;
 180 }
 181
 182 int _action_on_all_objects(ObjectStore *store, action_on_object_t &action, bool debug)
 183 {
 184   unsigned scanned = 0;
 185   int r = 0;
 186   vector<coll_t> colls_to_check;
 187   vector<coll_t> candidates;
 188   r = store->list_collections(candidates);
 189   if (r < 0) {
 190     cerr << "Error listing collections: " << cpp_strerror(r) << std::endl;
 191     return r;
 192   }
 193   for (vector<coll_t>::iterator i = candidates.begin();
 194        i != candidates.end();
 195        ++i) {
 196     if (i->is_pg()) {
 197       colls_to_check.push_back(*i);
 198     }
 199   }
 200
 201   if (debug)
 202     cerr << colls_to_check.size() << " pgs to scan" << std::endl;
 203   for (vector<coll_t>::iterator i = colls_to_check.begin();
 204        i != colls_to_check.end();
 205        ++i, ++scanned) {
 206     if (debug)
 207       cerr << "Scanning " << *i << ", " << scanned << "/"
 208            << colls_to_check.size() << " completed" << std::endl;
 209     r = _action_on_all_objects_in_pg(store, *i, action, debug);
 210     if (r < 0)
 211       return r;
 212   }
 213   return 0;
 214 }
 215
 216 int action_on_all_objects(ObjectStore *store, action_on_object_t &action, bool debug)
 217 {
 218   int r = _action_on_all_objects(store, action, debug);
 219   return r;
 220 }
 221
 222 struct pgid_object_list {
 223   list<pair<coll_t, ghobject_t> > _objects;
 224
 225   void insert(coll_t coll, ghobject_t &ghobj) {
 226     _objects.push_back(make_pair(coll, ghobj));
 227   }
 228
 229   void dump(Formatter *f, bool human_readable) const {
 230     if (!human_readable)
 231       f->open_array_section("pgid_objects");
 232     for (list<pair<coll_t, ghobject_t> >::const_iterator i = _objects.begin();
 233          i != _objects.end();
 234          ++i) {
 235       f->open_array_section("pgid_object");
 236       spg_t pgid;
 237       bool is_pg = i->first.is_pg(&pgid);
 238       if (is_pg)
 239         f->dump_string("pgid", stringify(pgid));
 240       if (!is_pg || !human_readable)
 241         f->dump_string("coll", i->first.to_str());
 242       f->open_object_section("ghobject");
 243       i->second.dump(f);
 244       f->close_section();
 245       f->close_section();
 246       if (human_readable) {
 247         f->flush(cout);
 248         cout << std::endl;
 249       }
 250     }
 251     if (!human_readable) {
 252       f->close_section();
 253       f->flush(cout);
 254       cout << std::endl;
 255     }
 256   }
 257 };
 258
 259 struct lookup_ghobject : public action_on_object_t {
 260   pgid_object_list _objects;
 261   const string _name;
 262   const boost::optional<std::string> _namespace;
 263   bool _need_snapset;
 264
 265   lookup_ghobject(const string& name, const boost::optional<std::string>& nspace, bool need_snapset = false) : _name(name),
 266                   _namespace(nspace), _need_snapset(need_snapset) { }
 267
 268   int call(ObjectStore *store, coll_t coll, ghobject_t &ghobj, object_info_t &oi) override {
 269     if (_need_snapset && !ghobj.hobj.has_snapset())
 270       return 0;
 271     if ((_name.length() == 0 || ghobj.hobj.oid.name == _name) &&
 272         (!_namespace || ghobj.hobj.nspace == _namespace))
 273       _objects.insert(coll, ghobj);
 274     return 0;
 275   }
 276
 277   int size() const {
 278     return _objects._objects.size();
 279   }
 280
 281   pair<coll_t, ghobject_t> pop() {
 282      pair<coll_t, ghobject_t> front = _objects._objects.front();
 283      _objects._objects.pop_front();
 284      return front;
 285   }
 286
 287   void dump(Formatter *f, bool human_readable) const {
 288     _objects.dump(f, human_readable);
 289   }
 290 };
 291
 292 ghobject_t infos_oid = OSD::make_infos_oid();
 293 ghobject_t log_oid;
 294 ghobject_t biginfo_oid;
 295
 296 int file_fd = fd_none;
 297 bool debug;
 298 super_header sh;
 299
 300 static int get_fd_data(int fd, bufferlist &bl)
 301 {
 302   uint64_t total = 0;
 303   do {
 304     ssize_t bytes = bl.read_fd(fd, max_read);
 305     if (bytes < 0) {
 306       cerr << "read_fd error " << cpp_strerror(bytes) << std::endl;
 307       return bytes;
 308     }
 309
 310     if (bytes == 0)
 311       break;
 312
 313     total += bytes;
 314   } while(true);
 315
 316   assert(bl.length() == total);
 317   return 0;
 318 }
 319
 320 int get_log(ObjectStore *fs, __u8 struct_ver,
 321             coll_t coll, spg_t pgid, const pg_info_t &info,
 322             PGLog::IndexedLog &log, pg_missing_t &missing)
 323 {
 324   try {
 325     ostringstream oss;
 326     assert(struct_ver > 0);
 327     PGLog::read_log_and_missing(fs, coll,
 328                     struct_ver >= 8 ? coll : coll_t::meta(),
 329                     struct_ver >= 8 ? pgid.make_pgmeta_oid() : log_oid,
 330                     info, log, missing,
 331                                 struct_ver < 9,
 332                                 oss,
 333                     g_ceph_context->_conf->osd_ignore_stale_divergent_priors);
 334     if (debug && oss.str().size())
 335       cerr << oss.str() << std::endl;
 336   }
 337   catch (const buffer::error &e) {
 338     cerr << "read_log_and_missing threw exception error " << e.what() << std::endl;
 339     return -EFAULT;
 340   }
 341   return 0;
 342 }
 343
 344 void dump_log(Formatter *formatter, ostream &out, pg_log_t &log,
 345               pg_missing_t &missing)
 346 {
 347   formatter->open_object_section("op_log");
 348   formatter->open_object_section("pg_log_t");
 349   log.dump(formatter);
 350   formatter->close_section();
 351   formatter->flush(out);
 352   formatter->open_object_section("pg_missing_t");
 353   missing.dump(formatter);
 354   formatter->close_section();
 355   formatter->close_section();
 356   formatter->flush(out);
 357 }
 358
 359 //Based on part of OSD::load_pgs()
 360 int finish_remove_pgs(ObjectStore *store)
 361 {
 362   vector<coll_t> ls;
 363   int r = store->list_collections(ls);
 364   if (r < 0) {
 365     cerr << "finish_remove_pgs: failed to list pgs: " << cpp_strerror(r)
 366       << std::endl;
 367     return r;
 368   }
 369
 370   for (vector<coll_t>::iterator it = ls.begin();
 371        it != ls.end();
 372        ++it) {
 373     spg_t pgid;
 374
 375     if (it->is_temp(&pgid) ||
 376        (it->is_pg(&pgid) && PG::_has_removal_flag(store, pgid))) {
 377       cout << "finish_remove_pgs " << *it << " removing " << pgid << std::endl;
 378       OSD::recursive_remove_collection(g_ceph_context, store, pgid, *it);
 379       continue;
 380     }
 381
 382     //cout << "finish_remove_pgs ignoring unrecognized " << *it << std::endl;
 383   }
 384   return 0;
 385 }
 386
 387 #pragma GCC diagnostic ignored "-Wpragmas"
 388 #pragma GCC diagnostic push
 389 #pragma GCC diagnostic ignored "-Wdeprecated-declarations"
 390
 391 int mark_pg_for_removal(ObjectStore *fs, spg_t pgid, ObjectStore::Transaction *t)
 392 {
 393   pg_info_t info(pgid);
 394   coll_t coll(pgid);
 395   ghobject_t pgmeta_oid(info.pgid.make_pgmeta_oid());
 396
 397   bufferlist bl;
 398   epoch_t map_epoch = 0;
 399   int r = PG::peek_map_epoch(fs, pgid, &map_epoch, &bl);
 400   if (r < 0)
 401     cerr << __func__ << " warning: peek_map_epoch reported error" << std::endl;
 402   PastIntervals past_intervals;
 403   __u8 struct_v;
 404   r = PG::read_info(fs, pgid, coll, bl, info, past_intervals, struct_v);
 405   if (r < 0) {
 406     cerr << __func__ << " error on read_info " << cpp_strerror(r) << std::endl;
 407     return r;
 408   }
 409   assert(struct_v >= 8);
 410   // new omap key
 411   cout << "setting '_remove' omap key" << std::endl;
 412   map<string,bufferlist> values;
 413   ::encode((char)1, values["_remove"]);
 414   t->omap_setkeys(coll, pgmeta_oid, values);
 415   return 0;
 416 }
 417
 418 #pragma GCC diagnostic pop
 419 #pragma GCC diagnostic warning "-Wpragmas"
 420
 421 int initiate_new_remove_pg(ObjectStore *store, spg_t r_pgid,
 422                            ObjectStore::Sequencer &osr)
 423 {
 424   if (!dry_run)
 425     finish_remove_pgs(store);
 426   if (!store->collection_exists(coll_t(r_pgid)))
 427     return -ENOENT;
 428
 429   cout << " marking collection for removal" << std::endl;
 430   if (dry_run)
 431     return 0;
 432   ObjectStore::Transaction rmt;
 433   int r = mark_pg_for_removal(store, r_pgid, &rmt);
 434   if (r < 0) {
 435     return r;
 436   }
 437   store->apply_transaction(&osr, std::move(rmt));
 438   finish_remove_pgs(store);
 439   return r;
 440 }
 441
 442 int write_info(ObjectStore::Transaction &t, epoch_t epoch, pg_info_t &info,
 443     PastIntervals &past_intervals)
 444 {
 445   //Empty for this
 446   coll_t coll(info.pgid);
 447   ghobject_t pgmeta_oid(info.pgid.make_pgmeta_oid());
 448   map<string,bufferlist> km;
 449   pg_info_t last_written_info;
 450   int ret = PG::_prepare_write_info(
 451     g_ceph_context,
 452     &km, epoch,
 453     info,
 454     last_written_info,
 455     past_intervals,
 456     true, true, false);
 457   if (ret) cerr << "Failed to write info" << std::endl;
 458   t.omap_setkeys(coll, pgmeta_oid, km);
 459   return ret;
 460 }
 461
 462 typedef map<eversion_t, hobject_t> divergent_priors_t;
 463
 464 int write_pg(ObjectStore::Transaction &t, epoch_t epoch, pg_info_t &info,
 465              pg_log_t &log, PastIntervals &past_intervals,
 466              divergent_priors_t &divergent,
 467              pg_missing_t &missing)
 468 {
 469   int ret = write_info(t, epoch, info, past_intervals);
 470   if (ret)
 471     return ret;
 472   coll_t coll(info.pgid);
 473   map<string,bufferlist> km;
 474
 475   if (!divergent.empty()) {
 476     assert(missing.get_items().empty());
 477     PGLog::write_log_and_missing_wo_missing(
 478       t, &km, log, coll, info.pgid.make_pgmeta_oid(), divergent, true);
 479   } else {
 480     pg_missing_tracker_t tmissing(missing);
 481     bool rebuilt_missing_set_with_deletes = missing.may_include_deletes;
 482     PGLog::write_log_and_missing(
 483       t, &km, log, coll, info.pgid.make_pgmeta_oid(), tmissing, true,
 484       &rebuilt_missing_set_with_deletes);
 485   }
 486   t.omap_setkeys(coll, info.pgid.make_pgmeta_oid(), km);
 487   return 0;
 488 }
 489
 490 int do_trim_pg_log(ObjectStore *store, const coll_t &coll,
 491                    pg_info_t &info, const spg_t &pgid,
 492                    ObjectStore::Sequencer &osr, epoch_t map_epoch,
 493                    PastIntervals &past_intervals)
 494 {
 495   ghobject_t oid = pgid.make_pgmeta_oid();
 496   struct stat st;
 497   int r = store->stat(coll, oid, &st);
 498   assert(r == 0);
 499   assert(st.st_size == 0);
 500
 501   cerr << "Log bounds are: " << "(" << info.log_tail << ","
 502        << info.last_update << "]" << std::endl;
 503
 504   uint64_t max_entries = g_ceph_context->_conf->osd_max_pg_log_entries;
 505   if (info.last_update.version - info.log_tail.version <= max_entries) {
 506     cerr << "Log not larger than osd_max_pg_log_entries " << max_entries << std::endl;
 507     return 0;
 508   }
 509
 510   assert(info.last_update.version > max_entries);
 511   version_t trim_to = info.last_update.version - max_entries;
 512   size_t trim_at_once = g_ceph_context->_conf->osd_pg_log_trim_max;
 513   eversion_t new_tail;
 514   bool done = false;
 515
 516   while (!done) {
 517     // gather keys so we can delete them in a batch without
 518     // affecting the iterator
 519     set<string> keys_to_trim;
 520     {
 521     ObjectMap::ObjectMapIterator p = store->get_omap_iterator(coll, oid);
 522     if (!p)
 523       break;
 524     for (p->seek_to_first(); p->valid(); p->next(false)) {
 525       if (p->key()[0] == '_')
 526         continue;
 527       if (p->key() == "can_rollback_to")
 528         continue;
 529       if (p->key() == "divergent_priors")
 530         continue;
 531       if (p->key() == "rollback_info_trimmed_to")
 532         continue;
 533       if (p->key() == "may_include_deletes_in_missing")
 534         continue;
 535       if (p->key().substr(0, 7) == string("missing"))
 536         continue;
 537       if (p->key().substr(0, 4) == string("dup_"))
 538         continue;
 539
 540       bufferlist bl = p->value();
 541       bufferlist::iterator bp = bl.begin();
 542       pg_log_entry_t e;
 543       try {
 544         e.decode_with_checksum(bp);
 545       } catch (const buffer::error &e) {
 546         cerr << "Error reading pg log entry: " << e << std::endl;
 547       }
 548       if (debug) {
 549         cerr << "read entry " << e << std::endl;
 550       }
 551       if (e.version.version > trim_to) {
 552         done = true;
 553         break;
 554       }
 555       keys_to_trim.insert(p->key());
 556       new_tail = e.version;
 557       if (keys_to_trim.size() >= trim_at_once)
 558         break;
 559     }
 560
 561     if (!p->valid())
 562       done = true;
 563     } // deconstruct ObjectMapIterator
 564
 565     // delete the keys
 566     if (!dry_run && !keys_to_trim.empty()) {
 567       cout << "Removing keys " << *keys_to_trim.begin() << " - " << *keys_to_trim.rbegin() << std::endl;
 568       ObjectStore::Transaction t;
 569       t.omap_rmkeys(coll, oid, keys_to_trim);
 570       int r = store->apply_transaction(&osr, std::move(t));
 571       if (r) {
 572         cerr << "Error trimming logs " << cpp_strerror(r) << std::endl;
 573       }
 574     }
 575   }
 576
 577   // update pg info with new tail
 578   if (!dry_run && new_tail !=  eversion_t()) {
 579     info.log_tail = new_tail;
 580     ObjectStore::Transaction t;
 581     int ret = write_info(t, map_epoch, info, past_intervals);
 582     if (ret)
 583       return ret;
 584     ret = store->apply_transaction(&osr, std::move(t));
 585     if (ret) {
 586       cerr << "Error updating pg info " << cpp_strerror(ret) << std::endl;
 587     }
 588   }
 589
 590   // compact the db since we just removed a bunch of data
 591   cerr << "Finished trimming, now compacting..." << std::endl;
 592   if (!dry_run)
 593     store->compact();
 594   return 0;
 595 }
 596
 597 const int OMAP_BATCH_SIZE = 25;
 598 void get_omap_batch(ObjectMap::ObjectMapIterator &iter, map<string, bufferlist> &oset)
 599 {
 600   oset.clear();
 601   for (int count = OMAP_BATCH_SIZE; count && iter->valid(); --count, iter->next()) {
 602     oset.insert(pair<string, bufferlist>(iter->key(), iter->value()));
 603   }
 604 }
 605
 606 int ObjectStoreTool::export_file(ObjectStore *store, coll_t cid, ghobject_t &obj)
 607 {
 608   struct stat st;
 609   mysize_t total;
 610   footer ft;
 611
 612   int ret = store->stat(cid, obj, &st);
 613   if (ret < 0)
 614     return ret;
 615
 616   cerr << "Read " << obj << std::endl;
 617
 618   total = st.st_size;
 619   if (debug)
 620     cerr << "size=" << total << std::endl;
 621
 622   object_begin objb(obj);
 623
 624   {
 625     bufferptr bp;
 626     bufferlist bl;
 627     ret = store->getattr(cid, obj, OI_ATTR, bp);
 628     if (ret < 0) {
 629       cerr << "getattr failure object_info " << ret << std::endl;
 630       return ret;
 631     }
 632     bl.push_back(bp);
 633     decode(objb.oi, bl);
 634     if (debug)
 635       cerr << "object_info: " << objb.oi << std::endl;
 636   }
 637
 638   // NOTE: we include whiteouts, lost, etc.
 639
 640   ret = write_section(TYPE_OBJECT_BEGIN, objb, file_fd);
 641   if (ret < 0)
 642     return ret;
 643
 644   uint64_t offset = 0;
 645   bufferlist rawdatabl;
 646   while(total > 0) {
 647     rawdatabl.clear();
 648     mysize_t len = max_read;
 649     if (len > total)
 650       len = total;
 651
 652     ret = store->read(cid, obj, offset, len, rawdatabl);
 653     if (ret < 0)
 654       return ret;
 655     if (ret == 0)
 656       return -EINVAL;
 657
 658     data_section dblock(offset, len, rawdatabl);
 659     if (debug)
 660       cerr << "data section offset=" << offset << " len=" << len << std::endl;
 661
 662     total -= ret;
 663     offset += ret;
 664
 665     ret = write_section(TYPE_DATA, dblock, file_fd);
 666     if (ret) return ret;
 667   }
 668
 669   //Handle attrs for this object
 670   map<string,bufferptr> aset;
 671   ret = store->getattrs(cid, obj, aset);
 672   if (ret) return ret;
 673   attr_section as(aset);
 674   ret = write_section(TYPE_ATTRS, as, file_fd);
 675   if (ret)
 676     return ret;
 677
 678   if (debug) {
 679     cerr << "attrs size " << aset.size() << std::endl;
 680   }
 681
 682   //Handle omap information
 683   bufferlist hdrbuf;
 684   ret = store->omap_get_header(cid, obj, &hdrbuf, true);
 685   if (ret < 0) {
 686     cerr << "omap_get_header: " << cpp_strerror(ret) << std::endl;
 687     return ret;
 688   }
 689
 690   omap_hdr_section ohs(hdrbuf);
 691   ret = write_section(TYPE_OMAP_HDR, ohs, file_fd);
 692   if (ret)
 693     return ret;
 694
 695   ObjectMap::ObjectMapIterator iter = store->get_omap_iterator(cid, obj);
 696   if (!iter) {
 697     ret = -ENOENT;
 698     cerr << "omap_get_iterator: " << cpp_strerror(ret) << std::endl;
 699     return ret;
 700   }
 701   iter->seek_to_first();
 702   int mapcount = 0;
 703   map<string, bufferlist> out;
 704   while(iter->valid()) {
 705     get_omap_batch(iter, out);
 706
 707     if (out.empty()) break;
 708
 709     mapcount += out.size();
 710     omap_section oms(out);
 711     ret = write_section(TYPE_OMAP, oms, file_fd);
 712     if (ret)
 713       return ret;
 714   }
 715   if (debug)
 716     cerr << "omap map size " << mapcount << std::endl;
 717
 718   ret = write_simple(TYPE_OBJECT_END, file_fd);
 719   if (ret)
 720     return ret;
 721
 722   return 0;
 723 }
 724
 725 int ObjectStoreTool::export_files(ObjectStore *store, coll_t coll)
 726 {
 727   ghobject_t next;
 728
 729   while (!next.is_max()) {
 730     vector<ghobject_t> objects;
 731     int r = store->collection_list(coll, next, ghobject_t::get_max(), 300,
 732       &objects, &next);
 733     if (r < 0)
 734       return r;
 735     for (vector<ghobject_t>::iterator i = objects.begin();
 736          i != objects.end();
 737          ++i) {
 738       assert(!i->hobj.is_meta());
 739       if (i->is_pgmeta() || i->hobj.is_temp()) {
 740         continue;
 741       }
 742       r = export_file(store, coll, *i);
 743       if (r < 0)
 744         return r;
 745     }
 746   }
 747   return 0;
 748 }
 749
 750 int set_inc_osdmap(ObjectStore *store, epoch_t e, bufferlist& bl, bool force,
 751                    ObjectStore::Sequencer &osr) {
 752   OSDMap::Incremental inc;
 753   bufferlist::iterator it = bl.begin();
 754   inc.decode(it);
 755   if (e == 0) {
 756     e = inc.epoch;
 757   } else if (e != inc.epoch) {
 758     cerr << "incremental.epoch mismatch: "
 759          << inc.epoch << " != " << e << std::endl;
 760     if (force) {
 761       cerr << "But will continue anyway." << std::endl;
 762     } else {
 763       return -EINVAL;
 764     }
 765   }
 766   const ghobject_t inc_oid = OSD::get_inc_osdmap_pobject_name(e);
 767   if (!store->exists(coll_t::meta(), inc_oid)) {
 768     cerr << "inc-osdmap (" << inc_oid << ") does not exist." << std::endl;
 769     if (!force) {
 770       return -ENOENT;
 771     }
 772     cout << "Creating a new epoch." << std::endl;
 773   }
 774   if (dry_run)
 775     return 0;
 776   ObjectStore::Transaction t;
 777   t.write(coll_t::meta(), inc_oid, 0, bl.length(), bl);
 778   t.truncate(coll_t::meta(), inc_oid, bl.length());
 779   int ret = store->apply_transaction(&osr, std::move(t));
 780   if (ret) {
 781     cerr << "Failed to set inc-osdmap (" << inc_oid << "): " << ret << std::endl;
 782   } else {
 783     cout << "Wrote inc-osdmap." << inc.epoch << std::endl;
 784   }
 785   return ret;
 786 }
 787
 788 int get_inc_osdmap(ObjectStore *store, epoch_t e, bufferlist& bl)
 789 {
 790   if (store->read(coll_t::meta(),
 791                   OSD::get_inc_osdmap_pobject_name(e),
 792                   0, 0, bl) < 0) {
 793     return -ENOENT;
 794   }
 795   return 0;
 796 }
 797
 798 int set_osdmap(ObjectStore *store, epoch_t e, bufferlist& bl, bool force,
 799                ObjectStore::Sequencer &osr) {
 800   OSDMap osdmap;
 801   osdmap.decode(bl);
 802   if (e == 0) {
 803     e = osdmap.get_epoch();
 804   } else if (e != osdmap.get_epoch()) {
 805     cerr << "osdmap.epoch mismatch: "
 806          << e << " != " << osdmap.get_epoch() << std::endl;
 807     if (force) {
 808       cerr << "But will continue anyway." << std::endl;
 809     } else {
 810       return -EINVAL;
 811     }
 812   }
 813   const ghobject_t full_oid = OSD::get_osdmap_pobject_name(e);
 814   if (!store->exists(coll_t::meta(), full_oid)) {
 815     cerr << "osdmap (" << full_oid << ") does not exist." << std::endl;
 816     if (!force) {
 817       return -ENOENT;
 818     }
 819     cout << "Creating a new epoch." << std::endl;
 820   }
 821   if (dry_run)
 822     return 0;
 823   ObjectStore::Transaction t;
 824   t.write(coll_t::meta(), full_oid, 0, bl.length(), bl);
 825   t.truncate(coll_t::meta(), full_oid, bl.length());
 826   int ret = store->apply_transaction(&osr, std::move(t));
 827   if (ret) {
 828     cerr << "Failed to set osdmap (" << full_oid << "): " << ret << std::endl;
 829   } else {
 830     cout << "Wrote osdmap." << osdmap.get_epoch() << std::endl;
 831   }
 832   return ret;
 833 }
 834
 835 int get_osdmap(ObjectStore *store, epoch_t e, OSDMap &osdmap, bufferlist& bl)
 836 {
 837   bool found = store->read(
 838       coll_t::meta(), OSD::get_osdmap_pobject_name(e), 0, 0, bl) >= 0;
 839   if (!found) {
 840     cerr << "Can't find OSDMap for pg epoch " << e << std::endl;
 841     return -ENOENT;
 842   }
 843   osdmap.decode(bl);
 844   if (debug)
 845     cerr << osdmap << std::endl;
 846   return 0;
 847 }
 848
 849 int add_osdmap(ObjectStore *store, metadata_section &ms)
 850 {
 851   return get_osdmap(store, ms.map_epoch, ms.osdmap, ms.osdmap_bl);
 852 }
 853
 854 int ObjectStoreTool::do_export(ObjectStore *fs, coll_t coll, spg_t pgid,
 855     pg_info_t &info, epoch_t map_epoch, __u8 struct_ver,
 856     const OSDSuperblock& superblock,
 857     PastIntervals &past_intervals)
 858 {
 859   PGLog::IndexedLog log;
 860   pg_missing_t missing;
 861
 862   cerr << "Exporting " << pgid << std::endl;
 863
 864   int ret = get_log(fs, struct_ver, coll, pgid, info, log, missing);
 865   if (ret > 0)
 866       return ret;
 867
 868   if (debug) {
 869     Formatter *formatter = Formatter::create("json-pretty");
 870     assert(formatter);
 871     dump_log(formatter, cerr, log, missing);
 872     delete formatter;
 873   }
 874   write_super();
 875
 876   pg_begin pgb(pgid, superblock);
 877   // Special case: If replicated pg don't require the importing OSD to have shard feature
 878   if (pgid.is_no_shard()) {
 879     pgb.superblock.compat_features.incompat.remove(CEPH_OSD_FEATURE_INCOMPAT_SHARDS);
 880   }
 881   ret = write_section(TYPE_PG_BEGIN, pgb, file_fd);
 882   if (ret)
 883     return ret;
 884
 885   // The metadata_section is now before files, so import can detect
 886   // errors and abort without wasting time.
 887   metadata_section ms(
 888     struct_ver,
 889     map_epoch,
 890     info,
 891     log,
 892     past_intervals,
 893     missing);
 894   ret = add_osdmap(fs, ms);
 895   if (ret)
 896     return ret;
 897   ret = write_section(TYPE_PG_METADATA, ms, file_fd);
 898   if (ret)
 899     return ret;
 900
 901   ret = export_files(fs, coll);
 902   if (ret) {
 903     cerr << "export_files error " << ret << std::endl;
 904     return ret;
 905   }
 906
 907   ret = write_simple(TYPE_PG_END, file_fd);
 908   if (ret)
 909     return ret;
 910
 911   return 0;
 912 }
 913
 914 int dump_data(Formatter *formatter, bufferlist &bl)
 915 {
 916   bufferlist::iterator ebliter = bl.begin();
 917   data_section ds;
 918   ds.decode(ebliter);
 919
 920   formatter->open_object_section("data_block");
 921   formatter->dump_unsigned("offset", ds.offset);
 922   formatter->dump_unsigned("len", ds.len);
 923   // XXX: Add option to dump data like od -cx ?
 924   formatter->close_section();
 925   formatter->flush(cout);
 926   return 0;
 927 }
 928
 929 int get_data(ObjectStore *store, coll_t coll, ghobject_t hoid,
 930     ObjectStore::Transaction *t, bufferlist &bl)
 931 {
 932   bufferlist::iterator ebliter = bl.begin();
 933   data_section ds;
 934   ds.decode(ebliter);
 935
 936   if (debug)
 937     cerr << "\tdata: offset " << ds.offset << " len " << ds.len << std::endl;
 938   t->write(coll, hoid, ds.offset, ds.len,  ds.databl);
 939   return 0;
 940 }
 941
 942 int dump_attrs(
 943   Formatter *formatter, ghobject_t hoid,
 944   bufferlist &bl)
 945 {
 946   bufferlist::iterator ebliter = bl.begin();
 947   attr_section as;
 948   as.decode(ebliter);
 949
 950   // This could have been handled in the caller if we didn't need to
 951   // support exports that didn't include object_info_t in object_begin.
 952   if (hoid.generation == ghobject_t::NO_GEN &&
 953       hoid.hobj.is_head()) {
 954     map<string,bufferlist>::iterator mi = as.data.find(SS_ATTR);
 955     if (mi != as.data.end()) {
 956       SnapSet snapset;
 957       auto p = mi->second.begin();
 958       snapset.decode(p);
 959       formatter->open_object_section("snapset");
 960       snapset.dump(formatter);
 961       formatter->close_section();
 962     } else {
 963       formatter->open_object_section("snapset");
 964       formatter->dump_string("error", "missing SS_ATTR");
 965       formatter->close_section();
 966     }
 967   }
 968
 969   formatter->open_object_section("attrs");
 970   formatter->open_array_section("user");
 971   for (auto kv : as.data) {
 972       // Skip system attributes
 973       if (('_' != kv.first.at(0)) || kv.first.size() == 1)
 974         continue;
 975       formatter->open_object_section("user_attr");
 976       formatter->dump_string("name", kv.first.substr(1));
 977       bool b64;
 978       formatter->dump_string("value", cleanbin(kv.second, b64));
 979       formatter->dump_bool("Base64", b64);
 980       formatter->close_section();
 981   }
 982   formatter->close_section();
 983   formatter->open_array_section("system");
 984   for (auto kv : as.data) {
 985       // Skip user attributes
 986       if (('_' == kv.first.at(0)) && kv.first.size() != 1)
 987         continue;
 988       formatter->open_object_section("sys_attr");
 989       formatter->dump_string("name", kv.first);
 990       formatter->close_section();
 991   }
 992   formatter->close_section();
 993   formatter->close_section();
 994   formatter->flush(cout);
 995
 996   return 0;
 997 }
 998
 999 int get_attrs(
1000   ObjectStore *store, coll_t coll, ghobject_t hoid,
1001   ObjectStore::Transaction *t, bufferlist &bl,
1002   OSDriver &driver, SnapMapper &snap_mapper)
1003 {
1004   bufferlist::iterator ebliter = bl.begin();
1005   attr_section as;
1006   as.decode(ebliter);
1007
1008   if (debug)
1009     cerr << "\tattrs: len " << as.data.size() << std::endl;
1010   t->setattrs(coll, hoid, as.data);
1011
1012   // This could have been handled in the caller if we didn't need to
1013   // support exports that didn't include object_info_t in object_begin.
1014   if (hoid.generation == ghobject_t::NO_GEN) {
1015     if (hoid.hobj.snap < CEPH_MAXSNAP) {
1016       map<string,bufferlist>::iterator mi = as.data.find(OI_ATTR);
1017       if (mi != as.data.end()) {
1018         object_info_t oi(mi->second);
1019
1020         if (debug)
1021           cerr << "object_info " << oi << std::endl;
1022
1023         OSDriver::OSTransaction _t(driver.get_transaction(t));
1024         set<snapid_t> oi_snaps(oi.legacy_snaps.begin(), oi.legacy_snaps.end());
1025         if (!oi_snaps.empty()) {
1026           if (debug)
1027             cerr << "\tsetting legacy snaps " << oi_snaps << std::endl;
1028           snap_mapper.add_oid(hoid.hobj, oi_snaps, &_t);
1029         }
1030       }
1031     } else {
1032       if (hoid.hobj.is_head()) {
1033         map<string,bufferlist>::iterator mi = as.data.find(SS_ATTR);
1034         if (mi != as.data.end()) {
1035           SnapSet snapset;
1036           auto p = mi->second.begin();
1037           snapset.decode(p);
1038           cout << "snapset " << snapset << std::endl;
1039           if (!snapset.is_legacy()) {
1040             for (auto& p : snapset.clone_snaps) {
1041               ghobject_t clone = hoid;
1042               clone.hobj.snap = p.first;
1043               set<snapid_t> snaps(p.second.begin(), p.second.end());
1044               if (!store->exists(coll, clone)) {
1045                 // no clone, skip.  this is probably a cache pool.  this works
1046                 // because we use a separate transaction per object and clones
1047                 // come before head in the archive.
1048                 if (debug)
1049                   cerr << "\tskipping missing " << clone << " (snaps "
1050                        << snaps << ")" << std::endl;
1051                 continue;
1052               }
1053               if (debug)
1054                 cerr << "\tsetting " << clone.hobj << " snaps " << snaps
1055                      << std::endl;
1056               OSDriver::OSTransaction _t(driver.get_transaction(t));
1057               assert(!snaps.empty());
1058               snap_mapper.add_oid(clone.hobj, snaps, &_t);
1059             }
1060           }
1061         } else {
1062           cerr << "missing SS_ATTR on " << hoid << std::endl;
1063         }
1064       }
1065     }
1066   }
1067
1068   return 0;
1069 }
1070
1071 int dump_omap_hdr(Formatter *formatter, bufferlist &bl)
1072 {
1073   bufferlist::iterator ebliter = bl.begin();
1074   omap_hdr_section oh;
1075   oh.decode(ebliter);
1076
1077   formatter->open_object_section("omap_header");
1078   formatter->dump_string("value", string(oh.hdr.c_str(), oh.hdr.length()));
1079   formatter->close_section();
1080   formatter->flush(cout);
1081   return 0;
1082 }
1083
1084 int get_omap_hdr(ObjectStore *store, coll_t coll, ghobject_t hoid,
1085     ObjectStore::Transaction *t, bufferlist &bl)
1086 {
1087   bufferlist::iterator ebliter = bl.begin();
1088   omap_hdr_section oh;
1089   oh.decode(ebliter);
1090
1091   if (debug)
1092     cerr << "\tomap header: " << string(oh.hdr.c_str(), oh.hdr.length())
1093       << std::endl;
1094   t->omap_setheader(coll, hoid, oh.hdr);
1095   return 0;
1096 }
1097
1098 int dump_omap(Formatter *formatter, bufferlist &bl)
1099 {
1100   bufferlist::iterator ebliter = bl.begin();
1101   omap_section os;
1102   os.decode(ebliter);
1103
1104   formatter->open_object_section("omaps");
1105   formatter->dump_unsigned("count", os.omap.size());
1106   formatter->open_array_section("data");
1107   for (auto o : os.omap) {
1108       formatter->open_object_section("omap");
1109       formatter->dump_string("name", o.first);
1110       bool b64;
1111       formatter->dump_string("value", cleanbin(o.second, b64));
1112       formatter->dump_bool("Base64", b64);
1113       formatter->close_section();
1114   }
1115   formatter->close_section();
1116   formatter->close_section();
1117   formatter->flush(cout);
1118   return 0;
1119 }
1120
1121 int get_omap(ObjectStore *store, coll_t coll, ghobject_t hoid,
1122     ObjectStore::Transaction *t, bufferlist &bl)
1123 {
1124   bufferlist::iterator ebliter = bl.begin();
1125   omap_section os;
1126   os.decode(ebliter);
1127
1128   if (debug)
1129     cerr << "\tomap: size " << os.omap.size() << std::endl;
1130   t->omap_setkeys(coll, hoid, os.omap);
1131   return 0;
1132 }
1133
1134 int ObjectStoreTool::dump_object(Formatter *formatter,
1135                                 bufferlist &bl)
1136 {
1137   bufferlist::iterator ebliter = bl.begin();
1138   object_begin ob;
1139   ob.decode(ebliter);
1140
1141   if (ob.hoid.hobj.is_temp()) {
1142     cerr << "ERROR: Export contains temporary object '" << ob.hoid << "'" << std::endl;
1143     return -EFAULT;
1144   }
1145
1146   formatter->open_object_section("object");
1147   formatter->open_object_section("oid");
1148   ob.hoid.dump(formatter);
1149   formatter->close_section();
1150   formatter->open_object_section("object_info");
1151   ob.oi.dump(formatter);
1152   formatter->close_section();
1153
1154   bufferlist ebl;
1155   bool done = false;
1156   while(!done) {
1157     sectiontype_t type;
1158     int ret = read_section(&type, &ebl);
1159     if (ret)
1160       return ret;
1161
1162     //cout << "\tdo_object: Section type " << hex << type << dec << std::endl;
1163     //cout << "\t\tsection size " << ebl.length() << std::endl;
1164     if (type >= END_OF_TYPES) {
1165       cout << "Skipping unknown object section type" << std::endl;
1166       continue;
1167     }
1168     switch(type) {
1169     case TYPE_DATA:
1170       if (dry_run) break;
1171       ret = dump_data(formatter, ebl);
1172       if (ret) return ret;
1173       break;
1174     case TYPE_ATTRS:
1175       if (dry_run) break;
1176       ret = dump_attrs(formatter, ob.hoid, ebl);
1177       if (ret) return ret;
1178       break;
1179     case TYPE_OMAP_HDR:
1180       if (dry_run) break;
1181       ret = dump_omap_hdr(formatter, ebl);
1182       if (ret) return ret;
1183       break;
1184     case TYPE_OMAP:
1185       if (dry_run) break;
1186       ret = dump_omap(formatter, ebl);
1187       if (ret) return ret;
1188       break;
1189     case TYPE_OBJECT_END:
1190       done = true;
1191       break;
1192     default:
1193       cerr << "Unknown section type " << type << std::endl;
1194       return -EFAULT;
1195     }
1196   }
1197   formatter->close_section();
1198   return 0;
1199 }
1200
1201 int ObjectStoreTool::get_object(ObjectStore *store, coll_t coll,
1202                                 bufferlist &bl, OSDMap &curmap,
1203                                 bool *skipped_objects,
1204                                 ObjectStore::Sequencer &osr)
1205 {
1206   ObjectStore::Transaction tran;
1207   ObjectStore::Transaction *t = &tran;
1208   bufferlist::iterator ebliter = bl.begin();
1209   object_begin ob;
1210   ob.decode(ebliter);
1211   OSDriver driver(
1212     store,
1213     coll_t(),
1214     OSD::make_snapmapper_oid());
1215   spg_t pg;
1216   coll.is_pg_prefix(&pg);
1217   SnapMapper mapper(g_ceph_context, &driver, 0, 0, 0, pg.shard);
1218
1219   if (ob.hoid.hobj.is_temp()) {
1220     cerr << "ERROR: Export contains temporary object '" << ob.hoid << "'" << std::endl;
1221     return -EFAULT;
1222   }
1223   assert(g_ceph_context);
1224   if (ob.hoid.hobj.nspace != g_ceph_context->_conf->osd_hit_set_namespace) {
1225     object_t oid = ob.hoid.hobj.oid;
1226     object_locator_t loc(ob.hoid.hobj);
1227     pg_t raw_pgid = curmap.object_locator_to_pg(oid, loc);
1228     pg_t pgid = curmap.raw_pg_to_pg(raw_pgid);
1229
1230     spg_t coll_pgid;
1231     if (coll.is_pg(&coll_pgid) == false) {
1232       cerr << "INTERNAL ERROR: Bad collection during import" << std::endl;
1233       return -EFAULT;
1234     }
1235     if (coll_pgid.shard != ob.hoid.shard_id) {
1236       cerr << "INTERNAL ERROR: Importing shard " << coll_pgid.shard
1237         << " but object shard is " << ob.hoid.shard_id << std::endl;
1238       return -EFAULT;
1239     }
1240
1241     if (coll_pgid.pgid != pgid) {
1242       cerr << "Skipping object '" << ob.hoid << "' which belongs in pg " << pgid << std::endl;
1243       *skipped_objects = true;
1244       skip_object(bl);
1245       return 0;
1246     }
1247   }
1248
1249   if (!dry_run)
1250     t->touch(coll, ob.hoid);
1251
1252   cout << "Write " << ob.hoid << std::endl;
1253
1254   bufferlist ebl;
1255   bool done = false;
1256   while(!done) {
1257     sectiontype_t type;
1258     int ret = read_section(&type, &ebl);
1259     if (ret)
1260       return ret;
1261
1262     //cout << "\tdo_object: Section type " << hex << type << dec << std::endl;
1263     //cout << "\t\tsection size " << ebl.length() << std::endl;
1264     if (type >= END_OF_TYPES) {
1265       cout << "Skipping unknown object section type" << std::endl;
1266       continue;
1267     }
1268     switch(type) {
1269     case TYPE_DATA:
1270       if (dry_run) break;
1271       ret = get_data(store, coll, ob.hoid, t, ebl);
1272       if (ret) return ret;
1273       break;
1274     case TYPE_ATTRS:
1275       if (dry_run) break;
1276       ret = get_attrs(store, coll, ob.hoid, t, ebl, driver, mapper);
1277       if (ret) return ret;
1278       break;
1279     case TYPE_OMAP_HDR:
1280       if (dry_run) break;
1281       ret = get_omap_hdr(store, coll, ob.hoid, t, ebl);
1282       if (ret) return ret;
1283       break;
1284     case TYPE_OMAP:
1285       if (dry_run) break;
1286       ret = get_omap(store, coll, ob.hoid, t, ebl);
1287       if (ret) return ret;
1288       break;
1289     case TYPE_OBJECT_END:
1290       done = true;
1291       break;
1292     default:
1293       cerr << "Unknown section type " << type << std::endl;
1294       return -EFAULT;
1295     }
1296   }
1297   if (!dry_run)
1298     store->apply_transaction(&osr, std::move(*t));
1299   return 0;
1300 }
1301
1302 int dump_pg_metadata(Formatter *formatter, bufferlist &bl, metadata_section &ms)
1303 {
1304   bufferlist::iterator ebliter = bl.begin();
1305   ms.decode(ebliter);
1306
1307   formatter->open_object_section("metadata_section");
1308
1309   formatter->dump_unsigned("pg_disk_version", (int)ms.struct_ver);
1310   formatter->dump_unsigned("map_epoch", ms.map_epoch);
1311
1312   formatter->open_object_section("OSDMap");
1313   ms.osdmap.dump(formatter);
1314   formatter->close_section();
1315   formatter->flush(cout);
1316   cout << std::endl;
1317
1318   formatter->open_object_section("info");
1319   ms.info.dump(formatter);
1320   formatter->close_section();
1321   formatter->flush(cout);
1322
1323   formatter->open_object_section("log");
1324   ms.log.dump(formatter);
1325   formatter->close_section();
1326   formatter->flush(cout);
1327
1328   formatter->open_object_section("pg_missing_t");
1329   ms.missing.dump(formatter);
1330   formatter->close_section();
1331
1332   // XXX: ms.past_intervals?
1333
1334   formatter->close_section();
1335   formatter->flush(cout);
1336
1337   if (ms.osdmap.get_epoch() != 0 && ms.map_epoch != ms.osdmap.get_epoch()) {
1338     cerr << "FATAL: Invalid OSDMap epoch in export data" << std::endl;
1339     return -EFAULT;
1340   }
1341
1342   return 0;
1343 }
1344
1345 int get_pg_metadata(ObjectStore *store, bufferlist &bl, metadata_section &ms,
1346     const OSDSuperblock& sb, OSDMap& curmap, spg_t pgid)
1347 {
1348   bufferlist::iterator ebliter = bl.begin();
1349   ms.decode(ebliter);
1350   spg_t old_pgid = ms.info.pgid;
1351   ms.info.pgid = pgid;
1352
1353 #if DIAGNOSTIC
1354   Formatter *formatter = new JSONFormatter(true);
1355   cout << "export pgid " << old_pgid << std::endl;
1356   cout << "struct_v " << (int)ms.struct_ver << std::endl;
1357   cout << "map epoch " << ms.map_epoch << std::endl;
1358
1359   formatter->open_object_section("importing OSDMap");
1360   ms.osdmap.dump(formatter);
1361   formatter->close_section();
1362   formatter->flush(cout);
1363   cout << std::endl;
1364
1365   cout << "osd current epoch " << sb.current_epoch << std::endl;
1366   formatter->open_object_section("current OSDMap");
1367   curmap.dump(formatter);
1368   formatter->close_section();
1369   formatter->flush(cout);
1370   cout << std::endl;
1371
1372   formatter->open_object_section("info");
1373   ms.info.dump(formatter);
1374   formatter->close_section();
1375   formatter->flush(cout);
1376   cout << std::endl;
1377
1378   formatter->open_object_section("log");
1379   ms.log.dump(formatter);
1380   formatter->close_section();
1381   formatter->flush(cout);
1382   cout << std::endl;
1383
1384   formatter->flush(cout);
1385   cout << std::endl;
1386 #endif
1387
1388   if (ms.osdmap.get_epoch() != 0 && ms.map_epoch != ms.osdmap.get_epoch()) {
1389     cerr << "FATAL: Invalid OSDMap epoch in export data" << std::endl;
1390     return -EFAULT;
1391   }
1392
1393   if (ms.map_epoch > sb.current_epoch) {
1394     cerr << "ERROR: Export PG's map_epoch " << ms.map_epoch << " > OSD's epoch " << sb.current_epoch << std::endl;
1395     cerr << "The OSD you are using is older than the exported PG" << std::endl;
1396     cerr << "Either use another OSD or join selected OSD to cluster to update it first" << std::endl;
1397     return -EINVAL;
1398   }
1399
1400   // Pool verified to exist for call to get_pg_num().
1401   unsigned new_pg_num = curmap.get_pg_num(pgid.pgid.pool());
1402
1403   if (pgid.pgid.ps() >= new_pg_num) {
1404     cerr << "Illegal pgid, the seed is larger than current pg_num" << std::endl;
1405     return -EINVAL;
1406   }
1407
1408   // Old exports didn't include OSDMap, see if we have a copy locally
1409   if (ms.osdmap.get_epoch() == 0) {
1410     OSDMap findmap;
1411     bufferlist findmap_bl;
1412     int ret = get_osdmap(store, ms.map_epoch, findmap, findmap_bl);
1413     if (ret == 0) {
1414       ms.osdmap.deepish_copy_from(findmap);
1415     } else {
1416       cerr << "WARNING: No OSDMap in old export,"
1417            " some objects may be ignored due to a split" << std::endl;
1418     }
1419   }
1420
1421   // Make sure old_pg_num is 0 in the unusual case that OSDMap not in export
1422   // nor can we find a local copy.
1423   unsigned old_pg_num = 0;
1424   if (ms.osdmap.get_epoch() != 0)
1425     old_pg_num = ms.osdmap.get_pg_num(pgid.pgid.pool());
1426
1427   if (debug) {
1428     cerr << "old_pg_num " << old_pg_num << std::endl;
1429     cerr << "new_pg_num " << new_pg_num << std::endl;
1430     cerr << ms.osdmap << std::endl;
1431     cerr << curmap << std::endl;
1432   }
1433
1434   // If we have managed to have a good OSDMap we can do these checks
1435   if (old_pg_num) {
1436     if (old_pgid.pgid.ps() >= old_pg_num) {
1437       cerr << "FATAL: pgid invalid for original map epoch" << std::endl;
1438       return -EFAULT;
1439     }
1440     if (pgid.pgid.ps() >= old_pg_num) {
1441       cout << "NOTICE: Post split pgid specified" << std::endl;
1442     } else {
1443       spg_t parent(pgid);
1444       if (parent.is_split(old_pg_num, new_pg_num, NULL)) {
1445             cerr << "WARNING: Split occurred, some objects may be ignored" << std::endl;
1446       }
1447     }
1448   }
1449
1450   if (debug) {
1451     cerr << "Import pgid " << ms.info.pgid << std::endl;
1452     cerr << "Previous past_intervals " << ms.past_intervals << std::endl;
1453     cerr << "history.same_interval_since " << ms.info.history.same_interval_since << std::endl;
1454   }
1455
1456   // Let osd recompute past_intervals and same_interval_since
1457   ms.past_intervals.clear();
1458   ms.info.history.same_interval_since =  0;
1459
1460   if (debug)
1461     cerr << "Changing pg epoch " << ms.map_epoch << " to " << sb.current_epoch << std::endl;
1462
1463   ms.map_epoch = sb.current_epoch;
1464
1465   return 0;
1466 }
1467
1468 // out: pg_log_t that only has entries that apply to import_pgid using curmap
1469 // reject: Entries rejected from "in" are in the reject.log.  Other fields not set.
1470 void filter_divergent_priors(spg_t import_pgid, const OSDMap &curmap,
1471   const string &hit_set_namespace, const divergent_priors_t &in,
1472   divergent_priors_t &out, divergent_priors_t &reject)
1473 {
1474   out.clear();
1475   reject.clear();
1476
1477   for (divergent_priors_t::const_iterator i = in.begin();
1478        i != in.end(); ++i) {
1479
1480     // Reject divergent priors for temporary objects
1481     if (i->second.is_temp()) {
1482       reject.insert(*i);
1483       continue;
1484     }
1485
1486     if (i->second.nspace != hit_set_namespace) {
1487       object_t oid = i->second.oid;
1488       object_locator_t loc(i->second);
1489       pg_t raw_pgid = curmap.object_locator_to_pg(oid, loc);
1490       pg_t pgid = curmap.raw_pg_to_pg(raw_pgid);
1491
1492       if (import_pgid.pgid == pgid) {
1493         out.insert(*i);
1494       } else {
1495         reject.insert(*i);
1496       }
1497     } else {
1498       out.insert(*i);
1499     }
1500   }
1501 }
1502
1503 int ObjectStoreTool::dump_import(Formatter *formatter)
1504 {
1505   bufferlist ebl;
1506   pg_info_t info;
1507   PGLog::IndexedLog log;
1508   //bool skipped_objects = false;
1509
1510   int ret = read_super();
1511   if (ret)
1512     return ret;
1513
1514   if (sh.magic != super_header::super_magic) {
1515     cerr << "Invalid magic number" << std::endl;
1516     return -EFAULT;
1517   }
1518
1519   if (sh.version > super_header::super_ver) {
1520     cerr << "Can't handle export format version=" << sh.version << std::endl;
1521     return -EINVAL;
1522   }
1523
1524   formatter->open_object_section("Export");
1525
1526   //First section must be TYPE_PG_BEGIN
1527   sectiontype_t type;
1528   ret = read_section(&type, &ebl);
1529   if (ret)
1530     return ret;
1531   if (type == TYPE_POOL_BEGIN) {
1532     cerr << "Dump of pool exports not supported" << std::endl;
1533     return -EINVAL;
1534   } else if (type != TYPE_PG_BEGIN) {
1535     cerr << "Invalid first section type " << std::to_string(type) << std::endl;
1536     return -EFAULT;
1537   }
1538
1539   bufferlist::iterator ebliter = ebl.begin();
1540   pg_begin pgb;
1541   pgb.decode(ebliter);
1542   spg_t pgid = pgb.pgid;
1543
1544   formatter->dump_string("pgid", stringify(pgid));
1545   formatter->dump_string("cluster_fsid", stringify(pgb.superblock.cluster_fsid));
1546   formatter->dump_string("features", stringify(pgb.superblock.compat_features));
1547
1548   bool done = false;
1549   bool found_metadata = false;
1550   metadata_section ms;
1551   bool objects_started = false;
1552   while(!done) {
1553     ret = read_section(&type, &ebl);
1554     if (ret)
1555       return ret;
1556
1557     if (debug) {
1558       cerr << "dump_import: Section type " << std::to_string(type) << std::endl;
1559     }
1560     if (type >= END_OF_TYPES) {
1561       cerr << "Skipping unknown section type" << std::endl;
1562       continue;
1563     }
1564     switch(type) {
1565     case TYPE_OBJECT_BEGIN:
1566       if (!objects_started) {
1567         formatter->open_array_section("objects");
1568         objects_started = true;
1569       }
1570       ret = dump_object(formatter, ebl);
1571       if (ret) return ret;
1572       break;
1573     case TYPE_PG_METADATA:
1574       if (objects_started)
1575         cerr << "WARNING: metadata_section out of order" << std::endl;
1576       ret = dump_pg_metadata(formatter, ebl, ms);
1577       if (ret) return ret;
1578       found_metadata = true;
1579       break;
1580     case TYPE_PG_END:
1581       if (objects_started) {
1582         formatter->close_section();
1583       }
1584       done = true;
1585       break;
1586     default:
1587       cerr << "Unknown section type " << std::to_string(type) << std::endl;
1588       return -EFAULT;
1589     }
1590   }
1591
1592   if (!found_metadata) {
1593     cerr << "Missing metadata section" << std::endl;
1594     return -EFAULT;
1595   }
1596
1597   formatter->close_section();
1598   formatter->flush(cout);
1599
1600   return 0;
1601 }
1602
1603 int ObjectStoreTool::do_import(ObjectStore *store, OSDSuperblock& sb,
1604                                bool force, std::string pgidstr,
1605                                ObjectStore::Sequencer &osr)
1606 {
1607   bufferlist ebl;
1608   pg_info_t info;
1609   PGLog::IndexedLog log;
1610   bool skipped_objects = false;
1611
1612   if (!dry_run)
1613     finish_remove_pgs(store);
1614
1615   int ret = read_super();
1616   if (ret)
1617     return ret;
1618
1619   if (sh.magic != super_header::super_magic) {
1620     cerr << "Invalid magic number" << std::endl;
1621     return -EFAULT;
1622   }
1623
1624   if (sh.version > super_header::super_ver) {
1625     cerr << "Can't handle export format version=" << sh.version << std::endl;
1626     return -EINVAL;
1627   }
1628
1629   //First section must be TYPE_PG_BEGIN
1630   sectiontype_t type;
1631   ret = read_section(&type, &ebl);
1632   if (ret)
1633     return ret;
1634   if (type == TYPE_POOL_BEGIN) {
1635     cerr << "Pool exports cannot be imported into a PG" << std::endl;
1636     return -EINVAL;
1637   } else if (type != TYPE_PG_BEGIN) {
1638     cerr << "Invalid first section type " << std::to_string(type) << std::endl;
1639     return -EFAULT;
1640   }
1641
1642   bufferlist::iterator ebliter = ebl.begin();
1643   pg_begin pgb;
1644   pgb.decode(ebliter);
1645   spg_t pgid = pgb.pgid;
1646   spg_t orig_pgid = pgid;
1647
1648   if (pgidstr.length()) {
1649     spg_t user_pgid;
1650
1651     bool ok = user_pgid.parse(pgidstr.c_str());
1652     // This succeeded in main() already
1653     assert(ok);
1654     if (pgid != user_pgid) {
1655       if (pgid.pool() != user_pgid.pool()) {
1656         cerr << "Can't specify a different pgid pool, must be " << pgid.pool() << std::endl;
1657         return -EINVAL;
1658       }
1659       if (pgid.is_no_shard() && !user_pgid.is_no_shard()) {
1660         cerr << "Can't specify a sharded pgid with a non-sharded export" << std::endl;
1661         return -EINVAL;
1662       }
1663       // Get shard from export information if not specified
1664       if (!pgid.is_no_shard() && user_pgid.is_no_shard()) {
1665         user_pgid.shard = pgid.shard;
1666       }
1667       if (pgid.shard != user_pgid.shard) {
1668         cerr << "Can't specify a different shard, must be " << pgid.shard << std::endl;
1669         return -EINVAL;
1670       }
1671       pgid = user_pgid;
1672     }
1673   }
1674
1675   if (!pgb.superblock.cluster_fsid.is_zero()
1676       && pgb.superblock.cluster_fsid != sb.cluster_fsid) {
1677     cerr << "Export came from different cluster with fsid "
1678          << pgb.superblock.cluster_fsid << std::endl;
1679     return -EINVAL;
1680   }
1681
1682   if (debug) {
1683     cerr << "Exported features: " << pgb.superblock.compat_features << std::endl;
1684   }
1685
1686   // Special case: Old export has SHARDS incompat feature on replicated pg, remove it
1687   if (pgid.is_no_shard())
1688     pgb.superblock.compat_features.incompat.remove(CEPH_OSD_FEATURE_INCOMPAT_SHARDS);
1689
1690   if (sb.compat_features.compare(pgb.superblock.compat_features) == -1) {
1691     CompatSet unsupported = sb.compat_features.unsupported(pgb.superblock.compat_features);
1692
1693     cerr << "Export has incompatible features set " << unsupported << std::endl;
1694
1695     // Let them import if they specify the --force option
1696     if (!force)
1697         return 11;  // Positive return means exit status
1698   }
1699
1700   // Don't import if pool no longer exists
1701   OSDMap curmap;
1702   bufferlist bl;
1703   ret = get_osdmap(store, sb.current_epoch, curmap, bl);
1704   if (ret) {
1705     cerr << "Can't find local OSDMap" << std::endl;
1706     return ret;
1707   }
1708   if (!curmap.have_pg_pool(pgid.pgid.m_pool)) {
1709     cerr << "Pool " << pgid.pgid.m_pool << " no longer exists" << std::endl;
1710     // Special exit code for this error, used by test code
1711     return 10;  // Positive return means exit status
1712   }
1713
1714   ghobject_t pgmeta_oid = pgid.make_pgmeta_oid();
1715   log_oid = OSD::make_pg_log_oid(pgid);
1716   biginfo_oid = OSD::make_pg_biginfo_oid(pgid);
1717
1718   //Check for PG already present.
1719   coll_t coll(pgid);
1720   if (store->collection_exists(coll)) {
1721     cerr << "pgid " << pgid << " already exists" << std::endl;
1722     return -EEXIST;
1723   }
1724
1725   if (!dry_run) {
1726     ObjectStore::Transaction t;
1727     PG::_create(t, pgid,
1728                 pgid.get_split_bits(curmap.get_pg_pool(pgid.pool())->get_pg_num()));
1729     PG::_init(t, pgid, NULL);
1730
1731     // mark this coll for removal until we're done
1732     map<string,bufferlist> values;
1733     ::encode((char)1, values["_remove"]);
1734     t.omap_setkeys(coll, pgid.make_pgmeta_oid(), values);
1735
1736     store->apply_transaction(&osr, std::move(t));
1737   }
1738
1739   cout << "Importing pgid " << pgid;
1740   if (orig_pgid != pgid) {
1741     cout << " exported as " << orig_pgid;
1742   }
1743   cout << std::endl;
1744
1745   bool done = false;
1746   bool found_metadata = false;
1747   metadata_section ms;
1748   while(!done) {
1749     ret = read_section(&type, &ebl);
1750     if (ret)
1751       return ret;
1752
1753     if (debug) {
1754       cout << __func__ << ": Section type " << std::to_string(type) << std::endl;
1755     }
1756     if (type >= END_OF_TYPES) {
1757       cout << "Skipping unknown section type" << std::endl;
1758       continue;
1759     }
1760     switch(type) {
1761     case TYPE_OBJECT_BEGIN:
1762       ret = get_object(store, coll, ebl, curmap, &skipped_objects, osr);
1763       if (ret) return ret;
1764       break;
1765     case TYPE_PG_METADATA:
1766       ret = get_pg_metadata(store, ebl, ms, sb, curmap, pgid);
1767       if (ret) return ret;
1768       found_metadata = true;
1769       break;
1770     case TYPE_PG_END:
1771       done = true;
1772       break;
1773     default:
1774       cerr << "Unknown section type " << std::to_string(type) << std::endl;
1775       return -EFAULT;
1776     }
1777   }
1778
1779   if (!found_metadata) {
1780     cerr << "Missing metadata section" << std::endl;
1781     return -EFAULT;
1782   }
1783
1784   ObjectStore::Transaction t;
1785   if (!dry_run) {
1786     pg_log_t newlog, reject;
1787     pg_log_t::filter_log(pgid, curmap, g_ceph_context->_conf->osd_hit_set_namespace,
1788       ms.log, newlog, reject);
1789     if (debug) {
1790       for (list<pg_log_entry_t>::iterator i = newlog.log.begin();
1791            i != newlog.log.end(); ++i)
1792         cerr << "Keeping log entry " << *i << std::endl;
1793       for (list<pg_log_entry_t>::iterator i = reject.log.begin();
1794            i != reject.log.end(); ++i)
1795         cerr << "Skipping log entry " << *i << std::endl;
1796     }
1797
1798     divergent_priors_t newdp, rejectdp;
1799     filter_divergent_priors(pgid, curmap, g_ceph_context->_conf->osd_hit_set_namespace,
1800       ms.divergent_priors, newdp, rejectdp);
1801     ms.divergent_priors = newdp;
1802     if (debug) {
1803       for (divergent_priors_t::iterator i = newdp.begin();
1804            i != newdp.end(); ++i)
1805         cerr << "Keeping divergent_prior " << *i << std::endl;
1806       for (divergent_priors_t::iterator i = rejectdp.begin();
1807            i != rejectdp.end(); ++i)
1808         cerr << "Skipping divergent_prior " << *i << std::endl;
1809     }
1810
1811     ms.missing.filter_objects([&](const hobject_t &obj) {
1812         if (obj.nspace == g_ceph_context->_conf->osd_hit_set_namespace)
1813           return false;
1814         assert(!obj.is_temp());
1815         object_t oid = obj.oid;
1816         object_locator_t loc(obj);
1817         pg_t raw_pgid = curmap.object_locator_to_pg(oid, loc);
1818         pg_t _pgid = curmap.raw_pg_to_pg(raw_pgid);
1819
1820         return pgid.pgid != _pgid;
1821       });
1822
1823
1824     if (debug) {
1825       pg_missing_t missing;
1826       Formatter *formatter = Formatter::create("json-pretty");
1827       dump_log(formatter, cerr, newlog, ms.missing);
1828       delete formatter;
1829     }
1830
1831     // Just like a split invalidate stats since the object count is changed
1832     if (skipped_objects)
1833       ms.info.stats.stats_invalid = true;
1834
1835     ret = write_pg(
1836       t,
1837       ms.map_epoch,
1838       ms.info,
1839       newlog,
1840       ms.past_intervals,
1841       ms.divergent_priors,
1842       ms.missing);
1843     if (ret) return ret;
1844   }
1845
1846   // done, clear removal flag
1847   if (debug)
1848     cerr << "done, clearing removal flag" << std::endl;
1849
1850   if (!dry_run) {
1851     set<string> remove;
1852     remove.insert("_remove");
1853     t.omap_rmkeys(coll, pgid.make_pgmeta_oid(), remove);
1854     store->apply_transaction(&osr, std::move(t));
1855   }
1856
1857   return 0;
1858 }
1859
1860 int do_list(ObjectStore *store, string pgidstr, string object, boost::optional<std::string> nspace,
1861             Formatter *formatter, bool debug, bool human_readable, bool head)
1862 {
1863   int r;
1864   lookup_ghobject lookup(object, nspace, head);
1865   if (pgidstr.length() > 0) {
1866     r = action_on_all_objects_in_pg(store, pgidstr, lookup, debug);
1867   } else {
1868     r = action_on_all_objects(store, lookup, debug);
1869   }
1870   if (r)
1871     return r;
1872   lookup.dump(formatter, human_readable);
1873   formatter->flush(cout);
1874   return 0;
1875 }
1876
1877 int do_meta(ObjectStore *store, string object, Formatter *formatter, bool debug, bool human_readable)
1878 {
1879   int r;
1880   boost::optional<std::string> nspace; // Not specified
1881   lookup_ghobject lookup(object, nspace);
1882   r = action_on_all_objects_in_exact_pg(store, coll_t::meta(), lookup, debug);
1883   if (r)
1884     return r;
1885   lookup.dump(formatter, human_readable);
1886   formatter->flush(cout);
1887   return 0;
1888 }
1889
1890 int remove_object(coll_t coll, ghobject_t &ghobj,
1891   SnapMapper &mapper,
1892   MapCacher::Transaction<std::string, bufferlist> *_t,
1893   ObjectStore::Transaction *t)
1894 {
1895   int r = mapper.remove_oid(ghobj.hobj, _t);
1896   if (r < 0 && r != -ENOENT) {
1897     cerr << "remove_oid returned " << cpp_strerror(r) << std::endl;
1898     return r;
1899   }
1900
1901   t->remove(coll, ghobj);
1902   return 0;
1903 }
1904
1905 int get_snapset(ObjectStore *store, coll_t coll, ghobject_t &ghobj, SnapSet &ss, bool silent);
1906
1907 int do_remove_object(ObjectStore *store, coll_t coll,
1908                      ghobject_t &ghobj, bool all, bool force,
1909                      ObjectStore::Sequencer &osr)
1910 {
1911   spg_t pg;
1912   coll.is_pg_prefix(&pg);
1913   OSDriver driver(
1914     store,
1915     coll_t(),
1916     OSD::make_snapmapper_oid());
1917   SnapMapper mapper(g_ceph_context, &driver, 0, 0, 0, pg.shard);
1918   struct stat st;
1919
1920   int r = store->stat(coll, ghobj, &st);
1921   if (r < 0) {
1922     cerr << "remove: " << cpp_strerror(r) << std::endl;
1923     return r;
1924   }
1925
1926   SnapSet ss;
1927   if (ghobj.hobj.has_snapset()) {
1928     r = get_snapset(store, coll, ghobj, ss, false);
1929     if (r < 0) {
1930       cerr << "Can't get snapset error " << cpp_strerror(r) << std::endl;
1931       return r;
1932     }
1933     if (!ss.snaps.empty() && !all) {
1934       if (force) {
1935         cout << "WARNING: only removing "
1936              << (ghobj.hobj.is_head() ? "head" : "snapdir")
1937              << " with snapshots present" << std::endl;
1938         ss.snaps.clear();
1939       } else {
1940         cerr << "Snapshots are present, use removeall to delete everything" << std::endl;
1941         return -EINVAL;
1942       }
1943     }
1944   }
1945
1946   ObjectStore::Transaction t;
1947   OSDriver::OSTransaction _t(driver.get_transaction(&t));
1948
1949   cout << "remove " << ghobj << std::endl;
1950
1951   if (!dry_run) {
1952     r = remove_object(coll, ghobj, mapper, &_t, &t);
1953     if (r < 0)
1954       return r;
1955   }
1956
1957   ghobject_t snapobj = ghobj;
1958   for (vector<snapid_t>::iterator i = ss.snaps.begin() ;
1959        i != ss.snaps.end() ; ++i) {
1960     snapobj.hobj.snap = *i;
1961     cout << "remove " << snapobj << std::endl;
1962     if (!dry_run) {
1963       r = remove_object(coll, snapobj, mapper, &_t, &t);
1964       if (r < 0)
1965         return r;
1966     }
1967   }
1968
1969   if (!dry_run)
1970     store->apply_transaction(&osr, std::move(t));
1971
1972   return 0;
1973 }
1974
1975 int do_list_attrs(ObjectStore *store, coll_t coll, ghobject_t &ghobj)
1976 {
1977   map<string,bufferptr> aset;
1978   int r = store->getattrs(coll, ghobj, aset);
1979   if (r < 0) {
1980     cerr << "getattrs: " << cpp_strerror(r) << std::endl;
1981     return r;
1982   }
1983
1984   for (map<string,bufferptr>::iterator i = aset.begin();i != aset.end(); ++i) {
1985     string key(i->first);
1986     if (outistty)
1987       key = cleanbin(key);
1988     cout << key << std::endl;
1989   }
1990   return 0;
1991 }
1992
1993 int do_list_omap(ObjectStore *store, coll_t coll, ghobject_t &ghobj)
1994 {
1995   ObjectMap::ObjectMapIterator iter = store->get_omap_iterator(coll, ghobj);
1996   if (!iter) {
1997     cerr << "omap_get_iterator: " << cpp_strerror(ENOENT) << std::endl;
1998     return -ENOENT;
1999   }
2000   iter->seek_to_first();
2001   map<string, bufferlist> oset;
2002   while(iter->valid()) {
2003     get_omap_batch(iter, oset);
2004
2005     for (map<string,bufferlist>::iterator i = oset.begin();i != oset.end(); ++i) {
2006       string key(i->first);
2007       if (outistty)
2008         key = cleanbin(key);
2009       cout << key << std::endl;
2010     }
2011   }
2012   return 0;
2013 }
2014
2015 int do_get_bytes(ObjectStore *store, coll_t coll, ghobject_t &ghobj, int fd)
2016 {
2017   struct stat st;
2018   mysize_t total;
2019
2020   int ret = store->stat(coll, ghobj, &st);
2021   if (ret < 0) {
2022     cerr << "get-bytes: " << cpp_strerror(ret) << std::endl;
2023     return ret;
2024   }
2025
2026   total = st.st_size;
2027   if (debug)
2028     cerr << "size=" << total << std::endl;
2029
2030   uint64_t offset = 0;
2031   bufferlist rawdatabl;
2032   while(total > 0) {
2033     rawdatabl.clear();
2034     mysize_t len = max_read;
2035     if (len > total)
2036       len = total;
2037
2038     ret = store->read(coll, ghobj, offset, len, rawdatabl);
2039     if (ret < 0)
2040       return ret;
2041     if (ret == 0)
2042       return -EINVAL;
2043
2044     if (debug)
2045       cerr << "data section offset=" << offset << " len=" << len << std::endl;
2046
2047     total -= ret;
2048     offset += ret;
2049
2050     ret = write(fd, rawdatabl.c_str(), ret);
2051     if (ret == -1) {
2052       perror("write");
2053       return -errno;
2054     }
2055   }
2056
2057   return 0;
2058 }
2059
2060 int do_set_bytes(ObjectStore *store, coll_t coll,
2061                  ghobject_t &ghobj, int fd,
2062                  ObjectStore::Sequencer &osr)
2063 {
2064   ObjectStore::Transaction tran;
2065   ObjectStore::Transaction *t = &tran;
2066
2067   if (debug)
2068     cerr << "Write " << ghobj << std::endl;
2069
2070   if (!dry_run) {
2071     t->touch(coll, ghobj);
2072     t->truncate(coll, ghobj, 0);
2073   }
2074
2075   uint64_t offset = 0;
2076   bufferlist rawdatabl;
2077   do {
2078     rawdatabl.clear();
2079     ssize_t bytes = rawdatabl.read_fd(fd, max_read);
2080     if (bytes < 0) {
2081       cerr << "read_fd error " << cpp_strerror(bytes) << std::endl;
2082       return bytes;
2083     }
2084
2085     if (bytes == 0)
2086       break;
2087
2088     if (debug)
2089       cerr << "\tdata: offset " << offset << " bytes " << bytes << std::endl;
2090     if (!dry_run)
2091       t->write(coll, ghobj, offset, bytes,  rawdatabl);
2092
2093     offset += bytes;
2094     // XXX: Should we apply_transaction() every once in a while for very large files
2095   } while(true);
2096
2097   if (!dry_run)
2098     store->apply_transaction(&osr, std::move(*t));
2099   return 0;
2100 }
2101
2102 int do_get_attr(ObjectStore *store, coll_t coll, ghobject_t &ghobj, string key)
2103 {
2104   bufferptr bp;
2105
2106   int r = store->getattr(coll, ghobj, key.c_str(), bp);
2107   if (r < 0) {
2108     cerr << "getattr: " << cpp_strerror(r) << std::endl;
2109     return r;
2110   }
2111
2112   string value(bp.c_str(), bp.length());
2113   if (outistty) {
2114     value = cleanbin(value);
2115     value.push_back('\n');
2116   }
2117   cout << value;
2118
2119   return 0;
2120 }
2121
2122 int do_set_attr(ObjectStore *store, coll_t coll,
2123                 ghobject_t &ghobj, string key, int fd,
2124                 ObjectStore::Sequencer &osr)
2125 {
2126   ObjectStore::Transaction tran;
2127   ObjectStore::Transaction *t = &tran;
2128   bufferlist bl;
2129
2130   if (debug)
2131     cerr << "Setattr " << ghobj << std::endl;
2132
2133   int ret = get_fd_data(fd, bl);
2134   if (ret < 0)
2135     return ret;
2136
2137   if (dry_run)
2138     return 0;
2139
2140   t->touch(coll, ghobj);
2141
2142   t->setattr(coll, ghobj, key,  bl);
2143
2144   store->apply_transaction(&osr, std::move(*t));
2145   return 0;
2146 }
2147
2148 int do_rm_attr(ObjectStore *store, coll_t coll,
2149                ghobject_t &ghobj, string key,
2150                ObjectStore::Sequencer &osr)
2151 {
2152   ObjectStore::Transaction tran;
2153   ObjectStore::Transaction *t = &tran;
2154
2155   if (debug)
2156     cerr << "Rmattr " << ghobj << std::endl;
2157
2158   if (dry_run)
2159     return 0;
2160
2161   t->rmattr(coll, ghobj, key);
2162
2163   store->apply_transaction(&osr, std::move(*t));
2164   return 0;
2165 }
2166
2167 int do_get_omap(ObjectStore *store, coll_t coll, ghobject_t &ghobj, string key)
2168 {
2169   set<string> keys;
2170   map<string, bufferlist> out;
2171
2172   keys.insert(key);
2173
2174   int r = store->omap_get_values(coll, ghobj, keys, &out);
2175   if (r < 0) {
2176     cerr << "omap_get_values: " << cpp_strerror(r) << std::endl;
2177     return r;
2178   }
2179
2180   if (out.empty()) {
2181     cerr << "Key not found" << std::endl;
2182     return -ENOENT;
2183   }
2184
2185   assert(out.size() == 1);
2186
2187   bufferlist bl = out.begin()->second;
2188   string value(bl.c_str(), bl.length());
2189   if (outistty) {
2190     value = cleanbin(value);
2191     value.push_back('\n');
2192   }
2193   cout << value;
2194
2195   return 0;
2196 }
2197
2198 int do_set_omap(ObjectStore *store, coll_t coll,
2199                 ghobject_t &ghobj, string key, int fd,
2200                 ObjectStore::Sequencer &osr)
2201 {
2202   ObjectStore::Transaction tran;
2203   ObjectStore::Transaction *t = &tran;
2204   map<string, bufferlist> attrset;
2205   bufferlist valbl;
2206
2207   if (debug)
2208     cerr << "Set_omap " << ghobj << std::endl;
2209
2210   int ret = get_fd_data(fd, valbl);
2211   if (ret < 0)
2212     return ret;
2213
2214   attrset.insert(pair<string, bufferlist>(key, valbl));
2215
2216   if (dry_run)
2217     return 0;
2218
2219   t->touch(coll, ghobj);
2220
2221   t->omap_setkeys(coll, ghobj, attrset);
2222
2223   store->apply_transaction(&osr, std::move(*t));
2224   return 0;
2225 }
2226
2227 int do_rm_omap(ObjectStore *store, coll_t coll,
2228                ghobject_t &ghobj, string key,
2229                ObjectStore::Sequencer &osr)
2230 {
2231   ObjectStore::Transaction tran;
2232   ObjectStore::Transaction *t = &tran;
2233   set<string> keys;
2234
2235   keys.insert(key);
2236
2237   if (debug)
2238     cerr << "Rm_omap " << ghobj << std::endl;
2239
2240   if (dry_run)
2241     return 0;
2242
2243   t->omap_rmkeys(coll, ghobj, keys);
2244
2245   store->apply_transaction(&osr, std::move(*t));
2246   return 0;
2247 }
2248
2249 int do_get_omaphdr(ObjectStore *store, coll_t coll, ghobject_t &ghobj)
2250 {
2251   bufferlist hdrbl;
2252
2253   int r = store->omap_get_header(coll, ghobj, &hdrbl, true);
2254   if (r < 0) {
2255     cerr << "omap_get_header: " << cpp_strerror(r) << std::endl;
2256     return r;
2257   }
2258
2259   string header(hdrbl.c_str(), hdrbl.length());
2260   if (outistty) {
2261     header = cleanbin(header);
2262     header.push_back('\n');
2263   }
2264   cout << header;
2265
2266   return 0;
2267 }
2268
2269 int do_set_omaphdr(ObjectStore *store, coll_t coll,
2270                    ghobject_t &ghobj, int fd,
2271                    ObjectStore::Sequencer &osr)
2272 {
2273   ObjectStore::Transaction tran;
2274   ObjectStore::Transaction *t = &tran;
2275   bufferlist hdrbl;
2276
2277   if (debug)
2278     cerr << "Omap_setheader " << ghobj << std::endl;
2279
2280   int ret = get_fd_data(fd, hdrbl);
2281   if (ret)
2282     return ret;
2283
2284   if (dry_run)
2285     return 0;
2286
2287   t->touch(coll, ghobj);
2288
2289   t->omap_setheader(coll, ghobj, hdrbl);
2290
2291   store->apply_transaction(&osr, std::move(*t));
2292   return 0;
2293 }
2294
2295 struct do_fix_lost : public action_on_object_t {
2296   ObjectStore::Sequencer *osr;
2297
2298   explicit do_fix_lost(ObjectStore::Sequencer *_osr) : osr(_osr) {}
2299
2300   int call(ObjectStore *store, coll_t coll,
2301                    ghobject_t &ghobj, object_info_t &oi) override {
2302     if (oi.is_lost()) {
2303       cout << coll << "/" << ghobj << " is lost";
2304       if (!dry_run)
2305         cout << ", fixing";
2306       cout << std::endl;
2307       if (dry_run)
2308         return 0;
2309       oi.clear_flag(object_info_t::FLAG_LOST);
2310       bufferlist bl;
2311       ::encode(oi, bl, -1);  /* fixme: using full features */
2312       ObjectStore::Transaction t;
2313       t.setattr(coll, ghobj, OI_ATTR, bl);
2314       int r = store->apply_transaction(osr, std::move(t));
2315       if (r < 0) {
2316         cerr << "Error getting fixing attr on : " << make_pair(coll, ghobj)
2317              << ", "
2318              << cpp_strerror(r) << std::endl;
2319         return r;
2320       }
2321     }
2322     return 0;
2323   }
2324 };
2325
2326 int get_snapset(ObjectStore *store, coll_t coll, ghobject_t &ghobj, SnapSet &ss, bool silent = false)
2327 {
2328   bufferlist attr;
2329   int r = store->getattr(coll, ghobj, SS_ATTR, attr);
2330   if (r < 0) {
2331     if (!silent)
2332       cerr << "Error getting snapset on : " << make_pair(coll, ghobj) << ", "
2333            << cpp_strerror(r) << std::endl;
2334     return r;
2335   }
2336   bufferlist::iterator bp = attr.begin();
2337   try {
2338     ::decode(ss, bp);
2339   } catch (...) {
2340     r = -EINVAL;
2341     cerr << "Error decoding snapset on : " << make_pair(coll, ghobj) << ", "
2342          << cpp_strerror(r) << std::endl;
2343     return r;
2344   }
2345   return 0;
2346 }
2347
2348 int print_obj_info(ObjectStore *store, coll_t coll, ghobject_t &ghobj, Formatter* formatter)
2349 {
2350   int r = 0;
2351   formatter->open_object_section("obj");
2352   formatter->open_object_section("id");
2353   ghobj.dump(formatter);
2354   formatter->close_section();
2355
2356   bufferlist attr;
2357   int gr = store->getattr(coll, ghobj, OI_ATTR, attr);
2358   if (gr < 0) {
2359     r = gr;
2360     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2361        << cpp_strerror(r) << std::endl;
2362   } else {
2363     object_info_t oi;
2364     bufferlist::iterator bp = attr.begin();
2365     try {
2366       ::decode(oi, bp);
2367       formatter->open_object_section("info");
2368       oi.dump(formatter);
2369       formatter->close_section();
2370     } catch (...) {
2371       r = -EINVAL;
2372       cerr << "Error decoding attr on : " << make_pair(coll, ghobj) << ", "
2373            << cpp_strerror(r) << std::endl;
2374     }
2375   }
2376   struct stat st;
2377   int sr =  store->stat(coll, ghobj, &st, true);
2378   if (sr < 0) {
2379     r = sr;
2380     cerr << "Error stat on : " << make_pair(coll, ghobj) << ", "
2381          << cpp_strerror(r) << std::endl;
2382   } else {
2383     formatter->open_object_section("stat");
2384     formatter->dump_int("size", st.st_size);
2385     formatter->dump_int("blksize", st.st_blksize);
2386     formatter->dump_int("blocks", st.st_blocks);
2387     formatter->dump_int("nlink", st.st_nlink);
2388     formatter->close_section();
2389   }
2390
2391   if (ghobj.hobj.has_snapset()) {
2392     SnapSet ss;
2393     int snr = get_snapset(store, coll, ghobj, ss);
2394     if (snr < 0) {
2395       r = snr;
2396     } else {
2397       formatter->open_object_section("SnapSet");
2398       ss.dump(formatter);
2399       formatter->close_section();
2400     }
2401   }
2402   formatter->close_section();
2403   formatter->flush(cout);
2404   cout << std::endl;
2405   return r;
2406 }
2407
2408 int set_size(ObjectStore *store, coll_t coll, ghobject_t &ghobj, uint64_t setsize, Formatter* formatter,
2409              ObjectStore::Sequencer &osr, bool corrupt)
2410 {
2411   if (ghobj.hobj.is_snapdir()) {
2412     cerr << "Can't set the size of a snapdir" << std::endl;
2413     return -EINVAL;
2414   }
2415   bufferlist attr;
2416   int r = store->getattr(coll, ghobj, OI_ATTR, attr);
2417   if (r < 0) {
2418     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2419        << cpp_strerror(r) << std::endl;
2420     return r;
2421   }
2422   object_info_t oi;
2423   bufferlist::iterator bp = attr.begin();
2424   try {
2425     ::decode(oi, bp);
2426   } catch (...) {
2427     r = -EINVAL;
2428     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2429          << cpp_strerror(r) << std::endl;
2430     return r;
2431   }
2432   struct stat st;
2433   r =  store->stat(coll, ghobj, &st, true);
2434   if (r < 0) {
2435     cerr << "Error stat on : " << make_pair(coll, ghobj) << ", "
2436          << cpp_strerror(r) << std::endl;
2437   }
2438   ghobject_t head(ghobj);
2439   SnapSet ss;
2440   bool found_head = true;
2441   map<snapid_t, uint64_t>::iterator csi;
2442   bool is_snap = ghobj.hobj.is_snap();
2443   if (is_snap) {
2444     head.hobj = head.hobj.get_head();
2445     r = get_snapset(store, coll, head, ss, true);
2446     if (r < 0 && r != -ENOENT) {
2447       // Requested get_snapset() silent, so if not -ENOENT show error
2448       cerr << "Error getting snapset on : " << make_pair(coll, head) << ", "
2449            << cpp_strerror(r) << std::endl;
2450       return r;
2451     }
2452     if (r == -ENOENT) {
2453       head.hobj = head.hobj.get_snapdir();
2454       r = get_snapset(store, coll, head, ss);
2455       if (r < 0)
2456         return r;
2457       found_head = false;
2458     } else {
2459       found_head = true;
2460     }
2461     csi = ss.clone_size.find(ghobj.hobj.snap);
2462     if (csi == ss.clone_size.end()) {
2463       cerr << "SnapSet is missing clone_size for snap " << ghobj.hobj.snap << std::endl;
2464       return -EINVAL;
2465     }
2466   }
2467   if ((uint64_t)st.st_size == setsize && oi.size == setsize
2468        && (!is_snap || csi->second == setsize)) {
2469     cout << "Size of object is already " << setsize << std::endl;
2470     return 0;
2471   }
2472   cout << "Setting size to " << setsize << ", stat size " << st.st_size
2473        << ", obj info size " << oi.size;
2474   if (is_snap) {
2475     cout << ", " << (found_head ? "head" : "snapdir")
2476          << " clone_size " << csi->second;
2477     csi->second = setsize;
2478   }
2479   cout << std::endl;
2480   if (!dry_run) {
2481     attr.clear();
2482     oi.size = setsize;
2483     ObjectStore::Transaction t;
2484     // Only modify object info if we want to corrupt it
2485     if (!corrupt && (uint64_t)st.st_size != setsize) {
2486       t.truncate(coll, ghobj, setsize);
2487       // Changing objectstore size will invalidate data_digest, so clear it.
2488       oi.clear_data_digest();
2489     }
2490     ::encode(oi, attr, -1);  /* fixme: using full features */
2491     t.setattr(coll, ghobj, OI_ATTR, attr);
2492     if (is_snap) {
2493       bufferlist snapattr;
2494       snapattr.clear();
2495       ::encode(ss, snapattr);
2496       t.setattr(coll, head, SS_ATTR, snapattr);
2497     }
2498     r = store->apply_transaction(&osr, std::move(t));
2499     if (r < 0) {
2500       cerr << "Error writing object info: " << make_pair(coll, ghobj) << ", "
2501          << cpp_strerror(r) << std::endl;
2502       return r;
2503     }
2504   }
2505   return 0;
2506 }
2507
2508 int clear_snapset(ObjectStore *store, coll_t coll, ghobject_t &ghobj,
2509                   string arg, ObjectStore::Sequencer &osr)
2510 {
2511   SnapSet ss;
2512   int ret = get_snapset(store, coll, ghobj, ss);
2513   if (ret < 0)
2514     return ret;
2515
2516   // Use "head" to set head_exists incorrectly
2517   if (arg == "corrupt" || arg == "head")
2518     ss.head_exists = !ghobj.hobj.is_head();
2519   else if (ss.head_exists != ghobj.hobj.is_head()) {
2520     cerr << "Correcting head_exists, set to "
2521          << (ghobj.hobj.is_head() ? "true" : "false") << std::endl;
2522     ss.head_exists = ghobj.hobj.is_head();
2523   }
2524   // Use "corrupt" to clear entire SnapSet
2525   // Use "seq" to just corrupt SnapSet.seq
2526   if (arg == "corrupt" || arg == "seq")
2527     ss.seq = 0;
2528   // Use "snaps" to just clear SnapSet.snaps
2529   if (arg == "corrupt" || arg == "snaps")
2530     ss.snaps.clear();
2531   // By default just clear clone, clone_overlap and clone_size
2532   if (arg == "corrupt")
2533     arg = "";
2534   if (arg == "" || arg == "clones")
2535     ss.clones.clear();
2536   if (arg == "" || arg == "clone_overlap")
2537     ss.clone_overlap.clear();
2538   if (arg == "" || arg == "clone_size")
2539     ss.clone_size.clear();
2540   // Break all clone sizes by adding 1
2541   if (arg == "size") {
2542     for (map<snapid_t, uint64_t>::iterator i = ss.clone_size.begin();
2543          i != ss.clone_size.end(); ++i)
2544       ++(i->second);
2545   }
2546
2547   if (!dry_run) {
2548     bufferlist bl;
2549     ::encode(ss, bl);
2550     ObjectStore::Transaction t;
2551     t.setattr(coll, ghobj, SS_ATTR, bl);
2552     int r = store->apply_transaction(&osr, std::move(t));
2553     if (r < 0) {
2554       cerr << "Error setting snapset on : " << make_pair(coll, ghobj) << ", "
2555            << cpp_strerror(r) << std::endl;
2556       return r;
2557     }
2558   }
2559   return 0;
2560 }
2561
2562 vector<snapid_t>::iterator find(vector<snapid_t> &v, snapid_t clid)
2563 {
2564   return std::find(v.begin(), v.end(), clid);
2565 }
2566
2567 map<snapid_t, interval_set<uint64_t> >::iterator
2568 find(map<snapid_t, interval_set<uint64_t> > &m, snapid_t clid)
2569 {
2570   return m.find(clid);
2571 }
2572
2573 map<snapid_t, uint64_t>::iterator find(map<snapid_t, uint64_t> &m,
2574                                        snapid_t clid)
2575 {
2576   return m.find(clid);
2577 }
2578
2579 template<class T>
2580 int remove_from(T &mv, string name, snapid_t cloneid, bool force)
2581 {
2582   typename T::iterator i = find(mv, cloneid);
2583   if (i != mv.end()) {
2584     mv.erase(i);
2585   } else {
2586     cerr << "Clone " << cloneid << " doesn't exist in " << name;
2587     if (force) {
2588       cerr << " (ignored)" << std::endl;
2589       return 0;
2590     }
2591     cerr << std::endl;
2592     return -EINVAL;
2593   }
2594   return 0;
2595 }
2596
2597 int remove_clone(ObjectStore *store, coll_t coll, ghobject_t &ghobj, snapid_t cloneid, bool force,
2598                      ObjectStore::Sequencer &osr)
2599 {
2600   // XXX: Don't allow this if in a cache tier or former cache tier
2601   // bool allow_incomplete_clones() const {
2602   //    return cache_mode != CACHEMODE_NONE || has_flag(FLAG_INCOMPLETE_CLONES);
2603
2604   SnapSet snapset;
2605   int ret = get_snapset(store, coll, ghobj, snapset);
2606   if (ret < 0)
2607     return ret;
2608
2609   // Derived from trim_object()
2610   // ...from snapset
2611   vector<snapid_t>::iterator p;
2612   for (p = snapset.clones.begin(); p != snapset.clones.end(); ++p)
2613     if (*p == cloneid)
2614       break;
2615   if (p == snapset.clones.end()) {
2616     cerr << "Clone " << cloneid << " not present";
2617     return -ENOENT;
2618   }
2619   if (p != snapset.clones.begin()) {
2620     // not the oldest... merge overlap into next older clone
2621     vector<snapid_t>::iterator n = p - 1;
2622     hobject_t prev_coid = ghobj.hobj;
2623     prev_coid.snap = *n;
2624     //bool adjust_prev_bytes = is_present_clone(prev_coid);
2625
2626     //if (adjust_prev_bytes)
2627     //  ctx->delta_stats.num_bytes -= snapset.get_clone_bytes(*n);
2628
2629     snapset.clone_overlap[*n].intersection_of(
2630         snapset.clone_overlap[*p]);
2631
2632     //if (adjust_prev_bytes)
2633     //  ctx->delta_stats.num_bytes += snapset.get_clone_bytes(*n);
2634   }
2635
2636   ret = remove_from(snapset.clones, "clones", cloneid, force);
2637   if (ret) return ret;
2638   ret = remove_from(snapset.clone_overlap, "clone_overlap", cloneid, force);
2639   if (ret) return ret;
2640   ret = remove_from(snapset.clone_size, "clone_size", cloneid, force);
2641   if (ret) return ret;
2642
2643   if (dry_run)
2644     return 0;
2645
2646   bufferlist bl;
2647   ::encode(snapset, bl);
2648   ObjectStore::Transaction t;
2649   t.setattr(coll, ghobj, SS_ATTR, bl);
2650   int r = store->apply_transaction(&osr, std::move(t));
2651   if (r < 0) {
2652     cerr << "Error setting snapset on : " << make_pair(coll, ghobj) << ", "
2653          << cpp_strerror(r) << std::endl;
2654     return r;
2655   }
2656   cout << "Removal of clone " << cloneid << " complete" << std::endl;
2657   cout << "Use pg repair after OSD restarted to correct stat information" << std::endl;
2658   return 0;
2659 }
2660
2661 int dup(string srcpath, ObjectStore *src, string dstpath, ObjectStore *dst)
2662 {
2663   cout << "dup from " << src->get_type() << ": " << srcpath << "\n"
2664        << "      to " << dst->get_type() << ": " << dstpath
2665        << std::endl;
2666   ObjectStore::Sequencer osr("dup");
2667   int num, i;
2668   vector<coll_t> collections;
2669   int r;
2670
2671   r = src->mount();
2672   if (r < 0) {
2673     cerr << "failed to mount src: " << cpp_strerror(r) << std::endl;
2674     return r;
2675   }
2676   r = dst->mount();
2677   if (r < 0) {
2678     cerr << "failed to mount dst: " << cpp_strerror(r) << std::endl;
2679     goto out_src;
2680   }
2681
2682   if (src->get_fsid() != dst->get_fsid()) {
2683     cerr << "src fsid " << src->get_fsid() << " != dest " << dst->get_fsid()
2684          << std::endl;
2685     goto out;
2686   }
2687   cout << "fsid " << src->get_fsid() << std::endl;
2688
2689   // make sure dst is empty
2690   r = dst->list_collections(collections);
2691   if (r < 0) {
2692     cerr << "error listing collections on dst: " << cpp_strerror(r) << std::endl;
2693     goto out;
2694   }
2695   if (!collections.empty()) {
2696     cerr << "destination store is not empty" << std::endl;
2697     goto out;
2698   }
2699
2700   r = src->list_collections(collections);
2701   if (r < 0) {
2702     cerr << "error listing collections on src: " << cpp_strerror(r) << std::endl;
2703     goto out;
2704   }
2705
2706   num = collections.size();
2707   cout << num << " collections" << std::endl;
2708   i = 1;
2709   for (auto cid : collections) {
2710     cout << i++ << "/" << num << " " << cid << std::endl;
2711     {
2712       ObjectStore::Transaction t;
2713       int bits = src->collection_bits(cid);
2714       if (bits < 0) {
2715         if (src->get_type() == "filestore" && cid.is_meta()) {
2716           bits = 0;
2717         } else {
2718           cerr << "cannot get bit count for collection " << cid << ": "
2719                << cpp_strerror(bits) << std::endl;
2720           goto out;
2721         }
2722       }
2723       t.create_collection(cid, bits);
2724       dst->apply_transaction(&osr, std::move(t));
2725     }
2726
2727     ghobject_t pos;
2728     uint64_t n = 0;
2729     uint64_t bytes = 0, keys = 0;
2730     while (true) {
2731       vector<ghobject_t> ls;
2732       r = src->collection_list(cid, pos, ghobject_t::get_max(), 1000, &ls, &pos);
2733       if (r < 0) {
2734         cerr << "collection_list on " << cid << " from " << pos << " got: "
2735              << cpp_strerror(r) << std::endl;
2736         goto out;
2737       }
2738       if (ls.empty()) {
2739         break;
2740       }
2741
2742       for (auto& oid : ls) {
2743         //cout << "  " << cid << " " << oid << std::endl;
2744         if (n % 100 == 0) {
2745           cout << "  " << std::setw(16) << n << " objects, "
2746                << std::setw(16) << bytes << " bytes, "
2747                << std::setw(16) << keys << " keys"
2748                << std::setw(1) << "\r" << std::flush;
2749         }
2750         n++;
2751
2752         ObjectStore::Transaction t;
2753         t.touch(cid, oid);
2754
2755         map<string,bufferptr> attrs;
2756         src->getattrs(cid, oid, attrs);
2757         if (!attrs.empty()) {
2758           t.setattrs(cid, oid, attrs);
2759         }
2760
2761         bufferlist bl;
2762         src->read(cid, oid, 0, 0, bl);
2763         if (bl.length()) {
2764           t.write(cid, oid, 0, bl.length(), bl);
2765           bytes += bl.length();
2766         }
2767
2768         bufferlist header;
2769         map<string,bufferlist> omap;
2770         src->omap_get(cid, oid, &header, &omap);
2771         if (header.length()) {
2772           t.omap_setheader(cid, oid, header);
2773           ++keys;
2774         }
2775         if (!omap.empty()) {
2776           keys += omap.size();
2777           t.omap_setkeys(cid, oid, omap);
2778         }
2779
2780         dst->apply_transaction(&osr, std::move(t));
2781       }
2782     }
2783     cout << "  " << std::setw(16) << n << " objects, "
2784          << std::setw(16) << bytes << " bytes, "
2785          << std::setw(16) << keys << " keys"
2786          << std::setw(1) << std::endl;
2787   }
2788
2789   // keyring
2790   cout << "keyring" << std::endl;
2791   {
2792     bufferlist bl;
2793     string s = srcpath + "/keyring";
2794     string err;
2795     r = bl.read_file(s.c_str(), &err);
2796     if (r < 0) {
2797       cerr << "failed to copy " << s << ": " << err << std::endl;
2798     } else {
2799       string d = dstpath + "/keyring";
2800       bl.write_file(d.c_str(), 0600);
2801     }
2802   }
2803
2804   // osd metadata
2805   cout << "duping osd metadata" << std::endl;
2806   {
2807     for (auto k : {"magic", "whoami", "ceph_fsid", "fsid"}) {
2808       string val;
2809       src->read_meta(k, &val);
2810       dst->write_meta(k, val);
2811     }
2812   }
2813
2814   dst->write_meta("ready", "ready");
2815
2816   cout << "done." << std::endl;
2817   r = 0;
2818  out:
2819   dst->umount();
2820  out_src:
2821   src->umount();
2822   return r;
2823 }
2824
2825 void usage(po::options_description &desc)
2826 {
2827     cerr << std::endl;
2828     cerr << desc << std::endl;
2829     cerr << std::endl;
2830     cerr << "Positional syntax:" << std::endl;
2831     cerr << std::endl;
2832     cerr << "ceph-objectstore-tool ... <object> (get|set)-bytes [file]" << std::endl;
2833     cerr << "ceph-objectstore-tool ... <object> set-(attr|omap) <key> [file]" << std::endl;
2834     cerr << "ceph-objectstore-tool ... <object> (get|rm)-(attr|omap) <key>" << std::endl;
2835     cerr << "ceph-objectstore-tool ... <object> get-omaphdr" << std::endl;
2836     cerr << "ceph-objectstore-tool ... <object> set-omaphdr [file]" << std::endl;
2837     cerr << "ceph-objectstore-tool ... <object> list-attrs" << std::endl;
2838     cerr << "ceph-objectstore-tool ... <object> list-omap" << std::endl;
2839     cerr << "ceph-objectstore-tool ... <object> remove|removeall" << std::endl;
2840     cerr << "ceph-objectstore-tool ... <object> dump" << std::endl;
2841     cerr << "ceph-objectstore-tool ... <object> set-size" << std::endl;
2842     cerr << "ceph-objectstore-tool ... <object> remove-clone-metadata <cloneid>" << std::endl;
2843     cerr << std::endl;
2844     cerr << "<object> can be a JSON object description as displayed" << std::endl;
2845     cerr << "by --op list." << std::endl;
2846     cerr << "<object> can be an object name which will be looked up in all" << std::endl;
2847     cerr << "the OSD's PGs." << std::endl;
2848     cerr << "<object> can be the empty string ('') which with a provided pgid " << std::endl;
2849     cerr << "specifies the pgmeta object" << std::endl;
2850     cerr << std::endl;
2851     cerr << "The optional [file] argument will read stdin or write stdout" << std::endl;
2852     cerr << "if not specified or if '-' specified." << std::endl;
2853 }
2854
2855 bool ends_with(const string& check, const string& ending)
2856 {
2857     return check.size() >= ending.size() && check.rfind(ending) == (check.size() - ending.size());
2858 }
2859
2860 // Based on FileStore::dump_journal(), set-up enough to only dump
2861 int mydump_journal(Formatter *f, string journalpath, bool m_journal_dio)
2862 {
2863   int r;
2864
2865   if (!journalpath.length())
2866     return -EINVAL;
2867
2868   FileJournal *journal = new FileJournal(g_ceph_context, uuid_d(), NULL, NULL,
2869                                          journalpath.c_str(), m_journal_dio);
2870   r = journal->_fdump(*f, false);
2871   delete journal;
2872   return r;
2873 }
2874
2875 int apply_layout_settings(ObjectStore *os, const OSDSuperblock &superblock,
2876                           const string &pool_name, const spg_t &pgid, bool dry_run)
2877 {
2878   int r = 0;
2879
2880   FileStore *fs = dynamic_cast<FileStore*>(os);
2881   if (!fs) {
2882     cerr << "Nothing to do for non-filestore backend" << std::endl;
2883     return 0; // making this return success makes testing easier
2884   }
2885
2886   OSDMap curmap;
2887   bufferlist bl;
2888   r = get_osdmap(os, superblock.current_epoch, curmap, bl);
2889   if (r) {
2890     cerr << "Can't find local OSDMap: " << cpp_strerror(r) << std::endl;
2891     return r;
2892   }
2893
2894   int64_t poolid = -1;
2895   if (pool_name.length()) {
2896     poolid = curmap.lookup_pg_pool_name(pool_name);
2897     if (poolid < 0) {
2898       cerr << "Couldn't find pool " << pool_name << ": " << cpp_strerror(poolid)
2899            << std::endl;
2900       return poolid;
2901     }
2902   }
2903
2904   vector<coll_t> collections, filtered_colls;
2905   r = os->list_collections(collections);
2906   if (r < 0) {
2907     cerr << "Error listing collections: " << cpp_strerror(r) << std::endl;
2908     return r;
2909   }
2910
2911   for (auto const &coll : collections) {
2912     spg_t coll_pgid;
2913     if (coll.is_pg(&coll_pgid) &&
2914         ((poolid >= 0 && coll_pgid.pool() == (uint64_t)poolid) ||
2915          coll_pgid == pgid)) {
2916       filtered_colls.push_back(coll);
2917     }
2918   }
2919
2920   size_t done = 0, total = filtered_colls.size();
2921   for (auto const &coll : filtered_colls) {
2922     if (dry_run) {
2923       cerr << "Would apply layout settings to " << coll << std::endl;
2924     } else {
2925       cerr << "Finished " << done << "/" << total << " collections" << "\r";
2926       r = fs->apply_layout_settings(coll);
2927       if (r < 0) {
2928         cerr << "Error applying layout settings to " << coll << std::endl;
2929         return r;
2930       }
2931     }
2932     ++done;
2933   }
2934
2935   cerr << "Finished " << total << "/" << total << " collections" << "\r" << std::endl;
2936   return r;
2937 }
2938
2939 int main(int argc, char **argv)
2940 {
2941   string dpath, jpath, pgidstr, op, file, mountpoint, mon_store_path, object;
2942   string target_data_path, fsid;
2943   string objcmd, arg1, arg2, type, format, argnspace, pool;
2944   boost::optional<std::string> nspace;
2945   spg_t pgid;
2946   unsigned epoch = 0;
2947   ghobject_t ghobj;
2948   bool human_readable;
2949   bool force;
2950   Formatter *formatter;
2951   bool head;
2952
2953   po::options_description desc("Allowed options");
2954   desc.add_options()
2955     ("help", "produce help message")
2956     ("type", po::value<string>(&type),
2957      "Arg is one of [bluestore, filestore (default), memstore]")
2958     ("data-path", po::value<string>(&dpath),
2959      "path to object store, mandatory")
2960     ("journal-path", po::value<string>(&jpath),
2961      "path to journal, use if tool can't find it")
2962     ("pgid", po::value<string>(&pgidstr),
2963      "PG id, mandatory for info, log, remove, export, export-remove, rm-past-intervals, mark-complete, trim-pg-log, and mandatory for apply-layout-settings if --pool is not specified")
2964     ("pool", po::value<string>(&pool),
2965      "Pool name, mandatory for apply-layout-settings if --pgid is not specified")
2966     ("op", po::value<string>(&op),
2967      "Arg is one of [info, log, remove, mkfs, fsck, repair, fuse, dup, export, export-remove, import, list, fix-lost, list-pgs, rm-past-intervals, dump-journal, dump-super, meta-list, "
2968      "get-osdmap, set-osdmap, get-inc-osdmap, set-inc-osdmap, mark-complete, apply-layout-settings, update-mon-db, dump-import, trim-pg-log]")
2969     ("epoch", po::value<unsigned>(&epoch),
2970      "epoch# for get-osdmap and get-inc-osdmap, the current epoch in use if not specified")
2971     ("file", po::value<string>(&file),
2972      "path of file to export, export-remove, import, get-osdmap, set-osdmap, get-inc-osdmap or set-inc-osdmap")
2973     ("mon-store-path", po::value<string>(&mon_store_path),
2974      "path of monstore to update-mon-db")
2975     ("fsid", po::value<string>(&fsid),
2976      "fsid for new store created by mkfs")
2977     ("target-data-path", po::value<string>(&target_data_path),
2978      "path of target object store (for --op dup)")
2979     ("mountpoint", po::value<string>(&mountpoint),
2980      "fuse mountpoint")
2981     ("format", po::value<string>(&format)->default_value("json-pretty"),
2982      "Output format which may be json, json-pretty, xml, xml-pretty")
2983     ("debug", "Enable diagnostic output to stderr")
2984     ("force", "Ignore some types of errors and proceed with operation - USE WITH CAUTION: CORRUPTION POSSIBLE NOW OR IN THE FUTURE")
2985     ("skip-journal-replay", "Disable journal replay")
2986     ("skip-mount-omap", "Disable mounting of omap")
2987     ("head", "Find head/snapdir when searching for objects by name")
2988     ("dry-run", "Don't modify the objectstore")
2989     ("namespace", po::value<string>(&argnspace), "Specify namespace when searching for objects")
2990     ;
2991
2992   po::options_description positional("Positional options");
2993   positional.add_options()
2994     ("object", po::value<string>(&object), "'' for pgmeta_oid, object name or ghobject in json")
2995     ("objcmd", po::value<string>(&objcmd), "command [(get|set)-bytes, (get|set|rm)-(attr|omap), (get|set)-omaphdr, list-attrs, list-omap, remove]")
2996     ("arg1", po::value<string>(&arg1), "arg1 based on cmd")
2997     ("arg2", po::value<string>(&arg2), "arg2 based on cmd")
2998     ;
2999
3000   po::options_description all;
3001   all.add(desc).add(positional);
3002
3003   po::positional_options_description pd;
3004   pd.add("object", 1).add("objcmd", 1).add("arg1", 1).add("arg2", 1);
3005
3006   vector<string> ceph_option_strings;
3007   po::variables_map vm;
3008   try {
3009     po::parsed_options parsed =
3010       po::command_line_parser(argc, argv).options(all).allow_unregistered().positional(pd).run();
3011     po::store( parsed, vm);
3012     po::notify(vm);
3013     ceph_option_strings = po::collect_unrecognized(parsed.options,
3014                                                    po::include_positional);
3015   } catch(po::error &e) {
3016     std::cerr << e.what() << std::endl;
3017     return 1;
3018   }
3019
3020   if (vm.count("help")) {
3021     usage(desc);
3022     return 1;
3023   }
3024
3025   debug = (vm.count("debug") > 0);
3026
3027   force = (vm.count("force") > 0);
3028
3029   if (vm.count("namespace"))
3030     nspace = argnspace;
3031
3032   dry_run = (vm.count("dry-run") > 0);
3033
3034   osflagbits_t flags = 0;
3035   if (dry_run || vm.count("skip-journal-replay"))
3036     flags |= SKIP_JOURNAL_REPLAY;
3037   if (vm.count("skip-mount-omap"))
3038     flags |= SKIP_MOUNT_OMAP;
3039   if (op == "update-mon-db")
3040     flags |= SKIP_JOURNAL_REPLAY;
3041
3042   head = (vm.count("head") > 0);
3043
3044   vector<const char *> ceph_options;
3045   env_to_vec(ceph_options);
3046   ceph_options.reserve(ceph_options.size() + ceph_option_strings.size());
3047   for (vector<string>::iterator i = ceph_option_strings.begin();
3048        i != ceph_option_strings.end();
3049        ++i) {
3050     ceph_options.push_back(i->c_str());
3051   }
3052
3053   char fn[PATH_MAX];
3054   snprintf(fn, sizeof(fn), "%s/type", dpath.c_str());
3055   int fd = ::open(fn, O_RDONLY);
3056   if (fd >= 0) {
3057     bufferlist bl;
3058     bl.read_fd(fd, 64);
3059     if (bl.length()) {
3060       string dp_type = string(bl.c_str(), bl.length() - 1);  // drop \n
3061       if (vm.count("type") && dp_type != "" && type != dp_type)
3062         cerr << "WARNING: Ignoring type \"" << type << "\" - found data-path type \""
3063              << dp_type << "\"" << std::endl;
3064       type = dp_type;
3065       //cout << "object store type is " << type << std::endl;
3066     }
3067     ::close(fd);
3068   }
3069   if (!vm.count("type") && type == "") {
3070     type = "filestore";
3071   }
3072   if (!vm.count("data-path") &&
3073      op != "dump-import" &&
3074      !(op == "dump-journal" && type == "filestore")) {
3075     cerr << "Must provide --data-path" << std::endl;
3076     usage(desc);
3077     return 1;
3078   }
3079   if (type == "filestore" && !vm.count("journal-path")) {
3080     jpath = dpath + "/journal";
3081   }
3082   if (!vm.count("op") && !vm.count("object")) {
3083     cerr << "Must provide --op or object command..." << std::endl;
3084     usage(desc);
3085     return 1;
3086   }
3087   if (op != "list" &&
3088       vm.count("op") && vm.count("object")) {
3089     cerr << "Can't specify both --op and object command syntax" << std::endl;
3090     usage(desc);
3091     return 1;
3092   }
3093   if (op == "apply-layout-settings" && !(vm.count("pool") ^ vm.count("pgid"))) {
3094     cerr << "apply-layout-settings requires either --pool or --pgid"
3095          << std::endl;
3096     usage(desc);
3097     return 1;
3098   }
3099   if (op != "list" && vm.count("object") && !vm.count("objcmd")) {
3100     cerr << "Invalid syntax, missing command" << std::endl;
3101     usage(desc);
3102     return 1;
3103   }
3104   if (op == "fuse" && mountpoint.length() == 0) {
3105     cerr << "Missing fuse mountpoint" << std::endl;
3106     usage(desc);
3107     return 1;
3108   }
3109   outistty = isatty(STDOUT_FILENO);
3110
3111   file_fd = fd_none;
3112   if ((op == "export" || op == "export-remove" || op == "get-osdmap" || op == "get-inc-osdmap") && !dry_run) {
3113     if (!vm.count("file") || file == "-") {
3114       if (outistty) {
3115         cerr << "stdout is a tty and no --file filename specified" << std::endl;
3116         return 1;
3117       }
3118       file_fd = STDOUT_FILENO;
3119     } else {
3120       file_fd = open(file.c_str(), O_WRONLY|O_CREAT|O_TRUNC, 0666);
3121     }
3122   } else if (op == "import" || op == "dump-import" || op == "set-osdmap" || op == "set-inc-osdmap") {
3123     if (!vm.count("file") || file == "-") {
3124       if (isatty(STDIN_FILENO)) {
3125         cerr << "stdin is a tty and no --file filename specified" << std::endl;
3126         return 1;
3127       }
3128       file_fd = STDIN_FILENO;
3129     } else {
3130       file_fd = open(file.c_str(), O_RDONLY);
3131     }
3132   }
3133
3134   ObjectStoreTool tool = ObjectStoreTool(file_fd, dry_run);
3135
3136   if (vm.count("file") && file_fd == fd_none && !dry_run) {
3137     cerr << "--file option only applies to import, dump-import, export, export-remove, "
3138          << "get-osdmap, set-osdmap, get-inc-osdmap or set-inc-osdmap" << std::endl;
3139     return 1;
3140   }
3141
3142   if (file_fd != fd_none && file_fd < 0) {
3143     string err = string("file: ") + file;
3144     perror(err.c_str());
3145     return 1;
3146   }
3147
3148   auto cct = global_init(
3149     NULL, ceph_options, CEPH_ENTITY_TYPE_OSD,
3150     CODE_ENVIRONMENT_UTILITY_NODOUT, 0);
3151     //CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
3152   common_init_finish(g_ceph_context);
3153   g_conf = g_ceph_context->_conf;
3154   if (debug) {
3155     g_conf->set_val_or_die("log_to_stderr", "true");
3156     g_conf->set_val_or_die("err_to_stderr", "true");
3157   }
3158   g_conf->apply_changes(NULL);
3159
3160   // Special list handling.  Treating pretty_format as human readable,
3161   // with one object per line and not an enclosing array.
3162   human_readable = ends_with(format, "-pretty");
3163   if ((op == "list" || op == "meta-list") && human_readable) {
3164     // Remove -pretty from end of format which we know is there
3165     format = format.substr(0, format.size() - strlen("-pretty"));
3166   }
3167
3168   formatter = Formatter::create(format);
3169   if (formatter == NULL) {
3170     cerr << "unrecognized format: " << format << std::endl;
3171     return 1;
3172   }
3173
3174   // Special handling for filestore journal, so we can dump it without mounting
3175   if (op == "dump-journal" && type == "filestore") {
3176     int ret = mydump_journal(formatter, jpath, g_conf->journal_dio);
3177     if (ret < 0) {
3178       cerr << "journal-path: " << jpath << ": "
3179            << cpp_strerror(ret) << std::endl;
3180       return 1;
3181     }
3182     formatter->flush(cout);
3183     return 0;
3184   }
3185
3186   if (op == "dump-import") {
3187     int ret = tool.dump_import(formatter);
3188     if (ret < 0) {
3189       cerr << "dump-import: "
3190            << cpp_strerror(ret) << std::endl;
3191       return 1;
3192     }
3193     return 0;
3194   }
3195
3196   //Verify that data-path really exists
3197   struct stat st;
3198   if (::stat(dpath.c_str(), &st) == -1) {
3199     string err = string("data-path: ") + dpath;
3200     perror(err.c_str());
3201     return 1;
3202   }
3203
3204   if (pgidstr.length() && !pgid.parse(pgidstr.c_str())) {
3205     cerr << "Invalid pgid '" << pgidstr << "' specified" << std::endl;
3206     return 1;
3207   }
3208
3209   //Verify that the journal-path really exists
3210   if (type == "filestore") {
3211     if (::stat(jpath.c_str(), &st) == -1) {
3212       string err = string("journal-path: ") + jpath;
3213       perror(err.c_str());
3214       return 1;
3215     }
3216     if (S_ISDIR(st.st_mode)) {
3217       cerr << "journal-path: " << jpath << ": "
3218            << cpp_strerror(EISDIR) << std::endl;
3219       return 1;
3220     }
3221   }
3222
3223   ObjectStore *fs = ObjectStore::create(g_ceph_context, type, dpath, jpath, flags);
3224   if (fs == NULL) {
3225     cerr << "Unable to create store of type " << type << std::endl;
3226     return 1;
3227   }
3228
3229   if (op == "fsck" || op == "fsck-deep") {
3230     int r = fs->fsck(op == "fsck-deep");
3231     if (r < 0) {
3232       cerr << "fsck failed: " << cpp_strerror(r) << std::endl;
3233       return 1;
3234     }
3235     if (r > 0) {
3236       cerr << "fsck found " << r << " errors" << std::endl;
3237       return 1;
3238     }
3239     cout << "fsck found no errors" << std::endl;
3240     return 0;
3241   }
3242   if (op == "repair" || op == "repair-deep") {
3243     int r = fs->repair(op == "repair-deep");
3244     if (r < 0) {
3245       cerr << "repair failed: " << cpp_strerror(r) << std::endl;
3246       return 1;
3247     }
3248     if (r > 0) {
3249       cerr << "repair found " << r << " errors" << std::endl;
3250       return 1;
3251     }
3252     cout << "repair found no errors" << std::endl;
3253     return 0;
3254   }
3255   if (op == "mkfs") {
3256     if (fsid.length()) {
3257       uuid_d f;
3258       bool r = f.parse(fsid.c_str());
3259       if (!r) {
3260         cerr << "failed to parse uuid '" << fsid << "'" << std::endl;
3261         return 1;
3262       }
3263       fs->set_fsid(f);
3264     }
3265     int r = fs->mkfs();
3266     if (r < 0) {
3267       cerr << "mkfs failed: " << cpp_strerror(r) << std::endl;
3268       return 1;
3269     }
3270     return 0;
3271   }
3272   if (op == "dup") {
3273     string target_type;
3274     char fn[PATH_MAX];
3275     snprintf(fn, sizeof(fn), "%s/type", target_data_path.c_str());
3276     int fd = ::open(fn, O_RDONLY);
3277     if (fd < 0) {
3278       cerr << "Unable to open " << target_data_path << "/type" << std::endl;
3279       exit(1);
3280     }
3281     bufferlist bl;
3282     bl.read_fd(fd, 64);
3283     if (bl.length()) {
3284       target_type = string(bl.c_str(), bl.length() - 1);  // drop \n
3285     }
3286     ::close(fd);
3287     ObjectStore *targetfs = ObjectStore::create(
3288       g_ceph_context, target_type,
3289       target_data_path, "", 0);
3290     if (targetfs == NULL) {
3291       cerr << "Unable to open store of type " << target_type << std::endl;
3292       return 1;
3293     }
3294     int r = dup(dpath, fs, target_data_path, targetfs);
3295     if (r < 0) {
3296       cerr << "dup failed: " << cpp_strerror(r) << std::endl;
3297       return 1;
3298     }
3299     return 0;
3300   }
3301
3302   ObjectStore::Sequencer *osr = new ObjectStore::Sequencer(__func__);
3303   int ret = fs->mount();
3304   if (ret < 0) {
3305     if (ret == -EBUSY) {
3306       cerr << "OSD has the store locked" << std::endl;
3307     } else {
3308       cerr << "Mount failed with '" << cpp_strerror(ret) << "'" << std::endl;
3309     }
3310     return 1;
3311   }
3312
3313   if (op == "fuse") {
3314 #ifdef HAVE_LIBFUSE
3315     FuseStore fuse(fs, mountpoint);
3316     cout << "mounting fuse at " << mountpoint << " ..." << std::endl;
3317     int r = fuse.main();
3318     if (r < 0) {
3319       cerr << "failed to mount fuse: " << cpp_strerror(r) << std::endl;
3320       return 1;
3321     }
3322 #else
3323     cerr << "fuse support not enabled" << std::endl;
3324 #endif
3325     return 0;
3326   }
3327
3328   vector<coll_t> ls;
3329   vector<coll_t>::iterator it;
3330   CompatSet supported;
3331
3332 #ifdef INTERNAL_TEST
3333   supported = get_test_compat_set();
3334 #else
3335   supported = OSD::get_osd_compat_set();
3336 #endif
3337
3338   bufferlist bl;
3339   OSDSuperblock superblock;
3340   bufferlist::iterator p;
3341   ret = fs->read(coll_t::meta(), OSD_SUPERBLOCK_GOBJECT, 0, 0, bl);
3342   if (ret < 0) {
3343     cerr << "Failure to read OSD superblock: " << cpp_strerror(ret) << std::endl;
3344     goto out;
3345   }
3346
3347   p = bl.begin();
3348   ::decode(superblock, p);
3349
3350   if (debug) {
3351     cerr << "Cluster fsid=" << superblock.cluster_fsid << std::endl;
3352   }
3353
3354   if (debug) {
3355     cerr << "Supported features: " << supported << std::endl;
3356     cerr << "On-disk features: " << superblock.compat_features << std::endl;
3357   }
3358   if (supported.compare(superblock.compat_features) == -1) {
3359     CompatSet unsupported = supported.unsupported(superblock.compat_features);
3360     cerr << "On-disk OSD incompatible features set "
3361       << unsupported << std::endl;
3362     ret = -EINVAL;
3363     goto out;
3364   }
3365
3366   if (op == "apply-layout-settings") {
3367     ret = apply_layout_settings(fs, superblock, pool, pgid, dry_run);
3368     goto out;
3369   }
3370
3371   if (op != "list" && vm.count("object")) {
3372     // Special case: Create pgmeta_oid if empty string specified
3373     // This can't conflict with any actual object names.
3374     if (object == "") {
3375       ghobj = pgid.make_pgmeta_oid();
3376     } else {
3377     json_spirit::Value v;
3378     try {
3379       if (!json_spirit::read(object, v) ||
3380           (v.type() != json_spirit::array_type && v.type() != json_spirit::obj_type)) {
3381         // Special: Need head/snapdir so set even if user didn't specify
3382         if (vm.count("objcmd") && (objcmd == "remove-clone-metadata"))
3383           head = true;
3384         lookup_ghobject lookup(object, nspace, head);
3385         if (pgidstr.length())
3386           ret = action_on_all_objects_in_exact_pg(fs, coll_t(pgid), lookup, debug);
3387         else
3388           ret = action_on_all_objects(fs, lookup, debug);
3389         if (ret) {
3390           throw std::runtime_error("Internal error");
3391         } else {
3392           if (lookup.size() != 1) {
3393             stringstream ss;
3394             if (lookup.size() == 0)
3395               ss << "No object id '" << object << "' found or invalid JSON specified";
3396             else
3397               ss << "Found " << lookup.size() << " objects with id '" << object
3398                  << "', please use a JSON spec from --op list instead";
3399             throw std::runtime_error(ss.str());
3400           }
3401           pair<coll_t, ghobject_t> found = lookup.pop();
3402           pgidstr = found.first.to_str();
3403           pgid.parse(pgidstr.c_str());
3404           ghobj = found.second;
3405         }
3406       } else {
3407         stringstream ss;
3408         if (pgidstr.length() == 0 && v.type() != json_spirit::array_type) {
3409           ss << "Without --pgid the object '" << object
3410              << "' must be a JSON array";
3411           throw std::runtime_error(ss.str());
3412         }
3413         if (v.type() == json_spirit::array_type) {
3414           json_spirit::Array array = v.get_array();
3415           if (array.size() != 2) {
3416             ss << "Object '" << object
3417                << "' must be a JSON array with 2 elements";
3418             throw std::runtime_error(ss.str());
3419           }
3420           vector<json_spirit::Value>::iterator i = array.begin();
3421           assert(i != array.end());
3422           if (i->type() != json_spirit::str_type) {
3423             ss << "Object '" << object
3424                << "' must be a JSON array with the first element a string";
3425             throw std::runtime_error(ss.str());
3426           }
3427           string object_pgidstr = i->get_str();
3428           if (object_pgidstr != "meta") {
3429             spg_t object_pgid;
3430             object_pgid.parse(object_pgidstr.c_str());
3431             if (pgidstr.length() > 0) {
3432               if (object_pgid != pgid) {
3433                 ss << "object '" << object
3434                    << "' has a pgid different from the --pgid="
3435                    << pgidstr << " option";
3436                 throw std::runtime_error(ss.str());
3437               }
3438             } else {
3439               pgidstr = object_pgidstr;
3440               pgid = object_pgid;
3441             }
3442           } else {
3443             pgidstr = object_pgidstr;
3444           }
3445           ++i;
3446           v = *i;
3447         }
3448         try {
3449           ghobj.decode(v);
3450         } catch (std::runtime_error& e) {
3451           ss << "Decode object JSON error: " << e.what();
3452           throw std::runtime_error(ss.str());
3453         }
3454         if (pgidstr != "meta" && (uint64_t)pgid.pgid.m_pool != (uint64_t)ghobj.hobj.pool) {
3455           cerr << "Object pool and pgid pool don't match" << std::endl;
3456           ret = 1;
3457           goto out;
3458         }
3459       }
3460     } catch (std::runtime_error& e) {
3461       cerr << e.what() << std::endl;
3462       ret = 1;
3463       goto out;
3464     }
3465     }
3466   }
3467
3468   // The ops which require --pgid option are checked here and
3469   // mentioned in the usage for --pgid.
3470   if ((op == "info" || op == "log" || op == "remove" || op == "export"
3471       || op == "export-remove" || op == "rm-past-intervals"
3472       || op == "mark-complete" || op == "trim-pg-log") &&
3473       pgidstr.length() == 0) {
3474     cerr << "Must provide pgid" << std::endl;
3475     usage(desc);
3476     ret = 1;
3477     goto out;
3478   }
3479
3480   if (op == "import") {
3481
3482     try {
3483       ret = tool.do_import(fs, superblock, force, pgidstr, *osr);
3484     }
3485     catch (const buffer::error &e) {
3486       cerr << "do_import threw exception error " << e.what() << std::endl;
3487       ret = -EFAULT;
3488     }
3489     if (ret == -EFAULT) {
3490       cerr << "Corrupt input for import" << std::endl;
3491     }
3492     if (ret == 0)
3493       cout << "Import successful" << std::endl;
3494     goto out;
3495   } else if (op == "dump-journal-mount") {
3496     // Undocumented feature to dump journal with mounted fs
3497     // This doesn't support the format option, but it uses the
3498     // ObjectStore::dump_journal() and mounts to get replay to run.
3499     ret = fs->dump_journal(cout);
3500     if (ret) {
3501       if (ret == -EOPNOTSUPP) {
3502         cerr << "Object store type \"" << type << "\" doesn't support journal dump" << std::endl;
3503       } else {
3504         cerr << "Journal dump failed with error " << cpp_strerror(ret) << std::endl;
3505       }
3506     }
3507     goto out;
3508   } else if (op == "get-osdmap") {
3509     bufferlist bl;
3510     OSDMap osdmap;
3511     if (epoch == 0) {
3512       epoch = superblock.current_epoch;
3513     }
3514     ret = get_osdmap(fs, epoch, osdmap, bl);
3515     if (ret) {
3516       cerr << "Failed to get osdmap#" << epoch << ": "
3517            << cpp_strerror(ret) << std::endl;
3518       goto out;
3519     }
3520     ret = bl.write_fd(file_fd);
3521     if (ret) {
3522       cerr << "Failed to write to " << file << ": " << cpp_strerror(ret) << std::endl;
3523     } else {
3524       cout << "osdmap#" << epoch << " exported." << std::endl;
3525     }
3526     goto out;
3527   } else if (op == "set-osdmap") {
3528     bufferlist bl;
3529     ret = get_fd_data(file_fd, bl);
3530     if (ret < 0) {
3531       cerr << "Failed to read osdmap " << cpp_strerror(ret) << std::endl;
3532     } else {
3533       ret = set_osdmap(fs, epoch, bl, force, *osr);
3534     }
3535     goto out;
3536   } else if (op == "get-inc-osdmap") {
3537     bufferlist bl;
3538     if (epoch == 0) {
3539       epoch = superblock.current_epoch;
3540     }
3541     ret = get_inc_osdmap(fs, epoch, bl);
3542     if (ret < 0) {
3543       cerr << "Failed to get incremental osdmap# " << epoch << ": "
3544            << cpp_strerror(ret) << std::endl;
3545       goto out;
3546     }
3547     ret = bl.write_fd(file_fd);
3548     if (ret) {
3549       cerr << "Failed to write to " << file << ": " << cpp_strerror(ret) << std::endl;
3550     } else {
3551       cout << "inc-osdmap#" << epoch << " exported." << std::endl;
3552     }
3553     goto out;
3554   } else if (op == "set-inc-osdmap") {
3555     bufferlist bl;
3556     ret = get_fd_data(file_fd, bl);
3557     if (ret < 0) {
3558       cerr << "Failed to read incremental osdmap  " << cpp_strerror(ret) << std::endl;
3559       goto out;
3560     } else {
3561       ret = set_inc_osdmap(fs, epoch, bl, force, *osr);
3562     }
3563     goto out;
3564   } else if (op == "update-mon-db") {
3565     if (!vm.count("mon-store-path")) {
3566       cerr << "Please specify the path to monitor db to update" << std::endl;
3567       ret = -EINVAL;
3568     } else {
3569       ret = update_mon_db(*fs, superblock, dpath + "/keyring", mon_store_path);
3570     }
3571     goto out;
3572   }
3573
3574   log_oid = OSD::make_pg_log_oid(pgid);
3575   biginfo_oid = OSD::make_pg_biginfo_oid(pgid);
3576
3577   if (op == "remove") {
3578     if (!force && !dry_run) {
3579       cerr << "Please use export-remove or you must use --force option" << std::endl;
3580       ret = -EINVAL;
3581       goto out;
3582     }
3583     ret = initiate_new_remove_pg(fs, pgid, *osr);
3584     if (ret < 0) {
3585       cerr << "PG '" << pgid << "' not found" << std::endl;
3586       goto out;
3587     }
3588     cout << "Remove successful" << std::endl;
3589     goto out;
3590   }
3591
3592   if (op == "fix-lost") {
3593     boost::scoped_ptr<action_on_object_t> action;
3594     action.reset(new do_fix_lost(osr));
3595     if (pgidstr.length())
3596       ret = action_on_all_objects_in_exact_pg(fs, coll_t(pgid), *action, debug);
3597     else
3598       ret = action_on_all_objects(fs, *action, debug);
3599     goto out;
3600   }
3601
3602   if (op == "list") {
3603     ret = do_list(fs, pgidstr, object, nspace, formatter, debug,
3604                   human_readable, head);
3605     if (ret < 0) {
3606       cerr << "do_list failed: " << cpp_strerror(ret) << std::endl;
3607     }
3608     goto out;
3609   }
3610
3611   if (op == "dump-super") {
3612     formatter->open_object_section("superblock");
3613     superblock.dump(formatter);
3614     formatter->close_section();
3615     formatter->flush(cout);
3616     cout << std::endl;
3617     goto out;
3618   }
3619
3620   if (op == "meta-list") {
3621     ret = do_meta(fs, object, formatter, debug, human_readable);
3622     if (ret < 0) {
3623       cerr << "do_meta failed: " << cpp_strerror(ret) << std::endl;
3624     }
3625     goto out;
3626   }
3627
3628   ret = fs->list_collections(ls);
3629   if (ret < 0) {
3630     cerr << "failed to list pgs: " << cpp_strerror(ret) << std::endl;
3631     goto out;
3632   }
3633
3634   if (debug && op == "list-pgs")
3635     cout << "Performing list-pgs operation" << std::endl;
3636
3637   // Find pg
3638   for (it = ls.begin(); it != ls.end(); ++it) {
3639     spg_t tmppgid;
3640
3641     if (pgidstr == "meta") {
3642       if (it->to_str() == "meta")
3643         break;
3644       else
3645         continue;
3646     }
3647
3648     if (!it->is_pg(&tmppgid)) {
3649       continue;
3650     }
3651
3652     if (it->is_temp(&tmppgid)) {
3653       continue;
3654     }
3655
3656     if (op != "list-pgs" && tmppgid != pgid) {
3657       continue;
3658     }
3659
3660     if (op != "list-pgs") {
3661       //Found!
3662       break;
3663     }
3664
3665     cout << tmppgid << std::endl;
3666   }
3667
3668   if (op == "list-pgs") {
3669     ret = 0;
3670     goto out;
3671   }
3672
3673   // If not an object command nor any of the ops handled below, then output this usage
3674   // before complaining about a bad pgid
3675   if (!vm.count("objcmd") && op != "export" && op != "export-remove" && op != "info" && op != "log" && op != "rm-past-intervals" && op != "mark-complete" && op != "trim-pg-log") {
3676     cerr << "Must provide --op (info, log, remove, mkfs, fsck, repair, export, export-remove, import, list, fix-lost, list-pgs, rm-past-intervals, dump-journal, dump-super, meta-list, "
3677       "get-osdmap, set-osdmap, get-inc-osdmap, set-inc-osdmap, mark-complete, dump-import, trim-pg-log)"
3678          << std::endl;
3679     usage(desc);
3680     ret = 1;
3681     goto out;
3682   }
3683   epoch_t map_epoch;
3684 // The following code for export, info, log require omap or !skip-mount-omap
3685   if (it != ls.end()) {
3686
3687     coll_t coll = *it;
3688
3689     if (vm.count("objcmd")) {
3690       ret = 0;
3691       if (objcmd == "remove" || objcmd == "removeall") {
3692         bool all = (objcmd == "removeall");
3693         ret = do_remove_object(fs, coll, ghobj, all, force, *osr);
3694         goto out;
3695       } else if (objcmd == "list-attrs") {
3696         ret = do_list_attrs(fs, coll, ghobj);
3697         goto out;
3698       } else if (objcmd == "list-omap") {
3699         ret = do_list_omap(fs, coll, ghobj);
3700         goto out;
3701       } else if (objcmd == "get-bytes" || objcmd == "set-bytes") {
3702         if (objcmd == "get-bytes") {
3703           int fd;
3704           if (vm.count("arg1") == 0 || arg1 == "-") {
3705             fd = STDOUT_FILENO;
3706           } else {
3707             fd = open(arg1.c_str(), O_WRONLY|O_TRUNC|O_CREAT|O_EXCL|O_LARGEFILE, 0666);
3708             if (fd == -1) {
3709               cerr << "open " << arg1 << " " << cpp_strerror(errno) << std::endl;
3710               ret = 1;
3711               goto out;
3712             }
3713           }
3714           ret = do_get_bytes(fs, coll, ghobj, fd);
3715           if (fd != STDOUT_FILENO)
3716             close(fd);
3717         } else {
3718           int fd;
3719           if (vm.count("arg1") == 0 || arg1 == "-") {
3720             // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3721             if (isatty(STDIN_FILENO)) {
3722                 cerr << "stdin is a tty and no file specified" << std::endl;
3723                 ret = 1;
3724                 goto out;
3725             }
3726             fd = STDIN_FILENO;
3727           } else {
3728             fd = open(arg1.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3729             if (fd == -1) {
3730               cerr << "open " << arg1 << " " << cpp_strerror(errno) << std::endl;
3731               ret = 1;
3732               goto out;
3733             }
3734           }
3735           ret = do_set_bytes(fs, coll, ghobj, fd, *osr);
3736           if (fd != STDIN_FILENO)
3737             close(fd);
3738         }
3739         goto out;
3740       } else if (objcmd == "get-attr") {
3741         if (vm.count("arg1") == 0) {
3742           usage(desc);
3743           ret = 1;
3744           goto out;
3745         }
3746         ret = do_get_attr(fs, coll, ghobj, arg1);
3747         goto out;
3748       } else if (objcmd == "set-attr") {
3749         if (vm.count("arg1") == 0) {
3750           usage(desc);
3751           ret = 1;
3752         }
3753
3754         int fd;
3755         if (vm.count("arg2") == 0 || arg2 == "-") {
3756           // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3757           if (isatty(STDIN_FILENO)) {
3758             cerr << "stdin is a tty and no file specified" << std::endl;
3759             ret = 1;
3760             goto out;
3761           }
3762           fd = STDIN_FILENO;
3763         } else {
3764           fd = open(arg2.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3765           if (fd == -1) {
3766             cerr << "open " << arg2 << " " << cpp_strerror(errno) << std::endl;
3767             ret = 1;
3768             goto out;
3769           }
3770         }
3771         ret = do_set_attr(fs, coll, ghobj, arg1, fd, *osr);
3772         if (fd != STDIN_FILENO)
3773           close(fd);
3774         goto out;
3775       } else if (objcmd == "rm-attr") {
3776         if (vm.count("arg1") == 0) {
3777           usage(desc);
3778           ret = 1;
3779           goto out;
3780         }
3781         ret = do_rm_attr(fs, coll, ghobj, arg1, *osr);
3782         goto out;
3783       } else if (objcmd == "get-omap") {
3784         if (vm.count("arg1") == 0) {
3785           usage(desc);
3786           ret = 1;
3787           goto out;
3788         }
3789         ret = do_get_omap(fs, coll, ghobj, arg1);
3790         goto out;
3791       } else if (objcmd == "set-omap") {
3792         if (vm.count("arg1") == 0) {
3793           usage(desc);
3794           ret = 1;
3795           goto out;
3796         }
3797         int fd;
3798         if (vm.count("arg2") == 0 || arg2 == "-") {
3799           // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3800           if (isatty(STDIN_FILENO)) {
3801             cerr << "stdin is a tty and no file specified" << std::endl;
3802             ret = 1;
3803             goto out;
3804           }
3805           fd = STDIN_FILENO;
3806         } else {
3807           fd = open(arg2.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3808           if (fd == -1) {
3809             cerr << "open " << arg2 << " " << cpp_strerror(errno) << std::endl;
3810             ret = 1;
3811             goto out;
3812           }
3813         }
3814         ret = do_set_omap(fs, coll, ghobj, arg1, fd, *osr);
3815         if (fd != STDIN_FILENO)
3816           close(fd);
3817         goto out;
3818       } else if (objcmd == "rm-omap") {
3819         if (vm.count("arg1") == 0) {
3820           usage(desc);
3821           ret = 1;
3822           goto out;
3823         }
3824         ret = do_rm_omap(fs, coll, ghobj, arg1, *osr);
3825         goto out;
3826       } else if (objcmd == "get-omaphdr") {
3827         if (vm.count("arg1")) {
3828           usage(desc);
3829           ret = 1;
3830           goto out;
3831         }
3832         ret = do_get_omaphdr(fs, coll, ghobj);
3833         goto out;
3834       } else if (objcmd == "set-omaphdr") {
3835         // Extra arg
3836         if (vm.count("arg2")) {
3837           usage(desc);
3838           ret = 1;
3839           goto out;
3840         }
3841         int fd;
3842         if (vm.count("arg1") == 0 || arg1 == "-") {
3843           // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3844           if (isatty(STDIN_FILENO)) {
3845             cerr << "stdin is a tty and no file specified" << std::endl;
3846             ret = 1;
3847             goto out;
3848           }
3849           fd = STDIN_FILENO;
3850         } else {
3851           fd = open(arg1.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3852           if (fd == -1) {
3853             cerr << "open " << arg1 << " " << cpp_strerror(errno) << std::endl;
3854             ret = 1;
3855             goto out;
3856           }
3857         }
3858         ret = do_set_omaphdr(fs, coll, ghobj, fd, *osr);
3859         if (fd != STDIN_FILENO)
3860           close(fd);
3861         goto out;
3862       } else if (objcmd == "dump") {
3863         // There should not be any other arguments
3864         if (vm.count("arg1") || vm.count("arg2")) {
3865           usage(desc);
3866           ret = 1;
3867           goto out;
3868         }
3869         ret = print_obj_info(fs, coll, ghobj, formatter);
3870         goto out;
3871       } else if (objcmd == "set-size" || objcmd == "corrupt-size") {
3872         // Undocumented testing feature
3873         bool corrupt = (objcmd == "corrupt-size");
3874         // Extra arg
3875         if (vm.count("arg1") == 0 || vm.count("arg2")) {
3876           usage(desc);
3877           ret = 1;
3878           goto out;
3879         }
3880         if (arg1.length() == 0 || !isdigit(arg1.c_str()[0])) {
3881           cerr << "Invalid size '" << arg1 << "' specified" << std::endl;
3882           ret = 1;
3883           goto out;
3884         }
3885         uint64_t size = atoll(arg1.c_str());
3886         ret = set_size(fs, coll, ghobj, size, formatter, *osr, corrupt);
3887         goto out;
3888       } else if (objcmd == "clear-snapset") {
3889         // UNDOCUMENTED: For testing zap SnapSet
3890         // IGNORE extra args since not in usage anyway
3891         if (!ghobj.hobj.has_snapset()) {
3892           cerr << "'" << objcmd << "' requires a head or snapdir object" << std::endl;
3893           ret = 1;
3894           goto out;
3895         }
3896         ret = clear_snapset(fs, coll, ghobj, arg1, *osr);
3897         goto out;
3898       } else if (objcmd == "remove-clone-metadata") {
3899         // Extra arg
3900         if (vm.count("arg1") == 0 || vm.count("arg2")) {
3901           usage(desc);
3902           ret = 1;
3903           goto out;
3904         }
3905         if (!ghobj.hobj.has_snapset()) {
3906           cerr << "'" << objcmd << "' requires a head or snapdir object" << std::endl;
3907           ret = 1;
3908           goto out;
3909         }
3910         if (arg1.length() == 0 || !isdigit(arg1.c_str()[0])) {
3911           cerr << "Invalid cloneid '" << arg1 << "' specified" << std::endl;
3912           ret = 1;
3913           goto out;
3914         }
3915         snapid_t cloneid = atoi(arg1.c_str());
3916         ret = remove_clone(fs, coll, ghobj, cloneid, force, *osr);
3917         goto out;
3918       }
3919       cerr << "Unknown object command '" << objcmd << "'" << std::endl;
3920       usage(desc);
3921       ret = 1;
3922       goto out;
3923     }
3924
3925     bufferlist bl;
3926     map_epoch = 0;
3927     ret = PG::peek_map_epoch(fs, pgid, &map_epoch, &bl);
3928     if (ret < 0)
3929       cerr << "peek_map_epoch reports error" << std::endl;
3930     if (debug)
3931       cerr << "map_epoch " << map_epoch << std::endl;
3932
3933     pg_info_t info(pgid);
3934     PastIntervals past_intervals;
3935     __u8 struct_ver;
3936     ret = PG::read_info(fs, pgid, coll, bl, info, past_intervals,
3937                       struct_ver);
3938     if (ret < 0) {
3939       cerr << "read_info error " << cpp_strerror(ret) << std::endl;
3940       goto out;
3941     }
3942     if (struct_ver < PG::compat_struct_v) {
3943       cerr << "PG is too old to upgrade, use older Ceph version" << std::endl;
3944       ret = -EFAULT;
3945       goto out;
3946     }
3947     if (debug)
3948       cerr << "struct_v " << (int)struct_ver << std::endl;
3949
3950     if (op == "export" || op == "export-remove") {
3951       ret = tool.do_export(fs, coll, pgid, info, map_epoch, struct_ver, superblock, past_intervals);
3952       if (ret == 0) {
3953         cerr << "Export successful" << std::endl;
3954         if (op == "export-remove") {
3955           ret = initiate_new_remove_pg(fs, pgid, *osr);
3956           // Export succeeded, so pgid is there
3957           assert(ret == 0);
3958           cerr << "Remove successful" << std::endl;
3959         }
3960       }
3961     } else if (op == "info") {
3962       formatter->open_object_section("info");
3963       info.dump(formatter);
3964       formatter->close_section();
3965       formatter->flush(cout);
3966       cout << std::endl;
3967     } else if (op == "log") {
3968       PGLog::IndexedLog log;
3969       pg_missing_t missing;
3970       ret = get_log(fs, struct_ver, coll, pgid, info, log, missing);
3971       if (ret < 0)
3972           goto out;
3973
3974       dump_log(formatter, cout, log, missing);
3975     } else if (op == "rm-past-intervals") {
3976       ObjectStore::Transaction tran;
3977       ObjectStore::Transaction *t = &tran;
3978
3979       if (struct_ver < PG::compat_struct_v) {
3980         cerr << "Can't remove past-intervals, version mismatch " << (int)struct_ver
3981           << " (pg)  < compat " << (int)PG::compat_struct_v << " (tool)"
3982           << std::endl;
3983         ret = -EFAULT;
3984         goto out;
3985       }
3986
3987       cout << "Remove past-intervals " << past_intervals << std::endl;
3988
3989       past_intervals.clear();
3990       if (dry_run) {
3991         ret = 0;
3992         goto out;
3993       }
3994       ret = write_info(*t, map_epoch, info, past_intervals);
3995
3996       if (ret == 0) {
3997         fs->apply_transaction(osr, std::move(*t));
3998         cout << "Removal succeeded" << std::endl;
3999       }
4000     } else if (op == "mark-complete") {
4001       ObjectStore::Transaction tran;
4002       ObjectStore::Transaction *t = &tran;
4003
4004       if (struct_ver < PG::compat_struct_v) {
4005         cerr << "Can't mark-complete, version mismatch " << (int)struct_ver
4006              << " (pg)  < compat " << (int)PG::compat_struct_v << " (tool)"
4007              << std::endl;
4008         ret = 1;
4009         goto out;
4010       }
4011
4012       cout << "Marking complete " << std::endl;
4013
4014       info.last_update = eversion_t(superblock.current_epoch, info.last_update.version + 1);
4015       info.last_backfill = hobject_t::get_max();
4016       info.last_epoch_started = superblock.current_epoch;
4017       info.history.last_epoch_started = superblock.current_epoch;
4018       info.history.last_epoch_clean = superblock.current_epoch;
4019       past_intervals.clear();
4020
4021       if (!dry_run) {
4022         ret = write_info(*t, map_epoch, info, past_intervals);
4023         if (ret != 0)
4024           goto out;
4025         fs->apply_transaction(osr, std::move(*t));
4026       }
4027       cout << "Marking complete succeeded" << std::endl;
4028     } else if (op == "trim-pg-log") {
4029       ret = do_trim_pg_log(fs, coll, info, pgid, *osr,
4030                            map_epoch, past_intervals);
4031       if (ret < 0) {
4032         cerr << "Error trimming pg log: " << cpp_strerror(ret) << std::endl;
4033         goto out;
4034       }
4035       cout << "Finished trimming pg log" << std::endl;
4036       goto out;
4037     } else {
4038       assert(!"Should have already checked for valid --op");
4039     }
4040   } else {
4041     cerr << "PG '" << pgid << "' not found" << std::endl;
4042     ret = -ENOENT;
4043   }
4044
4045 out:
4046   int r = fs->umount();
4047   delete osr;
4048   if (r < 0) {
4049     cerr << "umount failed: " << cpp_strerror(r) << std::endl;
4050     // If no previous error, then use umount() error
4051     if (ret == 0)
4052       ret = r;
4053   }
4054
4055   if (dry_run) {
4056     // Export output can go to stdout, so put this message on stderr
4057     if (op == "export")
4058       cerr << "dry-run: Nothing changed" << std::endl;
4059     else
4060       cout << "dry-run: Nothing changed" << std::endl;
4061   }
4062
4063   if (ret < 0)
4064     ret = 1;
4065   return ret;
4066 }