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 corrupt_info(ObjectStore *store, coll_t coll, ghobject_t &ghobj, Formatter* formatter,
2409              ObjectStore::Sequencer &osr)
2410 {
2411   bufferlist attr;
2412   int r = store->getattr(coll, ghobj, OI_ATTR, attr);
2413   if (r < 0) {
2414     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2415        << cpp_strerror(r) << std::endl;
2416     return r;
2417   }
2418   object_info_t oi;
2419   bufferlist::iterator bp = attr.begin();
2420   try {
2421     ::decode(oi, bp);
2422   } catch (...) {
2423     r = -EINVAL;
2424     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2425          << cpp_strerror(r) << std::endl;
2426     return r;
2427   }
2428   cout << "Corrupting info" << std::endl;
2429   if (!dry_run) {
2430     attr.clear();
2431     oi.alloc_hint_flags += 0xff;
2432     ObjectStore::Transaction t;
2433     ::encode(oi, attr, -1);  /* fixme: using full features */
2434     t.setattr(coll, ghobj, OI_ATTR, attr);
2435     r = store->apply_transaction(&osr, std::move(t));
2436     if (r < 0) {
2437       cerr << "Error writing object info: " << make_pair(coll, ghobj) << ", "
2438          << cpp_strerror(r) << std::endl;
2439       return r;
2440     }
2441   }
2442   return 0;
2443 }
2444
2445 int set_size(ObjectStore *store, coll_t coll, ghobject_t &ghobj, uint64_t setsize, Formatter* formatter,
2446              ObjectStore::Sequencer &osr, bool corrupt)
2447 {
2448   if (ghobj.hobj.is_snapdir()) {
2449     cerr << "Can't set the size of a snapdir" << std::endl;
2450     return -EINVAL;
2451   }
2452   bufferlist attr;
2453   int r = store->getattr(coll, ghobj, OI_ATTR, attr);
2454   if (r < 0) {
2455     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2456        << cpp_strerror(r) << std::endl;
2457     return r;
2458   }
2459   object_info_t oi;
2460   bufferlist::iterator bp = attr.begin();
2461   try {
2462     ::decode(oi, bp);
2463   } catch (...) {
2464     r = -EINVAL;
2465     cerr << "Error getting attr on : " << make_pair(coll, ghobj) << ", "
2466          << cpp_strerror(r) << std::endl;
2467     return r;
2468   }
2469   struct stat st;
2470   r =  store->stat(coll, ghobj, &st, true);
2471   if (r < 0) {
2472     cerr << "Error stat on : " << make_pair(coll, ghobj) << ", "
2473          << cpp_strerror(r) << std::endl;
2474   }
2475   ghobject_t head(ghobj);
2476   SnapSet ss;
2477   bool found_head = true;
2478   map<snapid_t, uint64_t>::iterator csi;
2479   bool is_snap = ghobj.hobj.is_snap();
2480   if (is_snap) {
2481     head.hobj = head.hobj.get_head();
2482     r = get_snapset(store, coll, head, ss, true);
2483     if (r < 0 && r != -ENOENT) {
2484       // Requested get_snapset() silent, so if not -ENOENT show error
2485       cerr << "Error getting snapset on : " << make_pair(coll, head) << ", "
2486            << cpp_strerror(r) << std::endl;
2487       return r;
2488     }
2489     if (r == -ENOENT) {
2490       head.hobj = head.hobj.get_snapdir();
2491       r = get_snapset(store, coll, head, ss);
2492       if (r < 0)
2493         return r;
2494       found_head = false;
2495     } else {
2496       found_head = true;
2497     }
2498     csi = ss.clone_size.find(ghobj.hobj.snap);
2499     if (csi == ss.clone_size.end()) {
2500       cerr << "SnapSet is missing clone_size for snap " << ghobj.hobj.snap << std::endl;
2501       return -EINVAL;
2502     }
2503   }
2504   if ((uint64_t)st.st_size == setsize && oi.size == setsize
2505        && (!is_snap || csi->second == setsize)) {
2506     cout << "Size of object is already " << setsize << std::endl;
2507     return 0;
2508   }
2509   cout << "Setting size to " << setsize << ", stat size " << st.st_size
2510        << ", obj info size " << oi.size;
2511   if (is_snap) {
2512     cout << ", " << (found_head ? "head" : "snapdir")
2513          << " clone_size " << csi->second;
2514     csi->second = setsize;
2515   }
2516   cout << std::endl;
2517   if (!dry_run) {
2518     attr.clear();
2519     oi.size = setsize;
2520     ObjectStore::Transaction t;
2521     // Only modify object info if we want to corrupt it
2522     if (!corrupt && (uint64_t)st.st_size != setsize) {
2523       t.truncate(coll, ghobj, setsize);
2524       // Changing objectstore size will invalidate data_digest, so clear it.
2525       oi.clear_data_digest();
2526     }
2527     ::encode(oi, attr, -1);  /* fixme: using full features */
2528     t.setattr(coll, ghobj, OI_ATTR, attr);
2529     if (is_snap) {
2530       bufferlist snapattr;
2531       snapattr.clear();
2532       ::encode(ss, snapattr);
2533       t.setattr(coll, head, SS_ATTR, snapattr);
2534     }
2535     r = store->apply_transaction(&osr, std::move(t));
2536     if (r < 0) {
2537       cerr << "Error writing object info: " << make_pair(coll, ghobj) << ", "
2538          << cpp_strerror(r) << std::endl;
2539       return r;
2540     }
2541   }
2542   return 0;
2543 }
2544
2545 int clear_snapset(ObjectStore *store, coll_t coll, ghobject_t &ghobj,
2546                   string arg, ObjectStore::Sequencer &osr)
2547 {
2548   SnapSet ss;
2549   int ret = get_snapset(store, coll, ghobj, ss);
2550   if (ret < 0)
2551     return ret;
2552
2553   // Use "head" to set head_exists incorrectly
2554   if (arg == "corrupt" || arg == "head")
2555     ss.head_exists = !ghobj.hobj.is_head();
2556   else if (ss.head_exists != ghobj.hobj.is_head()) {
2557     cerr << "Correcting head_exists, set to "
2558          << (ghobj.hobj.is_head() ? "true" : "false") << std::endl;
2559     ss.head_exists = ghobj.hobj.is_head();
2560   }
2561   // Use "corrupt" to clear entire SnapSet
2562   // Use "seq" to just corrupt SnapSet.seq
2563   if (arg == "corrupt" || arg == "seq")
2564     ss.seq = 0;
2565   // Use "snaps" to just clear SnapSet.snaps
2566   if (arg == "corrupt" || arg == "snaps")
2567     ss.snaps.clear();
2568   // By default just clear clone, clone_overlap and clone_size
2569   if (arg == "corrupt")
2570     arg = "";
2571   if (arg == "" || arg == "clones")
2572     ss.clones.clear();
2573   if (arg == "" || arg == "clone_overlap")
2574     ss.clone_overlap.clear();
2575   if (arg == "" || arg == "clone_size")
2576     ss.clone_size.clear();
2577   // Break all clone sizes by adding 1
2578   if (arg == "size") {
2579     for (map<snapid_t, uint64_t>::iterator i = ss.clone_size.begin();
2580          i != ss.clone_size.end(); ++i)
2581       ++(i->second);
2582   }
2583
2584   if (!dry_run) {
2585     bufferlist bl;
2586     ::encode(ss, bl);
2587     ObjectStore::Transaction t;
2588     t.setattr(coll, ghobj, SS_ATTR, bl);
2589     int r = store->apply_transaction(&osr, std::move(t));
2590     if (r < 0) {
2591       cerr << "Error setting snapset on : " << make_pair(coll, ghobj) << ", "
2592            << cpp_strerror(r) << std::endl;
2593       return r;
2594     }
2595   }
2596   return 0;
2597 }
2598
2599 vector<snapid_t>::iterator find(vector<snapid_t> &v, snapid_t clid)
2600 {
2601   return std::find(v.begin(), v.end(), clid);
2602 }
2603
2604 map<snapid_t, interval_set<uint64_t> >::iterator
2605 find(map<snapid_t, interval_set<uint64_t> > &m, snapid_t clid)
2606 {
2607   return m.find(clid);
2608 }
2609
2610 map<snapid_t, uint64_t>::iterator find(map<snapid_t, uint64_t> &m,
2611                                        snapid_t clid)
2612 {
2613   return m.find(clid);
2614 }
2615
2616 template<class T>
2617 int remove_from(T &mv, string name, snapid_t cloneid, bool force)
2618 {
2619   typename T::iterator i = find(mv, cloneid);
2620   if (i != mv.end()) {
2621     mv.erase(i);
2622   } else {
2623     cerr << "Clone " << cloneid << " doesn't exist in " << name;
2624     if (force) {
2625       cerr << " (ignored)" << std::endl;
2626       return 0;
2627     }
2628     cerr << std::endl;
2629     return -EINVAL;
2630   }
2631   return 0;
2632 }
2633
2634 int remove_clone(ObjectStore *store, coll_t coll, ghobject_t &ghobj, snapid_t cloneid, bool force,
2635                      ObjectStore::Sequencer &osr)
2636 {
2637   // XXX: Don't allow this if in a cache tier or former cache tier
2638   // bool allow_incomplete_clones() const {
2639   //    return cache_mode != CACHEMODE_NONE || has_flag(FLAG_INCOMPLETE_CLONES);
2640
2641   SnapSet snapset;
2642   int ret = get_snapset(store, coll, ghobj, snapset);
2643   if (ret < 0)
2644     return ret;
2645
2646   // Derived from trim_object()
2647   // ...from snapset
2648   vector<snapid_t>::iterator p;
2649   for (p = snapset.clones.begin(); p != snapset.clones.end(); ++p)
2650     if (*p == cloneid)
2651       break;
2652   if (p == snapset.clones.end()) {
2653     cerr << "Clone " << cloneid << " not present";
2654     return -ENOENT;
2655   }
2656   if (p != snapset.clones.begin()) {
2657     // not the oldest... merge overlap into next older clone
2658     vector<snapid_t>::iterator n = p - 1;
2659     hobject_t prev_coid = ghobj.hobj;
2660     prev_coid.snap = *n;
2661     //bool adjust_prev_bytes = is_present_clone(prev_coid);
2662
2663     //if (adjust_prev_bytes)
2664     //  ctx->delta_stats.num_bytes -= snapset.get_clone_bytes(*n);
2665
2666     snapset.clone_overlap[*n].intersection_of(
2667         snapset.clone_overlap[*p]);
2668
2669     //if (adjust_prev_bytes)
2670     //  ctx->delta_stats.num_bytes += snapset.get_clone_bytes(*n);
2671   }
2672
2673   ret = remove_from(snapset.clones, "clones", cloneid, force);
2674   if (ret) return ret;
2675   ret = remove_from(snapset.clone_overlap, "clone_overlap", cloneid, force);
2676   if (ret) return ret;
2677   ret = remove_from(snapset.clone_size, "clone_size", cloneid, force);
2678   if (ret) return ret;
2679
2680   if (dry_run)
2681     return 0;
2682
2683   bufferlist bl;
2684   ::encode(snapset, bl);
2685   ObjectStore::Transaction t;
2686   t.setattr(coll, ghobj, SS_ATTR, bl);
2687   int r = store->apply_transaction(&osr, std::move(t));
2688   if (r < 0) {
2689     cerr << "Error setting snapset on : " << make_pair(coll, ghobj) << ", "
2690          << cpp_strerror(r) << std::endl;
2691     return r;
2692   }
2693   cout << "Removal of clone " << cloneid << " complete" << std::endl;
2694   cout << "Use pg repair after OSD restarted to correct stat information" << std::endl;
2695   return 0;
2696 }
2697
2698 int dup(string srcpath, ObjectStore *src, string dstpath, ObjectStore *dst)
2699 {
2700   cout << "dup from " << src->get_type() << ": " << srcpath << "\n"
2701        << "      to " << dst->get_type() << ": " << dstpath
2702        << std::endl;
2703   ObjectStore::Sequencer osr("dup");
2704   int num, i;
2705   vector<coll_t> collections;
2706   int r;
2707
2708   r = src->mount();
2709   if (r < 0) {
2710     cerr << "failed to mount src: " << cpp_strerror(r) << std::endl;
2711     return r;
2712   }
2713   r = dst->mount();
2714   if (r < 0) {
2715     cerr << "failed to mount dst: " << cpp_strerror(r) << std::endl;
2716     goto out_src;
2717   }
2718
2719   if (src->get_fsid() != dst->get_fsid()) {
2720     cerr << "src fsid " << src->get_fsid() << " != dest " << dst->get_fsid()
2721          << std::endl;
2722     goto out;
2723   }
2724   cout << "fsid " << src->get_fsid() << std::endl;
2725
2726   // make sure dst is empty
2727   r = dst->list_collections(collections);
2728   if (r < 0) {
2729     cerr << "error listing collections on dst: " << cpp_strerror(r) << std::endl;
2730     goto out;
2731   }
2732   if (!collections.empty()) {
2733     cerr << "destination store is not empty" << std::endl;
2734     goto out;
2735   }
2736
2737   r = src->list_collections(collections);
2738   if (r < 0) {
2739     cerr << "error listing collections on src: " << cpp_strerror(r) << std::endl;
2740     goto out;
2741   }
2742
2743   num = collections.size();
2744   cout << num << " collections" << std::endl;
2745   i = 1;
2746   for (auto cid : collections) {
2747     cout << i++ << "/" << num << " " << cid << std::endl;
2748     {
2749       ObjectStore::Transaction t;
2750       int bits = src->collection_bits(cid);
2751       if (bits < 0) {
2752         if (src->get_type() == "filestore" && cid.is_meta()) {
2753           bits = 0;
2754         } else {
2755           cerr << "cannot get bit count for collection " << cid << ": "
2756                << cpp_strerror(bits) << std::endl;
2757           goto out;
2758         }
2759       }
2760       t.create_collection(cid, bits);
2761       dst->apply_transaction(&osr, std::move(t));
2762     }
2763
2764     ghobject_t pos;
2765     uint64_t n = 0;
2766     uint64_t bytes = 0, keys = 0;
2767     while (true) {
2768       vector<ghobject_t> ls;
2769       r = src->collection_list(cid, pos, ghobject_t::get_max(), 1000, &ls, &pos);
2770       if (r < 0) {
2771         cerr << "collection_list on " << cid << " from " << pos << " got: "
2772              << cpp_strerror(r) << std::endl;
2773         goto out;
2774       }
2775       if (ls.empty()) {
2776         break;
2777       }
2778
2779       for (auto& oid : ls) {
2780         //cout << "  " << cid << " " << oid << std::endl;
2781         if (n % 100 == 0) {
2782           cout << "  " << std::setw(16) << n << " objects, "
2783                << std::setw(16) << bytes << " bytes, "
2784                << std::setw(16) << keys << " keys"
2785                << std::setw(1) << "\r" << std::flush;
2786         }
2787         n++;
2788
2789         ObjectStore::Transaction t;
2790         t.touch(cid, oid);
2791
2792         map<string,bufferptr> attrs;
2793         src->getattrs(cid, oid, attrs);
2794         if (!attrs.empty()) {
2795           t.setattrs(cid, oid, attrs);
2796         }
2797
2798         bufferlist bl;
2799         src->read(cid, oid, 0, 0, bl);
2800         if (bl.length()) {
2801           t.write(cid, oid, 0, bl.length(), bl);
2802           bytes += bl.length();
2803         }
2804
2805         bufferlist header;
2806         map<string,bufferlist> omap;
2807         src->omap_get(cid, oid, &header, &omap);
2808         if (header.length()) {
2809           t.omap_setheader(cid, oid, header);
2810           ++keys;
2811         }
2812         if (!omap.empty()) {
2813           keys += omap.size();
2814           t.omap_setkeys(cid, oid, omap);
2815         }
2816
2817         dst->apply_transaction(&osr, std::move(t));
2818       }
2819     }
2820     cout << "  " << std::setw(16) << n << " objects, "
2821          << std::setw(16) << bytes << " bytes, "
2822          << std::setw(16) << keys << " keys"
2823          << std::setw(1) << std::endl;
2824   }
2825
2826   // keyring
2827   cout << "keyring" << std::endl;
2828   {
2829     bufferlist bl;
2830     string s = srcpath + "/keyring";
2831     string err;
2832     r = bl.read_file(s.c_str(), &err);
2833     if (r < 0) {
2834       cerr << "failed to copy " << s << ": " << err << std::endl;
2835     } else {
2836       string d = dstpath + "/keyring";
2837       bl.write_file(d.c_str(), 0600);
2838     }
2839   }
2840
2841   // osd metadata
2842   cout << "duping osd metadata" << std::endl;
2843   {
2844     for (auto k : {"magic", "whoami", "ceph_fsid", "fsid"}) {
2845       string val;
2846       src->read_meta(k, &val);
2847       dst->write_meta(k, val);
2848     }
2849   }
2850
2851   dst->write_meta("ready", "ready");
2852
2853   cout << "done." << std::endl;
2854   r = 0;
2855  out:
2856   dst->umount();
2857  out_src:
2858   src->umount();
2859   return r;
2860 }
2861
2862 void usage(po::options_description &desc)
2863 {
2864     cerr << std::endl;
2865     cerr << desc << std::endl;
2866     cerr << std::endl;
2867     cerr << "Positional syntax:" << std::endl;
2868     cerr << std::endl;
2869     cerr << "ceph-objectstore-tool ... <object> (get|set)-bytes [file]" << std::endl;
2870     cerr << "ceph-objectstore-tool ... <object> set-(attr|omap) <key> [file]" << std::endl;
2871     cerr << "ceph-objectstore-tool ... <object> (get|rm)-(attr|omap) <key>" << std::endl;
2872     cerr << "ceph-objectstore-tool ... <object> get-omaphdr" << std::endl;
2873     cerr << "ceph-objectstore-tool ... <object> set-omaphdr [file]" << std::endl;
2874     cerr << "ceph-objectstore-tool ... <object> list-attrs" << std::endl;
2875     cerr << "ceph-objectstore-tool ... <object> list-omap" << std::endl;
2876     cerr << "ceph-objectstore-tool ... <object> remove|removeall" << std::endl;
2877     cerr << "ceph-objectstore-tool ... <object> dump" << std::endl;
2878     cerr << "ceph-objectstore-tool ... <object> set-size" << std::endl;
2879     cerr << "ceph-objectstore-tool ... <object> remove-clone-metadata <cloneid>" << std::endl;
2880     cerr << std::endl;
2881     cerr << "<object> can be a JSON object description as displayed" << std::endl;
2882     cerr << "by --op list." << std::endl;
2883     cerr << "<object> can be an object name which will be looked up in all" << std::endl;
2884     cerr << "the OSD's PGs." << std::endl;
2885     cerr << "<object> can be the empty string ('') which with a provided pgid " << std::endl;
2886     cerr << "specifies the pgmeta object" << std::endl;
2887     cerr << std::endl;
2888     cerr << "The optional [file] argument will read stdin or write stdout" << std::endl;
2889     cerr << "if not specified or if '-' specified." << std::endl;
2890 }
2891
2892 bool ends_with(const string& check, const string& ending)
2893 {
2894     return check.size() >= ending.size() && check.rfind(ending) == (check.size() - ending.size());
2895 }
2896
2897 // Based on FileStore::dump_journal(), set-up enough to only dump
2898 int mydump_journal(Formatter *f, string journalpath, bool m_journal_dio)
2899 {
2900   int r;
2901
2902   if (!journalpath.length())
2903     return -EINVAL;
2904
2905   FileJournal *journal = new FileJournal(g_ceph_context, uuid_d(), NULL, NULL,
2906                                          journalpath.c_str(), m_journal_dio);
2907   r = journal->_fdump(*f, false);
2908   delete journal;
2909   return r;
2910 }
2911
2912 int apply_layout_settings(ObjectStore *os, const OSDSuperblock &superblock,
2913                           const string &pool_name, const spg_t &pgid, bool dry_run,
2914                           int target_level)
2915 {
2916   int r = 0;
2917
2918   FileStore *fs = dynamic_cast<FileStore*>(os);
2919   if (!fs) {
2920     cerr << "Nothing to do for non-filestore backend" << std::endl;
2921     return 0; // making this return success makes testing easier
2922   }
2923
2924   OSDMap curmap;
2925   bufferlist bl;
2926   r = get_osdmap(os, superblock.current_epoch, curmap, bl);
2927   if (r) {
2928     cerr << "Can't find local OSDMap: " << cpp_strerror(r) << std::endl;
2929     return r;
2930   }
2931
2932   int64_t poolid = -1;
2933   if (pool_name.length()) {
2934     poolid = curmap.lookup_pg_pool_name(pool_name);
2935     if (poolid < 0) {
2936       cerr << "Couldn't find pool " << pool_name << ": " << cpp_strerror(poolid)
2937            << std::endl;
2938       return poolid;
2939     }
2940   }
2941
2942   vector<coll_t> collections, filtered_colls;
2943   r = os->list_collections(collections);
2944   if (r < 0) {
2945     cerr << "Error listing collections: " << cpp_strerror(r) << std::endl;
2946     return r;
2947   }
2948
2949   for (auto const &coll : collections) {
2950     spg_t coll_pgid;
2951     if (coll.is_pg(&coll_pgid) &&
2952         ((poolid >= 0 && coll_pgid.pool() == (uint64_t)poolid) ||
2953          coll_pgid == pgid)) {
2954       filtered_colls.push_back(coll);
2955     }
2956   }
2957
2958   size_t done = 0, total = filtered_colls.size();
2959   for (auto const &coll : filtered_colls) {
2960     if (dry_run) {
2961       cerr << "Would apply layout settings to " << coll << std::endl;
2962     } else {
2963       cerr << "Finished " << done << "/" << total << " collections" << "\r";
2964       r = fs->apply_layout_settings(coll, target_level);
2965       if (r < 0) {
2966         cerr << "Error applying layout settings to " << coll << std::endl;
2967         return r;
2968       }
2969     }
2970     ++done;
2971   }
2972
2973   cerr << "Finished " << total << "/" << total << " collections" << "\r" << std::endl;
2974   return r;
2975 }
2976
2977 int main(int argc, char **argv)
2978 {
2979   string dpath, jpath, pgidstr, op, file, mountpoint, mon_store_path, object;
2980   string target_data_path, fsid;
2981   string objcmd, arg1, arg2, type, format, argnspace, pool;
2982   boost::optional<std::string> nspace;
2983   spg_t pgid;
2984   unsigned epoch = 0;
2985   ghobject_t ghobj;
2986   bool human_readable;
2987   bool force;
2988   Formatter *formatter;
2989   bool head;
2990
2991   po::options_description desc("Allowed options");
2992   desc.add_options()
2993     ("help", "produce help message")
2994     ("type", po::value<string>(&type),
2995      "Arg is one of [bluestore, filestore (default), memstore]")
2996     ("data-path", po::value<string>(&dpath),
2997      "path to object store, mandatory")
2998     ("journal-path", po::value<string>(&jpath),
2999      "path to journal, use if tool can't find it")
3000     ("pgid", po::value<string>(&pgidstr),
3001      "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")
3002     ("pool", po::value<string>(&pool),
3003      "Pool name, mandatory for apply-layout-settings if --pgid is not specified")
3004     ("op", po::value<string>(&op),
3005      "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, "
3006      "get-osdmap, set-osdmap, get-inc-osdmap, set-inc-osdmap, mark-complete, apply-layout-settings, update-mon-db, dump-import, trim-pg-log]")
3007     ("epoch", po::value<unsigned>(&epoch),
3008      "epoch# for get-osdmap and get-inc-osdmap, the current epoch in use if not specified")
3009     ("file", po::value<string>(&file),
3010      "path of file to export, export-remove, import, get-osdmap, set-osdmap, get-inc-osdmap or set-inc-osdmap")
3011     ("mon-store-path", po::value<string>(&mon_store_path),
3012      "path of monstore to update-mon-db")
3013     ("fsid", po::value<string>(&fsid),
3014      "fsid for new store created by mkfs")
3015     ("target-data-path", po::value<string>(&target_data_path),
3016      "path of target object store (for --op dup)")
3017     ("mountpoint", po::value<string>(&mountpoint),
3018      "fuse mountpoint")
3019     ("format", po::value<string>(&format)->default_value("json-pretty"),
3020      "Output format which may be json, json-pretty, xml, xml-pretty")
3021     ("debug", "Enable diagnostic output to stderr")
3022     ("force", "Ignore some types of errors and proceed with operation - USE WITH CAUTION: CORRUPTION POSSIBLE NOW OR IN THE FUTURE")
3023     ("skip-journal-replay", "Disable journal replay")
3024     ("skip-mount-omap", "Disable mounting of omap")
3025     ("head", "Find head/snapdir when searching for objects by name")
3026     ("dry-run", "Don't modify the objectstore")
3027     ("namespace", po::value<string>(&argnspace), "Specify namespace when searching for objects")
3028     ;
3029
3030   po::options_description positional("Positional options");
3031   positional.add_options()
3032     ("object", po::value<string>(&object), "'' for pgmeta_oid, object name or ghobject in json")
3033     ("objcmd", po::value<string>(&objcmd), "command [(get|set)-bytes, (get|set|rm)-(attr|omap), (get|set)-omaphdr, list-attrs, list-omap, remove]")
3034     ("arg1", po::value<string>(&arg1), "arg1 based on cmd, "
3035      "for apply-layout-settings: target hash level split to")
3036     ("arg2", po::value<string>(&arg2), "arg2 based on cmd")
3037     ;
3038
3039   po::options_description all;
3040   all.add(desc).add(positional);
3041
3042   po::positional_options_description pd;
3043   pd.add("object", 1).add("objcmd", 1).add("arg1", 1).add("arg2", 1);
3044
3045   vector<string> ceph_option_strings;
3046   po::variables_map vm;
3047   try {
3048     po::parsed_options parsed =
3049       po::command_line_parser(argc, argv).options(all).allow_unregistered().positional(pd).run();
3050     po::store( parsed, vm);
3051     po::notify(vm);
3052     ceph_option_strings = po::collect_unrecognized(parsed.options,
3053                                                    po::include_positional);
3054   } catch(po::error &e) {
3055     std::cerr << e.what() << std::endl;
3056     return 1;
3057   }
3058
3059   if (vm.count("help")) {
3060     usage(desc);
3061     return 1;
3062   }
3063
3064   debug = (vm.count("debug") > 0);
3065
3066   force = (vm.count("force") > 0);
3067
3068   if (vm.count("namespace"))
3069     nspace = argnspace;
3070
3071   dry_run = (vm.count("dry-run") > 0);
3072
3073   osflagbits_t flags = 0;
3074   if (dry_run || vm.count("skip-journal-replay"))
3075     flags |= SKIP_JOURNAL_REPLAY;
3076   if (vm.count("skip-mount-omap"))
3077     flags |= SKIP_MOUNT_OMAP;
3078   if (op == "update-mon-db")
3079     flags |= SKIP_JOURNAL_REPLAY;
3080
3081   head = (vm.count("head") > 0);
3082
3083   vector<const char *> ceph_options;
3084   env_to_vec(ceph_options);
3085   ceph_options.reserve(ceph_options.size() + ceph_option_strings.size());
3086   for (vector<string>::iterator i = ceph_option_strings.begin();
3087        i != ceph_option_strings.end();
3088        ++i) {
3089     ceph_options.push_back(i->c_str());
3090   }
3091
3092   char fn[PATH_MAX];
3093   snprintf(fn, sizeof(fn), "%s/type", dpath.c_str());
3094   int fd = ::open(fn, O_RDONLY);
3095   if (fd >= 0) {
3096     bufferlist bl;
3097     bl.read_fd(fd, 64);
3098     if (bl.length()) {
3099       string dp_type = string(bl.c_str(), bl.length() - 1);  // drop \n
3100       if (vm.count("type") && dp_type != "" && type != dp_type)
3101         cerr << "WARNING: Ignoring type \"" << type << "\" - found data-path type \""
3102              << dp_type << "\"" << std::endl;
3103       type = dp_type;
3104       //cout << "object store type is " << type << std::endl;
3105     }
3106     ::close(fd);
3107   }
3108   if (!vm.count("type") && type == "") {
3109     type = "filestore";
3110   }
3111   if (!vm.count("data-path") &&
3112      op != "dump-import" &&
3113      !(op == "dump-journal" && type == "filestore")) {
3114     cerr << "Must provide --data-path" << std::endl;
3115     usage(desc);
3116     return 1;
3117   }
3118   if (type == "filestore" && !vm.count("journal-path")) {
3119     jpath = dpath + "/journal";
3120   }
3121   if (!vm.count("op") && !vm.count("object")) {
3122     cerr << "Must provide --op or object command..." << std::endl;
3123     usage(desc);
3124     return 1;
3125   }
3126   if (op != "list" &&
3127       vm.count("op") && vm.count("object")) {
3128     cerr << "Can't specify both --op and object command syntax" << std::endl;
3129     usage(desc);
3130     return 1;
3131   }
3132   if (op == "apply-layout-settings" && !(vm.count("pool") ^ vm.count("pgid"))) {
3133     cerr << "apply-layout-settings requires either --pool or --pgid"
3134          << std::endl;
3135     usage(desc);
3136     return 1;
3137   }
3138   if (op != "list" && vm.count("object") && !vm.count("objcmd")) {
3139     cerr << "Invalid syntax, missing command" << std::endl;
3140     usage(desc);
3141     return 1;
3142   }
3143   if (op == "fuse" && mountpoint.length() == 0) {
3144     cerr << "Missing fuse mountpoint" << std::endl;
3145     usage(desc);
3146     return 1;
3147   }
3148   outistty = isatty(STDOUT_FILENO);
3149
3150   file_fd = fd_none;
3151   if ((op == "export" || op == "export-remove" || op == "get-osdmap" || op == "get-inc-osdmap") && !dry_run) {
3152     if (!vm.count("file") || file == "-") {
3153       if (outistty) {
3154         cerr << "stdout is a tty and no --file filename specified" << std::endl;
3155         return 1;
3156       }
3157       file_fd = STDOUT_FILENO;
3158     } else {
3159       file_fd = open(file.c_str(), O_WRONLY|O_CREAT|O_TRUNC, 0666);
3160     }
3161   } else if (op == "import" || op == "dump-import" || op == "set-osdmap" || op == "set-inc-osdmap") {
3162     if (!vm.count("file") || file == "-") {
3163       if (isatty(STDIN_FILENO)) {
3164         cerr << "stdin is a tty and no --file filename specified" << std::endl;
3165         return 1;
3166       }
3167       file_fd = STDIN_FILENO;
3168     } else {
3169       file_fd = open(file.c_str(), O_RDONLY);
3170     }
3171   }
3172
3173   ObjectStoreTool tool = ObjectStoreTool(file_fd, dry_run);
3174
3175   if (vm.count("file") && file_fd == fd_none && !dry_run) {
3176     cerr << "--file option only applies to import, dump-import, export, export-remove, "
3177          << "get-osdmap, set-osdmap, get-inc-osdmap or set-inc-osdmap" << std::endl;
3178     return 1;
3179   }
3180
3181   if (file_fd != fd_none && file_fd < 0) {
3182     string err = string("file: ") + file;
3183     perror(err.c_str());
3184     return 1;
3185   }
3186
3187   auto cct = global_init(
3188     NULL, ceph_options, CEPH_ENTITY_TYPE_OSD,
3189     CODE_ENVIRONMENT_UTILITY_NODOUT, 0);
3190     //CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
3191   common_init_finish(g_ceph_context);
3192   g_conf = g_ceph_context->_conf;
3193   if (debug) {
3194     g_conf->set_val_or_die("log_to_stderr", "true");
3195     g_conf->set_val_or_die("err_to_stderr", "true");
3196   }
3197   g_conf->apply_changes(NULL);
3198
3199   // Special list handling.  Treating pretty_format as human readable,
3200   // with one object per line and not an enclosing array.
3201   human_readable = ends_with(format, "-pretty");
3202   if ((op == "list" || op == "meta-list") && human_readable) {
3203     // Remove -pretty from end of format which we know is there
3204     format = format.substr(0, format.size() - strlen("-pretty"));
3205   }
3206
3207   formatter = Formatter::create(format);
3208   if (formatter == NULL) {
3209     cerr << "unrecognized format: " << format << std::endl;
3210     return 1;
3211   }
3212
3213   // Special handling for filestore journal, so we can dump it without mounting
3214   if (op == "dump-journal" && type == "filestore") {
3215     int ret = mydump_journal(formatter, jpath, g_conf->journal_dio);
3216     if (ret < 0) {
3217       cerr << "journal-path: " << jpath << ": "
3218            << cpp_strerror(ret) << std::endl;
3219       return 1;
3220     }
3221     formatter->flush(cout);
3222     return 0;
3223   }
3224
3225   if (op == "dump-import") {
3226     int ret = tool.dump_import(formatter);
3227     if (ret < 0) {
3228       cerr << "dump-import: "
3229            << cpp_strerror(ret) << std::endl;
3230       return 1;
3231     }
3232     return 0;
3233   }
3234
3235   //Verify that data-path really exists
3236   struct stat st;
3237   if (::stat(dpath.c_str(), &st) == -1) {
3238     string err = string("data-path: ") + dpath;
3239     perror(err.c_str());
3240     return 1;
3241   }
3242
3243   if (pgidstr.length() && !pgid.parse(pgidstr.c_str())) {
3244     cerr << "Invalid pgid '" << pgidstr << "' specified" << std::endl;
3245     return 1;
3246   }
3247
3248   //Verify that the journal-path really exists
3249   if (type == "filestore") {
3250     if (::stat(jpath.c_str(), &st) == -1) {
3251       string err = string("journal-path: ") + jpath;
3252       perror(err.c_str());
3253       return 1;
3254     }
3255     if (S_ISDIR(st.st_mode)) {
3256       cerr << "journal-path: " << jpath << ": "
3257            << cpp_strerror(EISDIR) << std::endl;
3258       return 1;
3259     }
3260   }
3261
3262   ObjectStore *fs = ObjectStore::create(g_ceph_context, type, dpath, jpath, flags);
3263   if (fs == NULL) {
3264     cerr << "Unable to create store of type " << type << std::endl;
3265     return 1;
3266   }
3267
3268   if (op == "fsck" || op == "fsck-deep") {
3269     int r = fs->fsck(op == "fsck-deep");
3270     if (r < 0) {
3271       cerr << "fsck failed: " << cpp_strerror(r) << std::endl;
3272       return 1;
3273     }
3274     if (r > 0) {
3275       cerr << "fsck found " << r << " errors" << std::endl;
3276       return 1;
3277     }
3278     cout << "fsck found no errors" << std::endl;
3279     return 0;
3280   }
3281   if (op == "repair" || op == "repair-deep") {
3282     int r = fs->repair(op == "repair-deep");
3283     if (r < 0) {
3284       cerr << "repair failed: " << cpp_strerror(r) << std::endl;
3285       return 1;
3286     }
3287     if (r > 0) {
3288       cerr << "repair found " << r << " errors" << std::endl;
3289       return 1;
3290     }
3291     cout << "repair found no errors" << std::endl;
3292     return 0;
3293   }
3294   if (op == "mkfs") {
3295     if (fsid.length()) {
3296       uuid_d f;
3297       bool r = f.parse(fsid.c_str());
3298       if (!r) {
3299         cerr << "failed to parse uuid '" << fsid << "'" << std::endl;
3300         return 1;
3301       }
3302       fs->set_fsid(f);
3303     }
3304     int r = fs->mkfs();
3305     if (r < 0) {
3306       cerr << "mkfs failed: " << cpp_strerror(r) << std::endl;
3307       return 1;
3308     }
3309     return 0;
3310   }
3311   if (op == "dup") {
3312     string target_type;
3313     char fn[PATH_MAX];
3314     snprintf(fn, sizeof(fn), "%s/type", target_data_path.c_str());
3315     int fd = ::open(fn, O_RDONLY);
3316     if (fd < 0) {
3317       cerr << "Unable to open " << target_data_path << "/type" << std::endl;
3318       exit(1);
3319     }
3320     bufferlist bl;
3321     bl.read_fd(fd, 64);
3322     if (bl.length()) {
3323       target_type = string(bl.c_str(), bl.length() - 1);  // drop \n
3324     }
3325     ::close(fd);
3326     ObjectStore *targetfs = ObjectStore::create(
3327       g_ceph_context, target_type,
3328       target_data_path, "", 0);
3329     if (targetfs == NULL) {
3330       cerr << "Unable to open store of type " << target_type << std::endl;
3331       return 1;
3332     }
3333     int r = dup(dpath, fs, target_data_path, targetfs);
3334     if (r < 0) {
3335       cerr << "dup failed: " << cpp_strerror(r) << std::endl;
3336       return 1;
3337     }
3338     return 0;
3339   }
3340
3341   ObjectStore::Sequencer *osr = new ObjectStore::Sequencer(__func__);
3342   int ret = fs->mount();
3343   if (ret < 0) {
3344     if (ret == -EBUSY) {
3345       cerr << "OSD has the store locked" << std::endl;
3346     } else {
3347       cerr << "Mount failed with '" << cpp_strerror(ret) << "'" << std::endl;
3348     }
3349     return 1;
3350   }
3351
3352   if (op == "fuse") {
3353 #ifdef HAVE_LIBFUSE
3354     FuseStore fuse(fs, mountpoint);
3355     cout << "mounting fuse at " << mountpoint << " ..." << std::endl;
3356     int r = fuse.main();
3357     if (r < 0) {
3358       cerr << "failed to mount fuse: " << cpp_strerror(r) << std::endl;
3359       return 1;
3360     }
3361 #else
3362     cerr << "fuse support not enabled" << std::endl;
3363 #endif
3364     return 0;
3365   }
3366
3367   vector<coll_t> ls;
3368   vector<coll_t>::iterator it;
3369   CompatSet supported;
3370
3371 #ifdef INTERNAL_TEST
3372   supported = get_test_compat_set();
3373 #else
3374   supported = OSD::get_osd_compat_set();
3375 #endif
3376
3377   bufferlist bl;
3378   OSDSuperblock superblock;
3379   bufferlist::iterator p;
3380   ret = fs->read(coll_t::meta(), OSD_SUPERBLOCK_GOBJECT, 0, 0, bl);
3381   if (ret < 0) {
3382     cerr << "Failure to read OSD superblock: " << cpp_strerror(ret) << std::endl;
3383     goto out;
3384   }
3385
3386   p = bl.begin();
3387   ::decode(superblock, p);
3388
3389   if (debug) {
3390     cerr << "Cluster fsid=" << superblock.cluster_fsid << std::endl;
3391   }
3392
3393   if (debug) {
3394     cerr << "Supported features: " << supported << std::endl;
3395     cerr << "On-disk features: " << superblock.compat_features << std::endl;
3396   }
3397   if (supported.compare(superblock.compat_features) == -1) {
3398     CompatSet unsupported = supported.unsupported(superblock.compat_features);
3399     cerr << "On-disk OSD incompatible features set "
3400       << unsupported << std::endl;
3401     ret = -EINVAL;
3402     goto out;
3403   }
3404
3405   if (op == "apply-layout-settings") {
3406     int target_level = 0;
3407     if (vm.count("arg1") && isdigit(arg1[0])) {
3408       target_level = atoi(arg1.c_str());
3409     }
3410     ret = apply_layout_settings(fs, superblock, pool, pgid, dry_run, target_level);
3411     goto out;
3412   }
3413
3414   if (op != "list" && vm.count("object")) {
3415     // Special case: Create pgmeta_oid if empty string specified
3416     // This can't conflict with any actual object names.
3417     if (object == "") {
3418       ghobj = pgid.make_pgmeta_oid();
3419     } else {
3420     json_spirit::Value v;
3421     try {
3422       if (!json_spirit::read(object, v) ||
3423           (v.type() != json_spirit::array_type && v.type() != json_spirit::obj_type)) {
3424         // Special: Need head/snapdir so set even if user didn't specify
3425         if (vm.count("objcmd") && (objcmd == "remove-clone-metadata"))
3426           head = true;
3427         lookup_ghobject lookup(object, nspace, head);
3428         if (pgidstr.length())
3429           ret = action_on_all_objects_in_exact_pg(fs, coll_t(pgid), lookup, debug);
3430         else
3431           ret = action_on_all_objects(fs, lookup, debug);
3432         if (ret) {
3433           throw std::runtime_error("Internal error");
3434         } else {
3435           if (lookup.size() != 1) {
3436             stringstream ss;
3437             if (lookup.size() == 0)
3438               ss << "No object id '" << object << "' found or invalid JSON specified";
3439             else
3440               ss << "Found " << lookup.size() << " objects with id '" << object
3441                  << "', please use a JSON spec from --op list instead";
3442             throw std::runtime_error(ss.str());
3443           }
3444           pair<coll_t, ghobject_t> found = lookup.pop();
3445           pgidstr = found.first.to_str();
3446           pgid.parse(pgidstr.c_str());
3447           ghobj = found.second;
3448         }
3449       } else {
3450         stringstream ss;
3451         if (pgidstr.length() == 0 && v.type() != json_spirit::array_type) {
3452           ss << "Without --pgid the object '" << object
3453              << "' must be a JSON array";
3454           throw std::runtime_error(ss.str());
3455         }
3456         if (v.type() == json_spirit::array_type) {
3457           json_spirit::Array array = v.get_array();
3458           if (array.size() != 2) {
3459             ss << "Object '" << object
3460                << "' must be a JSON array with 2 elements";
3461             throw std::runtime_error(ss.str());
3462           }
3463           vector<json_spirit::Value>::iterator i = array.begin();
3464           assert(i != array.end());
3465           if (i->type() != json_spirit::str_type) {
3466             ss << "Object '" << object
3467                << "' must be a JSON array with the first element a string";
3468             throw std::runtime_error(ss.str());
3469           }
3470           string object_pgidstr = i->get_str();
3471           if (object_pgidstr != "meta") {
3472             spg_t object_pgid;
3473             object_pgid.parse(object_pgidstr.c_str());
3474             if (pgidstr.length() > 0) {
3475               if (object_pgid != pgid) {
3476                 ss << "object '" << object
3477                    << "' has a pgid different from the --pgid="
3478                    << pgidstr << " option";
3479                 throw std::runtime_error(ss.str());
3480               }
3481             } else {
3482               pgidstr = object_pgidstr;
3483               pgid = object_pgid;
3484             }
3485           } else {
3486             pgidstr = object_pgidstr;
3487           }
3488           ++i;
3489           v = *i;
3490         }
3491         try {
3492           ghobj.decode(v);
3493         } catch (std::runtime_error& e) {
3494           ss << "Decode object JSON error: " << e.what();
3495           throw std::runtime_error(ss.str());
3496         }
3497         if (pgidstr != "meta" && (uint64_t)pgid.pgid.m_pool != (uint64_t)ghobj.hobj.pool) {
3498           cerr << "Object pool and pgid pool don't match" << std::endl;
3499           ret = 1;
3500           goto out;
3501         }
3502       }
3503     } catch (std::runtime_error& e) {
3504       cerr << e.what() << std::endl;
3505       ret = 1;
3506       goto out;
3507     }
3508     }
3509   }
3510
3511   // The ops which require --pgid option are checked here and
3512   // mentioned in the usage for --pgid.
3513   if ((op == "info" || op == "log" || op == "remove" || op == "export"
3514       || op == "export-remove" || op == "rm-past-intervals"
3515       || op == "mark-complete" || op == "trim-pg-log") &&
3516       pgidstr.length() == 0) {
3517     cerr << "Must provide pgid" << std::endl;
3518     usage(desc);
3519     ret = 1;
3520     goto out;
3521   }
3522
3523   if (op == "import") {
3524
3525     try {
3526       ret = tool.do_import(fs, superblock, force, pgidstr, *osr);
3527     }
3528     catch (const buffer::error &e) {
3529       cerr << "do_import threw exception error " << e.what() << std::endl;
3530       ret = -EFAULT;
3531     }
3532     if (ret == -EFAULT) {
3533       cerr << "Corrupt input for import" << std::endl;
3534     }
3535     if (ret == 0)
3536       cout << "Import successful" << std::endl;
3537     goto out;
3538   } else if (op == "dump-journal-mount") {
3539     // Undocumented feature to dump journal with mounted fs
3540     // This doesn't support the format option, but it uses the
3541     // ObjectStore::dump_journal() and mounts to get replay to run.
3542     ret = fs->dump_journal(cout);
3543     if (ret) {
3544       if (ret == -EOPNOTSUPP) {
3545         cerr << "Object store type \"" << type << "\" doesn't support journal dump" << std::endl;
3546       } else {
3547         cerr << "Journal dump failed with error " << cpp_strerror(ret) << std::endl;
3548       }
3549     }
3550     goto out;
3551   } else if (op == "get-osdmap") {
3552     bufferlist bl;
3553     OSDMap osdmap;
3554     if (epoch == 0) {
3555       epoch = superblock.current_epoch;
3556     }
3557     ret = get_osdmap(fs, epoch, osdmap, bl);
3558     if (ret) {
3559       cerr << "Failed to get osdmap#" << epoch << ": "
3560            << cpp_strerror(ret) << std::endl;
3561       goto out;
3562     }
3563     ret = bl.write_fd(file_fd);
3564     if (ret) {
3565       cerr << "Failed to write to " << file << ": " << cpp_strerror(ret) << std::endl;
3566     } else {
3567       cout << "osdmap#" << epoch << " exported." << std::endl;
3568     }
3569     goto out;
3570   } else if (op == "set-osdmap") {
3571     bufferlist bl;
3572     ret = get_fd_data(file_fd, bl);
3573     if (ret < 0) {
3574       cerr << "Failed to read osdmap " << cpp_strerror(ret) << std::endl;
3575     } else {
3576       ret = set_osdmap(fs, epoch, bl, force, *osr);
3577     }
3578     goto out;
3579   } else if (op == "get-inc-osdmap") {
3580     bufferlist bl;
3581     if (epoch == 0) {
3582       epoch = superblock.current_epoch;
3583     }
3584     ret = get_inc_osdmap(fs, epoch, bl);
3585     if (ret < 0) {
3586       cerr << "Failed to get incremental osdmap# " << epoch << ": "
3587            << cpp_strerror(ret) << std::endl;
3588       goto out;
3589     }
3590     ret = bl.write_fd(file_fd);
3591     if (ret) {
3592       cerr << "Failed to write to " << file << ": " << cpp_strerror(ret) << std::endl;
3593     } else {
3594       cout << "inc-osdmap#" << epoch << " exported." << std::endl;
3595     }
3596     goto out;
3597   } else if (op == "set-inc-osdmap") {
3598     bufferlist bl;
3599     ret = get_fd_data(file_fd, bl);
3600     if (ret < 0) {
3601       cerr << "Failed to read incremental osdmap  " << cpp_strerror(ret) << std::endl;
3602       goto out;
3603     } else {
3604       ret = set_inc_osdmap(fs, epoch, bl, force, *osr);
3605     }
3606     goto out;
3607   } else if (op == "update-mon-db") {
3608     if (!vm.count("mon-store-path")) {
3609       cerr << "Please specify the path to monitor db to update" << std::endl;
3610       ret = -EINVAL;
3611     } else {
3612       ret = update_mon_db(*fs, superblock, dpath + "/keyring", mon_store_path);
3613     }
3614     goto out;
3615   }
3616
3617   log_oid = OSD::make_pg_log_oid(pgid);
3618   biginfo_oid = OSD::make_pg_biginfo_oid(pgid);
3619
3620   if (op == "remove") {
3621     if (!force && !dry_run) {
3622       cerr << "Please use export-remove or you must use --force option" << std::endl;
3623       ret = -EINVAL;
3624       goto out;
3625     }
3626     ret = initiate_new_remove_pg(fs, pgid, *osr);
3627     if (ret < 0) {
3628       cerr << "PG '" << pgid << "' not found" << std::endl;
3629       goto out;
3630     }
3631     cout << "Remove successful" << std::endl;
3632     goto out;
3633   }
3634
3635   if (op == "fix-lost") {
3636     boost::scoped_ptr<action_on_object_t> action;
3637     action.reset(new do_fix_lost(osr));
3638     if (pgidstr.length())
3639       ret = action_on_all_objects_in_exact_pg(fs, coll_t(pgid), *action, debug);
3640     else
3641       ret = action_on_all_objects(fs, *action, debug);
3642     goto out;
3643   }
3644
3645   if (op == "list") {
3646     ret = do_list(fs, pgidstr, object, nspace, formatter, debug,
3647                   human_readable, head);
3648     if (ret < 0) {
3649       cerr << "do_list failed: " << cpp_strerror(ret) << std::endl;
3650     }
3651     goto out;
3652   }
3653
3654   if (op == "dump-super") {
3655     formatter->open_object_section("superblock");
3656     superblock.dump(formatter);
3657     formatter->close_section();
3658     formatter->flush(cout);
3659     cout << std::endl;
3660     goto out;
3661   }
3662
3663   if (op == "meta-list") {
3664     ret = do_meta(fs, object, formatter, debug, human_readable);
3665     if (ret < 0) {
3666       cerr << "do_meta failed: " << cpp_strerror(ret) << std::endl;
3667     }
3668     goto out;
3669   }
3670
3671   ret = fs->list_collections(ls);
3672   if (ret < 0) {
3673     cerr << "failed to list pgs: " << cpp_strerror(ret) << std::endl;
3674     goto out;
3675   }
3676
3677   if (debug && op == "list-pgs")
3678     cout << "Performing list-pgs operation" << std::endl;
3679
3680   // Find pg
3681   for (it = ls.begin(); it != ls.end(); ++it) {
3682     spg_t tmppgid;
3683
3684     if (pgidstr == "meta") {
3685       if (it->to_str() == "meta")
3686         break;
3687       else
3688         continue;
3689     }
3690
3691     if (!it->is_pg(&tmppgid)) {
3692       continue;
3693     }
3694
3695     if (it->is_temp(&tmppgid)) {
3696       continue;
3697     }
3698
3699     if (op != "list-pgs" && tmppgid != pgid) {
3700       continue;
3701     }
3702
3703     if (op != "list-pgs") {
3704       //Found!
3705       break;
3706     }
3707
3708     cout << tmppgid << std::endl;
3709   }
3710
3711   if (op == "list-pgs") {
3712     ret = 0;
3713     goto out;
3714   }
3715
3716   // If not an object command nor any of the ops handled below, then output this usage
3717   // before complaining about a bad pgid
3718   if (!vm.count("objcmd") && op != "export" && op != "export-remove" && op != "info" && op != "log" && op != "rm-past-intervals" && op != "mark-complete" && op != "trim-pg-log") {
3719     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, "
3720       "get-osdmap, set-osdmap, get-inc-osdmap, set-inc-osdmap, mark-complete, dump-import, trim-pg-log)"
3721          << std::endl;
3722     usage(desc);
3723     ret = 1;
3724     goto out;
3725   }
3726   epoch_t map_epoch;
3727 // The following code for export, info, log require omap or !skip-mount-omap
3728   if (it != ls.end()) {
3729
3730     coll_t coll = *it;
3731
3732     if (vm.count("objcmd")) {
3733       ret = 0;
3734       if (objcmd == "remove" || objcmd == "removeall") {
3735         bool all = (objcmd == "removeall");
3736         ret = do_remove_object(fs, coll, ghobj, all, force, *osr);
3737         goto out;
3738       } else if (objcmd == "list-attrs") {
3739         ret = do_list_attrs(fs, coll, ghobj);
3740         goto out;
3741       } else if (objcmd == "list-omap") {
3742         ret = do_list_omap(fs, coll, ghobj);
3743         goto out;
3744       } else if (objcmd == "get-bytes" || objcmd == "set-bytes") {
3745         if (objcmd == "get-bytes") {
3746           int fd;
3747           if (vm.count("arg1") == 0 || arg1 == "-") {
3748             fd = STDOUT_FILENO;
3749           } else {
3750             fd = open(arg1.c_str(), O_WRONLY|O_TRUNC|O_CREAT|O_EXCL|O_LARGEFILE, 0666);
3751             if (fd == -1) {
3752               cerr << "open " << arg1 << " " << cpp_strerror(errno) << std::endl;
3753               ret = 1;
3754               goto out;
3755             }
3756           }
3757           ret = do_get_bytes(fs, coll, ghobj, fd);
3758           if (fd != STDOUT_FILENO)
3759             close(fd);
3760         } else {
3761           int fd;
3762           if (vm.count("arg1") == 0 || arg1 == "-") {
3763             // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3764             if (isatty(STDIN_FILENO)) {
3765                 cerr << "stdin is a tty and no file specified" << std::endl;
3766                 ret = 1;
3767                 goto out;
3768             }
3769             fd = STDIN_FILENO;
3770           } else {
3771             fd = open(arg1.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3772             if (fd == -1) {
3773               cerr << "open " << arg1 << " " << cpp_strerror(errno) << std::endl;
3774               ret = 1;
3775               goto out;
3776             }
3777           }
3778           ret = do_set_bytes(fs, coll, ghobj, fd, *osr);
3779           if (fd != STDIN_FILENO)
3780             close(fd);
3781         }
3782         goto out;
3783       } else if (objcmd == "get-attr") {
3784         if (vm.count("arg1") == 0) {
3785           usage(desc);
3786           ret = 1;
3787           goto out;
3788         }
3789         ret = do_get_attr(fs, coll, ghobj, arg1);
3790         goto out;
3791       } else if (objcmd == "set-attr") {
3792         if (vm.count("arg1") == 0) {
3793           usage(desc);
3794           ret = 1;
3795         }
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_attr(fs, coll, ghobj, arg1, fd, *osr);
3815         if (fd != STDIN_FILENO)
3816           close(fd);
3817         goto out;
3818       } else if (objcmd == "rm-attr") {
3819         if (vm.count("arg1") == 0) {
3820           usage(desc);
3821           ret = 1;
3822           goto out;
3823         }
3824         ret = do_rm_attr(fs, coll, ghobj, arg1, *osr);
3825         goto out;
3826       } else if (objcmd == "get-omap") {
3827         if (vm.count("arg1") == 0) {
3828           usage(desc);
3829           ret = 1;
3830           goto out;
3831         }
3832         ret = do_get_omap(fs, coll, ghobj, arg1);
3833         goto out;
3834       } else if (objcmd == "set-omap") {
3835         if (vm.count("arg1") == 0) {
3836           usage(desc);
3837           ret = 1;
3838           goto out;
3839         }
3840         int fd;
3841         if (vm.count("arg2") == 0 || arg2 == "-") {
3842           // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3843           if (isatty(STDIN_FILENO)) {
3844             cerr << "stdin is a tty and no file specified" << std::endl;
3845             ret = 1;
3846             goto out;
3847           }
3848           fd = STDIN_FILENO;
3849         } else {
3850           fd = open(arg2.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3851           if (fd == -1) {
3852             cerr << "open " << arg2 << " " << cpp_strerror(errno) << std::endl;
3853             ret = 1;
3854             goto out;
3855           }
3856         }
3857         ret = do_set_omap(fs, coll, ghobj, arg1, fd, *osr);
3858         if (fd != STDIN_FILENO)
3859           close(fd);
3860         goto out;
3861       } else if (objcmd == "rm-omap") {
3862         if (vm.count("arg1") == 0) {
3863           usage(desc);
3864           ret = 1;
3865           goto out;
3866         }
3867         ret = do_rm_omap(fs, coll, ghobj, arg1, *osr);
3868         goto out;
3869       } else if (objcmd == "get-omaphdr") {
3870         if (vm.count("arg1")) {
3871           usage(desc);
3872           ret = 1;
3873           goto out;
3874         }
3875         ret = do_get_omaphdr(fs, coll, ghobj);
3876         goto out;
3877       } else if (objcmd == "set-omaphdr") {
3878         // Extra arg
3879         if (vm.count("arg2")) {
3880           usage(desc);
3881           ret = 1;
3882           goto out;
3883         }
3884         int fd;
3885         if (vm.count("arg1") == 0 || arg1 == "-") {
3886           // Since read_fd() doesn't handle ^D from a tty stdin, don't allow it.
3887           if (isatty(STDIN_FILENO)) {
3888             cerr << "stdin is a tty and no file specified" << std::endl;
3889             ret = 1;
3890             goto out;
3891           }
3892           fd = STDIN_FILENO;
3893         } else {
3894           fd = open(arg1.c_str(), O_RDONLY|O_LARGEFILE, 0666);
3895           if (fd == -1) {
3896             cerr << "open " << arg1 << " " << cpp_strerror(errno) << std::endl;
3897             ret = 1;
3898             goto out;
3899           }
3900         }
3901         ret = do_set_omaphdr(fs, coll, ghobj, fd, *osr);
3902         if (fd != STDIN_FILENO)
3903           close(fd);
3904         goto out;
3905       } else if (objcmd == "dump") {
3906         // There should not be any other arguments
3907         if (vm.count("arg1") || vm.count("arg2")) {
3908           usage(desc);
3909           ret = 1;
3910           goto out;
3911         }
3912         ret = print_obj_info(fs, coll, ghobj, formatter);
3913         goto out;
3914       } else if (objcmd == "corrupt-info") {   // Undocumented testing feature
3915         // There should not be any other arguments
3916         if (vm.count("arg1") || vm.count("arg2")) {
3917           usage(desc);
3918           ret = 1;
3919           goto out;
3920         }
3921         ret = corrupt_info(fs, coll, ghobj, formatter, *osr);
3922         goto out;
3923       } else if (objcmd == "set-size" || objcmd == "corrupt-size") {
3924         // Undocumented testing feature
3925         bool corrupt = (objcmd == "corrupt-size");
3926         // Extra arg
3927         if (vm.count("arg1") == 0 || vm.count("arg2")) {
3928           usage(desc);
3929           ret = 1;
3930           goto out;
3931         }
3932         if (arg1.length() == 0 || !isdigit(arg1.c_str()[0])) {
3933           cerr << "Invalid size '" << arg1 << "' specified" << std::endl;
3934           ret = 1;
3935           goto out;
3936         }
3937         uint64_t size = atoll(arg1.c_str());
3938         ret = set_size(fs, coll, ghobj, size, formatter, *osr, corrupt);
3939         goto out;
3940       } else if (objcmd == "clear-snapset") {
3941         // UNDOCUMENTED: For testing zap SnapSet
3942         // IGNORE extra args since not in usage anyway
3943         if (!ghobj.hobj.has_snapset()) {
3944           cerr << "'" << objcmd << "' requires a head or snapdir object" << std::endl;
3945           ret = 1;
3946           goto out;
3947         }
3948         ret = clear_snapset(fs, coll, ghobj, arg1, *osr);
3949         goto out;
3950       } else if (objcmd == "remove-clone-metadata") {
3951         // Extra arg
3952         if (vm.count("arg1") == 0 || vm.count("arg2")) {
3953           usage(desc);
3954           ret = 1;
3955           goto out;
3956         }
3957         if (!ghobj.hobj.has_snapset()) {
3958           cerr << "'" << objcmd << "' requires a head or snapdir object" << std::endl;
3959           ret = 1;
3960           goto out;
3961         }
3962         if (arg1.length() == 0 || !isdigit(arg1.c_str()[0])) {
3963           cerr << "Invalid cloneid '" << arg1 << "' specified" << std::endl;
3964           ret = 1;
3965           goto out;
3966         }
3967         snapid_t cloneid = atoi(arg1.c_str());
3968         ret = remove_clone(fs, coll, ghobj, cloneid, force, *osr);
3969         goto out;
3970       }
3971       cerr << "Unknown object command '" << objcmd << "'" << std::endl;
3972       usage(desc);
3973       ret = 1;
3974       goto out;
3975     }
3976
3977     bufferlist bl;
3978     map_epoch = 0;
3979     ret = PG::peek_map_epoch(fs, pgid, &map_epoch, &bl);
3980     if (ret < 0)
3981       cerr << "peek_map_epoch reports error" << std::endl;
3982     if (debug)
3983       cerr << "map_epoch " << map_epoch << std::endl;
3984
3985     pg_info_t info(pgid);
3986     PastIntervals past_intervals;
3987     __u8 struct_ver;
3988     ret = PG::read_info(fs, pgid, coll, bl, info, past_intervals,
3989                       struct_ver);
3990     if (ret < 0) {
3991       cerr << "read_info error " << cpp_strerror(ret) << std::endl;
3992       goto out;
3993     }
3994     if (struct_ver < PG::compat_struct_v) {
3995       cerr << "PG is too old to upgrade, use older Ceph version" << std::endl;
3996       ret = -EFAULT;
3997       goto out;
3998     }
3999     if (debug)
4000       cerr << "struct_v " << (int)struct_ver << std::endl;
4001
4002     if (op == "export" || op == "export-remove") {
4003       ret = tool.do_export(fs, coll, pgid, info, map_epoch, struct_ver, superblock, past_intervals);
4004       if (ret == 0) {
4005         cerr << "Export successful" << std::endl;
4006         if (op == "export-remove") {
4007           ret = initiate_new_remove_pg(fs, pgid, *osr);
4008           // Export succeeded, so pgid is there
4009           assert(ret == 0);
4010           cerr << "Remove successful" << std::endl;
4011         }
4012       }
4013     } else if (op == "info") {
4014       formatter->open_object_section("info");
4015       info.dump(formatter);
4016       formatter->close_section();
4017       formatter->flush(cout);
4018       cout << std::endl;
4019     } else if (op == "log") {
4020       PGLog::IndexedLog log;
4021       pg_missing_t missing;
4022       ret = get_log(fs, struct_ver, coll, pgid, info, log, missing);
4023       if (ret < 0)
4024           goto out;
4025
4026       dump_log(formatter, cout, log, missing);
4027     } else if (op == "rm-past-intervals") {
4028       ObjectStore::Transaction tran;
4029       ObjectStore::Transaction *t = &tran;
4030
4031       if (struct_ver < PG::compat_struct_v) {
4032         cerr << "Can't remove past-intervals, version mismatch " << (int)struct_ver
4033           << " (pg)  < compat " << (int)PG::compat_struct_v << " (tool)"
4034           << std::endl;
4035         ret = -EFAULT;
4036         goto out;
4037       }
4038
4039       cout << "Remove past-intervals " << past_intervals << std::endl;
4040
4041       past_intervals.clear();
4042       if (dry_run) {
4043         ret = 0;
4044         goto out;
4045       }
4046       ret = write_info(*t, map_epoch, info, past_intervals);
4047
4048       if (ret == 0) {
4049         fs->apply_transaction(osr, std::move(*t));
4050         cout << "Removal succeeded" << std::endl;
4051       }
4052     } else if (op == "mark-complete") {
4053       ObjectStore::Transaction tran;
4054       ObjectStore::Transaction *t = &tran;
4055
4056       if (struct_ver < PG::compat_struct_v) {
4057         cerr << "Can't mark-complete, version mismatch " << (int)struct_ver
4058              << " (pg)  < compat " << (int)PG::compat_struct_v << " (tool)"
4059              << std::endl;
4060         ret = 1;
4061         goto out;
4062       }
4063
4064       cout << "Marking complete " << std::endl;
4065
4066       info.last_update = eversion_t(superblock.current_epoch, info.last_update.version + 1);
4067       info.last_backfill = hobject_t::get_max();
4068       info.last_epoch_started = superblock.current_epoch;
4069       info.history.last_epoch_started = superblock.current_epoch;
4070       info.history.last_epoch_clean = superblock.current_epoch;
4071       past_intervals.clear();
4072
4073       if (!dry_run) {
4074         ret = write_info(*t, map_epoch, info, past_intervals);
4075         if (ret != 0)
4076           goto out;
4077         fs->apply_transaction(osr, std::move(*t));
4078       }
4079       cout << "Marking complete succeeded" << std::endl;
4080     } else if (op == "trim-pg-log") {
4081       ret = do_trim_pg_log(fs, coll, info, pgid, *osr,
4082                            map_epoch, past_intervals);
4083       if (ret < 0) {
4084         cerr << "Error trimming pg log: " << cpp_strerror(ret) << std::endl;
4085         goto out;
4086       }
4087       cout << "Finished trimming pg log" << std::endl;
4088       goto out;
4089     } else {
4090       assert(!"Should have already checked for valid --op");
4091     }
4092   } else {
4093     cerr << "PG '" << pgid << "' not found" << std::endl;
4094     ret = -ENOENT;
4095   }
4096
4097 out:
4098   int r = fs->umount();
4099   delete osr;
4100   if (r < 0) {
4101     cerr << "umount failed: " << cpp_strerror(r) << std::endl;
4102     // If no previous error, then use umount() error
4103     if (ret == 0)
4104       ret = r;
4105   }
4106
4107   if (dry_run) {
4108     // Export output can go to stdout, so put this message on stderr
4109     if (op == "export")
4110       cerr << "dry-run: Nothing changed" << std::endl;
4111     else
4112       cout << "dry-run: Nothing changed" << std::endl;
4113   }
4114
4115   if (ret < 0)
4116     ret = 1;
4117   return ret;
4118 }