src/ceph/src/rgw/rgw_rest_replica_log.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 eNovance SAS <licensing@enovance.com>
   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 #include "common/ceph_json.h"
  15 #include "common/strtol.h"
  16 #include "rgw_rest.h"
  17 #include "rgw_op.h"
  18 #include "rgw_rest_s3.h"
  19 #include "rgw_replica_log.h"
  20 #include "rgw_metadata.h"
  21 #include "rgw_bucket.h"
  22 #include "rgw_rest_replica_log.h"
  23 #include "rgw_client_io.h"
  24 #include "common/errno.h"
  25 #include "include/assert.h"
  26
  27 #define dout_context g_ceph_context
  28 #define dout_subsys ceph_subsys_rgw
  29 #define REPLICA_INPUT_MAX_LEN (512*1024)
  30
  31 static int parse_to_utime(string& in, utime_t& out) {
  32   uint64_t sec = 0;
  33   uint64_t nsec = 0;
  34   int ret = utime_t::parse_date(in.c_str(), &sec, &nsec);
  35   if (ret < 0)
  36     return ret;
  37
  38   out = utime_t(sec, nsec);
  39   return 0;
  40 }
  41
  42 void RGWOp_OBJLog_SetBounds::execute() {
  43   string id_str = s->info.args.get("id"),
  44          marker = s->info.args.get("marker"),
  45          time = s->info.args.get("time"),
  46          daemon_id = s->info.args.get("daemon_id");
  47
  48   if (id_str.empty() ||
  49       marker.empty() ||
  50       time.empty() ||
  51       daemon_id.empty()) {
  52     dout(5) << "Error - invalid parameter list" << dendl;
  53     http_ret = -EINVAL;
  54     return;
  55   }
  56
  57   int shard;
  58   string err;
  59   utime_t ut;
  60
  61   shard = (int)strict_strtol(id_str.c_str(), 10, &err);
  62   if (!err.empty()) {
  63     dout(5) << "Error parsing id parameter - " << id_str << ", err " << err << dendl;
  64     http_ret = -EINVAL;
  65     return;
  66   }
  67
  68   if (parse_to_utime(time, ut) < 0) {
  69     http_ret = -EINVAL;
  70     return;
  71   }
  72
  73   string pool;
  74   RGWReplicaObjectLogger rl(store, pool, prefix);
  75   bufferlist bl;
  76   list<RGWReplicaItemMarker> markers;
  77
  78   if ((http_ret = rgw_rest_get_json_input(store->ctx(), s, markers, REPLICA_INPUT_MAX_LEN, NULL)) < 0) {
  79     dout(5) << "Error - retrieving input data - " << http_ret << dendl;
  80     return;
  81   }
  82
  83   http_ret = rl.update_bound(shard, daemon_id, marker, ut, &markers);
  84 }
  85
  86 void RGWOp_OBJLog_GetBounds::execute() {
  87   string id = s->info.args.get("id");
  88
  89   if (id.empty()) {
  90     dout(5) << " Error - invalid parameter list" << dendl;
  91     http_ret = -EINVAL;
  92     return;
  93   }
  94
  95   int shard;
  96   string err;
  97
  98   shard = (int)strict_strtol(id.c_str(), 10, &err);
  99   if (!err.empty()) {
 100     dout(5) << "Error parsing id parameter - " << id << ", err " << err << dendl;
 101     http_ret = -EINVAL;
 102     return;
 103   }
 104
 105   string pool;
 106   RGWReplicaObjectLogger rl(store, pool, prefix);
 107   http_ret = rl.get_bounds(shard, bounds);
 108 }
 109
 110 void RGWOp_OBJLog_GetBounds::send_response() {
 111   set_req_state_err(s, http_ret);
 112   dump_errno(s);
 113   end_header(s);
 114
 115   if (http_ret < 0)
 116     return;
 117
 118   encode_json("bounds", bounds, s->formatter);
 119   flusher.flush();
 120 }
 121
 122 void RGWOp_OBJLog_DeleteBounds::execute() {
 123   string id = s->info.args.get("id"),
 124          daemon_id = s->info.args.get("daemon_id");
 125   bool purge_all;
 126
 127   s->info.args.get_bool("purge-all", &purge_all, false);
 128
 129   if (id.empty() ||
 130       (!purge_all && daemon_id.empty())) {
 131     dout(5) << "Error - invalid parameter list" << dendl;
 132     http_ret = -EINVAL;
 133     return;
 134   }
 135
 136   int shard;
 137   string err;
 138
 139   shard = (int)strict_strtol(id.c_str(), 10, &err);
 140   if (!err.empty()) {
 141     dout(5) << "Error parsing id parameter - " << id << ", err " << err << dendl;
 142     http_ret = -EINVAL;
 143   }
 144
 145   string pool;
 146   RGWReplicaObjectLogger rl(store, pool, prefix);
 147   http_ret = rl.delete_bound(shard, daemon_id, purge_all);
 148 }
 149
 150 static int bucket_instance_to_bucket(RGWRados *store, const string& bucket_instance, rgw_bucket& bucket) {
 151   RGWBucketInfo bucket_info;
 152   real_time mtime;
 153
 154   RGWObjectCtx obj_ctx(store);
 155   int r = store->get_bucket_instance_info(obj_ctx, bucket_instance, bucket_info, &mtime, NULL);
 156   if (r < 0) {
 157     dout(5) << "could not get bucket instance info for bucket=" << bucket_instance << ": " << cpp_strerror(r) << dendl;
 158     if (r == -ENOENT)
 159       return r;
 160     return -EINVAL;
 161   }
 162
 163   bucket = bucket_info.bucket;
 164   return 0;
 165 }
 166
 167 void RGWOp_BILog_SetBounds::execute() {
 168   string bucket_instance = s->info.args.get("bucket-instance"),
 169          marker = s->info.args.get("marker"),
 170          time = s->info.args.get("time"),
 171          daemon_id = s->info.args.get("daemon_id");
 172
 173   if (marker.empty() ||
 174       time.empty() ||
 175       daemon_id.empty()) {
 176     dout(5) << "Error - invalid parameter list" << dendl;
 177     http_ret = -EINVAL;
 178     return;
 179   }
 180
 181   utime_t ut;
 182
 183   if (parse_to_utime(time, ut) < 0) {
 184     http_ret = -EINVAL;
 185     return;
 186   }
 187
 188   int shard_id;
 189   http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
 190   if (http_ret < 0) {
 191     dout(5) << "failed to parse bucket instance" << dendl;
 192     return;
 193   }
 194
 195   rgw_bucket bucket;
 196
 197   if ((http_ret = bucket_instance_to_bucket(store, bucket_instance, bucket)) < 0) {
 198     return;
 199   }
 200
 201   RGWReplicaBucketLogger rl(store);
 202   bufferlist bl;
 203   list<RGWReplicaItemMarker> markers;
 204
 205   if ((http_ret = rgw_rest_get_json_input(store->ctx(), s, markers, REPLICA_INPUT_MAX_LEN, NULL)) < 0) {
 206     dout(5) << "Error - retrieving input data - " << http_ret << dendl;
 207     return;
 208   }
 209
 210   http_ret = rl.update_bound(bucket, shard_id, daemon_id, marker, ut, &markers);
 211 }
 212
 213 void RGWOp_BILog_GetBounds::execute() {
 214   string bucket_instance = s->info.args.get("bucket-instance");
 215   rgw_bucket bucket;
 216
 217   int shard_id;
 218
 219   http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
 220   if (http_ret < 0) {
 221     dout(5) << "failed to parse bucket instance" << dendl;
 222     return;
 223   }
 224
 225   if ((http_ret = bucket_instance_to_bucket(store, bucket_instance, bucket)) < 0)
 226     return;
 227
 228   RGWReplicaBucketLogger rl(store);
 229   http_ret = rl.get_bounds(bucket, shard_id, bounds);
 230 }
 231
 232 void RGWOp_BILog_GetBounds::send_response() {
 233   set_req_state_err(s, http_ret);
 234   dump_errno(s);
 235   end_header(s);
 236
 237   if (http_ret < 0)
 238     return;
 239
 240   encode_json("bounds", bounds, s->formatter);
 241   flusher.flush();
 242 }
 243
 244 void RGWOp_BILog_DeleteBounds::execute() {
 245   string bucket_instance = s->info.args.get("bucket-instance");
 246   string daemon_id = s->info.args.get("daemon_id");
 247   bool purge_all;
 248
 249   s->info.args.get_bool("purge-all", &purge_all, false);
 250
 251   if (daemon_id.empty() && !purge_all) {
 252     dout(5) << "Error - invalid parameter list" << dendl;
 253     http_ret = -EINVAL;
 254     return;
 255   }
 256
 257   int shard_id;
 258   http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
 259   if (http_ret < 0) {
 260     dout(5) << "failed to parse bucket instance" << dendl;
 261     return;
 262   }
 263
 264   rgw_bucket bucket;
 265
 266   if ((http_ret = bucket_instance_to_bucket(store, bucket_instance, bucket)) < 0) {
 267     return;
 268   }
 269
 270   RGWReplicaBucketLogger rl(store);
 271   http_ret = rl.delete_bound(bucket, shard_id, daemon_id, purge_all);
 272 }
 273
 274 RGWOp *RGWHandler_ReplicaLog::op_get() {
 275   bool exists;
 276   string type = s->info.args.get("type", &exists);
 277
 278   if (!exists) {
 279     return NULL;
 280   }
 281
 282   if (type.compare("metadata") == 0) {
 283     return new RGWOp_OBJLog_GetBounds(META_REPLICA_LOG_OBJ_PREFIX, "mdlog");
 284   } else if (type.compare("bucket-index") == 0) {
 285     return new RGWOp_BILog_GetBounds;
 286   } else if (type.compare("data") == 0) {
 287     return new RGWOp_OBJLog_GetBounds(DATA_REPLICA_LOG_OBJ_PREFIX, "datalog");
 288   }
 289   return NULL;
 290 }
 291
 292 RGWOp *RGWHandler_ReplicaLog::op_delete() {
 293   bool exists;
 294   string type = s->info.args.get("type", &exists);
 295
 296   if (!exists) {
 297     return NULL;
 298   }
 299
 300   if (type.compare("metadata") == 0)
 301     return new RGWOp_OBJLog_DeleteBounds(META_REPLICA_LOG_OBJ_PREFIX, "mdlog");
 302   else if (type.compare("bucket-index") == 0)
 303     return new RGWOp_BILog_DeleteBounds;
 304   else if (type.compare("data") == 0)
 305     return new RGWOp_OBJLog_DeleteBounds(DATA_REPLICA_LOG_OBJ_PREFIX, "datalog");
 306
 307   return NULL;
 308 }
 309
 310 RGWOp *RGWHandler_ReplicaLog::op_post() {
 311   bool exists;
 312   string type = s->info.args.get("type", &exists);
 313
 314   if (!exists) {
 315     return NULL;
 316   }
 317
 318   if (type.compare("metadata") == 0) {
 319     return new RGWOp_OBJLog_SetBounds(META_REPLICA_LOG_OBJ_PREFIX, "mdlog");
 320   } else if (type.compare("bucket-index") == 0) {
 321     return new RGWOp_BILog_SetBounds;
 322   } else if (type.compare("data") == 0) {
 323     return new RGWOp_OBJLog_SetBounds(DATA_REPLICA_LOG_OBJ_PREFIX, "datalog");
 324   }
 325   return NULL;
 326 }
 327