[ceph.git] / ceph / src / tools / rados / RadosImport.cc

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2015 Red Hat
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation.  See file COPYING.
 *
 */


#include "common/errno.h"

#include "osd/PGLog.h"
#include "RadosImport.h"

#define dout_context g_ceph_context
#define dout_subsys ceph_subsys_rados

using std::cerr;
using std::cout;
using std::map;
using std::string;

int RadosImport::import(std::string pool, bool no_overwrite)
{
  librados::IoCtx ioctx;
  librados::Rados cluster;

  char *id = getenv("CEPH_CLIENT_ID");
  if (id) cerr << "Client id is: " << id << std::endl;
  int ret = cluster.init(id);
  if (ret) {
    cerr << "Error " << ret << " in cluster.init" << std::endl;
    return ret;
  }
  ret = cluster.conf_read_file(NULL);
  if (ret) {
    cerr << "Error " << ret << " in cluster.conf_read_file" << std::endl;
    return ret;
  }
  ret = cluster.conf_parse_env(NULL);
  if (ret) {
    cerr << "Error " << ret << " in cluster.conf_read_env" << std::endl;
    return ret;
  }
  ret = cluster.connect();
  if (ret) {
    cerr << "Error " << ret << " in cluster.connect" << std::endl;
    return ret;
  }

  ret = cluster.ioctx_create(pool.c_str(), ioctx);
  if (ret < 0) {
    cerr << "ioctx_create " << pool << " failed with " << ret << std::endl;
    return ret;
  }

  return import(ioctx, no_overwrite);
}

int RadosImport::import(librados::IoCtx &io_ctx, bool no_overwrite)
{
  bufferlist ebl;
  pg_info_t info;
  PGLog::IndexedLog log;

  int ret = read_super();
  if (ret)
    return ret;

  if (sh.magic != super_header::super_magic) {
    cerr << "Invalid magic number: 0x"
      << std::hex << sh.magic << " vs. 0x" << super_header::super_magic
      << std::dec << std::endl;
    return -EFAULT;
  }

  if (sh.version > super_header::super_ver) {
    cerr << "Can't handle export format version=" << sh.version << std::endl;
    return -EINVAL;
  }

  //First section must be TYPE_PG_BEGIN
  sectiontype_t type;
  ret = read_section(&type, &ebl);
  if (ret)
    return ret;

  bool pool_mode = false;
  if (type == TYPE_POOL_BEGIN) {
    pool_mode = true;
    cout << "Importing pool" << std::endl;
  } else if (type == TYPE_PG_BEGIN) {
    auto ebliter = ebl.cbegin();
    pg_begin pgb;
    pgb.decode(ebliter);
    spg_t pgid = pgb.pgid;;
    if (!pgid.is_no_shard()) {
      cerr << "Importing Erasure Coded shard is not supported" << std::endl;
      return -EOPNOTSUPP;
    }
    dout(10) << "Exported features: " << pgb.superblock.compat_features << dendl;
    cout << "Importing from pgid " << pgid << std::endl;
  } else {
    cerr << "Invalid initial section code " << type << std::endl;
    return -EFAULT;
  }

  // XXX: How to check export features?
#if 0
  if (sb.compat_features.compare(pgb.superblock.compat_features) == -1) {
    cerr << "Export has incompatible features set "
      << pgb.superblock.compat_features << std::endl;
    return -EINVAL;
  }
#endif

#if defined(__linux__)
  if (file_fd != STDIN_FILENO)
    posix_fadvise(file_fd, 0, 0, POSIX_FADV_SEQUENTIAL);
#endif

  bool done = false;
  bool found_metadata = false;
  while(!done) {
    ret = read_section(&type, &ebl);
    if (ret)
      return ret;

    //cout << "do_import: Section type " << hex << type << dec << std::endl;
    if (type >= END_OF_TYPES) {
      cout << "Skipping unknown section type" << std::endl;
      continue;
    }
    switch(type) {
    case TYPE_OBJECT_BEGIN:
      ret = get_object_rados(io_ctx, ebl, no_overwrite);
      if (ret) {
        cerr << "Error inserting object: " << ret << std::endl;
        return ret;
      }
      break;
    case TYPE_PG_METADATA:
      dout(10) << "Don't care about the old metadata" << dendl;
      found_metadata = true;
      break;
    case TYPE_PG_END:
      done = true;
      break;
    case TYPE_POOL_END:
      done = true;
      break;
    default:
      return -EFAULT;
    }
  }

  if (!(pool_mode || found_metadata)) {
    cerr << "Missing metadata section!" << std::endl;
  }

#if defined(__linux__)
  if (file_fd != STDIN_FILENO)
    posix_fadvise(file_fd, 0, 0, POSIX_FADV_DONTNEED);
#endif
  return 0;
}

