Merge pull request #12113 from ceph/wip-rgw-zone-compression

rgw: move compression config into zone placement

Reviewed-by: Yehuda Sadeh <yehuda@redhat.com>

src/common/config_opts.h

@@ -1561,7 +1561,6 @@ OPTION(mon_mgr_beacon_grace, OPT_INT, 30)  // How long to wait to failover
 OPTION(rgw_list_bucket_min_readahead, OPT_INT, 1000) // minimum number of entries to read from rados for bucket listing
 
 OPTION(rgw_rest_getusage_op_compat, OPT_BOOL, false) // dump description of total stats for s3 GetUsage API
-OPTION(rgw_compression_type, OPT_STR, "none") // type of compressor, none to not use
 
 OPTION(mutex_perf_counter, OPT_BOOL, false) // enable/disable mutex perf counter
 OPTION(throttler_perf_counter, OPT_BOOL, true) // enable/disable throttler perf counter

src/rgw/rgw_admin.cc

@@ -5,6 +5,8 @@
 #include <sstream>
 #include <string>
 
+#include <boost/optional.hpp>
+
 #include "auth/Crypto.h"
 
 #include "common/armor.h"
@@ -232,6 +234,7 @@ void _usage()
   cout << "   --data_extra_pool=<pool>  placement target data extra (non-ec) pool\n";
   cout << "   --placement-index-type=<type>\n";
   cout << "                             placement target index type (normal, indexless, or #id)\n";
+  cout << "   --compression=<type>      placement target compression type (plugin name or empty/none)\n";
   cout << "   --tier-type=<type>        zone tier type\n";
   cout << "   --tier-config=<k>=<v>[,...]\n";
   cout << "                             set zone tier config keys, values\n";
@@ -2380,6 +2383,8 @@ int main(int argc, char **argv)
   RGWBucketIndexType placement_index_type = RGWBIType_Normal;
   bool index_type_specified = false;
 
+  boost::optional<std::string> compression_type;
+
   for (std::vector<const char*>::iterator i = args.begin(); i != args.end(); ) {
     if (ceph_argparse_double_dash(args, i)) {
       break;
@@ -2689,6 +2694,8 @@ int main(int argc, char **argv)
         }
       }
       index_type_specified = true;
+    } else if (ceph_argparse_witharg(args, i, &val, "--compression", (char*)NULL)) {
+      compression_type = val;
     } else if (strncmp(*i, "-", 1) == 0) {
       cerr << "ERROR: invalid flag " << *i << std::endl;
       return EINVAL;
@@ -4102,6 +4109,9 @@ int main(int argc, char **argv)
           if (index_type_specified) {
             info.index_type = placement_index_type;
           }
+          if (compression_type) {
+            info.compression_type = *compression_type;
+          }
         } else if (opt_cmd == OPT_ZONE_PLACEMENT_MODIFY) {
           auto p = zone.placement_pools.find(placement_id);
           if (p == zone.placement_pools.end()) {
@@ -4122,6 +4132,9 @@ int main(int argc, char **argv)
           if (index_type_specified) {
             info.index_type = placement_index_type;
           }
+          if (compression_type) {
+            info.compression_type = *compression_type;
+          }
         } else if (opt_cmd == OPT_ZONE_PLACEMENT_RM) {
           zone.placement_pools.erase(placement_id);
         }

src/rgw/rgw_compression.cc

@@ -64,8 +64,7 @@ RGWGetObj_Decompress::RGWGetObj_Decompress(CephContext* cct_,
 {
   compressor = Compressor::create(cct, cs_info->compression_type);
   if (!compressor.get())
-    lderr(cct) << "Cannot load compressor of type " << cs_info->compression_type 
-                     << " for rgw, check rgw_compression_type config option" << dendl;
+    lderr(cct) << "Cannot load compressor of type " << cs_info->compression_type << dendl;
 }
 
 int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len)
@@ -74,8 +73,7 @@ int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len
 
   if (!compressor.get()) {
     // if compressor isn't available - error, because cannot return decompressed data?
-    lderr(cct) << "Cannot load compressor of type " << cs_info->compression_type 
-                     << " for rgw, check rgw_compression_type config option" << dendl;
+    lderr(cct) << "Cannot load compressor of type " << cs_info->compression_type << dendl;
     return -EIO;
   }
   bufferlist out_bl, in_bl;

src/rgw/rgw_json_enc.cc

@@ -881,6 +881,7 @@ void RGWZonePlacementInfo::dump(Formatter *f) const
   encode_json("data_pool", data_pool, f);
   encode_json("data_extra_pool", data_extra_pool, f);
   encode_json("index_type", (uint32_t)index_type, f);
+  encode_json("compression", compression_type, f);
 }
 
 void RGWZonePlacementInfo::decode_json(JSONObj *obj)
@@ -891,6 +892,7 @@ void RGWZonePlacementInfo::decode_json(JSONObj *obj)
   uint32_t it;
   JSONDecoder::decode_json("index_type", it, obj);
   index_type = (RGWBucketIndexType)it;
+  JSONDecoder::decode_json("compression", compression_type, obj);
 }
 
 void RGWZoneParams::decode_json(JSONObj *obj)

src/rgw/rgw_op.cc

@@ -2807,10 +2807,10 @@ int RGWPutObj::get_data(const off_t fst, const off_t lst, bufferlist& bl)
   return ret;
 }
 
