ceph/src/cls/cephfs/cls_cephfs_client.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) 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
  17 #include "include/rados/librados.hpp"
  18 #include "mds/CInode.h"
  19
  20 #include "cls_cephfs_client.h"
  21
  22 using ceph::bufferlist;
  23 using ceph::decode;
  24
  25 #define XATTR_CEILING "scan_ceiling"
  26 #define XATTR_MAX_MTIME "scan_max_mtime"
  27 #define XATTR_MAX_SIZE "scan_max_size"
  28 #define XATTR_POOL_ID "scan_pool_id"
  29
  30 int ClsCephFSClient::accumulate_inode_metadata(
  31   librados::IoCtx &ctx,
  32   inodeno_t inode_no,
  33   const uint64_t obj_index,
  34   const uint64_t obj_size,
  35   const int64_t obj_pool_id,
  36   const time_t mtime)
  37 {
  38   AccumulateArgs args(
  39       obj_index,
  40       obj_size,
  41       mtime,
  42       XATTR_CEILING,
  43       XATTR_MAX_MTIME,
  44       XATTR_MAX_SIZE);
  45
  46   // Generate 0th object name, where we will accumulate sizes/mtimes
  47   object_t zeroth_object = InodeStore::get_object_name(inode_no, frag_t(), "");
  48
  49   // Construct a librados operation invoking our class method
  50   librados::ObjectWriteOperation op;
  51   bufferlist inbl;
  52   args.encode(inbl);
  53   op.exec("cephfs", "accumulate_inode_metadata", inbl);
  54
  55   if (obj_pool_id != -1) {
  56     bufferlist bl;
  57     encode(obj_pool_id, bl);
  58     op.setxattr(XATTR_POOL_ID, bl);
  59   }
  60
  61   // Execute op
  62   return ctx.operate(zeroth_object.name, &op);
  63 }
  64
  65 int ClsCephFSClient::delete_inode_accumulate_result(
  66     librados::IoCtx &ctx,
  67     const std::string &oid)
  68 {
  69   librados::ObjectWriteOperation op;
  70
  71   // Remove xattrs from object
  72   //
  73   op.rmxattr(XATTR_CEILING);
  74   op.rmxattr(XATTR_MAX_SIZE);
  75   op.rmxattr(XATTR_MAX_MTIME);
  76   op.rmxattr(XATTR_POOL_ID);
  77   op.set_op_flags2(librados::OP_FAILOK);
  78
  79   return (ctx.operate(oid, &op));
  80 }
  81
  82 int ClsCephFSClient::fetch_inode_accumulate_result(
  83   librados::IoCtx &ctx,
  84   const std::string &oid,
  85   inode_backtrace_t *backtrace,
  86   file_layout_t *layout,
  87   std::string *symlink,
  88   AccumulateResult *result)
  89 {
  90   ceph_assert(backtrace != NULL);
  91   ceph_assert(result != NULL);
  92
  93   librados::ObjectReadOperation op;
  94
  95   int scan_ceiling_r = 0;
  96   bufferlist scan_ceiling_bl;
  97   op.getxattr(XATTR_CEILING, &scan_ceiling_bl, &scan_ceiling_r);
  98
  99   int scan_max_size_r = 0;
 100   bufferlist scan_max_size_bl;
 101   op.getxattr(XATTR_MAX_SIZE, &scan_max_size_bl, &scan_max_size_r);
 102
 103   int scan_max_mtime_r = 0;
 104   bufferlist scan_max_mtime_bl;
 105   op.getxattr(XATTR_MAX_MTIME, &scan_max_mtime_bl, &scan_max_mtime_r);
 106
 107   int scan_pool_id_r = 0;
 108   bufferlist scan_pool_id_bl;
 109   op.getxattr(XATTR_POOL_ID, &scan_pool_id_bl, &scan_pool_id_r);
 110   op.set_op_flags2(librados::OP_FAILOK);
 111
 112   int parent_r = 0;
 113   bufferlist parent_bl;
 114   op.getxattr("parent", &parent_bl, &parent_r);
 115   op.set_op_flags2(librados::OP_FAILOK);
 116
 117   int layout_r = 0;
 118   bufferlist layout_bl;
 119   op.getxattr("layout", &layout_bl, &layout_r);
 120   op.set_op_flags2(librados::OP_FAILOK);
 121
 122   int symlink_r = 0;
 123   bufferlist symlink_bl;
 124   op.getxattr("symlink", &symlink_bl, &symlink_r);
 125   op.set_op_flags2(librados::OP_FAILOK);
 126
 127   bufferlist op_bl;
 128   int r = ctx.operate(oid, &op, &op_bl);
 129   if (r < 0) {
 130     return r;
 131   }
 132
 133   // Load scan_ceiling
 134   try {
 135     auto scan_ceiling_bl_iter = scan_ceiling_bl.cbegin();
 136     ObjCeiling ceiling;
 137     ceiling.decode(scan_ceiling_bl_iter);
 138     result->ceiling_obj_index = ceiling.id;
 139     result->ceiling_obj_size = ceiling.size;
 140   } catch (const ceph::buffer::error &err) {
 141     //dout(4) << "Invalid ceiling attr on '" << oid << "'" << dendl;
 142     return -EINVAL;
 143   }
 144
 145   // Load scan_max_size
 146   try {
 147     auto scan_max_size_bl_iter = scan_max_size_bl.cbegin();
 148     decode(result->max_obj_size, scan_max_size_bl_iter);
 149   } catch (const ceph::buffer::error &err) {
 150     //dout(4) << "Invalid size attr on '" << oid << "'" << dendl;
 151     return -EINVAL;
 152   }
 153
 154   // Load scan_pool_id
 155   if (scan_pool_id_bl.length()) {
 156     try {
 157       auto scan_pool_id_bl_iter = scan_pool_id_bl.cbegin();
 158       decode(result->obj_pool_id, scan_pool_id_bl_iter);
 159     } catch (const ceph::buffer::error &err) {
 160       //dout(4) << "Invalid pool_id attr on '" << oid << "'" << dendl;
 161       return -EINVAL;
 162     }
 163   }
 164
 165   // Load scan_max_mtime
 166   try {
 167     auto scan_max_mtime_bl_iter = scan_max_mtime_bl.cbegin();
 168     decode(result->max_mtime, scan_max_mtime_bl_iter);
 169   } catch (const ceph::buffer::error &err) {
 170     //dout(4) << "Invalid mtime attr on '" << oid << "'" << dendl;
 171     return -EINVAL;
 172   }
 173
 174   // Deserialize backtrace
 175   if (parent_bl.length()) {
 176     try {
 177       auto q = parent_bl.cbegin();
 178       backtrace->decode(q);
 179     } catch (ceph::buffer::error &e) {
 180       //dout(4) << "Corrupt backtrace on '" << oid << "': " << e << dendl;
 181       return -EINVAL;
 182     }
 183   }
 184
 185   // Deserialize layout
 186   if (layout_bl.length()) {
 187     try {
 188       auto q = layout_bl.cbegin();
 189       decode(*layout, q);
 190     } catch (ceph::buffer::error &e) {
 191       return -EINVAL;
 192     }
 193   }
 194
 195   // Deserialize symlink
 196   if (symlink_bl.length()) {
 197     try {
 198       auto q = symlink_bl.cbegin();
 199       decode(*symlink, q);
 200     } catch (ceph::buffer::error &e) {
 201       return -EINVAL;
 202     }
 203   }
 204
 205   return 0;
 206 }
 207
 208 void ClsCephFSClient::build_tag_filter(
 209           const std::string &scrub_tag,
 210           bufferlist *out_bl)
 211 {
 212   ceph_assert(out_bl != NULL);
 213
 214   // Leading part of bl is un-versioned string naming the filter
 215   encode(std::string("cephfs.inode_tag"), *out_bl);
 216
 217   // Filter-specific part of the bl: in our case this is a versioned structure
 218   InodeTagFilterArgs args;
 219   args.scrub_tag = scrub_tag;
 220   args.encode(*out_bl);
 221 }