int RadosImport::get_object_rados(librados::IoCtx &ioctx, bufferlist &bl, bool no_overwrite)
{
  auto ebliter = bl.cbegin();
  object_begin ob;
  ob.decode(ebliter);
  map<string,bufferlist>::iterator i;
  bufferlist abl;
  bool skipping;

  data_section ds;
  attr_section as;
  omap_hdr_section oh;
  omap_section os;

  ceph_assert(g_ceph_context);
  if (ob.hoid.hobj.nspace == g_ceph_context->_conf->osd_hit_set_namespace) {
    cout << "Skipping internal object " << ob.hoid << std::endl;
    skip_object(bl);
    return 0;
  }

  if (!ob.hoid.hobj.is_head()) {
    cout << "Skipping non-head for " << ob.hoid << std::endl;
    skip_object(bl);
    return 0;
  }

  ioctx.set_namespace(ob.hoid.hobj.get_namespace());
  ioctx.locator_set_key(ob.hoid.hobj.get_key());

  string msg("Write");
  skipping = false;
  if (dry_run) {
    uint64_t psize;
    time_t pmtime;
    int ret = ioctx.stat(ob.hoid.hobj.oid.name, &psize, &pmtime);
    if (ret == 0) {
      if (no_overwrite)
        // Could set skipping, but dry-run doesn't change anything either
        msg = "Skipping existing";
      else
        msg = "***Overwrite***";
    }
  } else {
    int ret = ioctx.create(ob.hoid.hobj.oid.name, true);
    if (ret && ret != -EEXIST) {
      cerr << "create failed: " << cpp_strerror(ret) << std::endl;
      return ret;
    }
    if (ret == -EEXIST) {
      if (no_overwrite) {
        msg = "Skipping existing";
        skipping = true;
      } else {
        msg = "***Overwrite***";
        ret = ioctx.remove(ob.hoid.hobj.oid.name);
        if (ret < 0) {
          cerr << "remove failed: " << cpp_strerror(ret) << std::endl;
          return ret;
        }
        ret = ioctx.create(ob.hoid.hobj.oid.name, true);
        // If object re-appeared after removal, let's just skip it
        if (ret == -EEXIST) {
          skipping = true;
          msg = "Skipping in-use object";
          ret = 0;
        }
        if (ret < 0) {
          cerr << "create failed: " << cpp_strerror(ret) << std::endl;
          return ret;
        }
      }
    }
  }

  cout << msg << " " << ob.hoid << std::endl;

  bool need_align = false;
  uint64_t alignment = 0;
  if (align) {
    need_align = true;
    alignment = align;
  } else {
    int ret = ioctx.pool_requires_alignment2(&need_align);
    if (ret < 0) {
      cerr << "pool_requires_alignment2 failed: " << cpp_strerror(ret)
        << std::endl;
      return ret;
    }

    if (need_align) {
      ret = ioctx.pool_required_alignment2(&alignment);
      if (ret < 0) {
        cerr << "pool_required_alignment2 failed: " << cpp_strerror(ret)
	  << std::endl;
	return ret;
      }
      ceph_assert(alignment != 0);
    }
  }

  if (need_align) {
    dout(10) << "alignment = " << alignment << dendl;
  }

  bufferlist ebl, databl;
  uint64_t in_offset = 0, out_offset = 0;
  bool done = false;
  while(!done) {
    sectiontype_t type;
    int ret = read_section(&type, &ebl);
    if (ret) {
      cerr << "Error reading section: " << ret << std::endl;
      return ret;
    }

    ebliter = ebl.cbegin();
    //cout << "\tdo_object: Section type " << hex << type << dec << std::endl;
    //cout << "\t\tsection size " << ebl.length() << std::endl;
    if (type >= END_OF_TYPES) {
      cout << "Skipping unknown object section type" << std::endl;
      continue;
    }
    switch(type) {
    case TYPE_DATA:
      ds.decode(ebliter);
      dout(10) << "\tdata: offset " << ds.offset << " len " << ds.len << dendl;
      if (need_align) {
        if (ds.offset != in_offset) {
          cerr << "Discontiguous object data in export" << std::endl;
          return -EFAULT;
        }
        ceph_assert(ds.databl.length() == ds.len);
        databl.claim_append(ds.databl);
        in_offset += ds.len;
        if (databl.length() >= alignment) {
          uint64_t rndlen = uint64_t(databl.length() / alignment) * alignment;
          dout(10) << "write offset=" << out_offset << " len=" << rndlen << dendl;
          if (!dry_run && !skipping) {
            ret = ioctx.write(ob.hoid.hobj.oid.name, databl, rndlen, out_offset);
            if (ret) {
              cerr << "write failed: " << cpp_strerror(ret) << std::endl;
              return ret;
            }
          }
          out_offset += rndlen;
          bufferlist n;
          if (databl.length() > rndlen) {
            ceph_assert(databl.length() - rndlen < alignment);
	    n.substr_of(databl, rndlen, databl.length() - rndlen);
          }
          databl = n;
        }
        break;
      }
      if (!dry_run && !skipping) {
        ret = ioctx.write(ob.hoid.hobj.oid.name, ds.databl, ds.len, ds.offset);
        if (ret) {
          cerr << "write failed: " << cpp_strerror(ret) << std::endl;
          return ret;
        }
      }
      break;
    case TYPE_ATTRS:
      as.decode(ebliter);

      dout(10) << "\tattrs: len " << as.data.size() << dendl;
      if (dry_run || skipping)
        break;
      for (std::map<string,bufferlist>::iterator i = as.data.begin();
          i != as.data.end(); ++i) {
	// The user xattrs that we want all begin with "_" with length > 1.
        // Drop key "_" and all attributes that do not start with '_'
        if (i->first == "_" || i->first[0] != '_')
          continue;
        ret = ioctx.setxattr(ob.hoid.hobj.oid.name, i->first.substr(1).c_str(), i->second);
        if (ret) {
          cerr << "setxattr failed: " << cpp_strerror(ret) << std::endl;
          if (ret != -EOPNOTSUPP)
            return ret;
        }
      }
      break;
    case TYPE_OMAP_HDR:
      oh.decode(ebliter);

      dout(10) << "\tomap header: " << string(oh.hdr.c_str(), oh.hdr.length())
        << dendl;
      if (dry_run || skipping)
        break;
      ret = ioctx.omap_set_header(ob.hoid.hobj.oid.name, oh.hdr);
      if (ret) {
        cerr << "omap_set_header failed: " << cpp_strerror(ret) << std::endl;
        if (ret != -EOPNOTSUPP)
          return ret;
      }
      break;
    case TYPE_OMAP:
      os.decode(ebliter);

      dout(10) << "\tomap: size " << os.omap.size() << dendl;
      if (dry_run || skipping)
        break;
      ret = ioctx.omap_set(ob.hoid.hobj.oid.name, os.omap);
      if (ret) {
        cerr << "omap_set failed: " << cpp_strerror(ret) << std::endl;
        if (ret != -EOPNOTSUPP)
          return ret;
      }
      break;
    case TYPE_OBJECT_END:
      done = true;
      if (need_align && databl.length() > 0) {
        ceph_assert(databl.length() < alignment);
        dout(10) << "END write offset=" << out_offset << " len=" << databl.length() << dendl;
        if (dry_run || skipping)
          break;
        ret = ioctx.write(ob.hoid.hobj.oid.name, databl, databl.length(), out_offset);
        if (ret) {
          cerr << "write failed: " << cpp_strerror(ret) << std::endl;
          return ret;
        }
      }
      break;
    default:
      cerr << "Unexpected section type " << type << std::endl;
      return -EFAULT;
    }
  }
  return 0;
}
Commit	Line	Data
7c673cae FG	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) 2015 Red Hat
	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
	16	#include "common/errno.h"
	17
	18	#include "osd/PGLog.h"
	19	#include "RadosImport.h"
	20
	21	#define dout_context g_ceph_context
	22	#define dout_subsys ceph_subsys_rados
	23