-// special handling for rgw_compression_type = "random" with multipart uploads
-static CompressorRef get_compressor_plugin(const req_state *s)
+// special handling for compression type = "random" with multipart uploads
+static CompressorRef get_compressor_plugin(const req_state *s,
+                                           const std::string& compression_type)
 {
-  const auto& compression_type = s->cct->_conf->rgw_compression_type;
   if (compression_type != "random") {
     return Compressor::create(s->cct, compression_type);
   }
@@ -2846,6 +2846,8 @@ void RGWPutObj::execute()
 
   off_t fst;
   off_t lst;
+  const auto& compression_type = store->get_zone_params().get_compression_type(
+      s->bucket_info.placement_rule);
   CompressorRef plugin;
   boost::optional<RGWPutObj_Compress> compressor;
 
@@ -2932,11 +2934,11 @@ void RGWPutObj::execute()
 
   fst = copy_source_range_fst;
   lst = copy_source_range_lst;
-  if (s->cct->_conf->rgw_compression_type != "none") {
-    plugin = get_compressor_plugin(s);
+  if (compression_type != "none") {
+    plugin = get_compressor_plugin(s, compression_type);
     if (!plugin) {
-      ldout(s->cct, 1) << "Cannot load plugin for rgw_compression_type "
-          << s->cct->_conf->rgw_compression_type << dendl;
+      ldout(s->cct, 1) << "Cannot load plugin for compression type "
+          << compression_type << dendl;
     } else {
       compressor.emplace(s->cct, plugin, filter);
       filter = &*compressor;
@@ -3239,12 +3241,13 @@ void RGWPostObj::execute()
   if (op_ret < 0)
     return;
 
-  const auto& compression_type = s->cct->_conf->rgw_compression_type;
+  const auto& compression_type = store->get_zone_params().get_compression_type(
+      s->bucket_info.placement_rule);
   CompressorRef plugin;
   if (compression_type != "none") {
     plugin = Compressor::create(s->cct, compression_type);
     if (!plugin) {
-      ldout(s->cct, 1) << "Cannot load plugin for rgw_compression_type "
+      ldout(s->cct, 1) << "Cannot load plugin for compression type "
           << compression_type << dendl;
     } else {
       compressor.emplace(s->cct, plugin, filter);

src/rgw/rgw_rados.cc

@@ -1654,6 +1654,17 @@ int RGWZoneParams::set_as_default(bool exclusive)
   return RGWSystemMetaObj::set_as_default(exclusive);
 }
 
+const string& RGWZoneParams::get_compression_type(const string& placement_rule) const
+{
+  static const std::string NONE{"none"};
+  auto p = placement_pools.find(placement_rule);
+  if (p == placement_pools.end()) {
+    return NONE;
+  }
+  const auto& type = p->second.compression_type;
+  return !type.empty() ? type : NONE;
+}
+
 void RGWPeriodMap::encode(bufferlist& bl) const {
   ENCODE_START(2, 1, bl);
   ::encode(id, bl);
@@ -7104,11 +7115,12 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& obj_ctx,
 
   RGWPutObjDataProcessor *filter = &processor;
 
-  const auto& compression_type = cct->_conf->rgw_compression_type;
+  const auto& compression_type = zone_params.get_compression_type(
+      dest_bucket_info.placement_rule);
   if (compression_type != "none") {
     plugin = Compressor::create(cct, compression_type);
     if (!plugin) {
-      ldout(cct, 1) << "Cannot load plugin for rgw_compression_type "
+      ldout(cct, 1) << "Cannot load plugin for compression type "
           << compression_type << dendl;
     } else {
       compressor = boost::in_place(cct, plugin, filter);

src/rgw/rgw_rados.h

@@ -933,20 +933,22 @@ struct RGWZonePlacementInfo {
   string data_pool;
   string data_extra_pool; /* if not set we should use data_pool */
   RGWBucketIndexType index_type;
+  std::string compression_type;
 
   RGWZonePlacementInfo() : index_type(RGWBIType_Normal) {}
 
   void encode(bufferlist& bl) const {
-    ENCODE_START(5, 1, bl);
+    ENCODE_START(6, 1, bl);
     ::encode(index_pool, bl);
     ::encode(data_pool, bl);
     ::encode(data_extra_pool, bl);
     ::encode((uint32_t)index_type, bl);
+    ::encode(compression_type, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(bufferlist::iterator& bl) {
-    DECODE_START(5, bl);
+    DECODE_START(6, bl);
     ::decode(index_pool, bl);
     ::decode(data_pool, bl);
     if (struct_v >= 4) {
@@ -957,6 +959,9 @@ struct RGWZonePlacementInfo {
       ::decode(it, bl);
       index_type = (RGWBucketIndexType)it;
     }
+    if (struct_v >= 6) {
+      ::decode(compression_type, bl);
+    }
     DECODE_FINISH(bl);
   }
   const string& get_data_extra_pool() {
@@ -1013,6 +1018,8 @@ struct RGWZoneParams : RGWSystemMetaObj {
   int create_default(bool old_format = false);
   int create(bool exclusive = true);
   int fix_pool_names();
+
+  const string& get_compression_type(const string& placement_rule) const;
 
   void encode(bufferlist& bl) const {
     ENCODE_START(8, 1, bl);

src/test/cli/radosgw-admin/help.t

@@ -182,6 +182,7 @@
      --data_extra_pool=<pool>  placement target data extra (non-ec) pool
      --placement-index-type=<type>
                                placement target index type (normal, indexless, or #id)
+     --compression=<type>      placement target compression type (plugin name or empty/none)
      --tier-type=<type>        zone tier type
      --tier-config=<k>=<v>[,...]
                                set zone tier config keys, values