20effc67 TL	24	using std::cerr;
	25	using std::cout;
	26	using std::map;
	27	using std::string;
	28
7c673cae FG	29	int RadosImport::import(std::string pool, bool no_overwrite)
	30	{
	31	librados::IoCtx ioctx;
	32	librados::Rados cluster;
	33
	34	char *id = getenv("CEPH_CLIENT_ID");
	35	if (id) cerr << "Client id is: " << id << std::endl;
	36	int ret = cluster.init(id);
	37	if (ret) {
	38	cerr << "Error " << ret << " in cluster.init" << std::endl;
	39	return ret;
	40	}
	41	ret = cluster.conf_read_file(NULL);
	42	if (ret) {
	43	cerr << "Error " << ret << " in cluster.conf_read_file" << std::endl;
	44	return ret;
	45	}
	46	ret = cluster.conf_parse_env(NULL);
	47	if (ret) {
	48	cerr << "Error " << ret << " in cluster.conf_read_env" << std::endl;
	49	return ret;
	50	}
	51	ret = cluster.connect();
	52	if (ret) {
	53	cerr << "Error " << ret << " in cluster.connect" << std::endl;
	54	return ret;
	55	}
	56
	57	ret = cluster.ioctx_create(pool.c_str(), ioctx);
	58	if (ret < 0) {
	59	cerr << "ioctx_create " << pool << " failed with " << ret << std::endl;
	60	return ret;
	61	}
	62
	63	return import(ioctx, no_overwrite);
	64	}
	65
	66	int RadosImport::import(librados::IoCtx &io_ctx, bool no_overwrite)
	67	{
	68	bufferlist ebl;
	69	pg_info_t info;
	70	PGLog::IndexedLog log;
	71
	72	int ret = read_super();
	73	if (ret)
	74	return ret;
	75
	76	if (sh.magic != super_header::super_magic) {
	77	cerr << "Invalid magic number: 0x"
	78	<< std::hex << sh.magic << " vs. 0x" << super_header::super_magic
	79	<< std::dec << std::endl;
	80	return -EFAULT;
	81	}
	82
	83	if (sh.version > super_header::super_ver) {
	84	cerr << "Can't handle export format version=" << sh.version << std::endl;
	85	return -EINVAL;
	86	}
	87
	88	//First section must be TYPE_PG_BEGIN
	89	sectiontype_t type;
	90	ret = read_section(&type, &ebl);
	91	if (ret)
	92	return ret;
93
94	bool pool_mode = false;
95	if (type == TYPE_POOL_BEGIN) {
96	pool_mode = true;
97	cout << "Importing pool" << std::endl;
98	} else if (type == TYPE_PG_BEGIN) {
11fdf7f2	99	auto ebliter = ebl.cbegin();
7c673cae FG	100	pg_begin pgb;
	101	pgb.decode(ebliter);
	102	spg_t pgid = pgb.pgid;;
	103	if (!pgid.is_no_shard()) {
	104	cerr << "Importing Erasure Coded shard is not supported" << std::endl;
	105	return -EOPNOTSUPP;
	106	}
	107	dout(10) << "Exported features: " << pgb.superblock.compat_features << dendl;
	108	cout << "Importing from pgid " << pgid << std::endl;
	109	} else {
	110	cerr << "Invalid initial section code " << type << std::endl;
	111	return -EFAULT;
	112	}
	113
	114	// XXX: How to check export features?
	115	#if 0
	116	if (sb.compat_features.compare(pgb.superblock.compat_features) == -1) {
	117	cerr << "Export has incompatible features set "
	118	<< pgb.superblock.compat_features << std::endl;
	119	return -EINVAL;
	120	}
	121	#endif
	122
	123	#if defined(__linux__)
	124	if (file_fd != STDIN_FILENO)
	125	posix_fadvise(file_fd, 0, 0, POSIX_FADV_SEQUENTIAL);
	126	#endif
	127
	128	bool done = false;
	129	bool found_metadata = false;
	130	while(!done) {
	131	ret = read_section(&type, &ebl);
	132	if (ret)
	133	return ret;
	134
	135	//cout << "do_import: Section type " << hex << type << dec << std::endl;
	136	if (type >= END_OF_TYPES) {
	137	cout << "Skipping unknown section type" << std::endl;
	138	continue;
	139	}
	140	switch(type) {
	141	case TYPE_OBJECT_BEGIN:
	142	ret = get_object_rados(io_ctx, ebl, no_overwrite);
	143	if (ret) {
	144	cerr << "Error inserting object: " << ret << std::endl;
	145	return ret;
	146	}
	147	break;
	148	case TYPE_PG_METADATA:
	149	dout(10) << "Don't care about the old metadata" << dendl;
	150	found_metadata = true;
	151	break;
	152	case TYPE_PG_END:
	153	done = true;
	154	break;
	155	case TYPE_POOL_END:
	156	done = true;
	157	break;
	158	default:
	159	return -EFAULT;
	160	}
	161	}
	162
	163	if (!(pool_mode \|\| found_metadata)) {
164	cerr << "Missing metadata section!" << std::endl;
165	}
166
167	#if defined(__linux__)
168	if (file_fd != STDIN_FILENO)
169	posix_fadvise(file_fd, 0, 0, POSIX_FADV_DONTNEED);
170	#endif
171	return 0;
172	}
173
174	int RadosImport::get_object_rados(librados::IoCtx &ioctx, bufferlist &bl, bool no_overwrite)
175	{
11fdf7f2	176	auto ebliter = bl.cbegin();
7c673cae FG	177	object_begin ob;
	178	ob.decode(ebliter);
	179	map<string,bufferlist>::iterator i;
	180	bufferlist abl;
	181	bool skipping;
	182
	183	data_section ds;
	184	attr_section as;
	185	omap_hdr_section oh;
	186	omap_section os;
	187
11fdf7f2	188	ceph_assert(g_ceph_context);
7c673cae FG	189	if (ob.hoid.hobj.nspace == g_ceph_context->_conf->osd_hit_set_namespace) {
	190	cout << "Skipping internal object " << ob.hoid << std::endl;
	191	skip_object(bl);
	192	return 0;
	193	}
	194
	195	if (!ob.hoid.hobj.is_head()) {
	196	cout << "Skipping non-head for " << ob.hoid << std::endl;
	197	skip_object(bl);
	198	return 0;
	199	}
	200
	201	ioctx.set_namespace(ob.hoid.hobj.get_namespace());
f6b5b4d7	202	ioctx.locator_set_key(ob.hoid.hobj.get_key());
7c673cae FG	203
	204	string msg("Write");
	205	skipping = false;
	206	if (dry_run) {
	207	uint64_t psize;
	208	time_t pmtime;
	209	int ret = ioctx.stat(ob.hoid.hobj.oid.name, &psize, &pmtime);
	210	if (ret == 0) {
	211	if (no_overwrite)
	212	// Could set skipping, but dry-run doesn't change anything either
	213	msg = "Skipping existing";
	214	else
	215	msg = "*Overwrite*";
	216	}
	217	} else {
	218	int ret = ioctx.create(ob.hoid.hobj.oid.name, true);
	219	if (ret && ret != -EEXIST) {
	220	cerr << "create failed: " << cpp_strerror(ret) << std::endl;
	221	return ret;
	222	}
	223	if (ret == -EEXIST) {
	224	if (no_overwrite) {
	225	msg = "Skipping existing";
	226	skipping = true;
	227	} else {
	228	msg = "*Overwrite*";
	229	ret = ioctx.remove(ob.hoid.hobj.oid.name);
	230	if (ret < 0) {
	231	cerr << "remove failed: " << cpp_strerror(ret) << std::endl;
	232	return ret;
	233	}
	234	ret = ioctx.create(ob.hoid.hobj.oid.name, true);
	235	// If object re-appeared after removal, let's just skip it
	236	if (ret == -EEXIST) {
	237	skipping = true;
	238	msg = "Skipping in-use object";
	239	ret = 0;
	240	}
	241	if (ret < 0) {
	242	cerr << "create failed: " << cpp_strerror(ret) << std::endl;
	243	return ret;
	244	}
	245	}
	246	}
	247	}
	248
	249	cout << msg << " " << ob.hoid << std::endl;
	250
	251	bool need_align = false;
	252	uint64_t alignment = 0;
	253	if (align) {
	254	need_align = true;
	255	alignment = align;
	256	} else {
	257	int ret = ioctx.pool_requires_alignment2(&need_align);
	258	if (ret < 0) {
	259	cerr << "pool_requires_alignment2 failed: " << cpp_strerror(ret)
	260	<< std::endl;
	261	return ret;
	262	}
	263
	264	if (need_align) {
	265	ret = ioctx.pool_required_alignment2(&alignment);
	266	if (ret < 0) {
267	cerr << "pool_required_alignment2 failed: " << cpp_strerror(ret)
268	<< std::endl;
269	return ret;
270	}
11fdf7f2	271	ceph_assert(alignment != 0);
7c673cae FG	272	}
	273	}
	274
	275	if (need_align) {
	276	dout(10) << "alignment = " << alignment << dendl;
	277	}
	278
	279	bufferlist ebl, databl;
	280	uint64_t in_offset = 0, out_offset = 0;
	281	bool done = false;
	282	while(!done) {
	283	sectiontype_t type;
	284	int ret = read_section(&type, &ebl);
	285	if (ret) {
	286	cerr << "Error reading section: " << ret << std::endl;
	287	return ret;
	288	}
	289
11fdf7f2	290	ebliter = ebl.cbegin();
7c673cae FG	291	//cout << "\tdo_object: Section type " << hex << type << dec << std::endl;
	292	//cout << "\t\tsection size " << ebl.length() << std::endl;
	293	if (type >= END_OF_TYPES) {
	294	cout << "Skipping unknown object section type" << std::endl;
	295	continue;
	296	}
	297	switch(type) {
	298	case TYPE_DATA:
	299	ds.decode(ebliter);
	300	dout(10) << "\tdata: offset " << ds.offset << " len " << ds.len << dendl;
	301	if (need_align) {
	302	if (ds.offset != in_offset) {
	303	cerr << "Discontiguous object data in export" << std::endl;
	304	return -EFAULT;
	305	}
11fdf7f2	306	ceph_assert(ds.databl.length() == ds.len);
7c673cae FG	307	databl.claim_append(ds.databl);
	308	in_offset += ds.len;
	309	if (databl.length() >= alignment) {
	310	uint64_t rndlen = uint64_t(databl.length() / alignment) * alignment;
	311	dout(10) << "write offset=" << out_offset << " len=" << rndlen << dendl;
	312	if (!dry_run && !skipping) {
	313	ret = ioctx.write(ob.hoid.hobj.oid.name, databl, rndlen, out_offset);
	314	if (ret) {
	315	cerr << "write failed: " << cpp_strerror(ret) << std::endl;
	316	return ret;
	317	}
	318	}
	319	out_offset += rndlen;
	320	bufferlist n;
	321	if (databl.length() > rndlen) {
11fdf7f2	322	ceph_assert(databl.length() - rndlen < alignment);
7c673cae FG	323	n.substr_of(databl, rndlen, databl.length() - rndlen);
	324	}
	325	databl = n;
	326	}
	327	break;
	328	}
	329	if (!dry_run && !skipping) {
	330	ret = ioctx.write(ob.hoid.hobj.oid.name, ds.databl, ds.len, ds.offset);
	331	if (ret) {
	332	cerr << "write failed: " << cpp_strerror(ret) << std::endl;
	333	return ret;
	334	}
	335	}
	336	break;
	337	case TYPE_ATTRS:
	338	as.decode(ebliter);
	339
	340	dout(10) << "\tattrs: len " << as.data.size() << dendl;
	341	if (dry_run \|\| skipping)
	342	break;
	343	for (std::map<string,bufferlist>::iterator i = as.data.begin();
	344	i != as.data.end(); ++i) {
	345	// The user xattrs that we want all begin with "_" with length > 1.
	346	// Drop key "_" and all attributes that do not start with '_'
	347	if (i->first == "_" \|\| i->first[0] != '_')
	348	continue;
	349	ret = ioctx.setxattr(ob.hoid.hobj.oid.name, i->first.substr(1).c_str(), i->second);
	350	if (ret) {
	351	cerr << "setxattr failed: " << cpp_strerror(ret) << std::endl;
	352	if (ret != -EOPNOTSUPP)
	353	return ret;
	354	}
	355	}
	356	break;
	357	case TYPE_OMAP_HDR:
	358	oh.decode(ebliter);
	359
	360	dout(10) << "\tomap header: " << string(oh.hdr.c_str(), oh.hdr.length())
	361	<< dendl;
	362	if (dry_run \|\| skipping)
	363	break;
	364	ret = ioctx.omap_set_header(ob.hoid.hobj.oid.name, oh.hdr);
	365	if (ret) {
	366	cerr << "omap_set_header failed: " << cpp_strerror(ret) << std::endl;
	367	if (ret != -EOPNOTSUPP)
	368	return ret;
	369	}
	370	break;
	371	case TYPE_OMAP:
	372	os.decode(ebliter);
	373
	374	dout(10) << "\tomap: size " << os.omap.size() << dendl;
	375	if (dry_run \|\| skipping)
	376	break;
	377	ret = ioctx.omap_set(ob.hoid.hobj.oid.name, os.omap);
	378	if (ret) {
	379	cerr << "omap_set failed: " << cpp_strerror(ret) << std::endl;
	380	if (ret != -EOPNOTSUPP)
	381	return ret;
	382	}
	383	break;
	384	case TYPE_OBJECT_END:
	385	done = true;
	386	if (need_align && databl.length() > 0) {
11fdf7f2	387	ceph_assert(databl.length() < alignment);
7c673cae FG	388	dout(10) << "END write offset=" << out_offset << " len=" << databl.length() << dendl;
	389	if (dry_run \|\| skipping)
	390	break;
	391	ret = ioctx.write(ob.hoid.hobj.oid.name, databl, databl.length(), out_offset);
	392	if (ret) {
	393	cerr << "write failed: " << cpp_strerror(ret) << std::endl;
	394	return ret;
	395	}
	396	}
	397	break;
	398	default:
	399	cerr << "Unexpected section type " << type << std::endl;
	400	return -EFAULT;
	401	}
	402	}
	403	return 0;
	404	}